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