napalm-bsd/tests/test_drivers.py

"""FreeBSDDriver and OpenBSDDriver: what netOrk's poll reads, on recorded output."""

from __future__ import annotations

from pathlib import Path

import pytest
from napalm.base import NetworkDriver
from napalm_device_types import OSDriver

from napalm_bsd import FreeBSDDriver, OpenBSDDriver
from napalm_bsd.base import BsdDriver

FIXTURES = Path(__file__).parent / "fixtures"
PS = "ps_axww_o_user_pid_ppid_cpu_mem_vsz_rss_stat_lstart_command.txt"


def _read(system: str, name: str) -> str:
    return (FIXTURES / system / name).read_text()


def _driver(cls, answers: dict[str, str]):
    """A driver whose commands are answered from *answers*; unknown ones print nothing."""
    driver = cls("host", "netork", "")
    asked: list[str] = []

    def out(command: str, *, privileged: bool = False, timeout: float = 60) -> str:
        asked.append(command)
        return answers.get(command, "").strip()

    driver._out = out  # type: ignore[method-assign]
    driver.asked = asked  # type: ignore[attr-defined]
    return driver


def _freebsd_host():
    s = "freebsd-host"
    return _driver(
        FreeBSDDriver,
        {
            "sysctl -n kern.hostname": "bsd-01.lab.example",
            "freebsd-version -u": "15.1-RELEASE",
            "freebsd-version -r": "15.1-RELEASE",
            FreeBSDDriver.PLATFORM_COMMAND: "maker=QEMU\nproduct=Standard PC (i440FX + PIIX, 1996)\nserial=",
            "sysctl -n kern.boottime": _read(s, "sysctl_n_kern.boottime.txt"),
            "date +%s": "1791408816",
            "ifconfig -a": _read(s, "ifconfig_a.txt"),
            "netstat -rnW": _read(s, "netstat_rn.txt"),
            "arp -an": _read(s, "arp_an.txt"),
            "cat /etc/passwd": _read(s, "cat_etc_passwd.txt"),
            "cat /etc/group": _read(s, "cat_etc_group.txt"),
            BsdDriver.PS_COMMAND: _read(s, PS),
            "cat /etc/crontab": _read(s, "cat_etc_crontab.txt"),
            "crontab -l": "0 4 * * * /home/netork/backup.sh\n",
        },
    )


def _openbsd_vm():
    s = "openbsd-vm"
    return _driver(
        OpenBSDDriver,
        {
            "hostname": "spike-openbsd",
            "uname -sr": "OpenBSD 7.9",
            "uname -v": "GENERIC.MP#449",
            OpenBSDDriver.PLATFORM_COMMAND: "vendor=QEMU\nproduct=Standard PC (i440FX + PIIX, 1996)\nserialno=",
            "sysctl -n kern.boottime": _read(s, "sysctl_n_kern.boottime.txt"),
            "date +%s": "1791406889",
            "ifconfig -a": _read(s, "ifconfig_a.txt"),
            "netstat -rn": _read(s, "netstat_rn.txt"),
            "arp -an": _read(s, "arp_an.txt"),
            "cat /etc/passwd": _read(s, "cat_etc_passwd.txt"),
            BsdDriver.PS_COMMAND: _read(s, PS),
        },
    )


class TestIdentity:
    @pytest.mark.parametrize(
        ("cls", "name", "label"),
        [(FreeBSDDriver, "freebsd", "FreeBSD"), (OpenBSDDriver, "openbsd", "OpenBSD")],
    )
    def test_names(self, cls, name, label):
        assert cls.DRIVER_NAME == name
        assert cls.TYPE_LABEL == label
        assert issubclass(cls, OSDriver)

    @pytest.mark.parametrize("cls", [FreeBSDDriver, OpenBSDDriver])
    @pytest.mark.parametrize(
        "getter",
        [
            "get_facts",
            "get_interfaces",
            "get_interfaces_ip",
            "get_route_to",
            "get_arp_table",
            "get_lldp_neighbors",
            "get_users",
            "get_processes",
            "get_cron_jobs",
        ],
    )
    def test_implements_what_the_poll_reads(self, cls, getter):
        """netOrk asks whether a getter is overridden, not merely present."""
        assert getattr(cls, getter) is not getattr(NetworkDriver, getter, None)


class TestFacts:
    def test_freebsd(self):
        facts = _freebsd_host().get_facts()
        assert facts == {
            "hostname": "bsd-01",
            "fqdn": "bsd-01.lab.example",
            "vendor": "QEMU",
            "model": "Standard PC (i440FX + PIIX, 1996)",
            "serial_number": "",
            "os_version": "FreeBSD 15.1-RELEASE",
            "uptime": 1791408816 - 1791405216,
            "interface_list": ["em0"],
            "running_kernel": "15.1-RELEASE",
        }

    def test_openbsd(self):
        facts = _openbsd_vm().get_facts()
        assert facts["hostname"] == facts["fqdn"] == "spike-openbsd"
        assert facts["os_version"] == "OpenBSD 7.9"
        assert facts["running_kernel"] == "GENERIC.MP#449"
        assert facts["uptime"] == 100
        assert facts["interface_list"] == ["vio0", "enc0", "pflog0"]

    def test_an_unreadable_boottime_is_no_uptime(self):
        driver = _freebsd_host()
        driver._out = lambda command, **kw: ""  # type: ignore[method-assign]
        assert driver.get_facts()["uptime"] == -1


class TestInterfaces:
    def test_napalm_shape(self):
        interfaces = _freebsd_host().get_interfaces()
        assert interfaces["em0"] == {
            "is_up": True,
            "is_enabled": True,
            "description": "",
            "last_flapped": -1.0,
            "speed": 1000.0,
            "mtu": 1500,
            "mac_address": "bc:24:11:20:b0:20",
        }

    def test_addresses(self):
        assert _openbsd_vm().get_interfaces_ip()["vio0"] == {
            "ipv4": {"10.0.2.15": {"prefix_length": 24}},
            "ipv6": {},
        }


class TestRoutesAndNeighbours:
    def test_routes_keyed_by_network(self):
        routes = _freebsd_host().get_route_to()
        default = routes["0.0.0.0/0"][0]
        assert default["next_hop"] == "192.0.2.1"
        assert default["outgoing_interface"] == "em0"
        assert default["protocol"] == "static"
        assert default["current_active"] is True

    def test_openbsd_reads_routes_without_W(self):
        driver = _openbsd_vm()
        assert "10.0.2.0/24" in driver.get_route_to()
        assert "netstat -rn" in driver.asked

    def test_arp(self):
        assert _openbsd_vm().get_arp_table()[0]["ip"] == "10.0.2.2"

    def test_no_lldp_daemon_means_no_neighbours(self):
        assert _freebsd_host().get_lldp_neighbors() == {}


class TestAccountsAndProcesses:
    def test_users(self):
        users = _freebsd_host().get_users()
        assert next(u for u in users if u["username"] == "root")["groups"] == ["wheel", "operator"]

    def test_processes(self):
        init = next(p for p in _openbsd_vm().get_processes() if p["pid"] == 1)
        assert init["command"] == "/sbin/init"

    def test_cron_jobs_from_the_system_and_the_login_users_crontab(self):
        jobs = _freebsd_host().get_cron_jobs()
        assert {"user": "root", "schedule": "0 * * * *", "command": "newsyslog"} in jobs
        assert {
            "user": "netork",
            "schedule": "0 4 * * *",
            "command": "/home/netork/backup.sh",
        } in jobs


class TestListeningSockets:
    def test_freebsd_as_root_is_attributed(self):
        driver = _driver(FreeBSDDriver, {})
        sockstat = _read("freebsd-vm", "sudo_sockstat_46lq_P_tcp_udp.txt")
        driver.run_command = lambda command, privileged=False, timeout=60, stdin=None: (  # type: ignore[method-assign]
            type("R", (), {"stdout": sockstat, "exit_code": 0})()
        )
        result = driver.get_listening_sockets()
        assert result["attributed"] is True
        assert len(result["sockets"]) == 7

    def test_openbsd_without_root_falls_back_to_netstat(self):
        driver = _driver(OpenBSDDriver, {})
        netstat = _read("openbsd-vm", "netstat_an.txt")
        calls = []

        def run(command, privileged=False, timeout=60, stdin=None):
            calls.append((command, privileged))
            if privileged:  # sudo -n without a password: refused
                return type("R", (), {"stdout": "", "exit_code": 1})()
            return type("R", (), {"stdout": netstat, "exit_code": 0})()

        driver.run_command = run  # type: ignore[method-assign]
        result = driver.get_listening_sockets()
        assert result["attributed"] is False
        assert ("tcp", "0.0.0.0", 22) in {
            (s["proto"], s["address"], s["port"]) for s in result["sockets"]
        }
        assert calls == [("fstat -n", True), (OpenBSDDriver.LISTENING_FALLBACK_COMMAND, False)]