diff --git a/napalm_opnsense/ping_mixin.py b/napalm_opnsense/ping_mixin.py index 1c411fd..72bf555 100644 --- a/napalm_opnsense/ping_mixin.py +++ b/napalm_opnsense/ping_mixin.py @@ -52,6 +52,10 @@ logger = logging.getLogger(__name__) _PING_API = "/api/diagnostics/ping" +#: Where the ping fields live when the model cannot be read. Matches what +#: OPNsense 24/25 ship: {"ping": {"settings": {...}}}. +_DEFAULT_PING_MODEL_PATH = ("ping", "settings") + #: Seconds between two ``search_jobs`` polls while waiting for probe results. _POLL_INTERVAL = 0.5 @@ -74,7 +78,7 @@ class OPNsensePingMixin: PING_SWEEP_MAX_TARGETS: int = 512 # Provided by the driver this mixin is mixed into. - _ping_model_node_cache: Optional[str] = None + _ping_model_path_cache: Optional[tuple[str, ...]] = None # ── NAPALM ─────────────────────────────────────────────────────────────── @@ -202,7 +206,7 @@ class OPNsensePingMixin: if source: settings["source_address"] = source - response = self._post(f"{_PING_API}/set", {self._ping_model_node(): settings}) + response = self._post(f"{_PING_API}/set", self._wrap_in_model(settings)) job_id = response.get("uuid") or response.get("id") if not job_id: raise RuntimeError(f"ping job creation failed for {destination}: {response}") @@ -262,27 +266,52 @@ class OPNsensePingMixin: result[str(job_id)] = row return result - def _ping_model_node(self) -> str: - """Name of the model's root node, i.e. the key ``set`` expects. + def _wrap_in_model(self, settings: dict[str, Any]) -> dict[str, Any]: + """Nest *settings* under the keys ``set`` expects, outermost last.""" + payload: dict[str, Any] = settings + for key in reversed(self._ping_model_path()): + payload = {key: payload} + return payload - Read once from ``GET /api/diagnostics/ping/get`` instead of hardcoded, - so a model rename in a future OPNsense release does not silently break - job creation. Falls back to ``"settings"``. + def _ping_model_path(self) -> tuple[str, ...]: + """Keys from the model root down to the fields, e.g. ``("ping", "settings")``. + + Read once from ``GET /api/diagnostics/ping/get`` rather than hardcoded, + so a model rename in a future release does not silently break job + creation. The response mirrors the model, so the path is the chain of + single-dict wrappers around the fields:: + + {"ping": {"settings": {"hostname": "", "fam": {...}, ...}}} + + The descent stops at the first level that holds more than one key — + that is the field level, where ``fam`` is a dict too and following it + would land inside a form field. + + Getting this wrong is expensive and quiet: OPNsense answers a job whose + hostname arrived at the wrong node with ``ping.settings.hostname: A + value is required``, and a sweep turns 254 such refusals into a network + that appears to hold nothing. """ - if self._ping_model_node_cache is not None: - return self._ping_model_node_cache + if self._ping_model_path_cache is not None: + return self._ping_model_path_cache - node = "settings" + path = _DEFAULT_PING_MODEL_PATH try: - response = self._get(f"{_PING_API}/get") - candidates = [key for key, value in response.items() if isinstance(value, dict)] - if len(candidates) == 1: - node = candidates[0] - except Exception as exc: # noqa: BLE001 - the default is a safe guess - logger.debug("Could not read ping model node, assuming %r: %s", node, exc) + node = self._get(f"{_PING_API}/get") + found: list[str] = [] + while isinstance(node, dict) and len(node) == 1: + key, value = next(iter(node.items())) + if not isinstance(value, dict): + break + found.append(key) + node = value + if found: + path = tuple(found) + except Exception as exc: # noqa: BLE001 - the default is what firmware ships + logger.debug("Could not read ping model path, assuming %r: %s", path, exc) - self._ping_model_node_cache = node - return node + self._ping_model_path_cache = path + return path # ── Parsing ────────────────────────────────────────────────────────────── diff --git a/tests/unit/test_ping.py b/tests/unit/test_ping.py index 5e6e6e1..666c581 100644 --- a/tests/unit/test_ping.py +++ b/tests/unit/test_ping.py @@ -20,10 +20,14 @@ from napalm_opnsense.opnsense import OPNsenseDriver class FakePingAPI: """Minimal stand-in for the OPNsense diagnostics ping controller.""" - def __init__(self, stats=None, model_node="settings"): + def __init__(self, stats=None, model_path=("ping", "settings")): #: hostname -> row returned by search_jobs self.stats = stats or {} - self.model_node = model_node + #: Keys from the model root down to the field level. A real OPNsense + #: nests them two deep ({"ping": {"settings": {...}}}); the tests used + #: to assume one, which is why nothing caught the payload going to the + #: wrong node. + self.model_path = tuple(model_path) self.created = [] # payloads passed to set self.started = [] # job ids passed to start self.stopped = [] # job ids passed to stop @@ -34,7 +38,14 @@ class FakePingAPI: def get(self, path): if path == "/api/diagnostics/ping/get": - return {self.model_node: {"hostname": "", "fam": "ip"}} + fields = { + "hostname": "", + "fam": {"ip": {"value": "IPv4", "selected": 1}}, + "source_address": "", + "packetsize": "", + "description": "", + } + return self._nest(fields) if path == "/api/diagnostics/ping/search_jobs": self.search_calls += 1 return {"rows": [self._row(jid, host) for jid, host in self.jobs.items()]} @@ -45,8 +56,7 @@ class FakePingAPI: self.created.append(data) self._next_id += 1 job_id = f"job-{self._next_id}" - settings = (data or {}).get(self.model_node, {}) - self.jobs[job_id] = settings.get("hostname", "") + self.jobs[job_id] = self.fields(data or {}).get("hostname", "") return {"result": "saved", "uuid": job_id} for action, sink in (("start", self.started), ("stop", self.stopped)): if path.startswith(f"/api/diagnostics/ping/{action}/"): @@ -59,12 +69,38 @@ class FakePingAPI: return {"status": "ok"} raise AssertionError(f"unexpected POST {path}") + def _nest(self, fields): + """Wrap *fields* in the model path, innermost first.""" + node = fields + for key in reversed(self.model_path): + node = {key: node} + return node + + def fields(self, payload): + """The field level of a posted payload, or {} if it went to the wrong node.""" + node = payload + for key in self.model_path: + if not isinstance(node, dict) or key not in node: + return {} + node = node[key] + return node if isinstance(node, dict) else {} + def _row(self, job_id, hostname): row = {"id": job_id, "hostname": hostname, "status": "running"} row.update(self.stats.get(hostname, {"send": 0, "received": 0})) return row +def _driver(fake): + """An OPNsenseDriver wired to *fake* instead of a real firewall.""" + with patch("napalm_opnsense.opnsense.requests.Session"): + drv = OPNsenseDriver(hostname="fw", username="k", password="s") + drv.session = MagicMock() + drv._get = fake.get + drv._post = fake.post + return drv + + def alive_row(rtt=1.5, send=1, received=1): return { "send": send, @@ -144,18 +180,14 @@ def test_ping_creates_job_with_destination_and_packetsize(driver, api): driver.ping("10.0.0.1", size=64, source="10.0.0.254") - assert api.created == [ - { - "settings": { - "hostname": "10.0.0.1", - "fam": "ip", - "packetsize": "64", - "interval": "1", - "source_address": "10.0.0.254", - "description": "netork ping sweep", - } - } - ] + assert api.fields(api.created[0]) == { + "hostname": "10.0.0.1", + "fam": "ip", + "packetsize": "64", + "interval": "1", + "source_address": "10.0.0.254", + "description": "netork ping sweep", + } def test_ping_uses_ipv6_family_for_v6_destination(driver, api): @@ -163,7 +195,7 @@ def test_ping_uses_ipv6_family_for_v6_destination(driver, api): driver.ping("2001:db8::1") - assert api.created[0]["settings"]["fam"] == "ip6" + assert api.fields(api.created[0])["fam"] == "ip6" def test_ping_starts_stops_and_removes_the_job(driver, api): @@ -200,19 +232,44 @@ def test_ping_returns_error_when_job_creation_fails(driver, api): assert "hostname" in result["error"] -def test_ping_honours_the_model_root_node_reported_by_the_api(): - fake = FakePingAPI(model_node="ping") +def test_ping_settings_go_to_the_node_the_api_asks_for(): + """A real OPNsense nests the ping model two deep — {"ping": {"settings": + {...}}} — and rejects a job whose hostname arrives anywhere else with + "ping.settings.hostname: A value is required". Posting the fields one level + too high made every probe fail, which a sweep reported as a silent network.""" + fake = FakePingAPI() fake.stats["10.0.0.1"] = alive_row() - with patch("napalm_opnsense.opnsense.requests.Session"): - drv = OPNsenseDriver(hostname="fw", username="k", password="s") - drv.session = MagicMock() - drv._get = fake.get - drv._post = fake.post + drv = _driver(fake) + + with patch("napalm_opnsense.ping_mixin.time.sleep"): + reply = drv.ping("10.0.0.1") + + assert fake.created[0] == { + "ping": { + "settings": { + "hostname": "10.0.0.1", + "fam": "ip", + "packetsize": "100", + "interval": "1", + "description": "netork ping sweep", + } + } + } + assert reply["success"]["probes_sent"] == 1 + + +def test_a_model_that_is_only_one_level_deep_still_works(): + """Older firmware exposes the fields directly under one node. The path is + read from the API rather than assumed, so both shapes work.""" + fake = FakePingAPI(model_path=("settings",)) + fake.stats["10.0.0.1"] = alive_row() + drv = _driver(fake) with patch("napalm_opnsense.ping_mixin.time.sleep"): drv.ping("10.0.0.1") - assert "ping" in fake.created[0] + assert list(fake.created[0]) == ["settings"] + assert fake.created[0]["settings"]["hostname"] == "10.0.0.1" # ---------------------------------------------------------------------------