From dd8977750deae9b86ec06814828e29d25fa5ce92 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 13 Oct 2026 01:22:22 +0200 Subject: [PATCH] chore: Python 3.13, mypy --strict in CI, eight bug fixes, docs Squashes three changes: ## chore: support Python 3.13 and make mypy --strict pass in CI Add 3.13 to the CI matrix and classifiers, run mypy in CI, fix the 14 strict errors it reported, and correct the README's minimum Python. ## fix: eight bugs found in a full review, each with a regression test - channel: run_on_transport lost output sshd sent after the exit status; it now reads stdout and stderr to EOF before returning. - ssh_server: a failed way back (restore or reload) raised out of harden_ssh_server after the host was touched; it is now reported in the result, with rolled_back False. - host_status: a newer RPM kernel (RHEL, Fedora) was never seen, because its release, not its arch, followed the last "-". - package_updates: an apt line wrapped at a blank lost the blank, and with it the package. - kernel: an empty payload (gzip or base64 missing) read as a report. - listening: "[fe80::1]%eth0:546", as ss prints it, kept its brackets. - ping_sweep: a reply whose "success" was not a dict aborted the sweep. - lag: a member of unknown speed (-1) lowered the LAG's speed. ## docs: bring the README and docstrings in line with the code The firewall worked example showed the runtime stubs the README forbids; the class table missed PhoneDriver and MediaDriver; ConfigLifecycleMixin is mixed in by drivers, not by a role base; services, mac_acl and packages named a class, a key and an exception that do not exist. --- .github/workflows/ci.yml | 6 +++++- README.md | 28 ++++++++++++++----------- napalm_device_types/_ucd_metrics.py | 16 +++++++------- napalm_device_types/base.py | 2 +- napalm_device_types/channel.py | 16 ++++++++++---- napalm_device_types/container_engine.py | 12 +++++------ napalm_device_types/host_status.py | 23 +++++++++++++++----- napalm_device_types/kernel.py | 5 ++++- napalm_device_types/lag.py | 2 +- napalm_device_types/listening.py | 8 ++++--- napalm_device_types/mac_acl.py | 2 +- napalm_device_types/package_updates.py | 2 +- napalm_device_types/packages.py | 4 ---- napalm_device_types/ping_sweep.py | 2 ++ napalm_device_types/services.py | 2 +- napalm_device_types/ssh_server.py | 11 +++++++--- pyproject.toml | 1 + tests/test_channel.py | 17 +++++++++++++++ tests/test_host_status.py | 7 +++++++ tests/test_kernel_facts.py | 8 +++++++ tests/test_lag_interfaces.py | 9 ++++++++ tests/test_listening_sockets.py | 1 + tests/test_package_updates.py | 10 +++++++++ tests/test_ping_sweep.py | 7 +++++++ tests/test_ssh_server.py | 14 +++++++++++++ 25 files changed, 163 insertions(+), 52 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9a9a6db..372e184 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,7 +12,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.10", "3.11", "3.12"] + python-version: ["3.10", "3.11", "3.12", "3.13"] steps: - name: Checkout uses: actions/checkout@v4 @@ -28,6 +28,10 @@ jobs: python -m pip install --upgrade pip python -m pip install -e ".[dev]" + - name: Type-check + run: | + python -m mypy napalm_device_types + - name: Run unit tests run: | python -m pytest -q --tb=short diff --git a/README.md b/README.md index 75cb466..d82f9f3 100644 --- a/README.md +++ b/README.md @@ -183,14 +183,15 @@ Concretely: matching/comparison/reconciliation algorithms, orchestration flows, generic data shapes belong here. Only the actual device communication belongs in `vendor/napalm-`. -**Worked example — firewall rule diff/apply** (`FirewallDriver`): +**Worked example — firewall rule diff/apply** (`FirewallRuleMixin`, in `FirewallDriver`): ```python -class FirewallDriver(DeviceTypeDriver): - # Abstract — every driver implements its own device communication. - def get_firewall_rules(self) -> List[FirewallRuleDict]: raise NotImplementedError - def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: raise NotImplementedError - def commit_firewall_rules(self) -> Dict[str, Any]: raise NotImplementedError +class FirewallRuleMixin: # mixed into FirewallDriver + # Declared only — every driver implements its own device communication. + if TYPE_CHECKING: + def get_firewall_rules(self) -> List[FirewallRuleDict]: ... + def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: ... + def commit_firewall_rules(self) -> Dict[str, Any]: ... # Concrete — the matching/comparison/orchestration algorithm is identical # for every firewall vendor, so it lives here once. @@ -242,7 +243,7 @@ detection vs. presentation there. pip install napalm-device-types ``` -Requires Python ≥ 3.9 and NAPALM ≥ 4.0. +Requires Python ≥ 3.10 (tested up to 3.13) and NAPALM ≥ 4.0. ## Available base classes @@ -255,12 +256,15 @@ Requires Python ≥ 3.9 and NAPALM ≥ 4.0. | `OSDriver` | General-purpose operating systems | Linux, BSD, macOS | | `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS | | `ResidentialGatewayDriver` | Router + firewall + AP in one box | OpenWrt, FritzBox | +| `PhoneDriver` | IP telephony endpoints | Yealink | +| `MediaDriver` | Speakers, streamers, media renderers | Sonos, Chromecast | -Mixins mixed into the classes above rather than used on their own: -`ConfigLifecycleMixin` (config load/compare/commit/rollback), `PingSweepMixin` -(subnet sweeps), and `DhcpServerMixin` (static DHCP reservations — mixed into -`FirewallDriver` and `ResidentialGatewayDriver`, since both commonly run the -DHCP server for their networks). +Mixins that come with the classes above: `PingSweepMixin` (subnet sweeps, in +every driver through `DeviceTypeDriver`) and `DhcpServerMixin` (static DHCP +reservations — mixed into `FirewallDriver` and `ResidentialGatewayDriver`, +since both commonly run the DHCP server for their networks). +`ConfigLifecycleMixin` (config load/compare/commit/rollback) is mixed in by a +driver itself. ## Usage diff --git a/napalm_device_types/_ucd_metrics.py b/napalm_device_types/_ucd_metrics.py index e67f0a8..67d03e2 100644 --- a/napalm_device_types/_ucd_metrics.py +++ b/napalm_device_types/_ucd_metrics.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio import re -from typing import Awaitable, Callable, Dict, Optional +from typing import Any, Awaitable, Callable, Dict, Optional # ── UCD-MIB OIDs ────────────────────────────────────────────────────────────── OID_SYS_UPTIME = "1.3.6.1.2.1.25.1.1.0" # hrSystemUptime (centiseconds) @@ -42,7 +42,7 @@ def ticks_to_seconds(raw: str | None) -> int | None: def build_if_metrics( - metrics: dict, + metrics: Dict[str, Any], descr: Dict[str, str], speed: Dict[str, str], in_oct: Dict[str, str], @@ -50,16 +50,16 @@ def build_if_metrics( in_err: Dict[str, str], out_err: Dict[str, str], *, - if_skip: re.Pattern | None = None, + if_skip: re.Pattern[str] | None = None, tx_err_is_drop: bool = False, ) -> None: if if_skip is None: if_skip = IF_SKIP_DEFAULT - interfaces: dict = {} + interfaces: Dict[str, Any] = {} for idx, name in descr.items(): if if_skip.match(name): continue - iface: dict = {"name": name} + iface: Dict[str, Any] = {"name": name} try: iface["rx_bytes"] = int(in_oct.get(idx, 0) or 0) iface["tx_bytes"] = int(out_oct.get(idx, 0) or 0) @@ -84,8 +84,8 @@ async def collect_ucd_metrics( snmp_walk: SnmpWalkFn, *, tx_err_is_drop: bool = False, - if_skip: re.Pattern | None = None, -) -> dict: + if_skip: re.Pattern[str] | None = None, +) -> Dict[str, Any]: """Collect UCD-MIB system metrics + IF-MIB interface counters in parallel.""" ( uptime_raw, cpu_idle_raw, @@ -116,7 +116,7 @@ async def collect_ucd_metrics( ), ) - metrics: dict = {} + metrics: Dict[str, Any] = {} secs = ticks_to_seconds(uptime_raw) if secs is not None: diff --git a/napalm_device_types/base.py b/napalm_device_types/base.py index f6f8766..01b8115 100644 --- a/napalm_device_types/base.py +++ b/napalm_device_types/base.py @@ -49,7 +49,7 @@ class PortSpec(NamedTuple): mandatory: bool = False -class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, CommandChannelMixin, NetworkDriver): +class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, CommandChannelMixin, NetworkDriver): # type: ignore[misc] """Common base for all netOrk device-type drivers. Sits between napalm.base.NetworkDriver and the type-specific abstract diff --git a/napalm_device_types/channel.py b/napalm_device_types/channel.py index cb007ca..6b534c9 100644 --- a/napalm_device_types/channel.py +++ b/napalm_device_types/channel.py @@ -28,7 +28,7 @@ from __future__ import annotations import socket import time -from typing import Any, NamedTuple, Optional, Protocol, TYPE_CHECKING +from typing import Any, NamedTuple, Optional, Protocol, Tuple, TYPE_CHECKING, cast #: How much of a stream's stderr is kept. Enough to tell "permission denied" #: from "no such command"; a stream that writes megabytes there keeps its tail. @@ -111,7 +111,7 @@ class ParamikoExecStream: def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes: self._channel.settimeout(timeout) try: - return self._channel.recv(max_bytes) + return cast(bytes, self._channel.recv(max_bytes)) except socket.timeout: raise TimeoutError(f"no data within {timeout}s") from None finally: @@ -130,7 +130,7 @@ class ParamikoExecStream: def exit_status(self) -> Optional[int]: if not self._channel.exit_status_ready(): return None - return self._channel.recv_exit_status() + return cast(int, self._channel.recv_exit_status()) @property def stderr(self) -> str: @@ -138,7 +138,7 @@ class ParamikoExecStream: return self._stderr.decode("utf-8", "replace") -def _collect(channel: Any, command: str, timeout: float) -> tuple: +def _collect(channel: Any, command: str, timeout: float) -> Tuple[bytes, bytes]: out, err = b"", b"" deadline = time.monotonic() + timeout while True: @@ -147,6 +147,14 @@ def _collect(channel: Any, command: str, timeout: float) -> tuple: elif channel.recv_stderr_ready(): err += channel.recv_stderr(_READ) elif channel.exit_status_ready(): + # sshd may send the exit status before the last output: read both to EOF. + try: + while chunk := channel.recv(_READ): + out += chunk + while chunk := channel.recv_stderr(_READ): + err += chunk + except socket.timeout: + raise TimeoutError(f"{command!r} did not finish within {timeout}s") from None return out, err elif time.monotonic() > deadline: raise TimeoutError(f"{command!r} did not finish within {timeout}s") diff --git a/napalm_device_types/container_engine.py b/napalm_device_types/container_engine.py index e41ac3f..265940a 100644 --- a/napalm_device_types/container_engine.py +++ b/napalm_device_types/container_engine.py @@ -29,7 +29,7 @@ stays truthful. from __future__ import annotations import shlex -from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING +from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING, cast from napalm_device_types.channel import ByteStream, CommandResult from napalm_device_types.models import ContainerEngineDict @@ -62,7 +62,7 @@ class ContainerEngineConnection: def open_api(self) -> ByteStream: """A stream to the engine's API: HTTP/1.1 over ``system dial-stdio``.""" - return self._driver.open_stream(self._command(["system", "dial-stdio"])) + return cast(ByteStream, self._driver.open_stream(self._command(["system", "dial-stdio"]))) def run_cli( self, @@ -77,9 +77,9 @@ class ContainerEngineConnection: *privileged* runs it as root, the way the driver gains root for any command: for a call the engine refused to the login user. """ - return self._driver.run_command( + return cast(CommandResult, self._driver.run_command( self._command(args), privileged=privileged, timeout=timeout, stdin=stdin - ) + )) def stream_cli( self, args: Sequence[str], *, merge_stderr: bool = False, privileged: bool = False @@ -90,9 +90,9 @@ class ContainerEngineConnection: progress there (``compose up``); *privileged* as for :meth:`run_cli`. """ command = self._command(args) - return self._driver.open_stream( + return cast(ByteStream, self._driver.open_stream( f"{command} 2>&1" if merge_stderr else command, privileged=privileged - ) + )) class ContainerEngineMixin: diff --git a/napalm_device_types/host_status.py b/napalm_device_types/host_status.py index 82f4ea9..ebf5a2b 100644 --- a/napalm_device_types/host_status.py +++ b/napalm_device_types/host_status.py @@ -157,18 +157,31 @@ def kernel_reboot_pending(running: str, installed: List[str]) -> Optional[str]: running one; otherwise None. The flavour is what follows the last ``-`` (``generic``, ``amd64``, ``pve``, - ``v8``); a kernel of another flavour is never a reason to reboot. + ``v8``), or on an RPM kernel (``5.14.0-427.13.1.el9_4.x86_64``) the arch + after the release; a kernel of another flavour is never a reason to reboot. """ - flavour = running.rsplit("-", 1)[-1] - same = [k for k in installed if k.rsplit("-", 1)[-1] == flavour] + version, flavour = _kernel_flavour(running) + same = [k for k in installed if _kernel_flavour(k)[1] == flavour] if not same: return None - newest = max(same, key=lambda k: _version_key(k.rsplit("-", 1)[0])) - if _version_key(newest.rsplit("-", 1)[0]) > _version_key(running.rsplit("-", 1)[0]): + newest = max(same, key=lambda k: _version_key(_kernel_flavour(k)[0])) + if _version_key(_kernel_flavour(newest)[0]) > _version_key(version): return newest return None +def _kernel_flavour(release: str) -> Tuple[str, str]: + """``6.8.0-87-generic`` -> ``("6.8.0-87", "generic")``, + ``5.14.0-427.13.1.el9_4.x86_64`` -> ``("5.14.0-427.13.1.el9_4", "x86_64")``.""" + base, sep, tail = release.rpartition("-") + if not sep: + return release, "" + if tail[:1].isdigit() and "." in tail: + rel, _, arch = tail.rpartition(".") + return f"{base}-{rel}", arch + return base, tail + + def _freebsd_reboot(sections: Dict[str, List[str]]) -> Optional[Tuple[bool, Optional[str]]]: """FreeBSD's answer from freebsd-version, or None on any other system.""" installed = (sections.get("freebsd-kernel") or [""])[0] diff --git a/napalm_device_types/kernel.py b/napalm_device_types/kernel.py index 5fd019c..463c2c4 100644 --- a/napalm_device_types/kernel.py +++ b/napalm_device_types/kernel.py @@ -70,9 +70,12 @@ def _report(output: str) -> str: raise ValueError("no kernel facts in the output") from None try: packed = base64.b64decode("".join(lines[start + 1 : end]), validate=True) - return gzip.decompress(packed).decode() + report = gzip.decompress(packed).decode() except (binascii.Error, OSError, EOFError, zlib.error, UnicodeDecodeError) as exc: raise ValueError(f"the kernel facts could not be decoded: {exc}") from exc + if "[release]" not in report.splitlines(): + raise ValueError("the kernel facts are empty: gzip or base64 is missing on the host") + return report def _sections(report: str) -> Dict[str, List[str]]: diff --git a/napalm_device_types/lag.py b/napalm_device_types/lag.py index d3a850c..1d715d0 100644 --- a/napalm_device_types/lag.py +++ b/napalm_device_types/lag.py @@ -46,7 +46,7 @@ def add_lag_interfaces( "is_enabled": any(interfaces[m].get("is_enabled") for m in members), "description": f"LAG ({', '.join(members)})", "last_flapped": -1.0, - "speed": sum(float(interfaces[m].get("speed") or 0) for m in members), + "speed": sum(max(float(interfaces[m].get("speed") or 0), 0.0) for m in members), "mtu": -1, "mac_address": "", "lag_members": members, diff --git a/napalm_device_types/listening.py b/napalm_device_types/listening.py index 68c5240..73a625e 100644 --- a/napalm_device_types/listening.py +++ b/napalm_device_types/listening.py @@ -110,13 +110,15 @@ def _sections(lines: List[str]) -> Dict[str, List[str]]: def _split_local(local: str) -> Optional[Tuple[str, Optional[str], int]]: - """``[fe80::1%eth0]:546`` -> ``("fe80::1", "eth0", 546)``; None if no port.""" + """``[fe80::1]%eth0:546`` (or ``[fe80::1%eth0]:546``) -> ``("fe80::1", "eth0", 546)``; + None if no port.""" host, sep, port = local.rpartition(":") if not sep or not port.isdigit(): return None - if host.startswith("[") and host.endswith("]"): - host = host[1:-1] address, _, zone = host.partition("%") + zone = zone.rstrip("]") + if address.startswith("["): + address = address[1:].rstrip("]") return address or "*", zone or None, int(port) diff --git a/napalm_device_types/mac_acl.py b/napalm_device_types/mac_acl.py index 942988e..a84e3cb 100644 --- a/napalm_device_types/mac_acl.py +++ b/napalm_device_types/mac_acl.py @@ -24,7 +24,7 @@ class MacAclMixin: Keys are SSID names. Each value contains: - * ssid (string) - SSID name (repeated for convenience) + * name (string) - SSID name (repeated for convenience) * policy (string) - ACL mode: * ``"allow"`` – whitelist: only listed MACs may associate diff --git a/napalm_device_types/package_updates.py b/napalm_device_types/package_updates.py index 4c3e4ce..7ca4697 100644 --- a/napalm_device_types/package_updates.py +++ b/napalm_device_types/package_updates.py @@ -43,7 +43,7 @@ def _joined_lines(output: str) -> List[str]: lines: List[str] = [] for line in output.splitlines(): if line.startswith(" ") and lines: - lines[-1] += line.strip() + lines[-1] += line[1:].rstrip() else: lines.append(line) return lines diff --git a/napalm_device_types/packages.py b/napalm_device_types/packages.py index 4d83f99..4a8f552 100644 --- a/napalm_device_types/packages.py +++ b/napalm_device_types/packages.py @@ -68,7 +68,6 @@ class PackageManagementMixin: :param name: Package name as known to the package manager. :param version: Exact version to install. An empty string (default) installs the latest available version. - :raises NotImplementedError: If the driver does not support package management. :raises ValueError: If the package name is unknown or the version does not exist in any configured feed. :raises RuntimeError: If the installation fails on the device side @@ -88,7 +87,6 @@ class PackageManagementMixin: The method blocks until the removal is complete. :param name: Package name to remove. - :raises NotImplementedError: If the driver does not support package management. :raises ValueError: If the package is not currently installed. :raises RuntimeError: If the removal fails on the device side (e.g. other packages depend on it). @@ -105,7 +103,6 @@ class PackageManagementMixin: dictionary. The structure is package-specific. :param name: Package name. - :raises NotImplementedError: If the driver does not support package management. :raises ValueError: If the package is not installed. Example:: @@ -140,7 +137,6 @@ class PackageManagementMixin: :param name: Package name. :param config: New configuration as a nested dictionary. - :raises NotImplementedError: If the driver does not support package management. :raises ValueError: If the package is not installed or the configuration contains invalid values. :raises RuntimeError: If the device rejects the configuration. diff --git a/napalm_device_types/ping_sweep.py b/napalm_device_types/ping_sweep.py index 10e19b3..ddd92c0 100644 --- a/napalm_device_types/ping_sweep.py +++ b/napalm_device_types/ping_sweep.py @@ -150,6 +150,8 @@ class PingSweepMixin: return {"ip": destination, "alive": False, "rtt_ms": None, "error": error} success = reply.get("success") or {} + if not isinstance(success, dict): + return {"ip": destination, "alive": False, "rtt_ms": None, "error": "malformed ping reply"} probes_sent = _as_int(success.get("probes_sent")) packet_loss = _as_int(success.get("packet_loss"), default=probes_sent) alive = bool(success.get("results")) or probes_sent > packet_loss diff --git a/napalm_device_types/services.py b/napalm_device_types/services.py index 20de39d..1c8a490 100644 --- a/napalm_device_types/services.py +++ b/napalm_device_types/services.py @@ -2,7 +2,7 @@ """Init-system services: listing them and starting/stopping them. Deliberately *not* the same thing as a NAS's exported shares, which live on -``StorageServiceCapability`` as ``get_storage_services``. Those two used to +``StorageDriver`` as ``get_storage_services``. Those two used to share the name ``get_services`` with incompatible return types, which is why a QNAP -- storage and OS at once -- could not satisfy both. diff --git a/napalm_device_types/ssh_server.py b/napalm_device_types/ssh_server.py index bb58a98..9b18673 100644 --- a/napalm_device_types/ssh_server.py +++ b/napalm_device_types/ssh_server.py @@ -418,9 +418,14 @@ class OpenSshServerMixin: def back(message: str, reload: bool) -> SshHardeningResult: say(message) say("Putting the previous configuration back") - restore() - if reload: - self._reload_ssh_server() + try: + restore() + if reload: + self._reload_ssh_server() + except Exception as exc: # noqa: BLE001 - report it, the caller must look + result.message = f"{message}; putting the previous configuration back failed: {exc}" + say(result.message) + return result result.rolled_back = True result.message = message return result diff --git a/pyproject.toml b/pyproject.toml index 2ea5663..9d792ec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,7 @@ classifiers = [ "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", "Topic :: System :: Networking", "Topic :: System :: Systems Administration", "Typing :: Typed", diff --git a/tests/test_channel.py b/tests/test_channel.py index 0096b9b..83d23c2 100644 --- a/tests/test_channel.py +++ b/tests/test_channel.py @@ -167,3 +167,20 @@ def test_close_closes_the_channel(): ParamikoExecStream(ch).close() assert ch.closed + + +def test_run_keeps_output_that_arrives_after_the_exit_status(): + """sshd may send exit-status before the last stdout bytes.""" + + class EarlyExit(FakeChannel): + def exit_status_ready(self): + return True + + def recv_ready(self): + return False + + channel = EarlyExit(stdout=[b"head ", b"tail"], stderr=[b"warn"]) + + result = run_on_transport(FakeTransport(channel), "cmd") + + assert result == CommandResult("head tail", "warn", 0) diff --git a/tests/test_host_status.py b/tests/test_host_status.py index d646117..e29e66d 100644 --- a/tests/test_host_status.py +++ b/tests/test_host_status.py @@ -131,6 +131,13 @@ class TestKernelRebootPending: ("7.0.14-19-pve", ["7.0.14-19-pve", "7.0.2-6-pve"], None), # Arch: the running kernel's modules were replaced by the upgrade. ("6.10.5-arch1-1", ["6.10.9-arch1-1"], "6.10.9-arch1-1"), + # RHEL/Fedora: the release changes with every update, the arch does not. + ( + "5.14.0-427.13.1.el9_4.x86_64", + ["5.14.0-427.13.1.el9_4.x86_64", "5.14.0-503.11.1.el9_5.x86_64"], + "5.14.0-503.11.1.el9_5.x86_64", + ), + ("6.9.7-200.fc40.x86_64", ["6.9.7-200.fc40.x86_64"], None), ], ) def test_the_newer_kernel_of_the_running_flavour(self, running, installed, newer): diff --git a/tests/test_kernel_facts.py b/tests/test_kernel_facts.py index c402676..0543faf 100644 --- a/tests/test_kernel_facts.py +++ b/tests/test_kernel_facts.py @@ -85,6 +85,14 @@ class TestParsing: with pytest.raises(ValueError): parse_kernel_facts("sh: gzip: not found\n") + def test_an_empty_payload_raises(self): + """gzip or base64 missing on the host: the frame is there, the report is not.""" + with pytest.raises(ValueError): + parse_kernel_facts("KFACTS_BEGIN\nKFACTS_END\n") + empty = base64.b64encode(gzip.compress(b"")).decode() + with pytest.raises(ValueError): + parse_kernel_facts(f"KFACTS_BEGIN\n{empty}\nKFACTS_END\n") + def test_a_damaged_payload_raises(self): with pytest.raises(ValueError): parse_kernel_facts("KFACTS_BEGIN\nnot base64 at all!\nKFACTS_END\n") diff --git a/tests/test_lag_interfaces.py b/tests/test_lag_interfaces.py index 1707c2d..852655c 100644 --- a/tests/test_lag_interfaces.py +++ b/tests/test_lag_interfaces.py @@ -74,3 +74,12 @@ def test_does_not_modify_its_input(): add_lag_interfaces(ifaces) assert list(ifaces) == ["3"] + + +def test_an_unknown_member_speed_does_not_make_the_lag_negative(): + ifaces = { + "3": _port(speed=-1.0, trunk_group="Trk3"), + "4": _port(speed=1000.0, trunk_group="Trk3"), + } + + assert add_lag_interfaces(ifaces)["Trk3"]["speed"] == 1000.0 diff --git a/tests/test_listening_sockets.py b/tests/test_listening_sockets.py index 86b228f..b67954f 100644 --- a/tests/test_listening_sockets.py +++ b/tests/test_listening_sockets.py @@ -84,6 +84,7 @@ class TestParsing: ("0.0.0.0%ens18:68", "0.0.0.0", "ens18"), ("[::ffff:127.0.0.1]:8125", "::ffff:127.0.0.1", None), ("[fe80::1%eth0]:546", "fe80::1", "eth0"), + ("[fe80::1]%eth0:546", "fe80::1", "eth0"), # what iproute2 ss prints (":::22", "::", None), # iproute2 4.9 prints IPv6 without brackets ], ) diff --git a/tests/test_package_updates.py b/tests/test_package_updates.py index 8a10b2c..b4c575d 100644 --- a/tests/test_package_updates.py +++ b/tests/test_package_updates.py @@ -81,6 +81,16 @@ class TestAptUpgradable: "3.0.13-0ubuntu3.5" ) + def test_a_line_wrapped_at_a_blank_keeps_the_blank(self): + wrapped = ( + "openssl/noble-updates 3.0.13-0ubuntu3.6\n" + " amd64 [upgradable from: 3.0.13-0ubuntu3.5]\n__APT_RC=0\n" + ) + + assert _by_name(parse_apt_upgradable(wrapped))["openssl"]["current_version"] == ( + "3.0.13-0ubuntu3.5" + ) + def test_one_package_for_several_architectures_is_one_entry(self): multiarch = ( "libc6/noble-updates 2.39-0ubuntu8.5 amd64 [upgradable from: 2.39-0ubuntu8.4]\n" diff --git a/tests/test_ping_sweep.py b/tests/test_ping_sweep.py index db2e4d9..90bef7d 100644 --- a/tests/test_ping_sweep.py +++ b/tests/test_ping_sweep.py @@ -219,3 +219,10 @@ def test_ping_sweep_stops_when_stop_requested(): assert result["scanned"] == 1 assert len(driver.calls) == 1 + + +@pytest.mark.parametrize("success", [True, "yes", ["x"]]) +def test_a_malformed_success_is_no_answer_not_an_abort(success): + entry = PingSweepMixin._parse_ping_reply("192.0.2.1", {"success": success}) + + assert entry == {"ip": "192.0.2.1", "alive": False, "rtt_ms": None, "error": "malformed ping reply"} diff --git a/tests/test_ssh_server.py b/tests/test_ssh_server.py index d6bc338..d91a974 100644 --- a/tests/test_ssh_server.py +++ b/tests/test_ssh_server.py @@ -335,3 +335,17 @@ def test_restoring_an_earlier_dropin_leaks_nothing_either(): host.files[DROPIN_PATH] = "# netOrk\nHostKeyAlgorithms ssh-ed25519\n" host.harden_ssh_server({"mac": ["hmac-sha1"]}, verify=lambda: False) assert host.files[DROPIN_PATH] == "# netOrk\nHostKeyAlgorithms ssh-ed25519\n" + + +def test_mixin_reports_a_failed_way_back_instead_of_raising(): + class ReloadFails(FakeHost): + def run_command(self, command, **kwargs): + if "reload" in command or "HUP" in command: + return CommandResult("", "Failed to reload ssh.service", 1) + return super().run_command(command, **kwargs) + + host = ReloadFails() + result = host.harden_ssh_server({"mac": ["hmac-sha1"]}) + + assert not result.applied and not result.rolled_back + assert "putting the previous configuration back failed" in result.message