133 lines
4.0 KiB
Python
133 lines
4.0 KiB
Python
import asyncio
|
|
import importlib
|
|
from unittest.mock import patch
|
|
|
|
|
|
drift_router = importlib.import_module("app.modules.drift.backend.router")
|
|
|
|
|
|
class DummyPayload(dict):
|
|
def __getattr__(self, item):
|
|
return self.get(item)
|
|
|
|
|
|
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"
|
|
|
|
|
|
def test_row_to_event_exposes_ip_and_site_client_metadata():
|
|
event = drift_router._row_to_event(
|
|
{
|
|
"id": 2,
|
|
"source_event_id": "uisp-abc-123",
|
|
"site_name": "Site A",
|
|
"customer_name": "Customer X",
|
|
"raw_json": {
|
|
"ip": "192.0.2.10",
|
|
"site_client_name": "Client Alpha",
|
|
"raw_item": {
|
|
"site_name": "Site A",
|
|
"client_name": "Client Alpha",
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
assert event["ip"] == "192.0.2.10"
|
|
assert event["site_client_name"] == "Client Alpha"
|
|
|
|
|
|
def test_row_to_event_extracts_ip_from_nested_identification_payload():
|
|
event = drift_router._row_to_event(
|
|
{
|
|
"id": 3,
|
|
"source_event_id": "uisp-abc-456",
|
|
"raw_json": {
|
|
"raw_item": {
|
|
"identification": {
|
|
"ip": "203.0.113.9",
|
|
"site": {"name": "North Site"},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert event["ip"] == "203.0.113.9"
|
|
assert event["site_client_name"] == "North Site"
|