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)]


class _Result:
    def __init__(self, stdout: str = "", exit_code: int = 0, stderr: str = "") -> None:
        self.stdout, self.exit_code, self.stderr = stdout, exit_code, stderr


def _channel(cls, answers: dict):
    """A driver whose run_command answers from *answers*: ``command -> stdout``
    or ``command -> (stdout, exit_code)``; ``(command, True)`` keys answer only
    privileged calls. Every call is recorded as ``(command, privileged)``."""
    driver = cls("host", "netork", "")
    calls: list[tuple[str, bool]] = []

    def run(command, *, privileged=False, timeout=60, stdin=None):
        calls.append((command, privileged))
        answer = answers.get((command, privileged), answers.get(command, ""))
        stdout, code = answer if isinstance(answer, tuple) else (answer, 0)
        return _Result(stdout, code)

    driver.run_command = run  # type: ignore[method-assign]
    driver.calls = calls  # type: ignore[attr-defined]
    return driver


class TestPackages:
    def test_freebsd_packages_with_their_repository(self):
        driver = _channel(
            FreeBSDDriver,
            {
                "pkg -N": "",
                FreeBSDDriver.PKG_QUERY: _read("freebsd-vm", "pkg_query_n_v_R_sb_c.txt"),
            },
        )
        packages = driver.get_packages()
        assert {p["source"] for p in packages} >= {"FreeBSD-base", "FreeBSD-ports"}

    def test_freebsd_without_pkg_has_no_packages(self):
        """A hand-installed classic system may never have bootstrapped pkg."""
        driver = _channel(FreeBSDDriver, {"pkg -N": ("pkg: pkg is not installed", 1)})
        assert driver.get_packages() == []
        assert driver.calls == [("pkg -N", False)]

    def test_openbsd_packages(self):
        driver = _channel(OpenBSDDriver, {"pkg_info": _read("openbsd-vm", "pkg_info_full.txt")})
        assert any(p["name"] == "pcre2" for p in driver.get_packages())

    @pytest.mark.parametrize(
        ("cls", "install", "uninstall"),
        [
            (FreeBSDDriver, "pkg install -y nginx", "pkg delete -y nginx"),
            (OpenBSDDriver, "pkg_add -I nginx", "pkg_delete nginx"),
        ],
    )
    def test_install_and_uninstall_as_root(self, cls, install, uninstall):
        driver = _channel(cls, {})
        driver.install_package("nginx")
        driver.uninstall_package("nginx")
        assert driver.calls == [(install, True), (uninstall, True)]

    def test_a_version_is_part_of_the_name(self):
        driver = _channel(FreeBSDDriver, {})
        driver.install_package("nginx", "1.28.0")
        assert driver.calls == [("pkg install -y nginx-1.28.0", True)]

    def test_a_failed_install_raises(self):
        driver = _channel(OpenBSDDriver, {("pkg_add -I nope", True): ("Can't find nope", 1)})
        with pytest.raises(RuntimeError, match="nope"):
            driver.install_package("nope")

    @pytest.mark.parametrize("name", ["", "nginx; reboot", "-f", "a b"])
    def test_a_name_that_is_not_a_package_is_refused(self, name):
        driver = _channel(FreeBSDDriver, {})
        with pytest.raises(ValueError):
            driver.install_package(name)
        assert driver.calls == []


class TestAvailableUpdates:
    def test_freebsd_marks_what_vuxml_knows(self):
        driver = _channel(
            FreeBSDDriver,
            {
                "pkg -N": "",
                ("pkg upgrade -n", True): _read("freebsd-vm", "sudo_pkg_upgrade_n_latest.txt"),
                ("pkg audit -Fq", True): ("python312-3.12.14\nexpat-2.8.2\n", 1),
                "pkg repos -l": _read("freebsd-vm", "pkg_repos_l.txt"),
            },
        )
        updates = {u["name"]: u for u in driver.get_available_updates()}
        assert len(updates) == 10
        assert updates["expat"]["security"] is True
        assert updates["pkg"]["security"] is False
        assert updates["pkg"]["origin"] == "FreeBSD-ports"
        assert "base-system" not in updates  # pkgbase: the base system is packages

    def test_freebsd_without_an_audit_cannot_tell(self):
        driver = _channel(
            FreeBSDDriver,
            {
                "pkg -N": "",
                ("pkg upgrade -n", True): _read("freebsd-vm", "sudo_pkg_upgrade_n_latest.txt"),
                ("pkg audit -Fq", True): ("pkg: cannot fetch", 3),
                "pkg repos -l": "FreeBSD-ports\nFreeBSD-base\n",
            },
        )
        assert {u["security"] for u in driver.get_available_updates()} == {None}

    def test_classic_freebsd_reports_fetched_base_updates_as_one_entry(self):
        driver = _channel(
            FreeBSDDriver,
            {
                "pkg -N": ("pkg: pkg is not installed", 1),
                ("freebsd-update --not-running-from-cron updatesready", True): ("", 0),
                "freebsd-version -u": "15.1-RELEASE",
            },
        )
        assert driver.get_available_updates() == [
            {
                "name": "base-system",
                "current_version": "15.1-RELEASE",
                "new_version": "fetched by freebsd-update",
                "origin": "freebsd-update",
                "security": None,
            }
        ]

    def test_classic_freebsd_without_fetched_updates(self):
        driver = _channel(
            FreeBSDDriver,
            {
                "pkg -N": ("", 1),
                ("freebsd-update --not-running-from-cron updatesready", True): ("", 2),
            },
        )
        assert driver.get_available_updates() == []

    def test_openbsd_packages_and_syspatches(self):
        driver = _channel(
            OpenBSDDriver,
            {
                ("pkg_add -u -n -v", True): _read("openbsd-vm", "sudo_pkg_add_u_n_v.txt"),
                ("syspatch -c", True): _read("openbsd-vm", "syspatch_c.txt"),
                ("syspatch -l", True): "",
                "uname -r": "7.9",
            },
        )
        updates = {u["name"]: u for u in driver.get_available_updates()}
        assert updates["pcre2"]["new_version"] == "10.48"
        assert updates["pcre2"]["security"] is None
        base = updates["base-system"]
        assert base["current_version"] == "7.9"
        assert base["new_version"].startswith("34 patches: 001_xserver")
        assert base["origin"] == "syspatch"
        assert base["security"] is None

    def test_openbsd_fully_patched(self):
        driver = _channel(
            OpenBSDDriver, {("pkg_add -u -n -v", True): "", ("syspatch -c", True): ""}
        )
        assert driver.get_available_updates() == []


class TestServices:
    def test_freebsd_reads_the_status_as_root(self):
        status = _read("freebsd-vm", "sudo_service_status_loop.txt")
        driver = _channel(FreeBSDDriver, {(FreeBSDDriver.SERVICE_STATUS_COMMAND, True): status})
        services = {s["name"]: s for s in driver.get_services()}
        assert services["cron"]["running"] is True

    def test_freebsd_without_root_reads_what_it_can(self):
        status = _read("freebsd-vm", "service_status_loop.txt")
        driver = _channel(
            FreeBSDDriver,
            {
                (FreeBSDDriver.SERVICE_STATUS_COMMAND, True): ("sudo: a password is required", 1),
                (FreeBSDDriver.SERVICE_STATUS_COMMAND, False): status,
            },
        )
        assert {s["name"] for s in driver.get_services()} >= {"sshd", "cron"}

    def test_openbsd(self):
        driver = _channel(
            OpenBSDDriver,
            {OpenBSDDriver.SERVICE_STATUS_COMMAND: _read("openbsd-vm", "rcctl_check_loop.txt")},
        )
        assert {s["name"]: s["running"] for s in driver.get_services()}["smtpd"] is True

    @pytest.mark.parametrize(
        ("cls", "command"),
        [(FreeBSDDriver, "service sshd restart"), (OpenBSDDriver, "rcctl restart sshd")],
    )
    def test_manage_as_root(self, cls, command):
        driver = _channel(cls, {})
        assert driver.manage_service("sshd", "restart") == {"success": True, "output": ""}
        assert driver.calls == [(command, True)]

    def test_a_failed_action_says_so(self):
        driver = _channel(
            OpenBSDDriver, {("rcctl start nope", True): ("rcctl: service nope does not exist", 2)}
        )
        result = driver.manage_service("nope", "start")
        assert result["success"] is False
        assert "does not exist" in result["output"]

    @pytest.mark.parametrize(
        ("name", "action"), [("sshd", "reload"), ("sshd;reboot", "start"), ("", "start")]
    )
    def test_refused_before_anything_is_sent(self, name, action):
        driver = _channel(FreeBSDDriver, {})
        with pytest.raises(ValueError):
            driver.manage_service(name, action)
        assert driver.calls == []