"""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, }, ] } ADDRESSES_RESPONSE = { "items": [ {"interface": "em0", "address": "192.0.2.10", "prefix": 24}, {"interface": "em0", "address": "2001:db8::1", "prefix": 64}, {"interface": "em1", "address": "203.0.113.5", "prefix": 30}, ] } 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_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 # --------------------------------------------------------------------------- # _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: "" result = driver.get_config() assert result["candidate"] == "" def test_candidate_contains_staged_routes(self, driver): driver._get = lambda path: "" 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: "" 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")