From 75e427f805b2c2798fd55832736480184b784fb3 Mon Sep 17 00:00:00 2001 From: Alexander Shevchenko Date: Tue, 6 Oct 2026 10:36:25 +0200 Subject: [PATCH] test(unifi): cover SSID extraction and scan mapping Cover wireless, wired, missing flags, missing SSIDs, and differing network names. Keep standalone imports valid when real pytz is installed. --- test/plugins/test_unifi_import.py | 44 ++++++++++++++++++++++++++++--- 1 file changed, 40 insertions(+), 4 deletions(-) diff --git a/test/plugins/test_unifi_import.py b/test/plugins/test_unifi_import.py index 4fe215554..8104f2ef2 100644 --- a/test/plugins/test_unifi_import.py +++ b/test/plugins/test_unifi_import.py @@ -1,7 +1,5 @@ """ -Tests for unifi_import/script.py - focused on the LOCK_FILE persistence fix -(moved from the ephemeral LOG_PATH/tmpfs to durable dbFolderPath, with a -one-time migration for existing installs). +Tests for UniFi client SSID import and persistent full-import lock files. Run from inside the NetAlertX container, or locally - NetAlertX-specific modules are stubbed out automatically before the script is imported. @@ -10,6 +8,7 @@ modules are stubbed out automatically before the script is imported. """ import importlib.util +import json import os import sys import tempfile @@ -44,7 +43,7 @@ _stub( is_mac=lambda v: isinstance(v, str) and len(v.split(":")) == 6, ) _stub("logger", mylog=lambda *a: None, Logger=MagicMock) -_stub("helper", get_setting_value=lambda k: "", normalize_string=lambda s: s) +_stub("helper", get_setting_value=lambda k: "UTC" if k == "TIMEZONE" else "", normalize_string=lambda s: s) if "pyunifi" not in sys.modules: _pyunifi = types.ModuleType("pyunifi") @@ -88,6 +87,43 @@ read_lock_file = script.read_lock_file set_lock_file_value = script.set_lock_file_value +@pytest.mark.parametrize( + "client_fields,expected_ssid", + [ + ({"is_wired": False, "essid": "example-wifi", "last_connection_network_name": "example-vlan"}, "example-wifi"), + ({"is_wired": False, "essid": "example-wifi", "connection_network_name": "example-vlan"}, "example-wifi"), + ({"essid": "example-wifi"}, "example-wifi"), + ({"is_wired": True, "essid": "old-wifi", "connection_network_name": "example-vlan"}, ""), + ({"is_wired": True}, ""), + ({"is_wired": False, "last_connection_network_name": "example-vlan"}, ""), + ({"is_wired": False, "essid": None}, ""), + ({"is_wired": False, "essid": ""}, ""), + ], +) +def test_collect_details_reports_wireless_ssid(client_fields, expected_ssid): + """Import the wireless SSID without treating network names as SSIDs.""" + objects = MagicMock() + client = {"mac": "aa:bb:cc:dd:ee:01", "ip": "192.0.2.1", **client_fields} + + script.collect_details( + {"cl": ""}, [client], set(), [], objects, "client", "", True + ) + + objects.add_object.assert_called_once() + assert objects.add_object.call_args.kwargs["extra"] == expected_ssid + + +def test_unifi_ssid_is_mapped_to_current_scan(): + """Route imported SSIDs through CurrentScan to the device update policy.""" + config_path = os.path.join(os.path.dirname(_SCRIPT_PATH), "config.json") + with open(config_path, encoding="utf-8") as config_file: + config = json.load(config_file) + + assert config["mapped_to_table"] == "CurrentScan" + extra = next(column for column in config["database_column_definitions"] if column["column"] == "extra") + assert extra.get("mapped_to_column") == "scanSSID" + + class TestMigrateLegacyLockFile: def test_migrates_legacy_file_to_new_location(self, tmp_path): legacy = tmp_path / "full_run.UNFIMP.lock"