diff --git a/server/plugins/unifi_import/config.json b/server/plugins/unifi_import/config.json index 762f61ac0..66ad56a55 100755 --- a/server/plugins/unifi_import/config.json +++ b/server/plugins/unifi_import/config.json @@ -360,21 +360,22 @@ }, { "column": "extra", + "mapped_to_column": "scanSSID", "css_classes": "col-sm-3", "default_value": "", "localized": ["name"], "name": [ { "language_code": "en_us", - "string": "Network" + "string": "SSID" }, { "language_code": "es_es", - "string": "Red" + "string": "SSID" }, { "language_code": "de_de", - "string": "Netzwerk" + "string": "SSID" } ], "options": [], diff --git a/server/plugins/unifi_import/script.py b/server/plugins/unifi_import/script.py index e22433562..cd3b14a01 100755 --- a/server/plugins/unifi_import/script.py +++ b/server/plugins/unifi_import/script.py @@ -171,6 +171,7 @@ def get_entries(plugin_objects: Plugin_Objects) -> Plugin_Objects: # ----------------------------------------------------------------------------- def collect_details(device_type, devices, online_macs, processed_macs, plugin_objects, device_label, device_vendor, force_import): + """Add eligible UniFi devices with their identity, topology, and wireless SSID to plugin results.""" for device in devices: mylog('verbose', [f'{json.dumps(device)}']) @@ -198,7 +199,7 @@ def collect_details(device_type, devices, online_macs, processed_macs, plugin_ob watched2=get_unifi_val(device, 'oui', device_vendor), watched3=deviceType, watched4=status, - extra=get_unifi_val(device, 'connection_network_name', ''), + extra='' if device.get('is_wired') is True else get_unifi_val(device, 'essid', ''), foreignKey="", helpVal1=parentMac, helpVal2=get_port(get_unifi_val(device, 'sw_port'), get_unifi_val(device, 'uplink_remote_port')), diff --git a/test/plugins/test_unifi_import.py b/test/plugins/test_unifi_import.py index 4fe215554..f5198c732 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 @@ """ import importlib.util +import json import os import sys import tempfile @@ -25,6 +24,7 @@ def _stub(name: str, **attrs): + """Register an absent dependency stub and track it for cleanup after importing the plugin.""" if name not in sys.modules: mod = types.ModuleType(name) for k, v in attrs.items(): @@ -44,7 +44,7 @@ def _stub(name: str, **attrs): 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 +88,43 @@ def _stub(name: str, **attrs): 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"