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

@@ -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

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-<name>`.
 
-**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
 

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:

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

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

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:

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]

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

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,

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)
 
 

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

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

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.

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

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.
 

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

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",

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)

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

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

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

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

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"

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"}

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