1246 lines
45 KiB
Python
1246 lines
45 KiB
Python
"""Unit tests for OPNsenseDriver — no real device required."""
|
||
|
||
import json
|
||
import pytest
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from napalm_opnsense.opnsense import OPNsenseDriver
|
||
from napalm.base.exceptions import ConnectionException, ConnectionClosedException
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.fixture
|
||
def driver():
|
||
"""Return a driver instance with a mocked requests.Session."""
|
||
with patch("napalm_opnsense.opnsense.requests.Session"):
|
||
drv = OPNsenseDriver(
|
||
hostname="opnsense.example.com",
|
||
username="api_key",
|
||
password="api_secret",
|
||
optional_args={"verify": False},
|
||
)
|
||
drv.session = MagicMock()
|
||
yield drv
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Sample API responses
|
||
# ---------------------------------------------------------------------------
|
||
|
||
STATUS_RESPONSE = {
|
||
"hostname": "opnsense01",
|
||
"version": "24.7",
|
||
"model": "OPNsense",
|
||
"serial": "ABC123",
|
||
"uptime": 12345,
|
||
}
|
||
|
||
INTERFACES_RESPONSE = {
|
||
"interfaces": [
|
||
{
|
||
"name": "em0",
|
||
"up": True,
|
||
"enabled": True,
|
||
"descr": "LAN",
|
||
"mac": "AA:BB:CC:DD:EE:FF",
|
||
"speed_mbps": 1000,
|
||
"mtu": 1500,
|
||
},
|
||
{
|
||
"name": "em1",
|
||
"up": False,
|
||
"enabled": True,
|
||
"descr": "WAN",
|
||
"mac": "AA:BB:CC:DD:EE:00",
|
||
"speed_mbps": None,
|
||
"mtu": 0,
|
||
},
|
||
]
|
||
}
|
||
|
||
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: "<opnsense/>"
|
||
result = driver.get_config()
|
||
assert result["candidate"] == ""
|
||
|
||
def test_candidate_contains_staged_routes(self, driver):
|
||
driver._get = lambda path: "<opnsense/>"
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
result = driver.get_config()
|
||
assert "WAN_GW" in result["candidate"]
|
||
|
||
def test_candidate_is_valid_json(self, driver):
|
||
driver._get = lambda path: "<opnsense/>"
|
||
driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
|
||
result = driver.get_config()
|
||
parsed = json.loads(result["candidate"])
|
||
assert isinstance(parsed, list)
|