8710f7f798
- Implemented the Drift module with a new FastAPI router and HTML templates for the frontend. - Created database tables for drift sources, devices, events, event history, and customer mappings. - Added functionality to manage and display drift events, including filtering and bulk mapping of monitors to customers. - Introduced tests for the Drift module, covering various functionalities including event blacklisting and UISP connector configuration. - Enhanced Telefoni service tests to ensure proper call termination handling.
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
import asyncio
|
|
import importlib
|
|
from unittest.mock import patch
|
|
|
|
|
|
drift_router = importlib.import_module("app.modules.drift.backend.router")
|
|
|
|
|
|
def test_get_uisp_connector_config_reads_settings():
|
|
with patch.object(drift_router, "execute_query_single", side_effect=[
|
|
{"value": "https://uisp.example.com"},
|
|
{"value": "secret-token"},
|
|
]):
|
|
config = drift_router._get_uisp_connector_config()
|
|
|
|
assert config == {
|
|
"base_url": "https://uisp.example.com",
|
|
"api_token": "secret-token",
|
|
}
|
|
|
|
|
|
def test_list_unmapped_monitors_uses_simple_query_without_group_by_error():
|
|
with patch.object(drift_router, "_ensure_schema"), patch.object(drift_router, "execute_query", return_value=[]) as mock_execute:
|
|
asyncio.run(drift_router.list_unmapped_monitors(limit=5))
|
|
|
|
sql = mock_execute.call_args[0][0]
|
|
assert "GROUP BY" not in sql
|
|
assert "LIMIT %s" in sql
|
|
|
|
|
|
def test_parse_blacklist_value_supports_json_and_dedupes():
|
|
raw = '["airCube-AC", "AIRCUBE-AC", " uisp-123 "]'
|
|
values = drift_router._parse_blacklist_value(raw)
|
|
assert values == ["aircube-ac", "uisp-123"]
|
|
|
|
|
|
def test_event_is_blacklisted_matches_device_name_and_source_id():
|
|
payload = {
|
|
"source_event_id": "uisp-abc-123",
|
|
"device": "airCube-AC",
|
|
"raw_json": {
|
|
"raw_item": {
|
|
"identification": {
|
|
"id": "abc-123",
|
|
"name": "airCube-AC",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
assert drift_router._event_is_blacklisted(payload, ["aircube-ac"])
|
|
assert drift_router._event_is_blacklisted(payload, ["abc-123"])
|
|
assert not drift_router._event_is_blacklisted(payload, ["some-other-device"])
|
|
|
|
|
|
def test_build_events_from_uisp_payload_exposes_direct_device_link():
|
|
events = drift_router._build_events_from_uisp_payload(
|
|
[
|
|
{
|
|
"identification": {"id": "abc-123", "name": "airCube-AC"},
|
|
"url": "https://uisp.example.com/nms/devices/abc-123",
|
|
}
|
|
],
|
|
source_id=1,
|
|
base_url="https://uisp.example.com",
|
|
)
|
|
|
|
assert events[0]["raw_json"]["device_link"] == "https://uisp.example.com/nms/devices/abc-123"
|
|
|
|
|
|
def test_row_to_event_extracts_device_link_from_existing_raw_payload():
|
|
event = drift_router._row_to_event(
|
|
{
|
|
"id": 1,
|
|
"source_event_id": "uisp-abc-123",
|
|
"raw_json": {
|
|
"raw_item": {
|
|
"url": "https://uisp.example.com/nms/devices/abc-123"
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert event["device_link"] == "https://uisp.example.com/nms/devices/abc-123"
|
|
assert event["source_link"] == "https://uisp.example.com/nms/devices/abc-123"
|