napalm-opnsense/tests/unit/test_public_contracts.py

"""What netOrk reached through ``_get``/``_post``, as public methods (netOrk #1113).

netOrk followed the firmware upgrade, edited Unbound's forwards and read FRR's
OSPF state through this driver's private REST helpers. The contracts are
napalm-device-types 4.17's; the OPNsense API calls and FRR's field names stay
here.
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from napalm_opnsense.opnsense import OPNsenseDriver


@pytest.fixture
def driver():
    with patch("napalm_opnsense.opnsense.requests.Session"):
        drv = OPNsenseDriver(hostname="fw", username="k", password="s", optional_args={"verify": False})
        drv.session = MagicMock()
        yield drv


class TestUpdateProgress:
    @pytest.mark.parametrize(
        "status, state",
        [("running", "running"), ("done", "done"), ("error", "failed"), ("", "idle"), (None, "idle")],
    )
    def test_the_upgrade_status_as_a_state(self, driver, status, state):
        st = {"status": status, "log": "***GOT REQUEST TO UPGRADE***\nfetching\n"}
        with patch.object(driver, "_get", return_value=st) as get:
            progress = driver.get_update_progress()

        get.assert_called_once_with("/api/core/firmware/upgradestatus")
        assert progress == {"state": state, "log": "***GOT REQUEST TO UPGRADE***\nfetching\n"}

    def test_the_packages_go_with_the_upgrade(self, driver):
        """netOrk named the packages it upgrades; ``apply_updates`` dropped them."""
        with patch.object(driver, "_post", return_value={"status": "ok"}) as post:
            result = driver.apply_updates(["opnsense", "openssl"])

        post.assert_called_once_with(
            "/api/core/firmware/upgrade", {"upgrade_packages": ["opnsense", "openssl"]}
        )
        assert result["success"] is True

    def test_no_packages_is_the_whole_upgrade(self, driver):
        with patch.object(driver, "_post", return_value={"status": "ok"}) as post:
            driver.apply_updates([])

        post.assert_called_once_with("/api/core/firmware/upgrade")


ROWS = [
    {"uuid": "u1", "enabled": "1", "type": "forward", "domain": "lab.example", "server": "10.0.0.53", "port": "53"},
    {"uuid": "u2", "enabled": "0", "type": "forward", "domain": "old.example", "server": "10.0.0.9", "port": ""},
    {"uuid": "u3", "enabled": "1", "type": "dot", "domain": "", "server": "9.9.9.9", "port": "853"},
]


class TestDnsForwards:
    def _post(self, calls):
        def post(path, data=None):
            calls.append((path, data))
            if path.endswith("searchForward"):
                return {"rows": ROWS}
            return {"result": "saved", "status": "ok"}

        return post

    def test_the_forwards_without_dns_over_tls(self, driver):
        calls = []
        with patch.object(driver, "_post", side_effect=self._post(calls)):
            forwards = driver.get_dns_forwards()

        assert forwards == [
            {"domain": "lab.example", "server": "10.0.0.53", "port": 53, "enabled": True},
            {"domain": "old.example", "server": "10.0.0.9", "port": 53, "enabled": False},
        ]

    def test_a_new_domain_is_added_and_applied(self, driver):
        calls = []
        with patch.object(driver, "_post", side_effect=self._post(calls)):
            driver.set_dns_forward("new.example.", "10.0.0.53")

        paths = [p for p, _ in calls]
        assert paths[-2:] == ["/api/unbound/settings/addForward", "/api/unbound/service/reconfigure"]
        added = calls[-2][1]["dot"]
        assert added == {"enabled": "1", "type": "forward", "domain": "new.example", "server": "10.0.0.53", "port": "53"}

    def test_a_known_domain_is_changed_not_doubled(self, driver):
        calls = []
        with patch.object(driver, "_post", side_effect=self._post(calls)):
            driver.set_dns_forward("lab.example", "10.0.0.54", port=5353)

        paths = [p for p, _ in calls]
        assert "/api/unbound/settings/setForward/u1" in paths
        assert "/api/unbound/settings/addForward" not in paths

    def test_a_refusal_is_raised(self, driver):
        def post(path, data=None):
            if path.endswith("searchForward"):
                return {"rows": []}
            return {"result": "failed", "validations": {"dot.server": "invalid"}}

        with patch.object(driver, "_post", side_effect=post), pytest.raises(RuntimeError, match="invalid"):
            driver.set_dns_forward("x.example", "not-an-ip")

    def test_removing_deletes_every_forward_of_the_domain(self, driver):
        calls = []
        with patch.object(driver, "_post", side_effect=self._post(calls)):
            driver.remove_dns_forward("lab.example.")

        paths = [p for p, _ in calls]
        assert paths[-2:] == ["/api/unbound/settings/delForward/u1", "/api/unbound/service/reconfigure"]

    def test_removing_an_unknown_domain_changes_nothing(self, driver):
        calls = []
        with patch.object(driver, "_post", side_effect=self._post(calls)):
            driver.remove_dns_forward("nothing.example")

        assert [p for p, _ in calls] == ["/api/unbound/settings/searchForward"]


OVERVIEW = {"response": {"routerId": "10.255.0.1", "areas": {"0.0.0.0": {}, "0.0.0.1": {}}}}
INTERFACES = {
    "response": {
        "interfaces": {
            "wg0": {
                "ipAddress": "10.255.0.1",
                "ipAddressPrefixlen": 24,
                "area": "0.0.0.0",
                "networkType": "POINTOPOINT",
                "authentication": "MessageDigest",
                "nbrCount": 2,
                "nbrAdjacentCount": 1,
                "cost": 10,
            }
        }
    }
}


class TestOspf:
    def test_frr_in_neutral_terms(self, driver):
        def get(path):
            return OVERVIEW if path.endswith("ospfoverview") else INTERFACES

        with patch.object(driver, "_get", side_effect=get):
            ospf = driver.get_ospf()

        assert ospf == {
            "router_id": "10.255.0.1",
            "areas": ["0.0.0.0", "0.0.0.1"],
            "interfaces": {
                "wg0": {
                    "address": "10.255.0.1",
                    "prefix_length": 24,
                    "area": "0.0.0.0",
                    "network_type": "POINTOPOINT",
                    "authentication": "MessageDigest",
                    "neighbors": 2,
                    "neighbors_adjacent": 1,
                    "cost": 10,
                }
            },
        }

    def test_no_ospf_running(self, driver):
        with patch.object(driver, "_get", return_value={"response": {}}):
            assert driver.get_ospf() == {"router_id": None, "areas": [], "interfaces": {}}