"""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 # --------------------------------------------------------------------------- # _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"} # --------------------------------------------------------------------------- # 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()