diff --git a/server/plugins/mikrotik_scan/mikrotik.py b/server/plugins/mikrotik_scan/mikrotik.py index 1bb566c8..3a6ca136 100755 --- a/server/plugins/mikrotik_scan/mikrotik.py +++ b/server/plugins/mikrotik_scan/mikrotik.py @@ -7,7 +7,7 @@ import sys INSTALL_PATH = os.getenv('NETALERTX_APP', '/app') sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"]) -from plugin_helper import Plugin_Objects # noqa: E402 [flake8 lint suppression] +from plugin_helper import Plugin_Objects, normalize_mac # noqa: E402 [flake8 lint suppression] from logger import mylog, Logger # noqa: E402 [flake8 lint suppression] from helper import get_setting_value # noqa: E402 [flake8 lint suppression] from const import logPath # noqa: E402 [flake8 lint suppression] @@ -64,16 +64,17 @@ def get_entries(plugin_objects: Plugin_Objects) -> Plugin_Objects: for lease in leases: lease_id = lease.get('.id') address = lease.get('address') - mac_address = lease.get('mac-address').lower() + raw_mac_address = lease.get('mac-address') host_name = lease.get('host-name') comment = lease.get('comment') last_seen = lease.get('last-seen') status = lease.get('status') device_name = comment or host_name or "(unknown)" - mylog('verbose', f"ID: {lease_id}, Address: {address}, MAC: {mac_address}, Host Name: {host_name}, Comment: {comment}, Last Seen: {last_seen}, Status: {status}") + mylog('verbose', f"ID: {lease_id}, Address: {address}, MAC: {raw_mac_address}, Host Name: {host_name}, Comment: {comment}, Last Seen: {last_seen}, Status: {status}") - if (status == "bound"): + if status == "bound" and raw_mac_address: + mac_address = normalize_mac(raw_mac_address) plugin_objects.add_object( primaryId = mac_address, secondaryId = address, diff --git a/test/plugins/test_mikrotik_scan.py b/test/plugins/test_mikrotik_scan.py new file mode 100644 index 00000000..f7d20505 --- /dev/null +++ b/test/plugins/test_mikrotik_scan.py @@ -0,0 +1,83 @@ +"""Tests for the MikroTik DHCP lease scanner.""" + +import importlib.util +import sys +import types +from pathlib import Path +from unittest.mock import MagicMock, patch + + +def _load_mikrotik_module(): + stubbed_module_names = [] + + def stub(name, **attributes): + module = types.ModuleType(name) + for attribute, value in attributes.items(): + setattr(module, attribute, value) + sys.modules[name] = module + stubbed_module_names.append(name) + + class TrapError(Exception): + pass + + stub( + "plugin_helper", + Plugin_Objects=MagicMock, + normalize_mac=lambda mac: mac.strip().lower(), + ) + stub("logger", mylog=MagicMock(), Logger=MagicMock()) + stub("helper", get_setting_value=MagicMock(return_value="UTC")) + stub("const", logPath="/tmp") + stub("conf", tz=None) + stub("pytz", timezone=MagicMock(return_value="UTC")) + stub("librouteros", connect=MagicMock()) + stub("librouteros.exceptions", TrapError=TrapError) + + module_path = Path(__file__).resolve().parents[2] / "server" / "plugins" / "mikrotik_scan" / "mikrotik.py" + spec = importlib.util.spec_from_file_location("mikrotik_scan", module_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + for name in stubbed_module_names: + sys.modules.pop(name, None) + + return module + + +mikrotik = _load_mikrotik_module() + + +def _lease(lease_id, address, mac_address, status="bound"): + lease = { + ".id": lease_id, + "address": address, + "host-name": f"device-{lease_id}", + "comment": "", + "last-seen": "1m", + "status": status, + } + if mac_address is not None: + lease["mac-address"] = mac_address + return lease + + +def test_disabled_lease_without_mac_does_not_abort_remaining_leases(): + leases = [ + _lease("*1", "192.168.1.2", "aa:bb:cc:dd:ee:01"), + _lease("*2", "192.168.1.5", None, status="waiting"), + _lease("*3", "192.168.1.8", "aa:bb:cc:dd:ee:03"), + ] + api = MagicMock(return_value=leases) + plugin_objects = MagicMock() + + mikrotik.MT_USER = "user" + mikrotik.MT_PASS = "password" + mikrotik.MT_HOST = "192.168.1.1" + mikrotik.MT_PORT = 8728 + + with patch.object(mikrotik, "connect", return_value=api): + result = mikrotik.get_entries(plugin_objects) + + assert result is plugin_objects + assert plugin_objects.add_object.call_count == 2 + assert [call.kwargs["primaryId"] for call in plugin_objects.add_object.call_args_list] == ["aa:bb:cc:dd:ee:01", "aa:bb:cc:dd:ee:03"]