napalm-opnsense/tests/unit/test_virtual_ips.py

"""The firewall's CARP addresses, with who holds each (netOrk #1121).

``/api/interfaces/overview/export`` -- the source ``get_interfaces_ip`` reads --
lists every address on an interface, and an address bound to a CARP VHID carries
the VHID and this node's CARP state. That covers both the CARP address itself and
an IP alias bound to its VHID: both float with the VHID's MASTER. An address
without a VHID stays on this node and is no virtual IP.

The overview has no CARP password in it; ``interfaces/vip_settings`` would, which
is why it is not read. The fixture is the shape of a real OPNsense 26.7 answer,
cut down, with documentation addresses.
"""

from __future__ import annotations

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

import pytest

from napalm_opnsense.opnsense import OPNsenseDriver
from napalm_opnsense.virtual_ips import carp_addresses

OVERVIEW = json.loads(
    (Path(__file__).parent / "fixtures" / "interfaces_overview_carp.json").read_text()
)


def _driver(overview=OVERVIEW) -> OPNsenseDriver:
    with patch("napalm_opnsense.opnsense.requests.Session"):
        drv = OPNsenseDriver(hostname="fw", username="k", password="s")
    drv.session = MagicMock()

    def get(path: str):
        if path == "/api/interfaces/overview/export":
            if isinstance(overview, Exception):
                raise overview
            return overview
        raise AssertionError(f"unexpected GET {path}")

    drv._get = get  # type: ignore[method-assign]
    return drv


def _by_address(entries) -> dict:
    return {e["address"]: e for e in entries}


class TestCarpAddresses:
    def test_every_address_bound_to_a_vhid_is_a_virtual_ip(self):
        assert [e["address"] for e in carp_addresses(OVERVIEW)] == [
            "192.0.2.1",
            "198.51.100.1",
            "198.51.100.140",
            "203.0.113.1",
            "2001:db8:8::1",
        ]

    def test_an_address_without_a_vhid_stays_on_this_node(self):
        addresses = {e["address"] for e in carp_addresses(OVERVIEW)}
        assert not addresses & {"192.0.2.2", "192.0.2.140", "127.0.0.1", "fe80::1"}

    def test_the_carp_address_carries_its_interface_and_state(self):
        assert _by_address(carp_addresses(OVERVIEW))["192.0.2.1"] == {
            "mechanism": "carp",
            "instance": "MGMT",
            "address": "192.0.2.1",
            "prefix_length": 24,
            "interface": "em0",
            "group_key": "vhid:108",
            "running": True,
            "holds": True,
        }

    def test_an_alias_on_a_vhid_follows_that_vhid(self):
        alias = _by_address(carp_addresses(OVERVIEW))["198.51.100.140"]
        assert (alias["group_key"], alias["prefix_length"], alias["holds"]) == (
            "vhid:140",
            32,
            True,
        )

    def test_a_vhid_not_up_neither_runs_nor_holds(self):
        entry = _by_address(carp_addresses(OVERVIEW))["203.0.113.1"]
        assert (entry["running"], entry["holds"]) == (False, False)

    def test_an_interface_without_a_description_is_named_by_its_identifier(self):
        assert _by_address(carp_addresses(OVERVIEW))["203.0.113.1"]["instance"] == "opt9"

    def test_on_a_backup_node_nothing_is_held(self):
        backup = json.loads(json.dumps(OVERVIEW).replace('"MASTER"', '"BACKUP"'))
        entries = carp_addresses(backup)
        assert entries and all(e["holds"] is False for e in entries)

    def test_the_answer_wrapped_in_an_object_reads_the_same(self):
        assert carp_addresses({"interfaces": OVERVIEW}) == carp_addresses(OVERVIEW)

    @pytest.mark.parametrize("answer", [[], {}, {"interfaces": None}])
    def test_a_firewall_without_interfaces_has_none(self, answer):
        assert carp_addresses(answer) == []

    @pytest.mark.parametrize("answer", ["nonsense", None, 42])
    def test_an_answer_of_another_shape_is_refused(self, answer):
        with pytest.raises(ValueError):
            carp_addresses(answer)

    def test_odd_entries_are_skipped(self):
        overview = [
            {"identifier": "opt1", "ipv4": ["192.0.2.9/24", {"vhid": "1"}, None]},
            "not-an-interface",
            {"identifier": "opt2", "ipv4": [{"ipaddr": "bogus", "vhid": "2"}]},
        ]
        assert carp_addresses(overview) == []


class TestGetVirtualIps:
    def test_a_firewall_reports_carp_and_does_not_look_for_the_linux_mechanisms(self):
        reading = _driver().get_virtual_ips()
        assert reading["vip_manager"] is None
        assert reading["keepalived"] is None
        assert len(reading["carp"]) == 5

    def test_a_firewall_without_carp_reports_an_empty_list(self):
        reading = _driver(overview=[OVERVIEW[3], OVERVIEW[4]]).get_virtual_ips()
        assert reading["carp"] == []

    def test_a_failed_request_is_raised_not_reported_as_none_declared(self):
        with pytest.raises(RuntimeError):
            _driver(overview=RuntimeError("API down")).get_virtual_ips()