Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions server/plugins/unifi_import/config.json
Original file line number Diff line number Diff line change
Expand Up @@ -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": [],
Expand Down
3 changes: 2 additions & 1 deletion server/plugins/unifi_import/script.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)}'])

Expand Down Expand Up @@ -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')),
Expand Down
45 changes: 41 additions & 4 deletions test/plugins/test_unifi_import.py
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -10,6 +8,7 @@
"""

import importlib.util
import json
import os
import sys
import tempfile
Expand All @@ -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():
Expand All @@ -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")
Expand Down Expand Up @@ -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"
Expand Down
Loading