diff --git a/registry.json b/registry.json index dfe06ec..5077db2 100644 --- a/registry.json +++ b/registry.json @@ -19,7 +19,7 @@ "description": "Monitor WAN1 and WAN2 on Ubiquiti UDM Pro/SE devices. Tracks link state, alive status, and failover events in the global DOCSight event log.", "author": "Oggy512", "repo": "https://github.com/itsDNNS/docsight-modules", - "version": "3.1.3", + "version": "3.1.4", "min_app_version": "2026.2", "type": "integration", "verified": true, diff --git a/tests/test_udm_wan_loader.py b/tests/test_udm_wan_loader.py new file mode 100644 index 0000000..b62b7dd --- /dev/null +++ b/tests/test_udm_wan_loader.py @@ -0,0 +1,178 @@ +import importlib.util +import shutil +import sys +import tempfile +import types +import unittest +from contextlib import contextmanager +from pathlib import Path + + +_MISSING = object() + + +@contextmanager +def isolated_modules(stubs, names): + saved = {name: sys.modules.get(name, _MISSING) for name in names} + try: + for name in names: + sys.modules.pop(name, None) + sys.modules.update(stubs) + yield + finally: + for name in names: + sys.modules.pop(name, None) + for name, module in saved.items(): + if module is not _MISSING: + sys.modules[name] = module + + +def module_stubs(config): + app = types.ModuleType("app") + app.__path__ = [] + + app_web = types.ModuleType("app.web") + app_web.require_auth = lambda function: function + app_web.get_config_manager = lambda: config + + app_collectors = types.ModuleType("app.collectors") + app_collectors.__path__ = [] + + app_collectors_base = types.ModuleType("app.collectors.base") + app_collectors_base.Collector = type("Collector", (), {}) + app_collectors_base.CollectorResult = type("CollectorResult", (), {}) + + app_tz = types.ModuleType("app.tz") + app_tz.utc_now = lambda: None + + flask = types.ModuleType("flask") + + class Blueprint: + def __init__(self, *args, **kwargs): + pass + + def route(self, *args, **kwargs): + return lambda function: function + + flask.Blueprint = Blueprint + flask.jsonify = lambda value: value + flask.render_template = lambda *args, **kwargs: None + + requests = types.ModuleType("requests") + requests.Session = type("Session", (), {}) + requests.exceptions = types.SimpleNamespace( + ConnectionError=type("ConnectionError", (Exception,), {}), + Timeout=type("Timeout", (Exception,), {}), + ) + + return { + "app": app, + "app.web": app_web, + "app.collectors": app_collectors, + "app.collectors.base": app_collectors_base, + "app.tz": app_tz, + "flask": flask, + "requests": requests, + } + + +class UdmWanSyntheticLoaderTests(unittest.TestCase): + def test_routes_reuse_host_loaded_collector(self): + for directory in ("community.udm_wan_monitor", "custom_udm_wan_monitor"): + with self.subTest(directory=directory): + self._assert_routes_reuse_host_loaded_collector(directory) + + def test_failed_fallback_removes_inserted_collector(self): + root = Path(__file__).resolve().parents[1] + directory = "custom_udm_wan_monitor" + routes_name = f"community_modules.{directory}.routes" + collector_name = f"app.modules.{directory}.collector" + stubs = module_stubs({}) + + with tempfile.TemporaryDirectory() as tmp: + installed = Path(tmp) / directory + installed.mkdir() + routes_path = shutil.copy2( + root / "udm-wan-monitor" / "routes.py", + installed / "routes.py", + ) + (installed / "collector.py").write_text( + 'raise RuntimeError("collector failed")\n', + encoding="utf-8", + ) + + with isolated_modules(stubs, self._temporary_names(directory, stubs)): + routes = self._load_leaf(routes_name, routes_path) + + with self.assertRaisesRegex(RuntimeError, "collector failed"): + routes._collector_mod() + + self.assertNotIn(collector_name, sys.modules) + + def _assert_routes_reuse_host_loaded_collector(self, directory): + root = Path(__file__).resolve().parents[1] + routes_name = f"community_modules.{directory}.routes" + collector_name = f"app.modules.{directory}.collector" + config = { + "udm_wan_enabled": True, + "udm_wan_host": "192.0.2.1", + "udm_wan_port": 8443, + "udm_wan_site": "default", + } + stubs = module_stubs(config) + + with tempfile.TemporaryDirectory() as tmp: + installed = Path(tmp) / directory + installed.mkdir() + routes_path = shutil.copy2( + root / "udm-wan-monitor" / "routes.py", + installed / "routes.py", + ) + collector_path = shutil.copy2( + root / "udm-wan-monitor" / "collector.py", + installed / "collector.py", + ) + + with isolated_modules(stubs, self._temporary_names(directory, stubs)): + routes = self._load_leaf(routes_name, routes_path) + collector = self._load_leaf(collector_name, collector_path) + + built_config = routes._build_cfg() + + self.assertEqual( + built_config, + { + "host": "192.0.2.1", + "base": "https://192.0.2.1:8443", + "username": "", + "password": "", + "site": "default", + "verify_ssl": False, + "enabled": True, + }, + ) + self.assertIs(routes._collector_mod(), collector) + + def _temporary_names(self, directory, stubs): + return set(stubs) | { + "app.modules", + f"app.modules.{directory}", + f"app.modules.{directory}.collector", + "app.modules.community.udm_wan_monitor.collector", + "community_modules", + f"community_modules.{directory}", + f"community_modules.{directory}.routes", + } + + def _load_leaf(self, name, path): + spec = importlib.util.spec_from_file_location(name, path) + self.assertIsNotNone(spec) + self.assertIsNotNone(spec.loader) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +if __name__ == "__main__": + unittest.main() diff --git a/udm-wan-monitor/collector.py b/udm-wan-monitor/collector.py index f510014..0e4f86a 100644 --- a/udm-wan-monitor/collector.py +++ b/udm-wan-monitor/collector.py @@ -1,5 +1,5 @@ """ -UDM WAN Monitor — Collector v3.1.1 +UDM WAN Monitor — Collector v3.1.4 Polls /proxy/network/api/s/{site}/stat/device and extracts WAN status from the UDM device entry. diff --git a/udm-wan-monitor/manifest.json b/udm-wan-monitor/manifest.json index a287eb1..7f404da 100644 --- a/udm-wan-monitor/manifest.json +++ b/udm-wan-monitor/manifest.json @@ -2,7 +2,7 @@ "id": "community.udm_wan_monitor", "name": "UDM WAN-Monitor", "description": "Überwacht WAN1 und WAN2 einer Ubiquiti UDM Pro und erzeugt Events im globalen DOCSight-Log.", - "version": "3.1.3", + "version": "3.1.4", "author": "Oggy512", "minAppVersion": "2026.2", "type": "integration", diff --git a/udm-wan-monitor/routes.py b/udm-wan-monitor/routes.py index fedb2c5..de7543a 100644 --- a/udm-wan-monitor/routes.py +++ b/udm-wan-monitor/routes.py @@ -1,5 +1,5 @@ """ -UDM WAN Monitor — Flask Routes v3.1.1 +UDM WAN Monitor — Flask Routes v3.1.4 GET /udm-wan → Standalone dashboard page GET /api/udm-wan/status → Latest cached data (JSON) @@ -39,15 +39,50 @@ def _collector(): pass return None +def _collector_mod(): + """Return the sibling collector module. + + DOCSight loads a community module's routes.py as a synthetic top-level + module (``community_modules..routes``) without registering parent + packages, so relative imports like ``from .collector import ...`` raise + ModuleNotFoundError at request time. Reuse the collector module the app + has already imported (``app.modules..collector``), falling back to + loading the file directly. + """ + import importlib.util # noqa: PLC0415 + import os # noqa: PLC0415 + import sys # noqa: PLC0415 + + directory = os.path.basename(os.path.dirname(os.path.abspath(__file__))) + name = f"app.modules.{directory}.collector" + mod = sys.modules.get(name) + if mod is None: + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "collector.py") + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise ImportError(f"Cannot load collector module from {path}") + mod = importlib.util.module_from_spec(spec) + sys.modules[name] = mod + try: + spec.loader.exec_module(mod) + except Exception: + if sys.modules.get(name) is mod: + del sys.modules[name] + raise + return mod + + def _build_cfg(): c = _cfg() - from .collector import _build_cfg_from # noqa: PLC0415 + _collector = _collector_mod() + _build_cfg_from = _collector._build_cfg_from d = _build_cfg_from(c) d["enabled"] = bool(c.get("udm_wan_enabled", False)) return d def _open_session(cfg): - from .collector import _login # noqa: PLC0415 + _collector = _collector_mod() + _login = _collector._login return _login(cfg) # ── Pages ────────────────────────────────────────────────────────────────────── @@ -87,7 +122,9 @@ def api_detail(): try: session = _open_session(cfg) - from .collector import _fetch_udm_device, parse_device # noqa: PLC0415 + _collector = _collector_mod() + _fetch_udm_device = _collector._fetch_udm_device + parse_device = _collector.parse_device device = _fetch_udm_device(session, cfg) parsed = parse_device(device) except req.exceptions.ConnectionError: @@ -153,7 +190,9 @@ def api_test(): return jsonify({"ok": False, "error": "Host not configured"}), 400 try: session = _open_session(cfg) - from .collector import _fetch_udm_device, parse_device # noqa: PLC0415 + _collector = _collector_mod() + _fetch_udm_device = _collector._fetch_udm_device + parse_device = _collector.parse_device device = _fetch_udm_device(session, cfg) parsed = parse_device(device) return jsonify({