Christian Manivong
4dd0fc2aee
`firmware/remove` acts on the OPNsense plugin set, and `get_packages` reads the same list — so software installed as a plain FreeBSD package is invisible to the one and unreachable by the other. The Wazuh agent is exactly that, on a driver the agent plugin lists as supported. Observed during a fleet-wide rollback on 2026-09-19: the `gw` device could not be handled through netOrk at all, and the request posted for it could never have succeeded. A name that is not a plugin now raises NotImplementedError rather than being POSTed. A request that cannot work reports failure for the wrong reason and sends whoever reads it looking in the wrong place; netOrk turns NotImplementedError into a 501, which is the accurate answer. Reaching plain packages would need shell access, and the credentials stored for these devices are frequently API-key only — that is a decision of its own, not a detail of this one. The injection guard still runs first: a malformed name is a ValueError before anything asks whether it is a plugin. netork#241
2597 lines
98 KiB
Python
2597 lines
98 KiB
Python
"""Unit tests for OPNsenseDriver — no real device required."""
|
||
|
||
import json
|
||
import pytest
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from napalm_opnsense.opnsense import OPNsenseDriver
|
||
from napalm.base.exceptions import ConnectionException, ConnectionClosedException
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.fixture
|
||
def driver():
|
||
"""Return a driver instance with a mocked requests.Session."""
|
||
with patch("napalm_opnsense.opnsense.requests.Session"):
|
||
drv = OPNsenseDriver(
|
||
hostname="opnsense.example.com",
|
||
username="api_key",
|
||
password="api_secret",
|
||
optional_args={"verify": False},
|
||
)
|
||
drv.session = MagicMock()
|
||
yield drv
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Sample API responses
|
||
# ---------------------------------------------------------------------------
|
||
|
||
STATUS_RESPONSE = {
|
||
"hostname": "opnsense01",
|
||
"version": "24.7",
|
||
"model": "OPNsense",
|
||
"serial": "ABC123",
|
||
"uptime": 12345,
|
||
}
|
||
|
||
INTERFACES_RESPONSE = {
|
||
"interfaces": [
|
||
{
|
||
"name": "em0",
|
||
"up": True,
|
||
"enabled": True,
|
||
"descr": "LAN",
|
||
"mac": "AA:BB:CC:DD:EE:FF",
|
||
"speed_mbps": 1000,
|
||
"mtu": 1500,
|
||
},
|
||
{
|
||
"name": "em1",
|
||
"up": False,
|
||
"enabled": True,
|
||
"descr": "WAN",
|
||
"mac": "AA:BB:CC:DD:EE:00",
|
||
"speed_mbps": None,
|
||
"mtu": 0,
|
||
},
|
||
]
|
||
}
|
||
|
||
# Bare-list shape with the "identifier" field: OPNsense's assigned interface
|
||
# name, distinct from the physical "device" the dict is keyed by. Wake-on-LAN
|
||
# is addressed by the former, everything else in this driver by the latter.
|
||
IDENTIFIED_INTERFACES_RESPONSE = [
|
||
{
|
||
"device": "em0",
|
||
"identifier": "lan",
|
||
"description": "LAN",
|
||
"status": "up",
|
||
"enabled": True,
|
||
"macaddr": "AA:BB:CC:DD:EE:FF",
|
||
"speed_mbps": 1000,
|
||
"mtu": 1500,
|
||
},
|
||
{
|
||
"device": "em1",
|
||
"identifier": "opt1",
|
||
"description": "GUEST",
|
||
"status": "down",
|
||
"enabled": True,
|
||
"macaddr": "AA:BB:CC:DD:EE:00",
|
||
"mtu": 1500,
|
||
},
|
||
]
|
||
|
||
# Real shape confirmed against a live OPNsense 24.7 instance's
|
||
# GET /api/interfaces/overview/export (see issue #14): a bare top-level list
|
||
# of interface dicts, addressed via "device" and CIDR "addr4"/"addr6"
|
||
# strings — not the "items"/"interface"/"address"/"prefix" shape this
|
||
# fixture previously assumed, which never matched get_interfaces_ip()'s
|
||
# actual parsing and made all four TestGetInterfacesIp tests fail.
|
||
ADDRESSES_RESPONSE = [
|
||
{"device": "em0", "addr4": "192.0.2.10/24", "addr6": "2001:db8::1/64"},
|
||
{"device": "em1", "addr4": "203.0.113.5/30", "addr6": ""},
|
||
]
|
||
|
||
ARP_RESPONSE = {
|
||
"arp": [
|
||
{"intf": "em0", "mac": "AA:BB:CC:DD:EE:01", "ip": "192.0.2.1", "expires": 900},
|
||
{"intf": "em0", "mac": "AA:BB:CC:DD:EE:02", "ip": "192.0.2.2", "expires": 600},
|
||
]
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _make_json_response(data):
|
||
mock_resp = MagicMock()
|
||
mock_resp.json.return_value = data
|
||
mock_resp.raise_for_status.return_value = None
|
||
return mock_resp
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# open() / close() / is_alive()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestOpenClose:
|
||
def test_open_raises_connection_exception_on_error(self):
|
||
drv = OPNsenseDriver(
|
||
hostname="unreachable.invalid",
|
||
username="k",
|
||
password="s",
|
||
optional_args={"verify": False},
|
||
)
|
||
with pytest.raises(ConnectionException):
|
||
drv.open()
|
||
|
||
def test_close_clears_session(self, driver):
|
||
driver.close()
|
||
assert driver.session is None
|
||
|
||
def test_close_is_idempotent(self, driver):
|
||
driver.close()
|
||
driver.close() # second call must not raise
|
||
|
||
|
||
class TestIsAlive:
|
||
def test_returns_false_when_no_session(self):
|
||
drv = OPNsenseDriver("host", "u", "p")
|
||
assert drv.is_alive() == {"is_alive": False}
|
||
|
||
def test_returns_true_on_successful_connection(self, driver):
|
||
with patch("napalm_opnsense.opnsense.socket.create_connection") as mock_conn:
|
||
mock_conn.return_value.__enter__ = MagicMock(return_value=None)
|
||
mock_conn.return_value.__exit__ = MagicMock(return_value=False)
|
||
result = driver.is_alive()
|
||
assert result == {"is_alive": True}
|
||
|
||
def test_returns_false_on_socket_error(self, driver):
|
||
with patch(
|
||
"napalm_opnsense.opnsense.socket.create_connection",
|
||
side_effect=OSError("refused"),
|
||
):
|
||
result = driver.is_alive()
|
||
assert result == {"is_alive": False}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _get()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestInternalGet:
|
||
def test_raises_when_no_session(self):
|
||
drv = OPNsenseDriver("host", "u", "p")
|
||
with pytest.raises(ConnectionClosedException):
|
||
drv._get("/api/core/system/status")
|
||
|
||
def test_calls_correct_url(self, driver):
|
||
driver.session.get.return_value = _make_json_response(STATUS_RESPONSE)
|
||
driver._get("/api/core/system/status")
|
||
driver.session.get.assert_called_once_with(
|
||
"https://opnsense.example.com/api/core/system/status",
|
||
timeout=60,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_facts()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGetFacts:
|
||
def test_returns_required_keys(self, driver):
|
||
driver._get = lambda path: (
|
||
STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
|
||
)
|
||
facts = driver.get_facts()
|
||
for key in ("vendor", "model", "hostname", "fqdn", "os_version",
|
||
"serial_number", "uptime", "interface_list"):
|
||
assert key in facts
|
||
|
||
def test_vendor_constant(self, driver):
|
||
driver._get = lambda path: (
|
||
STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
|
||
)
|
||
facts = driver.get_facts()
|
||
assert facts["vendor"] == "OPNsense"
|
||
|
||
def test_hostname_parsed(self, driver):
|
||
driver._get = lambda path: (
|
||
STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
|
||
)
|
||
facts = driver.get_facts()
|
||
assert facts["hostname"] == "opnsense01"
|
||
|
||
def test_os_version_parsed(self, driver):
|
||
driver._get = lambda path: (
|
||
STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
|
||
)
|
||
facts = driver.get_facts()
|
||
assert facts["os_version"] == "24.7"
|
||
|
||
def test_interface_list_populated(self, driver):
|
||
driver._get = lambda path: (
|
||
STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
|
||
)
|
||
facts = driver.get_facts()
|
||
assert "em0" in facts["interface_list"]
|
||
assert "em1" in facts["interface_list"]
|
||
|
||
def test_interface_list_empty_on_getter_failure(self, driver):
|
||
def fail_on_interfaces(path):
|
||
if "overview" in path:
|
||
raise RuntimeError("no endpoint")
|
||
return STATUS_RESPONSE
|
||
|
||
driver._get = fail_on_interfaces
|
||
facts = driver.get_facts()
|
||
assert facts["interface_list"] == []
|
||
|
||
def test_serial_number(self, driver):
|
||
driver._get = lambda path: (
|
||
STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
|
||
)
|
||
facts = driver.get_facts()
|
||
assert facts["serial_number"] == "ABC123"
|
||
|
||
def test_uptime(self, driver):
|
||
driver._get = lambda path: (
|
||
STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
|
||
)
|
||
facts = driver.get_facts()
|
||
assert facts["uptime"] == 12345
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_interfaces()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGetInterfaces:
|
||
def test_interface_count(self, driver):
|
||
driver._get = lambda path: INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert len(ifaces) == 2
|
||
|
||
def test_is_up_and_enabled(self, driver):
|
||
driver._get = lambda path: INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert ifaces["em0"]["is_up"] is True
|
||
assert ifaces["em0"]["is_enabled"] is True
|
||
assert ifaces["em1"]["is_up"] is False
|
||
|
||
def test_speed(self, driver):
|
||
driver._get = lambda path: INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert ifaces["em0"]["speed"] == 1000.0
|
||
assert ifaces["em1"]["speed"] == 0.0
|
||
|
||
def test_mac_address_lowercase(self, driver):
|
||
driver._get = lambda path: INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert ifaces["em0"]["mac_address"] == "aa:bb:cc:dd:ee:ff"
|
||
|
||
def test_description(self, driver):
|
||
driver._get = lambda path: INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert ifaces["em0"]["description"] == "LAN"
|
||
assert ifaces["em1"]["description"] == "WAN"
|
||
|
||
def test_last_flapped_is_negative_one(self, driver):
|
||
driver._get = lambda path: INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert ifaces["em0"]["last_flapped"] == -1.0
|
||
|
||
def test_identifier_passed_through(self, driver):
|
||
# OPNsense's *assigned* name, which is what send_wake_on_lan() needs —
|
||
# the dict key is the physical device ("em0"), which it rejects.
|
||
driver._get = lambda path: IDENTIFIED_INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert ifaces["em0"]["identifier"] == "lan"
|
||
assert ifaces["em1"]["identifier"] == "opt1"
|
||
|
||
def test_identifier_empty_when_unassigned(self, driver):
|
||
# An interface OPNsense has not assigned carries no identifier.
|
||
driver._get = lambda path: INTERFACES_RESPONSE
|
||
ifaces = driver.get_interfaces()
|
||
assert ifaces["em0"]["identifier"] == ""
|
||
|
||
def test_empty_response(self, driver):
|
||
driver._get = lambda path: {"interfaces": []}
|
||
assert driver.get_interfaces() == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_interfaces_ip()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGetInterfacesIp:
|
||
def test_entry_count(self, driver):
|
||
driver._get = lambda path: ADDRESSES_RESPONSE
|
||
result = driver.get_interfaces_ip()
|
||
# em0 has 2 addresses, em1 has 1
|
||
assert len(result) == 2
|
||
|
||
def test_ipv4_entry(self, driver):
|
||
driver._get = lambda path: ADDRESSES_RESPONSE
|
||
result = driver.get_interfaces_ip()
|
||
assert "192.0.2.10" in result["em0"]["ipv4"]
|
||
assert result["em0"]["ipv4"]["192.0.2.10"]["prefix_length"] == 24
|
||
|
||
def test_ipv6_entry(self, driver):
|
||
driver._get = lambda path: ADDRESSES_RESPONSE
|
||
result = driver.get_interfaces_ip()
|
||
assert "2001:db8::1" in result["em0"]["ipv6"]
|
||
assert result["em0"]["ipv6"]["2001:db8::1"]["prefix_length"] == 64
|
||
|
||
def test_second_interface(self, driver):
|
||
driver._get = lambda path: ADDRESSES_RESPONSE
|
||
result = driver.get_interfaces_ip()
|
||
assert "203.0.113.5" in result["em1"]["ipv4"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_arp_table()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGetArpTable:
|
||
def test_entry_count(self, driver):
|
||
driver._get = lambda path: ARP_RESPONSE
|
||
table = driver.get_arp_table()
|
||
assert len(table) == 2
|
||
|
||
def test_entry_structure(self, driver):
|
||
driver._get = lambda path: ARP_RESPONSE
|
||
entry = driver.get_arp_table()[0]
|
||
for key in ("interface", "mac", "ip", "age"):
|
||
assert key in entry
|
||
|
||
def test_ip_values(self, driver):
|
||
driver._get = lambda path: ARP_RESPONSE
|
||
ips = {e["ip"] for e in driver.get_arp_table()}
|
||
assert "192.0.2.1" in ips
|
||
assert "192.0.2.2" in ips
|
||
|
||
def test_mac_lowercase(self, driver):
|
||
driver._get = lambda path: ARP_RESPONSE
|
||
macs = {e["mac"] for e in driver.get_arp_table()}
|
||
assert all(m == m.lower() for m in macs)
|
||
|
||
def test_list_response_format(self, driver):
|
||
"""ARP endpoint may return a bare list instead of dict."""
|
||
bare_list = ARP_RESPONSE["arp"]
|
||
driver._get = lambda path: bare_list
|
||
table = driver.get_arp_table()
|
||
assert len(table) == 2
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_interfaces_counters()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
INTERFACE_STATISTICS_RESPONSE = {
|
||
"statistics": {
|
||
"em0": {
|
||
"input-packets": 10000,
|
||
"output-packets": 8000,
|
||
"input-bytes": 1000000,
|
||
"output-bytes": 800000,
|
||
"input-errors": 5,
|
||
"output-errors": 2,
|
||
"input-drops": 3,
|
||
"output-drops": 1,
|
||
"input-multicasts": 100,
|
||
"output-multicasts": 50,
|
||
"input-broadcasts": 20,
|
||
"output-broadcasts": 10,
|
||
},
|
||
"em1": {
|
||
"input-packets": 500,
|
||
"output-packets": 300,
|
||
"input-bytes": 50000,
|
||
"output-bytes": 30000,
|
||
"input-errors": 0,
|
||
"output-errors": 0,
|
||
"input-drops": 0,
|
||
"output-drops": 0,
|
||
"input-multicasts": 0,
|
||
"output-multicasts": 0,
|
||
"input-broadcasts": 0,
|
||
"output-broadcasts": 0,
|
||
},
|
||
}
|
||
}
|
||
|
||
|
||
class TestGetInterfacesCounters:
|
||
def test_interface_count(self, driver):
|
||
driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE
|
||
counters = driver.get_interfaces_counters()
|
||
assert len(counters) == 2
|
||
|
||
def test_required_keys(self, driver):
|
||
driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE
|
||
entry = driver.get_interfaces_counters()["em0"]
|
||
for key in (
|
||
"tx_errors", "rx_errors", "tx_discards", "rx_discards",
|
||
"tx_octets", "rx_octets",
|
||
"tx_unicast_packets", "rx_unicast_packets",
|
||
"tx_multicast_packets", "rx_multicast_packets",
|
||
"tx_broadcast_packets", "rx_broadcast_packets",
|
||
):
|
||
assert key in entry
|
||
|
||
def test_rx_octets(self, driver):
|
||
driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE
|
||
assert driver.get_interfaces_counters()["em0"]["rx_octets"] == 1000000
|
||
|
||
def test_tx_errors(self, driver):
|
||
driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE
|
||
assert driver.get_interfaces_counters()["em0"]["tx_errors"] == 2
|
||
|
||
def test_rx_errors(self, driver):
|
||
driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE
|
||
assert driver.get_interfaces_counters()["em0"]["rx_errors"] == 5
|
||
|
||
def test_zero_counters(self, driver):
|
||
driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE
|
||
em1 = driver.get_interfaces_counters()["em1"]
|
||
assert em1["tx_errors"] == 0
|
||
assert em1["rx_errors"] == 0
|
||
|
||
def test_empty_statistics(self, driver):
|
||
driver._get = lambda path: {"statistics": {}}
|
||
assert driver.get_interfaces_counters() == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_environment()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
SYSTEM_RESOURCES_RESPONSE = {
|
||
"cpu": {"used": "25"},
|
||
"memory": {"total": "4096000000", "used": "2048000000"},
|
||
}
|
||
|
||
SYSTEM_TEMP_RESPONSE = {
|
||
"data": [
|
||
{"device": "cpu0", "temperature": "52.5"},
|
||
{"device": "cpu1", "temperature": "48.0"},
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetEnvironment:
|
||
def test_required_top_keys(self, driver):
|
||
def fake_get(path):
|
||
if "temperature" in path:
|
||
return SYSTEM_TEMP_RESPONSE
|
||
return SYSTEM_RESOURCES_RESPONSE
|
||
|
||
driver._get = fake_get
|
||
env = driver.get_environment()
|
||
for key in ("fans", "temperature", "power", "cpu", "memory"):
|
||
assert key in env
|
||
|
||
def test_cpu_usage(self, driver):
|
||
driver._get = lambda path: (
|
||
SYSTEM_TEMP_RESPONSE if "temperature" in path else SYSTEM_RESOURCES_RESPONSE
|
||
)
|
||
env = driver.get_environment()
|
||
assert env["cpu"][0]["%usage"] == 25.0
|
||
|
||
def test_memory_values(self, driver):
|
||
driver._get = lambda path: (
|
||
SYSTEM_TEMP_RESPONSE if "temperature" in path else SYSTEM_RESOURCES_RESPONSE
|
||
)
|
||
env = driver.get_environment()
|
||
assert env["memory"]["used_ram"] == 2048000000
|
||
assert env["memory"]["available_ram"] == 2048000000
|
||
|
||
def test_temperature_sensors(self, driver):
|
||
driver._get = lambda path: (
|
||
SYSTEM_TEMP_RESPONSE if "temperature" in path else SYSTEM_RESOURCES_RESPONSE
|
||
)
|
||
env = driver.get_environment()
|
||
assert "cpu0" in env["temperature"]
|
||
assert env["temperature"]["cpu0"]["temperature"] == 52.5
|
||
|
||
def test_temperature_alert_thresholds(self, driver):
|
||
driver._get = lambda path: (
|
||
{"data": [{"device": "cpu0", "temperature": "85.0"}]}
|
||
if "temperature" in path else SYSTEM_RESOURCES_RESPONSE
|
||
)
|
||
env = driver.get_environment()
|
||
assert env["temperature"]["cpu0"]["is_alert"] is True
|
||
assert env["temperature"]["cpu0"]["is_critical"] is False
|
||
|
||
def test_temperature_endpoint_failing_gracefully(self, driver):
|
||
"""Driver must not raise if temperature endpoint is unavailable."""
|
||
def fake_get(path):
|
||
if "temperature" in path:
|
||
raise Exception("no sensor data")
|
||
return SYSTEM_RESOURCES_RESPONSE
|
||
|
||
driver._get = fake_get
|
||
env = driver.get_environment()
|
||
assert env["temperature"] == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_route_to()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
ROUTES_RESPONSE = {
|
||
"route": [
|
||
{
|
||
"network": "0.0.0.0/0",
|
||
"gateway": "192.0.2.1",
|
||
"flags": "UGS",
|
||
"netif": "em1",
|
||
"proto": "static",
|
||
"priority": 1,
|
||
},
|
||
{
|
||
"network": "192.0.2.0/24",
|
||
"gateway": "",
|
||
"flags": "U",
|
||
"netif": "em0",
|
||
"proto": "kernel",
|
||
"priority": 0,
|
||
},
|
||
{
|
||
"network": "198.51.100.0/24",
|
||
"gateway": "192.0.2.5",
|
||
"flags": "UGS",
|
||
"netif": "em0",
|
||
"proto": "static",
|
||
"priority": 1,
|
||
},
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetRouteTo:
|
||
def test_returns_all_routes_without_filter(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
routes = driver.get_route_to()
|
||
assert len(routes) == 3
|
||
|
||
def test_default_route_present(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
assert "0.0.0.0/0" in driver.get_route_to()
|
||
|
||
def test_next_hop(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
entry = driver.get_route_to()["0.0.0.0/0"][0]
|
||
assert entry["next_hop"] == "192.0.2.1"
|
||
|
||
def test_outgoing_interface(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
entry = driver.get_route_to()["0.0.0.0/0"][0]
|
||
assert entry["outgoing_interface"] == "em1"
|
||
|
||
def test_protocol_static(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
assert driver.get_route_to()["0.0.0.0/0"][0]["protocol"] == "static"
|
||
|
||
def test_protocol_kernel_mapped_to_connected(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
assert driver.get_route_to()["192.0.2.0/24"][0]["protocol"] == "connected"
|
||
|
||
def test_filter_by_destination(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
routes = driver.get_route_to(destination="0.0.0.0/0")
|
||
assert "0.0.0.0/0" in routes
|
||
assert "192.0.2.0/24" not in routes
|
||
|
||
def test_filter_by_protocol(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
routes = driver.get_route_to(protocol="static")
|
||
assert all(
|
||
e["protocol"] == "static"
|
||
for entries in routes.values()
|
||
for e in entries
|
||
)
|
||
|
||
def test_required_keys_in_entry(self, driver):
|
||
driver._get = lambda path: ROUTES_RESPONSE
|
||
entry = driver.get_route_to()["0.0.0.0/0"][0]
|
||
for key in (
|
||
"protocol", "current_active", "last_active", "age",
|
||
"next_hop", "outgoing_interface", "selected_next_hop",
|
||
"preference", "inactive_reason", "routing_table",
|
||
"protocol_attributes",
|
||
):
|
||
assert key in entry
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_ipv6_neighbors_table()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
NDP_RESPONSE = {
|
||
"rows": [
|
||
{
|
||
"intf": "em0",
|
||
"mac": "aa:bb:cc:dd:ee:01",
|
||
"ip": "fe80::1",
|
||
"expires": 120,
|
||
"state": "REACHABLE",
|
||
},
|
||
{
|
||
"intf": "em0",
|
||
"mac": "aa:bb:cc:dd:ee:02",
|
||
"ip": "2001:db8::1",
|
||
"expires": 60,
|
||
"state": "STALE",
|
||
},
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetIpv6NeighborsTable:
|
||
def test_entry_count(self, driver):
|
||
driver._get = lambda path: NDP_RESPONSE
|
||
assert len(driver.get_ipv6_neighbors_table()) == 2
|
||
|
||
def test_required_keys(self, driver):
|
||
driver._get = lambda path: NDP_RESPONSE
|
||
entry = driver.get_ipv6_neighbors_table()[0]
|
||
for key in ("interface", "mac", "ip", "age", "state"):
|
||
assert key in entry
|
||
|
||
def test_ip_values(self, driver):
|
||
driver._get = lambda path: NDP_RESPONSE
|
||
ips = {e["ip"] for e in driver.get_ipv6_neighbors_table()}
|
||
assert "fe80::1" in ips
|
||
assert "2001:db8::1" in ips
|
||
|
||
def test_mac_lowercase(self, driver):
|
||
driver._get = lambda path: NDP_RESPONSE
|
||
macs = {e["mac"] for e in driver.get_ipv6_neighbors_table()}
|
||
assert all(m == m.lower() for m in macs)
|
||
|
||
def test_list_format_response(self, driver):
|
||
"""Endpoint may return a bare list."""
|
||
driver._get = lambda path: NDP_RESPONSE["rows"]
|
||
assert len(driver.get_ipv6_neighbors_table()) == 2
|
||
|
||
def test_empty_response(self, driver):
|
||
driver._get = lambda path: {"rows": []}
|
||
assert driver.get_ipv6_neighbors_table() == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_lldp_neighbors() / get_lldp_neighbors_detail()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
LLDP_RESPONSE = {
|
||
"rows": [
|
||
{
|
||
"local_port": "em0",
|
||
"port_id": "eth1",
|
||
"chassis_id": "aa:bb:cc:dd:ee:ff",
|
||
"system_name": "core-sw-01",
|
||
"port_description": "uplink",
|
||
"system_description": "Cisco IOS",
|
||
"system_capabilities": "bridge, router",
|
||
"enabled_capabilities": "bridge",
|
||
}
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetLldpNeighbors:
|
||
def test_returns_neighbor(self, driver):
|
||
driver._get = lambda path: LLDP_RESPONSE
|
||
neighbors = driver.get_lldp_neighbors()
|
||
assert "em0" in neighbors
|
||
assert neighbors["em0"][0]["hostname"] == "core-sw-01"
|
||
assert neighbors["em0"][0]["port"] == "eth1"
|
||
|
||
def test_plugin_not_installed_returns_empty(self, driver):
|
||
driver._get = lambda path: (_ for _ in ()).throw(Exception("404"))
|
||
assert driver.get_lldp_neighbors() == {}
|
||
|
||
|
||
class TestGetLldpNeighborsDetail:
|
||
def test_required_keys(self, driver):
|
||
driver._get = lambda path: LLDP_RESPONSE
|
||
detail = driver.get_lldp_neighbors_detail()
|
||
entry = detail["em0"][0]
|
||
for key in (
|
||
"remote_chassis_id", "remote_system_name", "remote_port",
|
||
"remote_port_description", "remote_system_description",
|
||
"remote_system_capab", "remote_system_enable_capab",
|
||
):
|
||
assert key in entry
|
||
|
||
def test_chassis_id(self, driver):
|
||
driver._get = lambda path: LLDP_RESPONSE
|
||
assert driver.get_lldp_neighbors_detail()["em0"][0]["remote_chassis_id"] == "aa:bb:cc:dd:ee:ff"
|
||
|
||
def test_system_name(self, driver):
|
||
driver._get = lambda path: LLDP_RESPONSE
|
||
assert driver.get_lldp_neighbors_detail()["em0"][0]["remote_system_name"] == "core-sw-01"
|
||
|
||
def test_capabilities_parsed(self, driver):
|
||
driver._get = lambda path: LLDP_RESPONSE
|
||
capab = driver.get_lldp_neighbors_detail()["em0"][0]["remote_system_capab"]
|
||
assert "bridge" in capab
|
||
assert "router" in capab
|
||
|
||
def test_interface_filter(self, driver):
|
||
driver._get = lambda path: LLDP_RESPONSE
|
||
detail = driver.get_lldp_neighbors_detail(interface="em99")
|
||
assert detail == {}
|
||
|
||
def test_plugin_not_installed_returns_empty(self, driver):
|
||
driver._get = lambda path: (_ for _ in ()).throw(Exception("404"))
|
||
assert driver.get_lldp_neighbors_detail() == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_ntp_servers()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
NTP_STATUS_RESPONSE = {
|
||
"peers": [
|
||
{"address": "pool.ntp.org", "state": "synced"},
|
||
{"address": "time.cloudflare.com", "state": "candidate"},
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetNtpServers:
|
||
def test_entry_count(self, driver):
|
||
driver._get = lambda path: NTP_STATUS_RESPONSE
|
||
servers = driver.get_ntp_servers()
|
||
assert len(servers) == 2
|
||
|
||
def test_server_addresses(self, driver):
|
||
driver._get = lambda path: NTP_STATUS_RESPONSE
|
||
servers = driver.get_ntp_servers()
|
||
assert "pool.ntp.org" in servers
|
||
assert "time.cloudflare.com" in servers
|
||
|
||
def test_value_is_empty_dict(self, driver):
|
||
driver._get = lambda path: NTP_STATUS_RESPONSE
|
||
for v in driver.get_ntp_servers().values():
|
||
assert v == {}
|
||
|
||
def test_endpoint_failure_returns_empty(self, driver):
|
||
driver._get = lambda path: (_ for _ in ()).throw(Exception("service not running"))
|
||
assert driver.get_ntp_servers() == {}
|
||
|
||
def test_empty_peers(self, driver):
|
||
driver._get = lambda path: {"peers": []}
|
||
assert driver.get_ntp_servers() == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_vlans()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
VLAN_SEARCH_RESPONSE = {
|
||
"rows": [
|
||
{"tag": "10", "vlanif": "em0_vlan10", "if": "em0", "descr": "Management", "pcp": "0"},
|
||
{"tag": "20", "vlanif": "em0_vlan20 [LAN]", "if": "em0", "descr": "", "pcp": "0"},
|
||
{"tag": "100", "vlanif": "em1_vlan100", "if": "em1", "descr": "Guest WiFi", "pcp": "0"},
|
||
],
|
||
"rowCount": 3,
|
||
"total": 3,
|
||
"current": 1,
|
||
}
|
||
|
||
|
||
class TestGetVlans:
|
||
def test_returns_all_vlans(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
assert len(driver.get_vlans()) == 3
|
||
|
||
def test_keyed_by_tag_string(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
vlans = driver.get_vlans()
|
||
assert "10" in vlans
|
||
assert "20" in vlans
|
||
assert "100" in vlans
|
||
|
||
def test_name_from_descr(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
assert driver.get_vlans()["10"]["name"] == "Management"
|
||
|
||
def test_name_falls_back_to_vlanif_when_no_descr(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
# tag 20 has no descr — should use stripped vlanif
|
||
assert driver.get_vlans()["20"]["name"] == "em0_vlan20"
|
||
|
||
def test_interface_in_list(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
assert driver.get_vlans()["10"]["interfaces"] == ["em0_vlan10"]
|
||
|
||
def test_bracket_annotation_stripped_from_vlanif(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
# "em0_vlan20 [LAN]" must be stored as "em0_vlan20"
|
||
assert driver.get_vlans()["20"]["interfaces"] == ["em0_vlan20"]
|
||
|
||
def test_required_keys_present(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
for vlan in driver.get_vlans().values():
|
||
assert "name" in vlan
|
||
assert "interfaces" in vlan
|
||
|
||
def test_empty_response_returns_empty_dict(self, driver):
|
||
driver._get = lambda path: {"rows": [], "rowCount": 0, "total": 0}
|
||
assert driver.get_vlans() == {}
|
||
|
||
def test_multiple_vlans_on_different_parents(self, driver):
|
||
driver._get = lambda path: VLAN_SEARCH_RESPONSE
|
||
vlans = driver.get_vlans()
|
||
assert vlans["100"]["interfaces"] == ["em1_vlan100"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_bgp_neighbors()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BGP_CFG_RESPONSE = {
|
||
"bgp": {
|
||
"asnumber": "65000",
|
||
"routerid": "1.2.3.4",
|
||
"enabled": "1",
|
||
}
|
||
}
|
||
|
||
BGP_NEIGHBORS_RESPONSE = {
|
||
"response": {
|
||
"10.0.0.1": {
|
||
"remoteAs": 65001,
|
||
"localAs": 65000,
|
||
"nbrDesc": "upstream-peer",
|
||
"bgpState": "Established",
|
||
"bgpTimerUpMsec": 3723000,
|
||
"remoteRouterId": "10.0.0.1",
|
||
"adminShutdown": False,
|
||
"addressFamilyInfo": {
|
||
"ipv4Unicast": {
|
||
"sentPrefixCounter": 5,
|
||
"prefixReceivedCount": 20,
|
||
"acceptedPrefixCounter": 18,
|
||
}
|
||
},
|
||
},
|
||
"10.0.0.2": {
|
||
"remoteAs": 65002,
|
||
"localAs": 65000,
|
||
"nbrDesc": "",
|
||
"bgpState": "Active",
|
||
"bgpTimerUpMsec": 0,
|
||
"remoteRouterId": "",
|
||
"adminShutdown": True,
|
||
"addressFamilyInfo": {},
|
||
},
|
||
}
|
||
}
|
||
|
||
|
||
class TestGetBgpNeighbors:
|
||
def _fake_get(self, path):
|
||
if "diagnostics/bgpneighbors" in path:
|
||
return BGP_NEIGHBORS_RESPONSE
|
||
return BGP_CFG_RESPONSE
|
||
|
||
def test_returns_global_vrf(self, driver):
|
||
driver._get = self._fake_get
|
||
result = driver.get_bgp_neighbors()
|
||
assert "global" in result
|
||
|
||
def test_router_id(self, driver):
|
||
driver._get = self._fake_get
|
||
assert driver.get_bgp_neighbors()["global"]["router_id"] == "1.2.3.4"
|
||
|
||
def test_peer_count(self, driver):
|
||
driver._get = self._fake_get
|
||
peers = driver.get_bgp_neighbors()["global"]["peers"]
|
||
assert len(peers) == 2
|
||
|
||
def test_established_peer_is_up(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
assert peer["is_up"] is True
|
||
|
||
def test_active_peer_is_not_up(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.2"]
|
||
assert peer["is_up"] is False
|
||
|
||
def test_admin_shutdown_peer_is_disabled(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.2"]
|
||
assert peer["is_enabled"] is False
|
||
|
||
def test_non_shutdown_peer_is_enabled(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
assert peer["is_enabled"] is True
|
||
|
||
def test_uptime_established(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
assert peer["uptime"] == 3723 # 3723000 ms → 3723 s
|
||
|
||
def test_uptime_not_established(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.2"]
|
||
assert peer["uptime"] == -1
|
||
|
||
def test_remote_as(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
assert peer["remote_as"] == 65001
|
||
|
||
def test_local_as(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
assert peer["local_as"] == 65000
|
||
|
||
def test_remote_id(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
assert peer["remote_id"] == "10.0.0.1"
|
||
|
||
def test_description(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
assert peer["description"] == "upstream-peer"
|
||
|
||
def test_ipv4_prefix_counters(self, driver):
|
||
driver._get = self._fake_get
|
||
af = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]["address_family"]["ipv4"]
|
||
assert af["sent_prefixes"] == 5
|
||
assert af["received_prefixes"] == 20
|
||
assert af["accepted_prefixes"] == 18
|
||
|
||
def test_no_af_info_falls_back_to_ipv4_minus_one(self, driver):
|
||
driver._get = self._fake_get
|
||
af = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.2"]["address_family"]
|
||
assert "ipv4" in af
|
||
assert af["ipv4"]["sent_prefixes"] == -1
|
||
|
||
def test_ipv6_af_populated_when_present(self, driver):
|
||
def fake_get(path):
|
||
if "diagnostics/bgpneighbors" in path:
|
||
return {
|
||
"response": {
|
||
"2001:db8::1": {
|
||
"remoteAs": 65010,
|
||
"localAs": 65000,
|
||
"nbrDesc": "",
|
||
"bgpState": "Established",
|
||
"bgpTimerUpMsec": 1000,
|
||
"remoteRouterId": "2001:db8::1",
|
||
"adminShutdown": False,
|
||
"addressFamilyInfo": {
|
||
"ipv6Unicast": {
|
||
"sentPrefixCounter": 3,
|
||
"prefixReceivedCount": 7,
|
||
"acceptedPrefixCounter": 7,
|
||
}
|
||
},
|
||
}
|
||
}
|
||
}
|
||
return BGP_CFG_RESPONSE
|
||
|
||
driver._get = fake_get
|
||
af = driver.get_bgp_neighbors()["global"]["peers"]["2001:db8::1"]["address_family"]
|
||
assert "ipv6" in af
|
||
assert af["ipv6"]["sent_prefixes"] == 3
|
||
|
||
def test_plugin_absent_returns_empty(self, driver):
|
||
driver._get = lambda path: (_ for _ in ()).throw(Exception("404"))
|
||
assert driver.get_bgp_neighbors() == {}
|
||
|
||
def test_frr_not_running_returns_empty(self, driver):
|
||
def fake_get(path):
|
||
if "diagnostics" in path:
|
||
return {"response": "error"} # non-dict response
|
||
return BGP_CFG_RESPONSE
|
||
|
||
driver._get = fake_get
|
||
assert driver.get_bgp_neighbors() == {}
|
||
|
||
def test_required_peer_keys(self, driver):
|
||
driver._get = self._fake_get
|
||
peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
|
||
for key in ("local_as", "remote_as", "remote_id", "is_up", "is_enabled",
|
||
"description", "uptime", "address_family"):
|
||
assert key in peer
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_certificates()
|
||
#
|
||
# Field names verified 2026-07-15 against a live OPNsense 24.7 instance's
|
||
# actual POST /api/trust/cert/search response. Confirmed: valid_from/valid_to
|
||
# are Unix timestamps as strings (not a formatted date string), issuer comes
|
||
# from the resolved "%caref" label (not the raw "caref" ref-id), and rows
|
||
# also carry crt_payload/prv_payload/csr_payload -- get_certificates() must
|
||
# never surface those (private key material).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
CERT_SEARCH_RESPONSE = {
|
||
"rows": [
|
||
{
|
||
"refid": "69d55db78963e",
|
||
"descr": "gw.home.example.com (ACME Client)",
|
||
"commonname": "gw.home.example.com",
|
||
"caref": "6a24988658ae6",
|
||
"%caref": "YR1 (ACME Client)",
|
||
"cert_type": "server_cert",
|
||
"in_use": "1",
|
||
"valid_from": "1780779726",
|
||
"valid_to": "1788555725",
|
||
"crt_payload": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----\n",
|
||
"prv_payload": "-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----\n",
|
||
},
|
||
{
|
||
"refid": "def456",
|
||
"descr": "Internal cert",
|
||
"commonname": "",
|
||
"caref": "",
|
||
"%caref": "",
|
||
"cert_type": "server_cert",
|
||
"in_use": "0",
|
||
"valid_from": "1780000000",
|
||
"valid_to": "1790000000",
|
||
},
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetCertificates:
|
||
def test_returns_one_entry_per_row(self, driver):
|
||
driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE
|
||
certs = driver.get_certificates()
|
||
assert len(certs) == 2
|
||
|
||
def test_calls_trust_cert_search(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or CERT_SEARCH_RESPONSE
|
||
driver.get_certificates()
|
||
assert calls == ["/api/trust/cert/search"]
|
||
|
||
def test_maps_expected_fields(self, driver):
|
||
driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE
|
||
cert = driver.get_certificates()[0]
|
||
assert cert["name"] == "gw.home.example.com"
|
||
assert cert["issuer"] == "YR1 (ACME Client)"
|
||
assert cert["valid_from"] == 1780779726
|
||
assert cert["valid_to"] == 1788555725
|
||
assert cert["in_use_by"] == 1
|
||
|
||
def test_falls_back_to_descr_when_commonname_empty(self, driver):
|
||
driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE
|
||
cert = driver.get_certificates()[1]
|
||
assert cert["name"] == "Internal cert"
|
||
|
||
def test_missing_caref_defaults_to_empty_string(self, driver):
|
||
driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE
|
||
cert = driver.get_certificates()[1]
|
||
assert cert["issuer"] == ""
|
||
|
||
def test_never_includes_private_key_or_cert_payload(self, driver):
|
||
driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE
|
||
for cert in driver.get_certificates():
|
||
assert "prv_payload" not in cert
|
||
assert "crt_payload" not in cert
|
||
assert "csr_payload" not in cert
|
||
|
||
def test_no_rows_returns_empty_list(self, driver):
|
||
driver._post = lambda path, data=None: {"rows": []}
|
||
assert driver.get_certificates() == []
|
||
|
||
def test_api_error_returns_empty_list(self, driver):
|
||
driver._post = lambda path, data=None: (_ for _ in ()).throw(Exception("404"))
|
||
assert driver.get_certificates() == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_ddns_status()
|
||
#
|
||
# Endpoints verified 2026-07-15 against a live OPNsense 24.7 instance: the
|
||
# service id is "ddclient" (get_services()) but the API module is "dyndns",
|
||
# not "ddclient" -- /api/ddclient/* all 404, /api/dyndns/service/status and
|
||
# /api/dyndns/settings/get are the real paths. Scoped to enabled/running
|
||
# only: no live ddclient/dyndns account was configured on the test device
|
||
# to verify a per-account "registered IP" shape against, so that comparison
|
||
# is deliberately not implemented here (would be guesswork).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DYNDNS_STATUS_RESPONSE = {"status": "running"}
|
||
DYNDNS_SETTINGS_ENABLED = {"ddclient": {"general": {"enabled": "1"}}}
|
||
DYNDNS_SETTINGS_DISABLED = {"ddclient": {"general": {"enabled": "0"}}}
|
||
|
||
|
||
class TestGetDdnsStatus:
|
||
def _fake_get(self, status_response, settings_response):
|
||
def fake_get(path):
|
||
if "service/status" in path:
|
||
return status_response
|
||
return settings_response
|
||
|
||
return fake_get
|
||
|
||
def test_calls_expected_endpoints(self, driver):
|
||
calls = []
|
||
|
||
def fake_get(path):
|
||
calls.append(path)
|
||
if "service/status" in path:
|
||
return DYNDNS_STATUS_RESPONSE
|
||
return DYNDNS_SETTINGS_ENABLED
|
||
|
||
driver._get = fake_get
|
||
driver.get_ddns_status()
|
||
assert calls == ["/api/dyndns/service/status", "/api/dyndns/settings/get"]
|
||
|
||
def test_enabled_and_running(self, driver):
|
||
driver._get = self._fake_get(DYNDNS_STATUS_RESPONSE, DYNDNS_SETTINGS_ENABLED)
|
||
result = driver.get_ddns_status()
|
||
assert result == {"enabled": True, "running": True}
|
||
|
||
def test_enabled_but_not_running(self, driver):
|
||
driver._get = self._fake_get({"status": "stopped"}, DYNDNS_SETTINGS_ENABLED)
|
||
result = driver.get_ddns_status()
|
||
assert result == {"enabled": True, "running": False}
|
||
|
||
def test_disabled(self, driver):
|
||
driver._get = self._fake_get(DYNDNS_STATUS_RESPONSE, DYNDNS_SETTINGS_DISABLED)
|
||
result = driver.get_ddns_status()
|
||
assert result == {"enabled": False, "running": True}
|
||
|
||
def test_api_error_returns_none(self, driver):
|
||
driver._get = lambda path: (_ for _ in ()).throw(Exception("404"))
|
||
assert driver.get_ddns_status() is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# FreeRADIUS: get/create/delete radius clients and users
|
||
#
|
||
# Endpoints + field names verified 2026-07-15 against a live OPNsense 24.7
|
||
# instance via a real add -> search/get -> set -> del round trip (cleaned up
|
||
# immediately after). Key findings: the module is "freeradius" (matches the
|
||
# service id), the CRUD actions use a "search_<name>"/"add_<name>"/
|
||
# "set_<name>/<uuid>"/"del_<name>/<uuid>" suffix convention (not the plain
|
||
# "search"/"add" used by e.g. get_certificates' Trust module), the client's
|
||
# IP field is literally named "ip" (CIDR string, e.g. "203.0.113.1/32") --
|
||
# NOT "ipaddr" as first guessed (silently accepted-but-ignored by add_client
|
||
# rather than erroring). A config change only takes effect after
|
||
# POST /api/freeradius/service/reconfigure.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
RADIUS_CLIENTS_RESPONSE = {
|
||
"rows": [
|
||
{"uuid": "e5483f1b-936b-47ba-8cca-24d56ad643c6", "enabled": "1",
|
||
"name": "ap-lobby", "secret": "s3cr3t", "ip": "10.0.0.5/32"},
|
||
]
|
||
}
|
||
|
||
RADIUS_USERS_RESPONSE = {
|
||
"rows": [
|
||
{"uuid": "71da23fd-03f3-4e1f-a7e1-bb25645b981e", "enabled": "1",
|
||
"username": "jdoe", "password": "hunter2"},
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetRadiusClients:
|
||
def test_returns_one_entry_per_row(self, driver):
|
||
driver._get = lambda path: RADIUS_CLIENTS_RESPONSE
|
||
assert len(driver.get_radius_clients()) == 1
|
||
|
||
def test_maps_expected_fields(self, driver):
|
||
driver._get = lambda path: RADIUS_CLIENTS_RESPONSE
|
||
client = driver.get_radius_clients()[0]
|
||
assert client["id"] == "e5483f1b-936b-47ba-8cca-24d56ad643c6"
|
||
assert client["name"] == "ap-lobby"
|
||
assert client["ip"] == "10.0.0.5/32"
|
||
assert client["enabled"] is True
|
||
|
||
def test_calls_search_client(self, driver):
|
||
calls = []
|
||
driver._get = lambda path: calls.append(path) or RADIUS_CLIENTS_RESPONSE
|
||
driver.get_radius_clients()
|
||
assert calls == ["/api/freeradius/client/search_client"]
|
||
|
||
def test_api_error_returns_empty_list(self, driver):
|
||
driver._get = lambda path: (_ for _ in ()).throw(Exception("404"))
|
||
assert driver.get_radius_clients() == []
|
||
|
||
|
||
class TestCreateRadiusClient:
|
||
def test_success_returns_remote_id(self, driver):
|
||
driver._post = lambda path, data=None: {"result": "saved"}
|
||
driver._get = lambda path: RADIUS_CLIENTS_RESPONSE
|
||
result = driver.create_radius_client("ap-lobby", "10.0.0.5/32", "s3cr3t")
|
||
assert result == {"success": True, "id": "e5483f1b-936b-47ba-8cca-24d56ad643c6"}
|
||
|
||
def test_success_calls_add_then_reconfigure(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
driver._get = lambda path: RADIUS_CLIENTS_RESPONSE
|
||
driver.create_radius_client("ap-lobby", "10.0.0.5/32", "s3cr3t")
|
||
assert calls[0] == (
|
||
"/api/freeradius/client/add_client",
|
||
{"client": {"name": "ap-lobby", "ip": "10.0.0.5/32", "secret": "s3cr3t"}},
|
||
)
|
||
assert calls[1] == ("/api/freeradius/service/reconfigure", None)
|
||
|
||
def test_validation_failure_does_not_reconfigure(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {
|
||
"result": "failed",
|
||
"validations": {"client.name": "A value is required."},
|
||
}
|
||
result = driver.create_radius_client("", "10.0.0.5/32", "s3cr3t")
|
||
assert result["success"] is False
|
||
assert result["validations"] == {"client.name": "A value is required."}
|
||
assert calls == ["/api/freeradius/client/add_client"]
|
||
|
||
def test_created_entry_not_found_after_add_returns_no_id(self, driver):
|
||
driver._post = lambda path, data=None: {"result": "saved"}
|
||
driver._get = lambda path: {"rows": []}
|
||
result = driver.create_radius_client("ap-lobby", "10.0.0.5/32", "s3cr3t")
|
||
assert result == {"success": True, "id": None}
|
||
|
||
|
||
class TestDeleteRadiusClient:
|
||
def test_success_reconfigures_service(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {"result": "deleted"}
|
||
result = driver.delete_radius_client("e5483f1b-936b-47ba-8cca-24d56ad643c6")
|
||
assert result == {"success": True}
|
||
assert calls == [
|
||
"/api/freeradius/client/del_client/e5483f1b-936b-47ba-8cca-24d56ad643c6",
|
||
"/api/freeradius/service/reconfigure",
|
||
]
|
||
|
||
def test_not_found_does_not_reconfigure(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {"result": "failed"}
|
||
result = driver.delete_radius_client("does-not-exist")
|
||
assert result == {"success": False}
|
||
assert calls == ["/api/freeradius/client/del_client/does-not-exist"]
|
||
|
||
|
||
class TestGetRadiusUsers:
|
||
def test_returns_one_entry_per_row(self, driver):
|
||
driver._get = lambda path: RADIUS_USERS_RESPONSE
|
||
assert len(driver.get_radius_users()) == 1
|
||
|
||
def test_maps_expected_fields(self, driver):
|
||
driver._get = lambda path: RADIUS_USERS_RESPONSE
|
||
user = driver.get_radius_users()[0]
|
||
assert user["id"] == "71da23fd-03f3-4e1f-a7e1-bb25645b981e"
|
||
assert user["username"] == "jdoe"
|
||
assert user["enabled"] is True
|
||
|
||
def test_never_includes_password(self, driver):
|
||
driver._get = lambda path: RADIUS_USERS_RESPONSE
|
||
for user in driver.get_radius_users():
|
||
assert "password" not in user
|
||
|
||
def test_calls_search_user(self, driver):
|
||
calls = []
|
||
driver._get = lambda path: calls.append(path) or RADIUS_USERS_RESPONSE
|
||
driver.get_radius_users()
|
||
assert calls == ["/api/freeradius/user/search_user"]
|
||
|
||
def test_api_error_returns_empty_list(self, driver):
|
||
driver._get = lambda path: (_ for _ in ()).throw(Exception("404"))
|
||
assert driver.get_radius_users() == []
|
||
|
||
|
||
class TestCreateRadiusUser:
|
||
def test_success_returns_remote_id(self, driver):
|
||
driver._post = lambda path, data=None: {"result": "saved"}
|
||
driver._get = lambda path: RADIUS_USERS_RESPONSE
|
||
result = driver.create_radius_user("jdoe", "hunter2")
|
||
assert result == {"success": True, "id": "71da23fd-03f3-4e1f-a7e1-bb25645b981e"}
|
||
|
||
def test_success_calls_add_then_reconfigure(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
driver._get = lambda path: RADIUS_USERS_RESPONSE
|
||
driver.create_radius_user("jdoe", "hunter2")
|
||
assert calls[0] == (
|
||
"/api/freeradius/user/add_user",
|
||
{"user": {"username": "jdoe", "password": "hunter2"}},
|
||
)
|
||
assert calls[1] == ("/api/freeradius/service/reconfigure", None)
|
||
|
||
def test_validation_failure_does_not_reconfigure(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {
|
||
"result": "failed",
|
||
"validations": {"user.username": "A value is required."},
|
||
}
|
||
result = driver.create_radius_user("", "hunter2")
|
||
assert result["success"] is False
|
||
assert calls == ["/api/freeradius/user/add_user"]
|
||
|
||
def test_created_entry_not_found_after_add_returns_no_id(self, driver):
|
||
driver._post = lambda path, data=None: {"result": "saved"}
|
||
driver._get = lambda path: {"rows": []}
|
||
result = driver.create_radius_user("jdoe", "hunter2")
|
||
assert result == {"success": True, "id": None}
|
||
|
||
|
||
class TestDeleteRadiusUser:
|
||
def test_success_reconfigures_service(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {"result": "deleted"}
|
||
result = driver.delete_radius_user("71da23fd-03f3-4e1f-a7e1-bb25645b981e")
|
||
assert result == {"success": True}
|
||
assert calls == [
|
||
"/api/freeradius/user/del_user/71da23fd-03f3-4e1f-a7e1-bb25645b981e",
|
||
"/api/freeradius/service/reconfigure",
|
||
]
|
||
|
||
def test_not_found_does_not_reconfigure(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {"result": "failed"}
|
||
result = driver.delete_radius_user("does-not-exist")
|
||
assert result == {"success": False}
|
||
assert calls == ["/api/freeradius/user/del_user/does-not-exist"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _post()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestInternalPost:
|
||
def test_raises_when_no_session(self):
|
||
drv = OPNsenseDriver("host", "u", "p")
|
||
with pytest.raises(ConnectionClosedException):
|
||
drv._post("/api/routes/routes/reconfigure")
|
||
|
||
def test_calls_correct_url(self, driver):
|
||
driver.session.post.return_value = _make_json_response({"result": "ok"})
|
||
driver._post("/api/routes/routes/reconfigure")
|
||
driver.session.post.assert_called_once_with(
|
||
"https://opnsense.example.com/api/routes/routes/reconfigure",
|
||
json={},
|
||
timeout=60,
|
||
)
|
||
|
||
def test_sends_json_payload(self, driver):
|
||
driver.session.post.return_value = _make_json_response({"uuid": "abc-123"})
|
||
payload = {"route": {"network": "10.0.0.0/8", "gateway": "WAN_GW"}}
|
||
driver._post("/api/routes/routes/addroute", payload)
|
||
driver.session.post.assert_called_once_with(
|
||
"https://opnsense.example.com/api/routes/routes/addroute",
|
||
json=payload,
|
||
timeout=60,
|
||
)
|
||
|
||
def test_returns_parsed_json(self, driver):
|
||
driver.session.post.return_value = _make_json_response({"result": "saved"})
|
||
result = driver._post("/api/routes/routes/reconfigure")
|
||
assert result == {"result": "saved"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# send_wake_on_lan()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestSendWakeOnLan:
|
||
def test_raises_value_error_without_interface(self, driver):
|
||
with pytest.raises(ValueError):
|
||
driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF")
|
||
driver.session.post.assert_not_called()
|
||
|
||
def test_sends_correct_request(self, driver):
|
||
driver.session.post.return_value = _make_json_response({"status": ""})
|
||
driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||
driver.session.post.assert_called_once_with(
|
||
"https://opnsense.example.com/api/wol/wol/set",
|
||
json={"wake": {"interface": "lan", "mac": "AA:BB:CC:DD:EE:FF"}},
|
||
timeout=60,
|
||
)
|
||
|
||
def test_success_on_empty_status(self, driver):
|
||
# OPNsense's WolController returns {"status": trim(configd output)} on
|
||
# success — the underlying `wol` CLI tool typically prints nothing.
|
||
driver.session.post.return_value = _make_json_response({"status": ""})
|
||
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||
assert result == {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"}
|
||
|
||
def test_returns_failure_on_error_status(self, driver):
|
||
driver.session.post.return_value = _make_json_response(
|
||
{"status": "error", "error_msg": "Incorrect IPv4 configuration on interface"}
|
||
)
|
||
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="opt1")
|
||
assert result == {"success": False, "output": "Incorrect IPv4 configuration on interface"}
|
||
|
||
def test_returns_failure_on_empty_response(self, driver):
|
||
# OPNsense's model validation (bad MAC/interface) silently returns {}
|
||
# with HTTP 200 rather than an error status.
|
||
driver.session.post.return_value = _make_json_response({})
|
||
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||
assert result["success"] is False
|
||
|
||
def test_returns_failure_on_request_exception(self, driver):
|
||
# e.g. HTTP 404 — the os-wol plugin is not installed on this firewall.
|
||
driver.session.post.side_effect = Exception("404 Client Error: Not Found")
|
||
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||
assert result["success"] is False
|
||
assert "404" in result["output"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# load_merge_candidate()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
VALID_ROUTES_CONFIG = json.dumps([
|
||
{"network": "10.0.0.0/8", "gateway": "WAN_GW", "descr": "internal"},
|
||
{"network": "0.0.0.0/0", "gateway": "WAN_GW", "descr": "default"},
|
||
])
|
||
|
||
|
||
class TestLoadMergeCandidate:
|
||
def test_accepts_valid_json_string(self, driver):
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
assert driver._candidate_config is not None
|
||
assert len(driver._candidate_config) == 2
|
||
|
||
def test_parses_required_keys(self, driver):
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
first = driver._candidate_config[0]
|
||
assert first["network"] == "10.0.0.0/8"
|
||
assert first["gateway"] == "WAN_GW"
|
||
|
||
def test_reads_from_file(self, driver, tmp_path):
|
||
cfg_file = tmp_path / "routes.json"
|
||
cfg_file.write_text(VALID_ROUTES_CONFIG)
|
||
driver.load_merge_candidate(filename=str(cfg_file))
|
||
assert len(driver._candidate_config) == 2
|
||
|
||
def test_raises_if_both_args_given(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
with pytest.raises(MergeConfigException):
|
||
driver.load_merge_candidate(filename="f.json", config="{}")
|
||
|
||
def test_raises_if_no_args_given(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
with pytest.raises(MergeConfigException):
|
||
driver.load_merge_candidate()
|
||
|
||
def test_raises_on_invalid_json(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
with pytest.raises(MergeConfigException, match="Invalid JSON"):
|
||
driver.load_merge_candidate(config="not json {{{")
|
||
|
||
def test_raises_if_not_a_list(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
with pytest.raises(MergeConfigException, match="array"):
|
||
driver.load_merge_candidate(config='{"network": "1.0.0.0/8"}')
|
||
|
||
def test_raises_if_route_missing_network(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
bad = json.dumps([{"gateway": "GW1"}])
|
||
with pytest.raises(MergeConfigException, match="'network'"):
|
||
driver.load_merge_candidate(config=bad)
|
||
|
||
def test_raises_if_route_missing_gateway(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
bad = json.dumps([{"network": "10.0.0.0/8"}])
|
||
with pytest.raises(MergeConfigException, match="'gateway'"):
|
||
driver.load_merge_candidate(config=bad)
|
||
|
||
def test_raises_on_missing_file(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
with pytest.raises(MergeConfigException, match="Cannot read"):
|
||
driver.load_merge_candidate(filename="/nonexistent/path/routes.json")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# compare_config()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
SEARCH_ROUTE_RESPONSE = {
|
||
"rows": [
|
||
{"network": "192.168.1.0/24", "gateway": "LAN_GW", "descr": "lan", "disabled": "0"},
|
||
]
|
||
}
|
||
|
||
|
||
class TestCompareConfig:
|
||
def test_returns_empty_string_without_candidate(self, driver):
|
||
assert driver.compare_config() == ""
|
||
|
||
def test_returns_diff_string(self, driver):
|
||
driver._get = lambda path: SEARCH_ROUTE_RESPONSE
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
diff = driver.compare_config()
|
||
assert "---" in diff
|
||
assert "+++" in diff
|
||
|
||
def test_diff_shows_added_routes(self, driver):
|
||
driver._get = lambda path: SEARCH_ROUTE_RESPONSE
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
diff = driver.compare_config()
|
||
assert "WAN_GW" in diff
|
||
|
||
def test_empty_diff_when_config_matches(self, driver):
|
||
same_config = json.dumps([
|
||
{"network": "192.168.1.0/24", "gateway": "LAN_GW", "descr": "lan", "disabled": "0"}
|
||
])
|
||
driver._get = lambda path: SEARCH_ROUTE_RESPONSE
|
||
driver.load_merge_candidate(config=same_config)
|
||
diff = driver.compare_config()
|
||
assert diff == ""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# commit_config()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BACKUPS_RESPONSE = {
|
||
"items": [
|
||
{"id": "config-opnsense01-1234567890.xml", "time": "1234567890", "description": "before change"},
|
||
{"id": "config-opnsense01-1234567800.xml", "time": "1234567800", "description": "initial"},
|
||
]
|
||
}
|
||
|
||
|
||
class TestCommitConfig:
|
||
def test_raises_without_candidate(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
with pytest.raises(MergeConfigException, match="No candidate"):
|
||
driver.commit_config()
|
||
|
||
def test_posts_each_route_and_reconfigure(self, driver):
|
||
# _get for backup list + POST for 2 routes + POST reconfigure
|
||
driver._get = lambda path: BACKUPS_RESPONSE
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
driver.session.post.return_value = _make_json_response({})
|
||
driver.commit_config()
|
||
assert driver.session.post.call_count == 3 # addroute×2 + reconfigure
|
||
|
||
def test_records_pre_commit_backup_id(self, driver):
|
||
driver._get = lambda path: BACKUPS_RESPONSE
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
driver.session.post.return_value = _make_json_response({})
|
||
driver.commit_config()
|
||
assert driver._pre_commit_backup_id == "config-opnsense01-1234567890.xml"
|
||
|
||
def test_clears_candidate_after_commit(self, driver):
|
||
driver._get = lambda path: BACKUPS_RESPONSE
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
driver.session.post.return_value = _make_json_response({})
|
||
driver.commit_config()
|
||
assert driver._candidate_config is None
|
||
|
||
def test_records_none_backup_when_no_backups_exist(self, driver):
|
||
driver._get = lambda path: {"items": []}
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
driver.session.post.return_value = _make_json_response({})
|
||
driver.commit_config()
|
||
assert driver._pre_commit_backup_id is None
|
||
|
||
def test_raises_on_api_error(self, driver):
|
||
from napalm.base.exceptions import MergeConfigException
|
||
import requests as _req
|
||
driver._get = lambda path: BACKUPS_RESPONSE
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
driver.session.post.side_effect = _req.exceptions.RequestException("timeout")
|
||
with pytest.raises(MergeConfigException, match="Failed to apply"):
|
||
driver.commit_config()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# discard_config()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestDiscardConfig:
|
||
def test_clears_candidate(self, driver):
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
driver.discard_config()
|
||
assert driver._candidate_config is None
|
||
|
||
def test_idempotent_when_no_candidate(self, driver):
|
||
driver.discard_config() # must not raise
|
||
assert driver._candidate_config is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# rollback()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestRollback:
|
||
def test_noop_when_no_backups_and_no_commit(self, driver):
|
||
driver._get = lambda path: {"items": []}
|
||
driver.rollback() # must not raise
|
||
driver.session.post.assert_not_called()
|
||
|
||
def test_uses_pre_commit_backup_id(self, driver):
|
||
driver._pre_commit_backup_id = "config-opnsense01-1234567890.xml"
|
||
driver.session.post.return_value = _make_json_response({"status": "ok"})
|
||
driver.rollback()
|
||
url = driver.session.post.call_args[0][0]
|
||
assert "config-opnsense01-1234567890.xml" in url
|
||
assert "revert_backup" in url
|
||
|
||
def test_falls_back_to_latest_backup_without_commit(self, driver):
|
||
driver._get = lambda path: BACKUPS_RESPONSE
|
||
driver.session.post.return_value = _make_json_response({"status": "ok"})
|
||
driver.rollback()
|
||
url = driver.session.post.call_args[0][0]
|
||
assert "config-opnsense01-1234567890.xml" in url
|
||
|
||
def test_clears_pre_commit_backup_id_after_rollback(self, driver):
|
||
driver._pre_commit_backup_id = "config-opnsense01-1234567890.xml"
|
||
driver.session.post.return_value = _make_json_response({})
|
||
driver.rollback()
|
||
assert driver._pre_commit_backup_id is None
|
||
|
||
def test_calls_revert_backup_exactly_once(self, driver):
|
||
driver._pre_commit_backup_id = "config-opnsense01-1234567890.xml"
|
||
driver.session.post.return_value = _make_json_response({})
|
||
driver.rollback()
|
||
assert driver.session.post.call_count == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_config() — candidate slot
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGetConfigCandidate:
|
||
def test_candidate_empty_without_staged_config(self, driver):
|
||
driver._get = lambda path: "<opnsense/>"
|
||
result = driver.get_config()
|
||
assert result["candidate"] == ""
|
||
|
||
def test_candidate_contains_staged_routes(self, driver):
|
||
driver._get = lambda path: "<opnsense/>"
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
result = driver.get_config()
|
||
assert "WAN_GW" in result["candidate"]
|
||
|
||
def test_candidate_is_valid_json(self, driver):
|
||
driver._get = lambda path: "<opnsense/>"
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
result = driver.get_config()
|
||
parsed = json.loads(result["candidate"])
|
||
assert isinstance(parsed, list)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# create_dhcp_reservation()
|
||
#
|
||
# Payload/response shapes below are taken verbatim from a live probe against
|
||
# a real OPNsense box running the Kea DHCPv4 (os-kea) plugin — searchSubnet,
|
||
# searchReservation, addReservation, delReservation, and service/reconfigure
|
||
# were all exercised live (including a real add + verify + delete cycle) to
|
||
# confirm the exact request/response schema before writing this driver
|
||
# method and these tests against it.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
KEA_SUBNETS_RESPONSE = {
|
||
"rows": [
|
||
{"uuid": "82766878-c5ac-41f3-b7b0-e24d2419beb3", "subnet": "172.22.0.0/24"},
|
||
{"uuid": "6854cab3-ebb5-4031-b987-0edcc6723546", "subnet": "172.22.8.0/24"},
|
||
]
|
||
}
|
||
|
||
|
||
class TestCreateDhcpReservation:
|
||
def test_adds_new_reservation_when_none_exists(self, driver):
|
||
driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE)
|
||
driver.session.post.side_effect = [
|
||
_make_json_response({"rows": []}), # searchReservation — no match
|
||
_make_json_response({"result": "saved", "uuid": "new-uuid-123"}), # addReservation
|
||
_make_json_response({"status": "ok"}), # service/reconfigure
|
||
]
|
||
|
||
driver.create_dhcp_reservation(
|
||
mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253", hostname="new-vm"
|
||
)
|
||
|
||
add_call = driver.session.post.call_args_list[1]
|
||
assert add_call.args[0] == "https://opnsense.example.com/api/kea/dhcpv4/addReservation"
|
||
payload = add_call.kwargs["json"]["reservation"]
|
||
assert payload["subnet"] == "6854cab3-ebb5-4031-b987-0edcc6723546"
|
||
assert payload["ip_address"] == "172.22.8.253"
|
||
assert payload["hw_address"] == "02:aa:bb:cc:dd:ee"
|
||
assert payload["hostname"] == "new-vm"
|
||
assert payload["description"] == "[netork]"
|
||
|
||
reconfigure_call = driver.session.post.call_args_list[2]
|
||
assert reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure"
|
||
|
||
def test_updates_existing_reservation_for_same_ip(self, driver):
|
||
driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE)
|
||
driver.session.post.side_effect = [
|
||
_make_json_response(
|
||
{"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]}
|
||
),
|
||
_make_json_response({"result": "saved", "uuid": "existing-uuid-456"}),
|
||
_make_json_response({"status": "ok"}),
|
||
]
|
||
|
||
driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")
|
||
|
||
set_call = driver.session.post.call_args_list[1]
|
||
assert (
|
||
set_call.args[0]
|
||
== "https://opnsense.example.com/api/kea/dhcpv4/setReservation/existing-uuid-456"
|
||
)
|
||
|
||
def test_raises_when_ip_not_in_any_kea_subnet(self, driver):
|
||
driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE)
|
||
|
||
with pytest.raises(ValueError, match="No Kea-managed subnet"):
|
||
driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="10.99.99.99")
|
||
|
||
driver.session.post.assert_not_called()
|
||
|
||
def test_raises_when_kea_rejects_reservation(self, driver):
|
||
driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE)
|
||
driver.session.post.side_effect = [
|
||
_make_json_response({"rows": []}),
|
||
_make_json_response(
|
||
{
|
||
"result": "failed",
|
||
"validations": {"reservation.ip_address": "Address not in specified subnet"},
|
||
}
|
||
),
|
||
]
|
||
|
||
with pytest.raises(RuntimeError, match="Kea rejected"):
|
||
driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")
|
||
|
||
# reconfigure must NOT be called after a rejected reservation
|
||
assert driver.session.post.call_count == 2
|
||
|
||
def test_raises_when_kea_plugin_unavailable(self, driver):
|
||
driver.session.get.side_effect = Exception("404 Not Found")
|
||
|
||
with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"):
|
||
driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")
|
||
|
||
driver.session.post.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# delete_dhcp_reservation_and_lease()
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDeleteDhcpReservationAndLease:
|
||
def test_deletes_existing_reservation_and_lease(self, driver):
|
||
driver.session.post.side_effect = [
|
||
_make_json_response(
|
||
{"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]}
|
||
), # searchReservation — match
|
||
_make_json_response({"result": "deleted"}), # delReservation
|
||
_make_json_response({"status": "ok"}), # service/reconfigure
|
||
_make_json_response({"result": "deleted"}), # delLease
|
||
]
|
||
driver.session.get.return_value = _make_json_response(
|
||
{"rows": [{"address": "172.22.8.253", "hwaddr": "02:aa:bb:cc:dd:ee"}]}
|
||
)
|
||
|
||
result = driver.delete_dhcp_reservation_and_lease(
|
||
mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253"
|
||
)
|
||
|
||
assert result == {
|
||
"reservation_found": True,
|
||
"reservation_deleted": True,
|
||
"lease_found": True,
|
||
"lease_deleted": True,
|
||
}
|
||
del_res_call = driver.session.post.call_args_list[1]
|
||
assert (
|
||
del_res_call.args[0]
|
||
== "https://opnsense.example.com/api/kea/dhcpv4/delReservation/existing-uuid-456"
|
||
)
|
||
reconfigure_call = driver.session.post.call_args_list[2]
|
||
assert reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure"
|
||
del_lease_call = driver.session.post.call_args_list[3]
|
||
# IP is a URL path segment, not a JSON body field — verified live
|
||
# against a real OPNsense instance; a {"ips": [ip]} body (the
|
||
# natural reading of the documented del_lease($ips=null) signature)
|
||
# returns {"status": "error", "message": "Missing lease IP parameter"}.
|
||
assert (
|
||
del_lease_call.args[0]
|
||
== "https://opnsense.example.com/api/kea/leases4/del_lease/172.22.8.253"
|
||
)
|
||
|
||
def test_noop_when_no_reservation_and_no_lease_found(self, driver):
|
||
driver.session.post.return_value = _make_json_response({"rows": []})
|
||
driver.session.get.return_value = _make_json_response({"rows": []})
|
||
|
||
result = driver.delete_dhcp_reservation_and_lease(
|
||
mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253"
|
||
)
|
||
|
||
assert result == {
|
||
"reservation_found": False,
|
||
"reservation_deleted": False,
|
||
"lease_found": False,
|
||
"lease_deleted": False,
|
||
}
|
||
# Only searchReservation was called — no delReservation/reconfigure/delLease.
|
||
assert driver.session.post.call_count == 1
|
||
|
||
def test_reservation_delete_raises_when_kea_rejects(self, driver):
|
||
driver.session.post.side_effect = [
|
||
_make_json_response(
|
||
{"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]}
|
||
),
|
||
_make_json_response({"result": "not found"}), # delReservation rejected
|
||
]
|
||
|
||
with pytest.raises(RuntimeError, match="Kea rejected reservation delete"):
|
||
driver.delete_dhcp_reservation_and_lease(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")
|
||
|
||
# reconfigure must NOT be called after a rejected delete
|
||
assert driver.session.post.call_count == 2
|
||
|
||
def test_lease_delete_failure_is_non_fatal(self, driver):
|
||
driver.session.post.side_effect = [
|
||
_make_json_response(
|
||
{"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]}
|
||
),
|
||
_make_json_response({"result": "deleted"}),
|
||
_make_json_response({"status": "ok"}),
|
||
]
|
||
driver.session.get.side_effect = Exception("connection reset")
|
||
|
||
result = driver.delete_dhcp_reservation_and_lease(
|
||
mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253"
|
||
)
|
||
|
||
assert result["reservation_found"] is True
|
||
assert result["reservation_deleted"] is True
|
||
assert result["lease_found"] is False
|
||
assert result["lease_deleted"] is False
|
||
|
||
def test_raises_when_kea_plugin_unavailable(self, driver):
|
||
driver.session.post.side_effect = Exception("404 Not Found")
|
||
|
||
with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"):
|
||
driver.delete_dhcp_reservation_and_lease(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# apply_firewall_rule / commit_firewall_rules
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestApplyFirewallRule:
|
||
def test_add_calls_addRule_with_translated_fields(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
rule = {
|
||
"description": "allow_mgmt_to_fw_gui",
|
||
"action": "pass",
|
||
"interface": "lan",
|
||
"direction": "in",
|
||
"protocol": "tcp",
|
||
"source_net": "MGMT_NET",
|
||
"source_port": "",
|
||
"destination_net": "(self)",
|
||
"destination_port": "https",
|
||
"enabled": True,
|
||
"quick": True,
|
||
"log": False,
|
||
}
|
||
driver.apply_firewall_rule(rule)
|
||
|
||
assert len(calls) == 1
|
||
path, payload = calls[0]
|
||
assert path == "/api/firewall/filter/addRule"
|
||
assert payload["rule"]["enabled"] == "1"
|
||
assert payload["rule"]["quick"] == "1"
|
||
assert payload["rule"]["log"] == "0"
|
||
assert payload["rule"]["action"] == "pass"
|
||
assert payload["rule"]["interface"] == "lan"
|
||
assert payload["rule"]["source_net"] == "MGMT_NET"
|
||
assert payload["rule"]["destination_net"] == "(self)"
|
||
assert payload["rule"]["destination_port"] == "https"
|
||
assert payload["rule"]["descr"] == "allow_mgmt_to_fw_gui"
|
||
assert payload["rule"]["floating"] == "no"
|
||
|
||
def test_update_calls_setRule_with_uuid(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_firewall_rule({"description": "x", "action": "pass"}, uuid="abc-123")
|
||
|
||
assert calls[0][0] == "/api/firewall/filter/setRule/abc-123"
|
||
|
||
def test_empty_interface_is_a_floating_rule(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_firewall_rule({"description": "x", "action": "pass", "interface": ""})
|
||
|
||
assert calls[0][1]["rule"]["floating"] == "yes"
|
||
|
||
def test_disabled_quick_log_flags_translate_to_zero(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_firewall_rule(
|
||
{
|
||
"description": "x",
|
||
"action": "block",
|
||
"enabled": False,
|
||
"quick": False,
|
||
"log": True,
|
||
}
|
||
)
|
||
|
||
rule = calls[0][1]["rule"]
|
||
assert rule["enabled"] == "0"
|
||
assert rule["quick"] == "0"
|
||
assert rule["log"] == "1"
|
||
|
||
|
||
class TestCommitFirewallRules:
|
||
def test_calls_filter_apply(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"status": "ok"}
|
||
|
||
result = driver.commit_firewall_rules()
|
||
|
||
assert calls == [("/api/firewall/filter/apply", {})]
|
||
assert result == {"status": "ok"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DHCP reservations (Kea) — the vendor-neutral DhcpServerMixin contract
|
||
# ---------------------------------------------------------------------------
|
||
|
||
KEA_RESV_SUBNETS_RESPONSE = {
|
||
"rows": [
|
||
{"uuid": "sub-mgmt", "subnet": "10.10.20.0/24"},
|
||
{"uuid": "sub-users", "subnet": "10.30.20.0/24"},
|
||
]
|
||
}
|
||
|
||
KEA_RESV_ROWS_RESPONSE = {
|
||
"rows": [
|
||
{
|
||
"uuid": "res-1",
|
||
"subnet": "10.10.20.0/24",
|
||
"ip_address": "10.10.20.50",
|
||
"hw_address": "aa:bb:cc:dd:ee:01",
|
||
"hostname": "nas",
|
||
"description": "Home NAS",
|
||
},
|
||
{
|
||
"uuid": "res-2",
|
||
"subnet": "10.30.20.0/24",
|
||
"ip_address": "10.30.20.80",
|
||
"hw_address": "AA:BB:CC:DD:EE:02",
|
||
"hostname": "",
|
||
"description": "",
|
||
},
|
||
]
|
||
}
|
||
|
||
|
||
class TestGetDhcpReservations:
|
||
def test_maps_kea_rows_to_vendor_neutral_dicts(self, driver):
|
||
driver._post = lambda path, data=None: KEA_RESV_ROWS_RESPONSE
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
result = driver.get_dhcp_reservations()
|
||
|
||
assert result[0] == {
|
||
"uuid": "res-1",
|
||
"mac": "aa:bb:cc:dd:ee:01",
|
||
"ip": "10.10.20.50",
|
||
"hostname": "nas",
|
||
"description": "Home NAS",
|
||
"subnet": "10.10.20.0/24",
|
||
}
|
||
assert result[1]["hostname"] == ""
|
||
assert result[1]["description"] == ""
|
||
|
||
def test_requests_all_rows_not_just_first_page(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or KEA_RESV_ROWS_RESPONSE
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
driver.get_dhcp_reservations()
|
||
|
||
assert calls[0][0] == "/api/kea/dhcpv4/searchReservation"
|
||
assert calls[0][1]["rowCount"] == -1
|
||
|
||
def test_resolves_subnet_uuid_to_cidr(self, driver):
|
||
# Depending on version, searchReservation returns the relation's UUID
|
||
# rather than its display value — the caller must still get a CIDR.
|
||
driver._post = lambda path, data=None: {
|
||
"rows": [
|
||
{
|
||
"uuid": "res-1",
|
||
"subnet": "sub-users",
|
||
"ip_address": "10.30.20.50",
|
||
"hw_address": "aa:bb:cc:dd:ee:03",
|
||
"hostname": "printer",
|
||
"description": "",
|
||
}
|
||
]
|
||
}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
result = driver.get_dhcp_reservations()
|
||
|
||
assert result[0]["subnet"] == "10.30.20.0/24"
|
||
|
||
def test_unresolvable_subnet_degrades_to_empty_string(self, driver):
|
||
driver._post = lambda path, data=None: {
|
||
"rows": [
|
||
{
|
||
"uuid": "res-1",
|
||
"subnet": "sub-gone",
|
||
"ip_address": "10.99.0.5",
|
||
"hw_address": "aa:bb:cc:dd:ee:04",
|
||
"hostname": "",
|
||
"description": "",
|
||
}
|
||
]
|
||
}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
assert driver.get_dhcp_reservations()[0]["subnet"] == ""
|
||
|
||
def test_empty_device_returns_empty_list(self, driver):
|
||
driver._post = lambda path, data=None: {"rows": []}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
assert driver.get_dhcp_reservations() == []
|
||
|
||
def test_kea_plugin_unavailable_raises_runtime_error(self, driver):
|
||
def _boom(path, data=None):
|
||
raise RuntimeError("404 Not Found")
|
||
|
||
driver._post = _boom
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"):
|
||
driver.get_dhcp_reservations()
|
||
|
||
|
||
class TestApplyDhcpReservation:
|
||
def _reservation(self, **overrides):
|
||
base = {
|
||
"uuid": "",
|
||
"mac": "aa:bb:cc:dd:ee:01",
|
||
"ip": "10.10.20.50",
|
||
"hostname": "nas",
|
||
"description": "Home NAS",
|
||
"subnet": "10.10.20.0/24",
|
||
}
|
||
base.update(overrides)
|
||
return base
|
||
|
||
def test_add_posts_to_add_reservation(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
result = driver.apply_dhcp_reservation(self._reservation())
|
||
|
||
assert result == {"success": True}
|
||
assert calls[0][0] == "/api/kea/dhcpv4/addReservation"
|
||
assert calls[0][1] == {
|
||
"reservation": {
|
||
"subnet": "sub-mgmt",
|
||
"ip_address": "10.10.20.50",
|
||
"hw_address": "aa:bb:cc:dd:ee:01",
|
||
"hostname": "nas",
|
||
"description": "Home NAS",
|
||
}
|
||
}
|
||
|
||
def test_update_posts_to_set_reservation_with_uuid(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {"result": "saved"}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
driver.apply_dhcp_reservation(self._reservation(), uuid="res-1")
|
||
|
||
assert calls == ["/api/kea/dhcpv4/setReservation/res-1"]
|
||
|
||
def test_does_not_reconfigure_on_every_write(self, driver):
|
||
# Committing is a separate step — one reconfigure per ruleset, not per
|
||
# reservation.
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {"result": "saved"}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
driver.apply_dhcp_reservation(self._reservation())
|
||
|
||
assert not any("reconfigure" in path for path in calls)
|
||
|
||
def test_unknown_subnet_raises_value_error(self, driver):
|
||
driver._post = lambda path, data=None: {"result": "saved"}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
with pytest.raises(ValueError, match="No Kea-managed subnet"):
|
||
driver.apply_dhcp_reservation(self._reservation(subnet="192.168.99.0/24"))
|
||
|
||
def test_falls_back_to_containing_subnet_when_cidr_absent(self, driver):
|
||
# A caller that only knows the IP (no subnet field) should still work:
|
||
# Kea's own subnet list decides which subnet the address belongs to.
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
driver.apply_dhcp_reservation(self._reservation(subnet=""))
|
||
|
||
assert calls[0][1]["reservation"]["subnet"] == "sub-mgmt"
|
||
|
||
def test_rejected_write_raises_runtime_error(self, driver):
|
||
driver._post = lambda path, data=None: {
|
||
"result": "failed",
|
||
"validations": {"reservation.ip_address": "Duplicate entry exists."},
|
||
}
|
||
driver._get = lambda path: KEA_RESV_SUBNETS_RESPONSE
|
||
|
||
with pytest.raises(RuntimeError, match="Kea rejected"):
|
||
driver.apply_dhcp_reservation(self._reservation())
|
||
|
||
|
||
class TestCommitDhcpReservations:
|
||
def test_posts_reconfigure(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {"status": "ok"}
|
||
|
||
result = driver.commit_dhcp_reservations()
|
||
|
||
assert calls == ["/api/kea/service/reconfigure"]
|
||
assert result == {"success": True}
|
||
|
||
|
||
# ── Kea DHCPv4 subnets ────────────────────────────────────────────────────────
|
||
|
||
KEA_SUBNET_SEARCH_RESPONSE = {
|
||
"rows": [
|
||
{"uuid": "sub-home", "subnet": "10.10.20.0/24", "description": "Home"},
|
||
{"uuid": "sub-guest", "subnet": "10.10.40.0/24", "description": "Guest"},
|
||
]
|
||
}
|
||
|
||
# getSubnet wraps the record under `subnet4` -- captured from OPNsense 25.x,
|
||
# not guessed. Unlike searchSubnet it carries the options, and OPNsense
|
||
# renders AsList fields as comma-separated strings.
|
||
KEA_SUBNET_DETAIL_RESPONSE = {
|
||
"subnet4": {
|
||
"subnet": "10.10.20.0/24",
|
||
"description": "Home",
|
||
"pools": "10.10.20.100-10.10.20.200",
|
||
"match-client-id": "1",
|
||
"option_data": {
|
||
"routers": "10.10.20.1",
|
||
"domain_name_servers": "10.10.20.1,10.10.20.2",
|
||
"domain_name": "home.local",
|
||
"domain_search": "home.local,office.local",
|
||
"ntp_servers": "",
|
||
},
|
||
}
|
||
}
|
||
|
||
|
||
class TestGetDhcpSubnets:
|
||
def _wire(self, driver, detail=None):
|
||
driver._get = lambda path: (
|
||
KEA_SUBNET_SEARCH_RESPONSE
|
||
if "searchSubnet" in path
|
||
else (detail or KEA_SUBNET_DETAIL_RESPONSE)
|
||
)
|
||
|
||
def test_maps_kea_subnet_to_vendor_neutral_dict(self, driver):
|
||
self._wire(driver)
|
||
|
||
result = driver.get_dhcp_subnets()
|
||
|
||
assert result[0] == {
|
||
"uuid": "sub-home",
|
||
"subnet": "10.10.20.0/24",
|
||
"description": "Home",
|
||
"pools": ["10.10.20.100-10.10.20.200"],
|
||
"option_data": {
|
||
"routers": ["10.10.20.1"],
|
||
"domain_name_servers": ["10.10.20.1", "10.10.20.2"],
|
||
"domain_name": "home.local",
|
||
"domain_search": ["home.local", "office.local"],
|
||
},
|
||
"match_client_id": True,
|
||
}
|
||
|
||
def test_fetches_detail_per_subnet(self, driver):
|
||
paths = []
|
||
driver._get = lambda path: (
|
||
paths.append(path)
|
||
or (KEA_SUBNET_SEARCH_RESPONSE if "searchSubnet" in path else KEA_SUBNET_DETAIL_RESPONSE)
|
||
)
|
||
|
||
driver.get_dhcp_subnets()
|
||
|
||
assert "/api/kea/dhcpv4/getSubnet/sub-home" in paths
|
||
assert "/api/kea/dhcpv4/getSubnet/sub-guest" in paths
|
||
|
||
def test_empty_option_is_omitted_not_reported_as_empty(self, driver):
|
||
"""An option Kea does not carry must read as "unset", not as []."""
|
||
self._wire(driver)
|
||
result = driver.get_dhcp_subnets()
|
||
assert "ntp_servers" not in result[0]["option_data"]
|
||
|
||
def test_multiline_pools_become_a_list(self, driver):
|
||
detail = {
|
||
"subnet4": {
|
||
"subnet": "10.10.20.0/24",
|
||
"description": "",
|
||
"pools": "10.10.20.100-10.10.20.150\n10.10.20.180-10.10.20.200",
|
||
"option_data": {},
|
||
}
|
||
}
|
||
self._wire(driver, detail)
|
||
assert driver.get_dhcp_subnets()[0]["pools"] == [
|
||
"10.10.20.100-10.10.20.150",
|
||
"10.10.20.180-10.10.20.200",
|
||
]
|
||
|
||
def test_selection_map_shape_is_accepted(self, driver):
|
||
"""Some OPNsense versions render list fields as a selection map."""
|
||
detail = {
|
||
"subnet4": {
|
||
"subnet": "10.10.20.0/24",
|
||
"description": "",
|
||
"pools": "",
|
||
"option_data": {
|
||
"domain_search": {
|
||
"home.local": {"value": "home.local", "selected": 1},
|
||
"office.local": {"value": "office.local", "selected": 0},
|
||
}
|
||
},
|
||
}
|
||
}
|
||
self._wire(driver, detail)
|
||
assert driver.get_dhcp_subnets()[0]["option_data"]["domain_search"] == ["home.local"]
|
||
|
||
def test_a_subnet_whose_detail_fails_is_skipped_not_fatal(self, driver):
|
||
def _get(path):
|
||
if "searchSubnet" in path:
|
||
return KEA_SUBNET_SEARCH_RESPONSE
|
||
if "sub-home" in path:
|
||
raise RuntimeError("boom")
|
||
return KEA_SUBNET_DETAIL_RESPONSE
|
||
|
||
driver._get = _get
|
||
result = driver.get_dhcp_subnets()
|
||
assert [s["uuid"] for s in result] == ["sub-guest"]
|
||
|
||
def test_missing_plugin_raises(self, driver):
|
||
def _get(path):
|
||
raise RuntimeError("404")
|
||
|
||
driver._get = _get
|
||
with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"):
|
||
driver.get_dhcp_subnets()
|
||
|
||
|
||
class TestApplyDhcpSubnet:
|
||
def test_update_uses_set_endpoint_with_uuid(self, driver):
|
||
calls = []
|
||
driver._get = lambda path: KEA_SUBNET_DETAIL_RESPONSE
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_dhcp_subnet(
|
||
{"subnet": "10.10.20.0/24", "option_data": {"domain_search": ["a.local"]}},
|
||
uuid="sub-home",
|
||
)
|
||
|
||
assert calls[0][0] == "/api/kea/dhcpv4/setSubnet/sub-home"
|
||
|
||
def test_create_uses_add_endpoint(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_dhcp_subnet({"subnet": "10.10.99.0/24", "option_data": {}})
|
||
|
||
assert calls[0][0] == "/api/kea/dhcpv4/addSubnet"
|
||
|
||
def test_unnamed_options_are_preserved_on_update(self, driver):
|
||
"""The whole point of the partial-update contract.
|
||
|
||
Kea autocollects routers/domain_name_servers. Setting only
|
||
domain_search must not blank them, or every managed subnet loses its
|
||
gateway the first time netOrk touches it.
|
||
"""
|
||
calls = []
|
||
driver._get = lambda path: KEA_SUBNET_DETAIL_RESPONSE
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_dhcp_subnet(
|
||
{"subnet": "10.10.20.0/24", "option_data": {"domain_search": ["new.local"]}},
|
||
uuid="sub-home",
|
||
)
|
||
|
||
sent = calls[0][1]["subnet"]["option_data"]
|
||
assert sent["domain_search"] == "new.local"
|
||
assert sent["routers"] == "10.10.20.1"
|
||
assert sent["domain_name_servers"] == "10.10.20.1,10.10.20.2"
|
||
|
||
def test_lists_are_serialised_comma_separated(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_dhcp_subnet(
|
||
{
|
||
"subnet": "10.10.99.0/24",
|
||
"option_data": {"domain_search": ["a.local", "b.local"]},
|
||
}
|
||
)
|
||
|
||
assert calls[0][1]["subnet"]["option_data"]["domain_search"] == "a.local,b.local"
|
||
|
||
def test_pools_are_serialised_newline_separated(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_dhcp_subnet(
|
||
{
|
||
"subnet": "10.10.99.0/24",
|
||
"pools": ["10.10.99.10-10.10.99.20", "10.10.99.30-10.10.99.40"],
|
||
"option_data": {},
|
||
}
|
||
)
|
||
|
||
assert (
|
||
calls[0][1]["subnet"]["pools"]
|
||
== "10.10.99.10-10.10.99.20\n10.10.99.30-10.10.99.40"
|
||
)
|
||
|
||
def test_explicit_option_data_disables_autocollect(self, driver):
|
||
"""Kea would otherwise re-fill routers/DNS/NTP and fight the desired state."""
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_dhcp_subnet(
|
||
{"subnet": "10.10.99.0/24", "option_data": {"domain_search": ["a.local"]}}
|
||
)
|
||
|
||
assert calls[0][1]["subnet"]["option_data_autocollect"] == "0"
|
||
|
||
def test_no_options_leaves_autocollect_alone(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"}
|
||
|
||
driver.apply_dhcp_subnet({"subnet": "10.10.99.0/24", "option_data": {}})
|
||
|
||
assert "option_data_autocollect" not in calls[0][1]["subnet"]
|
||
|
||
def test_rejection_raises(self, driver):
|
||
driver._post = lambda path, data=None: {"result": "failed", "validations": {"x": "bad"}}
|
||
|
||
with pytest.raises(RuntimeError, match="10.10.99.0/24"):
|
||
driver.apply_dhcp_subnet({"subnet": "10.10.99.0/24", "option_data": {}})
|
||
|
||
|
||
class TestCommitDhcpSubnets:
|
||
def test_reconfigures_kea(self, driver):
|
||
calls = []
|
||
driver._post = lambda path, data=None: calls.append(path) or {}
|
||
|
||
assert driver.commit_dhcp_subnets() == {"success": True}
|
||
assert calls == ["/api/kea/service/reconfigure"]
|
||
|
||
|
||
class TestKeaSubnetRecordKey:
|
||
"""`getSubnet` wraps its record under `subnet4`. Reading the wrong key is
|
||
not a cosmetic miss: the record comes back empty, so every subnet reports
|
||
no pools and no options, and the partial-option merge on update has
|
||
nothing to preserve."""
|
||
|
||
def test_reads_the_subnet4_wrapper(self, driver):
|
||
record = driver._kea_subnet_record({"subnet4": {"subnet": "10.0.0.0/24"}})
|
||
assert record == {"subnet": "10.0.0.0/24"}
|
||
|
||
def test_falls_back_to_the_legacy_subnet_wrapper(self, driver):
|
||
record = driver._kea_subnet_record({"subnet": {"subnet": "10.0.0.0/24"}})
|
||
assert record == {"subnet": "10.0.0.0/24"}
|
||
|
||
def test_unknown_shape_yields_an_empty_record(self, driver):
|
||
assert driver._kea_subnet_record({"something_else": {}}) == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DNS host overrides
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _override_row(uuid, host="opnsense-office", domain="office.local",
|
||
ip="10.10.0.1", description="[netork]", **extra):
|
||
row = {
|
||
"uuid": uuid,
|
||
"hostname": host,
|
||
"domain": domain,
|
||
"server": ip,
|
||
"rr": "A",
|
||
"description": description,
|
||
"enabled": "1",
|
||
"addptr": "1",
|
||
"isAlias": False,
|
||
}
|
||
row.update(extra)
|
||
return row
|
||
|
||
|
||
class TestUnboundHostOverrideInventory:
|
||
"""The inventory is what `sync_dns_zone` deletes from, so it has to report
|
||
the device as it is. Collapsing rows that merely look alike hides genuine
|
||
duplicates and — worse — puts them out of reach of the cleanup pass."""
|
||
|
||
def test_reports_every_duplicate_under_its_own_uuid(self, driver):
|
||
driver._detect_active_dns = lambda: "unbound"
|
||
driver._post = lambda path, data=None: {
|
||
"rows": [_override_row("uuid-1"), _override_row("uuid-2"), _override_row("uuid-3")]
|
||
}
|
||
|
||
entries = driver._get_unbound_host_overrides()
|
||
|
||
assert [e["uuid"] for e in entries] == ["uuid-1", "uuid-2", "uuid-3"]
|
||
|
||
def test_drops_alias_rows(self, driver):
|
||
driver._post = lambda path, data=None: {
|
||
"rows": [
|
||
_override_row("parent"),
|
||
_override_row("alias", host="www", isAlias=True),
|
||
]
|
||
}
|
||
|
||
entries = driver._get_unbound_host_overrides()
|
||
|
||
assert [e["uuid"] for e in entries] == ["parent"]
|
||
|
||
def test_falls_back_to_logical_dedup_when_the_flag_is_absent(self, driver):
|
||
"""Releases that predate `isAlias` give no way to tell an alias from a
|
||
copy, so the old collapse stays the safer read there."""
|
||
rows = [_override_row("uuid-1"), _override_row("uuid-2")]
|
||
for row in rows:
|
||
del row["isAlias"]
|
||
driver._post = lambda path, data=None: {"rows": rows}
|
||
|
||
entries = driver._get_unbound_host_overrides()
|
||
|
||
assert [e["uuid"] for e in entries] == ["uuid-1"]
|
||
|
||
def test_reads_the_auto_ptr_flag_from_addptr(self, driver):
|
||
driver._post = lambda path, data=None: {"rows": [_override_row("u", addptr="0")]}
|
||
|
||
assert driver._get_unbound_host_overrides()[0]["ptrrecord"] is False
|
||
|
||
def test_falls_back_to_the_legacy_ptrrecord_field(self, driver):
|
||
row = _override_row("u", ptrrecord="0")
|
||
del row["addptr"]
|
||
driver._post = lambda path, data=None: {"rows": [row]}
|
||
|
||
assert driver._get_unbound_host_overrides()[0]["ptrrecord"] is False
|
||
|
||
|
||
class TestSyncDnsZone:
|
||
def _record_calls(self, driver, rows):
|
||
calls = []
|
||
|
||
def _post(path, data=None):
|
||
calls.append((path, data))
|
||
if path.endswith("searchhostoverride"):
|
||
return {"rows": rows}
|
||
return {"result": "saved"}
|
||
|
||
driver._detect_active_dns = lambda: "unbound"
|
||
driver._post = _post
|
||
return calls
|
||
|
||
def test_deletes_every_managed_copy_not_just_one(self, driver):
|
||
rows = [_override_row("uuid-1"), _override_row("uuid-2"), _override_row("uuid-3")]
|
||
calls = self._record_calls(driver, rows)
|
||
|
||
driver.sync_dns_zone("office.local", [
|
||
{"hostname": "opnsense-office", "ip": "10.10.0.1", "record_type": "A", "enabled": True}
|
||
])
|
||
|
||
deleted = [p for p, _ in calls if "delhostoverride" in p]
|
||
assert sorted(p.rsplit("/", 1)[-1] for p in deleted) == ["uuid-1", "uuid-2", "uuid-3"]
|
||
assert len([p for p, _ in calls if p.endswith("addhostoverride")]) == 1
|
||
|
||
def test_leaves_records_it_does_not_manage_alone(self, driver):
|
||
rows = [
|
||
_override_row("managed"),
|
||
_override_row("by-hand", host="nas", description="added by hand"),
|
||
]
|
||
calls = self._record_calls(driver, rows)
|
||
|
||
driver.sync_dns_zone("office.local", [])
|
||
|
||
deleted = [p.rsplit("/", 1)[-1] for p, _ in calls if "delhostoverride" in p]
|
||
assert deleted == ["managed"]
|
||
|
||
def test_pushes_one_row_per_logical_record(self, driver):
|
||
"""Two netOrk rows for one host is a state the DB can legitimately be
|
||
in; the device still ends up with a single override."""
|
||
calls = self._record_calls(driver, [])
|
||
|
||
driver.sync_dns_zone("office.local", [
|
||
{"hostname": "opnsense-office", "ip": "10.10.0.1", "record_type": "A", "enabled": True},
|
||
{"hostname": "opnsense-office", "ip": "10.10.0.1", "record_type": "A", "enabled": True},
|
||
])
|
||
|
||
added = [d for p, d in calls if p.endswith("addhostoverride")]
|
||
assert len(added) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# uninstall_package – says what it cannot do
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestUninstallPackage:
|
||
"""`firmware/remove` reaches OPNsense plugins and nothing else.
|
||
|
||
`get_packages` reads `/api/core/firmware/info` filtered to the plugin list,
|
||
and `firmware/remove` acts on the same set. Software installed as a plain
|
||
FreeBSD package is invisible to the first and unreachable by the second —
|
||
the Wazuh agent being exactly that, on a driver the agent plugin lists as
|
||
supported.
|
||
|
||
Observed during a fleet-wide rollback on 2026-09-19: the `gw` device could
|
||
not be handled through netOrk at all, and the request that was posted for
|
||
it could never have succeeded.
|
||
|
||
So it refuses instead of posting. A request that cannot work is worse than
|
||
an honest no: the caller stops looking for the real problem. The API turns
|
||
NotImplementedError into 501, which is the accurate answer.
|
||
"""
|
||
|
||
def test_a_plugin_is_removed(self, driver):
|
||
driver._post = MagicMock(return_value={"status": "ok"})
|
||
|
||
result = driver.uninstall_package("os-wazuh-agent")
|
||
|
||
assert result["success"] is True
|
||
driver._post.assert_called_once()
|
||
|
||
def test_a_freebsd_package_is_refused(self, driver):
|
||
driver._post = MagicMock()
|
||
|
||
with pytest.raises(NotImplementedError, match="plugin"):
|
||
driver.uninstall_package("wazuh-agent")
|
||
|
||
driver._post.assert_not_called()
|
||
|
||
def test_the_refusal_names_the_package(self, driver):
|
||
"""Whoever reads the 501 needs to know which name was rejected."""
|
||
with pytest.raises(NotImplementedError, match="wazuh-agent"):
|
||
driver.uninstall_package("wazuh-agent")
|
||
|
||
def test_a_malformed_name_is_still_a_ValueError(self, driver):
|
||
"""Refusing non-plugins must not swallow the injection guard."""
|
||
with pytest.raises(ValueError):
|
||
driver.uninstall_package("os-thing; rm -rf /")
|