napalm-opnsense/tests/unit/test_identity_and_held_addresses.py

"""What the firewall is called, which version it runs, and which addresses it serves.

``/api/core/system/status`` on OPNsense 26.7 carries neither name nor version
(``metadata`` and ``subsystems`` only), so every firewall read as nameless and
"unknown". ``/api/diagnostics/system/system_information`` has both: ``name`` is
the FQDN, ``versions[0]`` reads ``"OPNsense 26.7.7-amd64"``.

An address bound to a CARP VHID sits on the interface of every node, in state
MASTER or BACKUP. Only the MASTER answers for it, so only there is it one of the
node's addresses -- the same as keepalived, where only the master has the address
at all (netOrk #1121: a virtual IP's badge sits on its holder's interface).
"""

from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import patch

import pytest

from napalm_opnsense.opnsense import OPNsenseDriver

OVERVIEW = json.loads(
    (Path(__file__).parent / "fixtures" / "interfaces_overview_carp.json").read_text()
)
SYSTEM_INFORMATION = {
    "name": "gw-02.home.manivong.de",
    "versions": ["OPNsense 26.7.7-amd64", "FreeBSD 15.1-RELEASE-p4", "OpenSSL 3.5.9"],
}
STATUS_26_7 = {"metadata": {"translations": {}}, "subsystems": {}}


@pytest.fixture
def driver() -> OPNsenseDriver:
    with patch("napalm_opnsense.opnsense.requests.Session"):
        return OPNsenseDriver(hostname="fw", username="k", password="s")


def _answers(driver, system_information=SYSTEM_INFORMATION, status=STATUS_26_7, overview=OVERVIEW):
    def get(path: str):
        if path == "/api/diagnostics/system/system_information":
            if isinstance(system_information, Exception):
                raise system_information
            return system_information
        if path == "/api/core/system/status":
            return status
        if path == "/api/interfaces/overview/export":
            return overview
        raise AssertionError(f"unexpected GET {path}")

    driver._get = get
    return driver


class TestIdentity:
    def test_the_name_is_the_short_host_and_the_fqdn_the_whole(self, driver):
        facts = _answers(driver).get_facts()
        assert (facts["hostname"], facts["fqdn"]) == ("gw-02", "gw-02.home.manivong.de")

    def test_the_version_comes_without_the_architecture(self, driver):
        assert _answers(driver).get_facts()["os_version"] == "26.7.7"

    @pytest.mark.parametrize(
        "line, version",
        [
            ("OPNsense 26.7.7_1-aarch64", "26.7.7_1"),
            ("OPNsense 25.1", "25.1"),
            ("OPNsense 24.7.12-i386", "24.7.12"),
        ],
    )
    def test_version_spellings(self, driver, line, version):
        info = {"name": "fw", "versions": [line]}
        assert _answers(driver, system_information=info).get_facts()["os_version"] == version

    def test_a_name_without_a_domain_is_both(self, driver):
        facts = _answers(driver, system_information={"name": "fw", "versions": []}).get_facts()
        assert (facts["hostname"], facts["fqdn"]) == ("fw", "fw")

    def test_without_system_information_the_status_answer_still_counts(self, driver):
        """An older OPNsense, or an API key without the diagnostics privilege."""
        status = {"hostname": "opnsense01", "version": "24.7"}
        facts = _answers(driver, system_information=RuntimeError("403"), status=status).get_facts()
        assert (facts["hostname"], facts["os_version"]) == ("opnsense01", "24.7")

    def test_nothing_known_is_empty_and_unknown(self, driver):
        facts = _answers(driver, system_information={}).get_facts()
        assert (facts["hostname"], facts["fqdn"], facts["os_version"]) == ("", "", "unknown")


class TestHeldAddresses:
    def test_the_master_serves_its_carp_addresses(self, driver):
        result = _answers(driver).get_interfaces_ip()
        assert set(result["em0"]["ipv4"]) == {"192.0.2.2", "192.0.2.1"}
        assert result["em0"]["ipv4"]["192.0.2.1"] == {"prefix_length": 24}
        assert set(result["vlan0040"]["ipv4"]) == {
            "198.51.100.2",
            "198.51.100.1",
            "198.51.100.140",
        }

    def test_a_backup_serves_its_own_address_only(self, driver):
        backup = json.loads(json.dumps(OVERVIEW).replace('"MASTER"', '"BACKUP"'))
        result = _answers(driver, overview=backup).get_interfaces_ip()
        assert set(result["em0"]["ipv4"]) == {"192.0.2.2"}
        assert set(result["vlan0040"]["ipv4"]) == {"198.51.100.2"}

    def test_a_vhid_that_is_not_up_is_not_served(self, driver):
        result = _answers(driver).get_interfaces_ip()
        assert set(result["vlan0011"]["ipv4"]) == {"203.0.113.2"}

    def test_a_held_ipv6_carp_address_is_served(self, driver):
        result = _answers(driver).get_interfaces_ip()
        assert "2001:db8:8::1" in result["em0"]["ipv6"]

    def test_an_address_without_a_vhid_is_left_as_before(self, driver):
        """Only CARP addresses are added; a plain alias is a separate question."""
        result = _answers(driver).get_interfaces_ip()
        assert set(result["vlan0020"]["ipv4"]) == {"192.0.2.130"}

    def test_without_a_primary_address_the_list_still_leaves_out_a_backup(self, driver):
        overview = [
            {
                "device": "vlan0099",
                "ipv4": [
                    {"ipaddr": "192.0.2.66/26"},
                    {"ipaddr": "192.0.2.65/26", "vhid": "199", "status": "BACKUP"},
                ],
            }
        ]
        result = _answers(driver, overview=overview).get_interfaces_ip()
        assert set(result["vlan0099"]["ipv4"]) == {"192.0.2.66"}