fix(ping): post the settings where the API expects them, not one node above
Every ping against a live OPNsense failed:
ping job creation failed for 10.30.0.1: {'result': 'failed',
'validations': {'ping.settings.hostname': 'A value is required.'}}
_ping_model_node read GET /api/diagnostics/ping/get and took the single
dict-valued key as the node to post under. The real model nests two levels:
{"ping": {"settings": {"hostname": "", "fam": {"ip": {...}, "ip6": {...}},
"source_address": "", "packetsize": "", ...}}}
so the helper answered "ping" and the job was created with the fields sitting
where the settings node belongs. The hostname never arrived, and the firewall
said so on every single call.
_ping_model_path walks the whole chain of single-dict wrappers and stops at the
first level holding more than one key — the field level, where fam is a dict
too and one more step would land inside a form field. _wrap_in_model nests the
settings accordingly, so a one-level model keeps working and the default, for
when /get cannot be read, is what current firmware ships.
The tests missed this because FakePingAPI answered /get with a one-level model
and read the posted payload back through the same assumption: the fake agreed
with the code about a shape neither of them shares with a device. It now speaks
what an OPNsense speaks, reads the payload through the model path, and a second
test keeps the one-level case covered.
Verified against a live firewall: the job is accepted ({"result": "ok"}) and
10.30.0.1 answers 3 of 3 at 0.116 ms.
Closes #3
This commit is contained in:
+83
-26
@@ -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"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user