Merge branch 'main' into feat/apt-proxy-verify

The NAPALM driver standard landed on main (#52, 4.21.1). apt_proxy follows
it: builtin generics, strict mypy, complexity 10 at most (parse_apt_proxy,
same_proxy_url and the comment-out step split into helpers), 100 % covered.
Its TypedDicts live in apt_proxy.py, since models.py is over 1000 lines and
may not grow. Version 4.22.0 with its CHANGELOG section.

tests/quality_baseline.json

@@ -0,0 +1,39 @@
+{
+ "coverage_floor": 95.0,
+ "function_complexity": {
+  "napalm_device_types/_ucd_metrics.py::build_if_metrics": 20,
+  "napalm_device_types/_ucd_metrics.py::collect_ucd_metrics": 17,
+  "napalm_device_types/host_firewall.py::_iptables_rule": 11,
+  "napalm_device_types/host_firewall.py::_module_match": 13,
+  "napalm_device_types/host_firewall.py::_module_options": 14,
+  "napalm_device_types/host_firewall.py::_nft_field": 15,
+  "napalm_device_types/host_firewall.py::_nft_match": 12,
+  "napalm_device_types/host_firewall.py::_nft_rule": 16,
+  "napalm_device_types/host_firewall.py::_parse_nft": 14,
+  "napalm_device_types/host_firewall.py::rule_text": 14,
+  "napalm_device_types/host_status.py::_reboot": 12,
+  "napalm_device_types/lag.py::add_lag_interfaces": 11,
+  "napalm_device_types/listening.py::parse_listening_sockets": 13,
+  "napalm_device_types/service_clusters.py::_patroni": 11,
+  "napalm_device_types/ssh_server.py::plan_ssh_hardening": 38,
+  "napalm_device_types/virtual_ips.py::_keepalived": 12
+ },
+ "function_lines": {
+  "napalm_device_types/_ucd_metrics.py::collect_ucd_metrics": 119,
+  "napalm_device_types/dhcp.py::DhcpServerMixin.diff_dhcp_reservations": 54,
+  "napalm_device_types/dhcp.py::DhcpServerMixin.diff_dhcp_subnets": 60,
+  "napalm_device_types/firewall_rules.py::FirewallRuleMixin.diff_firewall_rules": 56,
+  "napalm_device_types/hypervisor.py::HypervisorDriver.create_vm_from_cloud_init": 84,
+  "napalm_device_types/hypervisor.py::HypervisorDriver.get_vm_config": 77,
+  "napalm_device_types/ssh_server.py::OpenSshServerMixin.harden_ssh_server": 52,
+  "napalm_device_types/ssh_server.py::plan_ssh_hardening": 60,
+  "napalm_device_types/storage.py::StorageDriver.get_volumes": 52,
+  "napalm_device_types/switch.py::SwitchDriver.get_dot1x_ports": 51,
+  "napalm_device_types/switch.py::SwitchDriver.get_poe_status": 63
+ },
+ "mypy_relaxed": [],
+ "python_file_lines": {
+  "napalm_device_types/models.py": 1504
+ },
+ "untyped_public": []
+}

tests/test_apt_proxy.py

@@ -19,6 +19,7 @@ from napalm_device_types.apt_proxy import (
     APT_PROXY_COMMAND,
     AptProxyMixin,
     apt_fetch_command,
+    comment_out_command,
     parse_apt_fetch,
     parse_apt_proxy,
     same_proxy_url,
@@ -108,9 +109,7 @@ class TestWhereItIsSet:
         ]
 
     def test_a_nested_setting_is_named_but_not_editable(self):
-        report = parse_apt_proxy(
-            _report(parts=[f'{_D}01proxy:2:    Proxy "http://nested:3142";'])
-        )
+        report = parse_apt_proxy(_report(parts=[f'{_D}01proxy:2:    Proxy "http://nested:3142";']))
 
         assert report["settings"] == [
             {"file": f"{_D}01proxy", "line": 2, "value": "http://nested:3142", "editable": False}
@@ -255,7 +254,9 @@ class TestFetch:
 
 class TestFetchCommand:
     def test_past_the_proxy_to_the_url_itself(self):
-        command = apt_fetch_command("http://c:3142/acng-report.html", direct=True, expect="Cache NG")
+        command = apt_fetch_command(
+            "http://c:3142/acng-report.html", direct=True, expect="Cache NG"
+        )
 
         assert "'Acquire::http::Proxy::c=DIRECT'" in command
         assert "grep -c -F -- 'Cache NG'" in command
@@ -271,7 +272,10 @@ class TestFetchCommand:
 
 
 def test_the_commands_are_valid_posix_sh():
-    for command in (APT_PROXY_COMMAND, apt_fetch_command("http://c:3142/x", direct=True, expect="y")):
+    for command in (
+        APT_PROXY_COMMAND,
+        apt_fetch_command("http://c:3142/x", direct=True, expect="y"),
+    ):
         subprocess.run(["sh", "-n", "-c", command], check=True)
 
 
@@ -473,3 +477,57 @@ class TestConfigurePackageProxy:
 
         assert result["success"] is False
         assert host.files == {}
+
+
+class TestWhatTheParserSkips:
+    def test_a_line_that_is_no_grep_output(self):
+        assert (
+            parse_apt_proxy(_report(parts=["grep: /etc/apt/apt.conf.d/x: Permission denied"]))[
+                "settings"
+            ]
+            == []
+        )
+
+    def test_a_dump_line_of_another_key(self):
+        report = parse_apt_proxy(_report(dump=['Acquire::http::Timeout "10";']))
+        assert report["proxy"] is None
+
+    def test_a_fetch_report_without_an_exit_status_raises(self):
+        with pytest.raises(ValueError, match="exit status"):
+            parse_apt_fetch("APROXY_BEGIN\n[fetch]\nGet:1 x\nAPROXY_END\n")
+
+
+class _Plain(AptProxyMixin):
+    """A driver that says nothing about its package manager: apt is assumed."""
+
+
+def test_without_its_own_answer_a_host_is_an_apt_host():
+    assert _Plain()._is_apt_host() is True
+
+
+class TestNotAnAptHost:
+    def test_no_check(self):
+        with pytest.raises(NotImplementedError):
+            _Host(apt=False).check_package_proxy("http://c:3142/x")
+
+    def test_no_removal(self):
+        result = _Host(apt=False).configure_package_proxy("")
+        assert result["success"] is False
+        assert "not an apt host" in result["output"]
+
+
+def test_a_comment_out_that_fails_stops_the_fix():
+    host = _Host({f"{_D}90curtin-aptproxy": 'Acquire::http::Proxy "http://old:3128";\n'})
+    host._run_apt_root_command = lambda command, done: {"ok": False, "output": "sed: denied"}
+
+    result = host.configure_package_proxy("http://c:3142")
+
+    assert result["success"] is False
+    assert result["steps"][-1] == {"step": "conflicts", "ok": False, "output": "sed: denied"}
+
+
+def test_the_comment_out_command():
+    assert comment_out_command("/etc/apt/apt.conf", [2, 4]) == (
+        "sed -i -e '2{/^[[:space:]]*Acquire::http::Proxy[[:space:]]/Is|^|// |}' "
+        "-e '4{/^[[:space:]]*Acquire::http::Proxy[[:space:]]/Is|^|// |}' /etc/apt/apt.conf"
+    )

tests/test_channel.py

@@ -8,8 +8,6 @@ paramiko channel is faked, so the tests pin behaviour, not paramiko.
 
 from __future__ import annotations
 
-import socket
-
 import pytest
 
 from napalm_device_types.channel import (
@@ -56,7 +54,7 @@ class FakeChannel:
         if self.out:
             return self.out.pop(0)
         if self.hang:
-            raise socket.timeout()
+            raise TimeoutError()
         return b""
 
     def recv_stderr_ready(self):

tests/test_container_engine.py

@@ -32,7 +32,9 @@ class FakeDriver(ContainerEngineMixin):
     def run_command(self, command, *, privileged=False, timeout=60, stdin=None):
         self.commands.append((command, privileged, timeout, stdin))
         found = any(command == f"command -v {b}" for b in self._binaries())
-        return CommandResult(stdout="/usr/bin/x\n" if found else "", stderr="", exit_code=0 if found else 1)
+        return CommandResult(
+            stdout="/usr/bin/x\n" if found else "", stderr="", exit_code=0 if found else 1
+        )
 
     def open_stream(self, command, *, privileged=False):
         self.streams.append((command, privileged))
@@ -86,7 +88,9 @@ def test_cli_arguments_are_quoted_and_prefixed_by_the_binary():
 def test_cli_stdin_and_timeout_reach_the_channel():
     driver = FakeDriver()
 
-    driver.open_container_engine("docker").run_cli(["login", "--password-stdin"], stdin=b"x", timeout=30)
+    driver.open_container_engine("docker").run_cli(
+        ["login", "--password-stdin"], stdin=b"x", timeout=30
+    )
 
     _command, _privileged, timeout, stdin = driver.commands[-1]
     assert (timeout, stdin) == (30, b"x")

tests/test_dhcp_diff_apply.py

@@ -9,9 +9,10 @@ principle: generic vs. device-specific logic" for why this logic lives here
 and not in a vendor driver.
 """
 
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import pytest
+
 from napalm_device_types import DhcpServerMixin, FirewallDriver, ResidentialGatewayDriver
 from napalm_device_types.dhcp import normalize_mac
 from napalm_device_types.models import DhcpReservationDict
@@ -33,21 +34,21 @@ def _reservation(**overrides: Any) -> DhcpReservationDict:
 class _FakeDhcpServer(DhcpServerMixin):
     """In-memory fake — no network, no Kea/UCI specifics."""
 
-    def __init__(self, live: Optional[List[DhcpReservationDict]] = None) -> None:
+    def __init__(self, live: list[DhcpReservationDict] | None = None) -> None:
         self.live = live or []
-        self.applied: List[Dict[str, Any]] = []
+        self.applied: list[dict[str, Any]] = []
         self.commits = 0
 
-    def get_dhcp_reservations(self) -> List[DhcpReservationDict]:
+    def get_dhcp_reservations(self) -> list[DhcpReservationDict]:
         return self.live
 
     def apply_dhcp_reservation(
-        self, reservation: DhcpReservationDict, *, uuid: Optional[str] = None
-    ) -> Dict[str, Any]:
+        self, reservation: DhcpReservationDict, *, uuid: str | None = None
+    ) -> dict[str, Any]:
         self.applied.append({"reservation": reservation, "uuid": uuid})
         return {"success": True}
 
-    def commit_dhcp_reservations(self) -> Dict[str, Any]:
+    def commit_dhcp_reservations(self) -> dict[str, Any]:
         self.commits += 1
         return {"success": True}
 

tests/test_dhcp_ha_config_sync.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """Two facts an HA pair of servers reports about itself.
 
 `get_dhcp_ha` is what a DHCP server's own HA configuration says: the peers it

tests/test_dhcp_subnet_diff_apply.py

@@ -11,9 +11,10 @@ A subnet is a heavier object than a reservation: a wrong `pools` or
 below lean on the never-delete and preserve-unmanaged-options guarantees.
 """
 
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import pytest
+
 from napalm_device_types import DhcpServerMixin
 from napalm_device_types.models import DhcpSubnetDict
 
@@ -35,21 +36,21 @@ def _subnet(**overrides: Any) -> DhcpSubnetDict:
 
 
 class _FakeDhcpServer(DhcpServerMixin):
-    def __init__(self, live: Optional[List[DhcpSubnetDict]] = None) -> None:
+    def __init__(self, live: list[DhcpSubnetDict] | None = None) -> None:
         self.live = live or []
-        self.applied: List[Dict[str, Any]] = []
+        self.applied: list[dict[str, Any]] = []
         self.commits = 0
 
-    def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
+    def get_dhcp_subnets(self) -> list[DhcpSubnetDict]:
         return self.live
 
     def apply_dhcp_subnet(
-        self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
-    ) -> Dict[str, Any]:
+        self, subnet: DhcpSubnetDict, *, uuid: str | None = None
+    ) -> dict[str, Any]:
         self.applied.append({"subnet": subnet, "uuid": uuid})
         return {"success": True}
 
-    def commit_dhcp_subnets(self) -> Dict[str, Any]:
+    def commit_dhcp_subnets(self) -> dict[str, Any]:
         self.commits += 1
         return {"success": True}
 
@@ -215,9 +216,7 @@ class _BareDriver(DhcpServerMixin):
     pass
 
 
-@pytest.mark.parametrize(
-    "method", ["get_dhcp_subnets", "apply_dhcp_subnet", "commit_dhcp_subnets"]
-)
+@pytest.mark.parametrize("method", ["get_dhcp_subnets", "apply_dhcp_subnet", "commit_dhcp_subnets"])
 def test_device_specific_methods_are_absent_until_implemented(method: str) -> None:
     """Declared under TYPE_CHECKING, so they do not exist until a driver adds them."""
     assert not hasattr(_BareDriver, method)

tests/test_fact_getters.py

@@ -31,7 +31,7 @@ def test_a_missing_license_may_say_what_the_device_withholds_without_it():
     # firmware inventory until DCMS is activated: partly licensed, not unlicensed.
     hints = get_type_hints(LicenseDict)
     # Python 3.10 keeps NotRequired[...] around the type; later versions strip it.
-    assert "List[str]" in str(hints["withholds"])
+    assert "list[str]" in str(hints["withholds"])
     assert "withholds" not in LicenseDict.__required_keys__
 
 

tests/test_fingerprint_base.py

@@ -3,21 +3,21 @@
 from __future__ import annotations
 
 import pytest
+from napalm.base import NetworkDriver
+
 from napalm_device_types import (
     AccessPointDriver,
     DeviceTypeDriver,
     FingerprintRule,
     FirewallDriver,
-    rules_max_weight,
     HypervisorDriver,
     OSDriver,
     PortSpec,
     ResidentialGatewayDriver,
     StorageDriver,
     SwitchDriver,
+    rules_max_weight,
 )
-from napalm.base import NetworkDriver
-
 
 # ── Hierarchy ─────────────────────────────────────────────────────────────────
 
@@ -49,6 +49,7 @@ def test_device_type_driver_inherits_network_driver():
 
 # ── Default attributes ────────────────────────────────────────────────────────
 
+
 def test_defaults_on_device_type_driver():
     assert DeviceTypeDriver.VENDOR == ""
     assert DeviceTypeDriver.DRIVER_NAME == ""
@@ -72,6 +73,7 @@ def test_fingerprint_attributes_inherited(cls):
 
 # ── FingerprintRule ───────────────────────────────────────────────────────────
 
+
 def test_fingerprint_rule_defaults():
     r = FingerprintRule("fritz!box")
     assert r.pattern == "fritz!box"
@@ -102,6 +104,7 @@ def test_fingerprint_rule_group():
 
 # ── rules_max_weight: the most a set of rules can add ─────────────────────────
 
+
 def test_max_weight_sums_rules_that_can_match_together():
     rules = [FingerprintRule("proxmox virtual environment", 9.0), FingerprintRule("pve", 2.0)]
     assert rules_max_weight(rules) == 11.0
@@ -146,6 +149,7 @@ def test_fingerprint_rule_is_immutable():
 
 # ── PortSpec ──────────────────────────────────────────────────────────────────
 
+
 def test_port_spec_defaults():
     p = PortSpec("https", 8006)
     assert p.scheme == "https"
@@ -169,6 +173,7 @@ def test_port_spec_is_immutable():
 
 # ── Concrete driver subclass ──────────────────────────────────────────────────
 
+
 def test_concrete_driver_overrides_fingerprint():
     class MyDriver(FirewallDriver):
         VENDOR = "Acme"

tests/test_firewall_diff_apply.py

@@ -9,9 +9,10 @@ device-specific logic" for why this logic lives here and not in a vendor
 driver.
 """
 
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import pytest
+
 from napalm_device_types import FirewallDriver
 from napalm_device_types.models import FirewallRuleDict
 
@@ -39,21 +40,21 @@ def _rule(**overrides: Any) -> FirewallRuleDict:
 class _FakeFirewall(FirewallDriver):
     """In-memory fake — no network, no OPNsense/vendor specifics."""
 
-    def __init__(self, live_rules: Optional[List[FirewallRuleDict]] = None) -> None:
+    def __init__(self, live_rules: list[FirewallRuleDict] | None = None) -> None:
         self.live_rules = live_rules or []
-        self.applied: List[Dict[str, Any]] = []
+        self.applied: list[dict[str, Any]] = []
         self.committed = False
 
-    def get_firewall_rules(self) -> List[FirewallRuleDict]:
+    def get_firewall_rules(self) -> list[FirewallRuleDict]:
         return self.live_rules
 
     def apply_firewall_rule(
-        self, rule: FirewallRuleDict, *, uuid: Optional[str] = None
-    ) -> Dict[str, Any]:
+        self, rule: FirewallRuleDict, *, uuid: str | None = None
+    ) -> dict[str, Any]:
         self.applied.append({"rule": rule, "uuid": uuid})
         return {"success": True}
 
-    def commit_firewall_rules(self) -> Dict[str, Any]:
+    def commit_firewall_rules(self) -> dict[str, Any]:
         self.committed = True
         return {"success": True}
 

tests/test_fix_templates.py

@@ -17,12 +17,12 @@ from napalm_device_types import (
     ManagementControllerDriver,
     MediaDriver,
     OSDriver,
+    PackageProxyMixin,
     PhoneDriver,
     ResidentialGatewayDriver,
+    SnmpAgentMixin,
     StorageDriver,
     SwitchDriver,
-    PackageProxyMixin,
-    SnmpAgentMixin,
     run_fix_steps,
 )
 
@@ -97,7 +97,12 @@ class _Agent(SnmpAgentMixin):
 
 
 class TestSnmpAgent:
-    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.0.0.5"], "contact": "ops", "location": "rack 1"}
+    SETTINGS = {
+        "community": "c0mm",
+        "allowed_sources": ["10.0.0.5"],
+        "contact": "ops",
+        "location": "rack 1",
+    }
 
     def test_install_configure_open_and_start_with_the_caller_s_values(self):
         agent = _Agent()
@@ -153,7 +158,10 @@ class TestEnsurePackage:
 
     def test_an_install_that_reports_failure_fails(self):
         driver = _Packages(installed=[])
-        driver.install_package = lambda name, version="": {"success": False, "output": "E: no candidate"}
+        driver.install_package = lambda name, version="": {
+            "success": False,
+            "output": "E: no candidate",
+        }
 
         result = driver.ensure_package("lldpd")
 
@@ -164,7 +172,12 @@ class TestEnsurePackage:
 @pytest.mark.parametrize("base", ROLE_BASES, ids=lambda b: b.__name__)
 @pytest.mark.parametrize(
     "method",
-    ["configure_snmp_agent", "ensure_package", "configure_package_proxy", "grant_container_engine_access"],
+    [
+        "configure_snmp_agent",
+        "ensure_package",
+        "configure_package_proxy",
+        "grant_container_engine_access",
+    ],
 )
 def test_no_role_base_offers_a_fix(base, method):
     assert getattr(base, method, None) is None

tests/test_host_status.py

@@ -452,7 +452,10 @@ class TestBusyCheck:
 
     UBUNTU_IDLE = [
         ("systemd", "/sbin/init"),
-        ("unattended-upgr", "/usr/bin/python3 /usr/share/unattended-upgrades/unattended-upgrade-shutdown --wait-for-signal"),
+        (
+            "unattended-upgr",
+            "/usr/bin/python3 /usr/share/unattended-upgrades/unattended-upgrade-shutdown --wait-for-signal",
+        ),
         ("sshd", "sshd: netork [priv]"),
     ]
 

tests/test_lag_interfaces.py

@@ -3,7 +3,9 @@
 from napalm_device_types import add_lag_interfaces
 
 
-def _port(is_up: bool = True, is_enabled: bool = True, speed: float = 1000.0, trunk_group: str = "") -> dict:
+def _port(
+    is_up: bool = True, is_enabled: bool = True, speed: float = 1000.0, trunk_group: str = ""
+) -> dict:
     port = {
         "is_up": is_up,
         "is_enabled": is_enabled,

tests/test_listening_sockets.py

@@ -89,7 +89,7 @@ class TestParsing:
         ],
     )
     def test_every_address_form(self, local: str, address: str, interface):
-        line = f"tcp LISTEN 0 128 {local} *:* users:((\"x\",pid=1,fd=3))\n"
+        line = f'tcp LISTEN 0 128 {local} *:* users:(("x",pid=1,fd=3))\n'
 
         [socket] = parse_listening_sockets(_wire(line, "1 0::/system.slice/x.service\n"))
 
@@ -188,7 +188,7 @@ class TestFailures:
 
     def test_a_report_cut_short_raises(self):
         with pytest.raises(ValueError):
-            parse_listening_sockets(_wire()[:-len("SOCK_END\n")])
+            parse_listening_sockets(_wire()[: -len("SOCK_END\n")])
 
     def test_a_host_without_ss_or_netstat_is_unavailable(self):
         with pytest.raises(ListeningSocketsUnavailable):

tests/test_ping_sweep.py

@@ -5,7 +5,6 @@ from napalm.base import NetworkDriver
 
 from napalm_device_types import DeviceTypeDriver, PingSweepMixin, driver_supports_ping
 
-
 # ── Fakes ─────────────────────────────────────────────────────────────────────
 
 
@@ -225,4 +224,9 @@ def test_ping_sweep_stops_when_stop_requested():
 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"}
+    assert entry == {
+        "ip": "192.0.2.1",
+        "alive": False,
+        "rtt_ms": None,
+        "error": "malformed ping reply",
+    }

tests/test_port_forwards.py

@@ -19,7 +19,7 @@ from napalm_device_types.nat_vpn import NatVpnMixin
 def test_a_firewall_and_a_gateway_share_the_declaration():
     assert issubclass(FirewallDriver, NatVpnMixin)
     assert issubclass(ResidentialGatewayDriver, NatVpnMixin)
-    assert "def get_port_forwards(self) -> List[PortForwardDict]" in inspect.getsource(NatVpnMixin)
+    assert "def get_port_forwards(self) -> list[PortForwardDict]" in inspect.getsource(NatVpnMixin)
 
 
 def test_it_is_declared_once():

tests/test_provisioning.py

@@ -4,6 +4,7 @@ the same for every hypervisor (NetOrk/netork#794)."""
 from __future__ import annotations
 
 import pytest
+
 from napalm_device_types import HypervisorDriver
 from napalm_device_types.provisioning import (
     QEMU_GUEST_AGENTS,

tests/test_role_contracts.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """The contract that makes multi-role drivers safe.
 
 A role base declares what a device of that kind can be asked for. It must not

tests/test_send_wake_on_lan.py

@@ -9,6 +9,7 @@ sibling base.
 """
 
 import pytest
+
 from napalm_device_types import FirewallDriver
 
 

tests/test_service_clusters.py

@@ -35,14 +35,29 @@ SECRET = "s3cr3t-do-not-leak"
 
 def _wire(*sections: str, systemd: bool = True, docker: str = "[docker]", noise: str = "") -> str:
     head = "[systemd]" if systemd else "[no-systemd]"
-    body = "\n".join([head, *sections[:1], docker, *sections[1:]]) if sections else f"{head}\n{docker}"
+    body = (
+        "\n".join([head, *sections[:1], docker, *sections[1:]]) if sections else f"{head}\n{docker}"
+    )
     return f"{noise}SVCC_BEGIN\n{body}\nSVCC_END\n"
 
 
 DB01_CLUSTER = {
     "members": [
-        {"name": "db-01", "role": "leader", "state": "running", "host": "10.7.234.12", "port": 5432},
-        {"name": "db-02", "role": "sync_standby", "state": "streaming", "host": "10.7.234.11", "port": 5432, "lag": 0},
+        {
+            "name": "db-01",
+            "role": "leader",
+            "state": "running",
+            "host": "10.7.234.12",
+            "port": 5432,
+        },
+        {
+            "name": "db-02",
+            "role": "sync_standby",
+            "state": "streaming",
+            "host": "10.7.234.11",
+            "port": 5432,
+            "lag": 0,
+        },
     ],
     "scope": "netork-db",
 }
@@ -79,10 +94,24 @@ class TestPatroni:
                 "cluster": "netork-db",
                 "running": True,
                 "members": [
-                    {"name": "db-01", "address": "10.7.234.12", "port": 5432, "role": "leader",
-                     "state": "running", "this_host": True, "config_from": None},
-                    {"name": "db-02", "address": "10.7.234.11", "port": 5432, "role": "sync_standby",
-                     "state": "streaming", "this_host": False, "config_from": None},
+                    {
+                        "name": "db-01",
+                        "address": "10.7.234.12",
+                        "port": 5432,
+                        "role": "leader",
+                        "state": "running",
+                        "this_host": True,
+                        "config_from": None,
+                    },
+                    {
+                        "name": "db-02",
+                        "address": "10.7.234.11",
+                        "port": 5432,
+                        "role": "sync_standby",
+                        "state": "streaming",
+                        "this_host": False,
+                        "config_from": None,
+                    },
                 ],
             }
         ]
@@ -91,8 +120,11 @@ class TestPatroni:
         reading = parse_service_clusters(
             _wire(
                 _patroni(
-                    "active active", "scope 'netork-db'  # the cluster", "name db-01",
-                    "restapi 10.7.234.12:8008", "rest-unreachable",
+                    "active active",
+                    "scope 'netork-db'  # the cluster",
+                    "name db-01",
+                    "restapi 10.7.234.12:8008",
+                    "rest-unreachable",
                 )
             )
         )
@@ -100,8 +132,15 @@ class TestPatroni:
         (cluster,) = reading["patroni"]
         assert cluster["cluster"] == "netork-db"
         assert cluster["members"] == [
-            {"name": "db-01", "address": "10.7.234.12", "port": None, "role": None,
-             "state": None, "this_host": True, "config_from": None}
+            {
+                "name": "db-01",
+                "address": "10.7.234.12",
+                "port": None,
+                "role": None,
+                "state": None,
+                "this_host": True,
+                "config_from": None,
+            }
         ]
 
     def test_an_unreadable_config_is_not_looked_at(self):
@@ -112,7 +151,9 @@ class TestPatroni:
         assert reading["patroni"] is None
 
     def test_a_config_without_a_scope_declares_no_cluster(self):
-        assert parse_service_clusters(_wire(_patroni("active active", "name db-01")))["patroni"] == []
+        assert (
+            parse_service_clusters(_wire(_patroni("active active", "name db-01")))["patroni"] == []
+        )
 
     def test_a_host_without_systemd_is_not_looked_at(self):
         assert parse_service_clusters(_wire(systemd=False))["patroni"] is None
@@ -127,8 +168,10 @@ class TestAdguardhomeSync:
             _wire(
                 "",
                 _container(
-                    "adguardhome-sync", "ghcr.io/bakito/adguardhome-sync:v0.9.3",
-                    "ORIGIN_URL http://10.9.100.5:3000", "REPLICA1_URL http://10.9.100.6:3000",
+                    "adguardhome-sync",
+                    "ghcr.io/bakito/adguardhome-sync:v0.9.3",
+                    "ORIGIN_URL http://10.9.100.5:3000",
+                    "REPLICA1_URL http://10.9.100.6:3000",
                     "REPLICA2_URL https://dns-03.example.org",
                 ),
             )
@@ -136,9 +179,13 @@ class TestAdguardhomeSync:
 
         (cluster,) = reading["adguardhome_sync"]
         assert (cluster["mechanism"], cluster["service"], cluster["cluster"]) == (
-            "adguardhome-sync", "adguardhome", "10.9.100.5:3000",
+            "adguardhome-sync",
+            "adguardhome",
+            "10.9.100.5:3000",
         )
-        assert [(m["address"], m["port"], m["role"], m["config_from"]) for m in cluster["members"]] == [
+        assert [
+            (m["address"], m["port"], m["role"], m["config_from"]) for m in cluster["members"]
+        ] == [
             ("10.9.100.5", 3000, "origin", None),
             ("10.9.100.6", 3000, "replica", 0),
             ("dns-03.example.org", 443, "replica", 0),
@@ -147,8 +194,15 @@ class TestAdguardhomeSync:
 
     def test_a_single_replica_url_counts(self):
         reading = parse_service_clusters(
-            _wire("", _container("sync", "bakito/adguardhome-sync", "ORIGIN_URL http://a:3000",
-                                 "REPLICA_URL http://b:3000"))
+            _wire(
+                "",
+                _container(
+                    "sync",
+                    "bakito/adguardhome-sync",
+                    "ORIGIN_URL http://a:3000",
+                    "REPLICA_URL http://b:3000",
+                ),
+            )
         )
 
         assert [m["address"] for m in reading["adguardhome_sync"][0]["members"]] == ["a", "b"]
@@ -163,36 +217,68 @@ class TestAdguardhomeSync:
 class TestBunkerweb:
     def test_a_worker_takes_its_configuration_from_the_manager_it_admits(self):
         reading = parse_service_clusters(
-            _wire("", _container("bunkerweb", "bunkerity/bunkerweb@sha256:2026", "API_WHITELIST_IP 10.50.100.5"))
+            _wire(
+                "",
+                _container(
+                    "bunkerweb", "bunkerity/bunkerweb@sha256:2026", "API_WHITELIST_IP 10.50.100.5"
+                ),
+            )
         )
 
         (cluster,) = reading["bunkerweb"]
         assert (cluster["service"], cluster["cluster"]) == ("bunkerweb", "10.50.100.5")
         assert cluster["members"] == [
-            {"name": None, "address": None, "port": 5000, "role": "worker", "state": None,
-             "this_host": True, "config_from": 1},
-            {"name": None, "address": "10.50.100.5", "port": None, "role": "manager", "state": None,
-             "this_host": False, "config_from": None},
+            {
+                "name": None,
+                "address": None,
+                "port": 5000,
+                "role": "worker",
+                "state": None,
+                "this_host": True,
+                "config_from": 1,
+            },
+            {
+                "name": None,
+                "address": "10.50.100.5",
+                "port": None,
+                "role": "manager",
+                "state": None,
+                "this_host": False,
+                "config_from": None,
+            },
         ]
 
     def test_the_default_whitelist_names_no_manager(self):
         """127.0.0.0/8 and a docker network are BunkerWeb's default, not a peer."""
         reading = parse_service_clusters(
-            _wire("", _container("bw", "bunkerity/bunkerweb:1.6", "API_WHITELIST_IP 127.0.0.0/8 10.20.30.0/24"))
+            _wire(
+                "",
+                _container(
+                    "bw", "bunkerity/bunkerweb:1.6", "API_WHITELIST_IP 127.0.0.0/8 10.20.30.0/24"
+                ),
+            )
         )
 
         assert reading["bunkerweb"] == []
 
     def test_a_manager_names_its_instances(self):
         reading = parse_service_clusters(
-            _wire("", _container("bw", "bunkerity/bunkerweb-all-in-one:1.6",
-                                 "BUNKERWEB_INSTANCES 10.50.100.6:5000 bw-03")),
+            _wire(
+                "",
+                _container(
+                    "bw",
+                    "bunkerity/bunkerweb-all-in-one:1.6",
+                    "BUNKERWEB_INSTANCES 10.50.100.6:5000 bw-03",
+                ),
+            ),
         )
 
         (cluster,) = reading["bunkerweb"]
         assert cluster["cluster"] is None  # the manager cannot name its own address
-        assert [(m["address"], m["port"], m["role"], m["this_host"], m["config_from"])
-                for m in cluster["members"]] == [
+        assert [
+            (m["address"], m["port"], m["role"], m["this_host"], m["config_from"])
+            for m in cluster["members"]
+        ] == [
             (None, None, "manager", True, None),
             ("10.50.100.6", 5000, "worker", False, 0),
             ("bw-03", 5000, "worker", False, 0),
@@ -200,7 +286,9 @@ class TestBunkerweb:
 
     def test_a_manager_without_instances_is_no_cluster(self):
         """proxy-01's all-in-one keeps its instances in its database."""
-        reading = parse_service_clusters(_wire("", _container("bw", "bunkerity/bunkerweb-all-in-one@sha256:2b34")))
+        reading = parse_service_clusters(
+            _wire("", _container("bw", "bunkerity/bunkerweb-all-in-one@sha256:2b34"))
+        )
 
         assert reading["bunkerweb"] == []
 
@@ -298,7 +386,7 @@ class _Host:
             "docker",
             'case "$1" in\n'
             f'  ps) {cat} "$d/containers" ;;\n'
-            '  inspect) for last; do :; done;\n'
+            "  inspect) for last; do :; done;\n"
             '    case "$3" in *Config.Image*) '
             f'{cat} "$d/image-$last" ;; *) {cat} "$d/env-$last" ;; esac ;;\n'
             "esac\n",
@@ -362,7 +450,9 @@ def host(request, tmp_path) -> _Host:
 class TestTheCommandOnAHost:
     def test_patroni_as_on_db_01(self, host):
         config = host.file("etc/patroni/config.yml", PATRONI_YML)
-        host.unit("patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config)))
+        host.unit(
+            "patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config))
+        )
         host.rest(DB01_CLUSTER)
 
         out = host.run()
@@ -379,7 +469,9 @@ class TestTheCommandOnAHost:
 
     def test_patroni_whose_rest_api_does_not_answer(self, host):
         config = host.file("etc/patroni/config.yml", PATRONI_YML)
-        host.unit("patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config)))
+        host.unit(
+            "patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config))
+        )
 
         (cluster,) = parse_service_clusters(host.run())["patroni"]
 
@@ -391,7 +483,9 @@ class TestTheCommandOnAHost:
     def test_a_patroni_config_it_cannot_read_is_not_looked_at(self, host):
         config = host.file("etc/patroni/config.yml", PATRONI_YML)
         config.chmod(0)
-        host.unit("patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config)))
+        host.unit(
+            "patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config))
+        )
 
         out = host.run()
 
@@ -400,14 +494,19 @@ class TestTheCommandOnAHost:
 
     def test_the_sync_and_a_worker_without_their_secrets(self, host):
         host.container(
-            "adguardhome-sync", "ghcr.io/bakito/adguardhome-sync:v0.9.3",
-            "ORIGIN_URL=http://10.9.100.5:3000", f"ORIGIN_PASSWORD={SECRET}",
-            f"REPLICA1_URL=http://admin:{SECRET}@10.9.100.6:3000", f"REPLICA1_PASSWORD={SECRET}",
+            "adguardhome-sync",
+            "ghcr.io/bakito/adguardhome-sync:v0.9.3",
+            "ORIGIN_URL=http://10.9.100.5:3000",
+            f"ORIGIN_PASSWORD={SECRET}",
+            f"REPLICA1_URL=http://admin:{SECRET}@10.9.100.6:3000",
+            f"REPLICA1_PASSWORD={SECRET}",
             "CRON=*/5 * * * *",
         )
         host.container(
-            "bunkerweb", "bunkerity/bunkerweb@sha256:2026",
-            "API_WHITELIST_IP=10.50.100.5", f"DATABASE_URI=mariadb+pymysql://bw:{SECRET}@db/bw",
+            "bunkerweb",
+            "bunkerity/bunkerweb@sha256:2026",
+            "API_WHITELIST_IP=10.50.100.5",
+            f"DATABASE_URI=mariadb+pymysql://bw:{SECRET}@db/bw",
         )
         host.container("zabbix", "zabbix/zabbix-agent2:alpine-latest", f"ZBX_PSK={SECRET}")
 
@@ -417,7 +516,8 @@ class TestTheCommandOnAHost:
         assert "admin" not in out
         reading = parse_service_clusters(out)
         assert [m["address"] for m in reading["adguardhome_sync"][0]["members"]] == [
-            "10.9.100.5", "10.9.100.6",
+            "10.9.100.5",
+            "10.9.100.6",
         ]
         assert reading["bunkerweb"][0]["cluster"] == "10.50.100.5"
         assert reading["patroni"] == []
@@ -483,8 +583,13 @@ class TestTheSource:
 
     def test_a_container_by_its_name(self):
         reading = parse_service_clusters(
-            _wire("", _container("adguardhome-sync", "bakito/adguardhome-sync", "ORIGIN_URL http://a:3000"),
-                  _container("bw", "bunkerity/bunkerweb", "API_WHITELIST_IP 10.0.0.5"))
+            _wire(
+                "",
+                _container(
+                    "adguardhome-sync", "bakito/adguardhome-sync", "ORIGIN_URL http://a:3000"
+                ),
+                _container("bw", "bunkerity/bunkerweb", "API_WHITELIST_IP 10.0.0.5"),
+            )
         )
 
         assert reading["adguardhome_sync"][0]["source"] == "adguardhome-sync"

tests/test_ssh_host_keys.py

@@ -252,7 +252,7 @@ class TestNetmiko:
                         other = paramiko.ECDSAKey.generate()
                         raise paramiko.BadHostKeyException("10.0.0.5", host_key, other)
                     except paramiko.SSHException as exc:
-                        raise RuntimeError(f"A paramiko SSHException occurred: {exc}")
+                        raise RuntimeError(f"A paramiko SSHException occurred: {exc}") from exc
                 except RuntimeError as exc:
                     raise ConnectionException(str(exc)) from exc
         assert caught.value.observation.seen == SshHostKey.of(host_key)
@@ -261,7 +261,9 @@ class TestNetmiko:
 
     def test_any_other_failure_passes_unchanged(self):
         with pytest.raises(ConnectionException, match="timed out"):
-            with HostKeyCheck({KNOWN_KEYS_ARG: {"22": ED25519.as_dict()}}, "10.0.0.5", 22).netmiko({}):
+            with HostKeyCheck({KNOWN_KEYS_ARG: {"22": ED25519.as_dict()}}, "10.0.0.5", 22).netmiko(
+                {}
+            ):
                 raise ConnectionException("timed out")
 
     def test_observe_reports_the_key_netmiko_saw(self, host_key):

tests/test_ssh_server.py

@@ -9,7 +9,6 @@ cannot get through -- runs without a machine.
 from __future__ import annotations
 
 import re
-from typing import Dict, List, Optional
 
 import pytest
 
@@ -69,7 +68,11 @@ def test_parse_reads_the_four_lists_in_server_order():
     eff = parse_sshd_effective(SSHD_T)
     assert eff["kex"][:2] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"]
     assert eff["hostkey"] == ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512", "rsa-sha2-256"]
-    assert eff["cipher"] == ["chacha20-poly1305@openssh.com", "aes128-ctr", "aes256-gcm@openssh.com"]
+    assert eff["cipher"] == [
+        "chacha20-poly1305@openssh.com",
+        "aes128-ctr",
+        "aes256-gcm@openssh.com",
+    ]
     assert eff["mac"][-1] == "hmac-sha2-256"
 
 
@@ -82,7 +85,9 @@ def test_parse_refuses_output_without_the_lists():
 
 
 def test_plan_removes_what_was_asked_and_keeps_the_order():
-    plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1", "umac-64-etm@openssh.com"]})
+    plan = plan_ssh_hardening(
+        parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1", "umac-64-etm@openssh.com"]}
+    )
     assert plan.remove == {"mac": ["umac-64-etm@openssh.com", "hmac-sha1"]}
     assert plan.keep["mac"] == ["hmac-sha2-256-etm@openssh.com", "hmac-sha2-256"]
     assert plan.changed_classes == ["mac"]
@@ -113,20 +118,38 @@ def test_plan_keeps_one_algorithm_the_managing_client_still_speaks():
     know would leave the server reachable, but not by netOrk."""
     plan = plan_ssh_hardening(
         parse_sshd_effective(SSHD_T),
-        {"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384",
-                 "diffie-hellman-group14-sha256"]},
+        {
+            "kex": [
+                "curve25519-sha256@libssh.org",
+                "ecdh-sha2-nistp256",
+                "ecdh-sha2-nistp384",
+                "diffie-hellman-group14-sha256",
+            ]
+        },
         client=PARAMIKO,
     )
     assert "curve25519-sha256@libssh.org" in plan.keep["kex"]
-    assert [(r.algorithm, r.reason) for r in plan.refused] == [("curve25519-sha256@libssh.org", "client")]
-    assert plan.remove["kex"] == ["ecdh-sha2-nistp256", "ecdh-sha2-nistp384", "diffie-hellman-group14-sha256"]
+    assert [(r.algorithm, r.reason) for r in plan.refused] == [
+        ("curve25519-sha256@libssh.org", "client")
+    ]
+    assert plan.remove["kex"] == [
+        "ecdh-sha2-nistp256",
+        "ecdh-sha2-nistp384",
+        "diffie-hellman-group14-sha256",
+    ]
 
 
 def test_plan_without_a_client_list_only_guards_the_last_algorithm():
     plan = plan_ssh_hardening(
         parse_sshd_effective(SSHD_T),
-        {"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384",
-                 "diffie-hellman-group14-sha256"]},
+        {
+            "kex": [
+                "curve25519-sha256@libssh.org",
+                "ecdh-sha2-nistp256",
+                "ecdh-sha2-nistp384",
+                "diffie-hellman-group14-sha256",
+            ]
+        },
     )
     assert plan.refused == []
     assert plan.keep["kex"] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"]
@@ -171,12 +194,18 @@ def test_plan_removes_a_host_key_algorithm_when_a_usable_one_stays():
 def test_plan_leaves_host_keys_alone_when_the_keys_are_unknown():
     plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"hostkey": ["ecdsa-sha2-nistp256"]})
     assert plan.changed_classes == []
-    assert [(r.algorithm, r.reason) for r in plan.refused] == [("ecdsa-sha2-nistp256", "host_keys_unknown")]
+    assert [(r.algorithm, r.reason) for r in plan.refused] == [
+        ("ecdsa-sha2-nistp256", "host_keys_unknown")
+    ]
 
 
 def test_host_key_algorithms_from_ssh_keygen():
     assert host_key_algorithms("\n".join(KEYGEN.values())) == {
-        "ecdsa-sha2-nistp256", "ssh-ed25519", "rsa-sha2-512", "rsa-sha2-256", "ssh-rsa",
+        "ecdsa-sha2-nistp256",
+        "ssh-ed25519",
+        "rsa-sha2-512",
+        "rsa-sha2-256",
+        "ssh-rsa",
     }
     assert host_key_algorithms("521 SHA256:x no comment (ECDSA)") == {"ecdsa-sha2-nistp521"}
     assert host_key_algorithms("") is None
@@ -204,11 +233,15 @@ def test_include_is_found_only_in_the_global_section():
     assert has_dropin_include("Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n")
     assert has_dropin_include("  include   /etc/ssh/sshd_config.d/*.conf\n")
     assert not has_dropin_include("Port 22\n")
-    assert not has_dropin_include("Port 22\nMatch User x\n  Include /etc/ssh/sshd_config.d/*.conf\n")
+    assert not has_dropin_include(
+        "Port 22\nMatch User x\n  Include /etc/ssh/sshd_config.d/*.conf\n"
+    )
 
 
 def test_main_config_gets_the_lists_first_and_the_old_lines_disabled():
-    main = "# defaults\nPort 22\nMACs hmac-sha1,hmac-sha2-256\nMatch User backup\n  MACs hmac-sha1\n"
+    main = (
+        "# defaults\nPort 22\nMACs hmac-sha1,hmac-sha2-256\nMatch User backup\n  MACs hmac-sha1\n"
+    )
     text = render_main_config(main, {"mac": ["hmac-sha2-256"]}, ["mac"])
     lines = text.splitlines()
     first_directive = next(line for line in lines if line and not line.startswith("#"))
@@ -227,28 +260,34 @@ def test_main_config_rewrite_is_idempotent():
 
 # ── the mixin, against a fake host ───────────────────────────────────────
 
-_DIRECTIVE = {"kexalgorithms": "kex", "hostkeyalgorithms": "hostkey", "ciphers": "cipher", "macs": "mac"}
+_DIRECTIVE = {
+    "kexalgorithms": "kex",
+    "hostkeyalgorithms": "hostkey",
+    "ciphers": "cipher",
+    "macs": "mac",
+}
 
 
 class FakeHost(OpenSshServerMixin):
     """A host with files, an sshd that reads them first-match-wins, and a log of what ran."""
 
-    def __init__(self, main: str = "Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n",
-                 sshd_t: str = SSHD_T) -> None:
-        self.files: Dict[str, str] = {MAIN_CONFIG: main}
-        self.modes: Dict[str, str] = {MAIN_CONFIG: "644"}
+    def __init__(
+        self, main: str = "Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n", sshd_t: str = SSHD_T
+    ) -> None:
+        self.files: dict[str, str] = {MAIN_CONFIG: main}
+        self.modes: dict[str, str] = {MAIN_CONFIG: "644"}
         self.sshd_t = sshd_t
         self.defaults = parse_sshd_effective(sshd_t)
         self.keys = {p: KEYGEN[p] for p in parse_sshd_hostkeys(sshd_t)}
         self.keygen_ok = True
-        self.ran: List[str] = []
+        self.ran: list[str] = []
         self.syntax_ok = True
         self.reloads = 0
 
-    def _effective(self) -> Dict[str, List[str]]:
+    def _effective(self) -> dict[str, list[str]]:
         eff = {k: list(v) for k, v in self.defaults.items()}
         seen = set()
-        lines: List[str] = []
+        lines: list[str] = []
         for line in self.files[MAIN_CONFIG].splitlines():
             if re.match(r"\s*include\s", line, re.I):
                 lines += self.files.get(DROPIN_PATH, "").splitlines()
@@ -269,8 +308,14 @@ class FakeHost(OpenSshServerMixin):
     #: or a cached timestamp): the password the driver put in front of the data.
     SUDO_PASSWORD = "hunter2"
 
-    def run_command(self, command: str, *, privileged: bool = False, timeout: float = 60,
-                    stdin: Optional[bytes] = None) -> CommandResult:
+    def run_command(
+        self,
+        command: str,
+        *,
+        privileged: bool = False,
+        timeout: float = 60,
+        stdin: bytes | None = None,
+    ) -> CommandResult:
         self.ran.append(command)
         if privileged and stdin is not None:
             stdin = f"{self.SUDO_PASSWORD}\n".encode() + stdin
@@ -294,14 +339,20 @@ class FakeHost(OpenSshServerMixin):
             if not self.keygen_ok:
                 return CommandResult("", "Permission denied", 1)
             paths = re.search(r"for k in (.*?); do", command).group(1).replace("'", "").split()
-            return CommandResult("".join(self.keys[p] + "\n" for p in paths if p in self.keys), "", 0)
+            return CommandResult(
+                "".join(self.keys[p] + "\n" for p in paths if p in self.keys), "", 0
+            )
         if command.startswith("sshd -T") or "sshd -T" in command:
             eff = self._effective()
             out = "".join(f"{d} {','.join(eff[k])}\n" for d, k in _DIRECTIVE.items())
-            out += "".join(line + "\n" for line in self.sshd_t.splitlines() if line.startswith("hostkey "))
+            out += "".join(
+                line + "\n" for line in self.sshd_t.splitlines() if line.startswith("hostkey ")
+            )
             return CommandResult(out, "", 0)
         if "sshd -t" in command:
-            return CommandResult("", "" if self.syntax_ok else "bad line 3", 0 if self.syntax_ok else 255)
+            return CommandResult(
+                "", "" if self.syntax_ok else "bad line 3", 0 if self.syntax_ok else 255
+            )
         m = re.match(r"cat -- '?([^' ]+)'?$", command)
         if m:
             path = m.group(1)
@@ -323,7 +374,7 @@ class FakeHost(OpenSshServerMixin):
 
 def test_mixin_writes_the_dropin_checks_it_and_reloads():
     host = FakeHost()
-    lines: List[str] = []
+    lines: list[str] = []
     result = host.harden_ssh_server({"mac": ["hmac-sha1"]}, client=PARAMIKO, log=lines.append)
     assert result.applied and not result.rolled_back
     assert result.path == DROPIN_PATH
@@ -439,7 +490,7 @@ def test_mixin_reports_a_failed_way_back_instead_of_raising():
 
 def test_mixin_refuses_to_lock_out_a_host_with_only_ed25519_and_rsa_keys():
     host = FakeHost(sshd_t=SSHD_T_FEW_KEYS)
-    lines: List[str] = []
+    lines: list[str] = []
     result = host.harden_ssh_server(
         {"hostkey": ["ssh-ed25519", "rsa-sha2-256", "rsa-sha2-512"]}, log=lines.append
     )

tests/test_system_identity.py

@@ -38,8 +38,20 @@ def test_a_supermicro_placeholder_serial_is_dropped():
 
 @pytest.mark.parametrize(
     "value",
-    ["", "None", "N/A", "Not Specified", "To be filled by O.E.M.", "Default string",
-     "System Serial Number", "0123456789", "1234567890", "0", "00000000", "Not Applicable"],
+    [
+        "",
+        "None",
+        "N/A",
+        "Not Specified",
+        "To be filled by O.E.M.",
+        "Default string",
+        "System Serial Number",
+        "0123456789",
+        "1234567890",
+        "0",
+        "00000000",
+        "Not Applicable",
+    ],
 )
 def test_placeholders(value):
     assert is_placeholder_identity(value)
@@ -47,9 +59,12 @@ def test_placeholders(value):
 
 @pytest.mark.parametrize(
     "uuid",
-    ["00000000-0000-0000-0000-000000000000", "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF",
-     # A UUID many cheap boards ship, the same on every one of them.
-     "03000200-0400-0500-0006-000700080009"],
+    [
+        "00000000-0000-0000-0000-000000000000",
+        "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF",
+        # A UUID many cheap boards ship, the same on every one of them.
+        "03000200-0400-0500-0006-000700080009",
+    ],
 )
 def test_placeholder_uuids(uuid):
     assert parse_system_identity(_out(uuid, "", ""))["system_uuid"] == ""
@@ -63,7 +78,10 @@ def test_real_values_are_not_placeholders():
 def test_unreadable_files_read_as_empty():
     # product_serial and product_uuid are root's alone; without root they come back empty.
     assert parse_system_identity(_out("", "", "BS1")) == {
-        "system_uuid": "", "product_serial": "", "board_serial": "BS1"}
+        "system_uuid": "",
+        "product_serial": "",
+        "board_serial": "BS1",
+    }
 
 
 def test_output_without_its_frame_is_not_a_reading():

tests/test_systemd_services.py

@@ -409,4 +409,4 @@ class TestSystemdServicesMixin:
 
     def test_not_every_os_driver_has_it(self):
         assert not hasattr(OSDriver, "manage_service")
-        assert callable(getattr(SystemdServicesMixin, "manage_service"))
+        assert callable(SystemdServicesMixin.manage_service)

tests/test_type_label.py

@@ -1,6 +1,7 @@
 """Tests for TYPE_LABEL class attribute on all base driver classes."""
 
 import pytest
+
 from napalm_device_types import (
     AccessPointDriver,
     FirewallDriver,
@@ -12,7 +13,6 @@ from napalm_device_types import (
     SwitchDriver,
 )
 
-
 BASE_CLASSES = [
     (AccessPointDriver, "Access Point"),
     (FirewallDriver, "Firewall"),

tests/test_ucd_metrics.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import asyncio
-from typing import Dict, Optional
 
 from napalm_device_types._ucd_metrics import (
     OID_IF_DESCR,
@@ -24,11 +23,11 @@ LEGACY = {
 }
 
 
-def _collect(tables: Dict[str, Dict[str, str]]) -> dict:
-    async def get(oid: str) -> Optional[str]:
+def _collect(tables: dict[str, dict[str, str]]) -> dict:
+    async def get(oid: str) -> str | None:
         return None
 
-    async def walk(oid: str) -> Dict[str, str]:
+    async def walk(oid: str) -> dict[str, str]:
         return tables.get(oid, {})
 
     return asyncio.run(collect_ucd_metrics(get, walk))
@@ -44,7 +43,11 @@ def test_the_64_bit_counters_and_high_speed_win():
 
     eth0 = _collect(tables)["interfaces"]["eth0"]
 
-    assert (eth0["rx_bytes"], eth0["tx_bytes"], eth0["speed_mbps"]) == (8589934592, 17179869184, 10000)
+    assert (eth0["rx_bytes"], eth0["tx_bytes"], eth0["speed_mbps"]) == (
+        8589934592,
+        17179869184,
+        10000,
+    )
 
 
 def test_an_agent_without_ifxtable_falls_back_to_the_32_bit_counters():

tests/test_virtual_ips.py

@@ -152,9 +152,7 @@ class TestVipManager:
         assert parse_virtual_ips(_wire(section))["vip_manager"] == []
 
     def test_a_template_unit_is_named_after_its_instance(self):
-        section = _vip_manager(
-            "active active", "cfg ip 10.0.0.10", unit="vip-manager@pg16.service"
-        )
+        section = _vip_manager("active active", "cfg ip 10.0.0.10", unit="vip-manager@pg16.service")
 
         [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
 
@@ -496,8 +494,10 @@ class TestTheCommandOnAHost:
         assert SECRET not in out
         assert "MASTER" not in out and "priority" not in out
         reading = parse_virtual_ips(out)
-        assert [(e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
-                for e in reading["keepalived"]] == [
+        assert [
+            (e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
+            for e in reading["keepalived"]
+        ] == [
             ("VI_WEB", "192.0.2.10", 24, "eth0", "vrid:51"),
             ("VI_DB", "192.0.2.20", 32, "eth1", "vrid:52"),
             ("VI_DB", "2001:db8::20", 64, "eth1", "vrid:52"),
@@ -593,8 +593,10 @@ class TestTheCommandOnAHost:
         out = host.run()
 
         assert SECRET not in out
-        assert [(e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
-                for e in parse_virtual_ips(out)["keepalived"]] == [
+        assert [
+            (e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
+            for e in parse_virtual_ips(out)["keepalived"]
+        ] == [
             ("VI_1", "192.0.2.50", 32, "eth0", None),
             ("VI_2", "192.0.2.60", 24, "eth1", "vrid:53"),
         ]

tests/test_vm_cpu_types.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 
 import ast
 import inspect
-from typing import List, get_type_hints
+from typing import get_type_hints
 
 import napalm_device_types.hypervisor as hypervisor_module
 from napalm_device_types import HypervisorDriver
@@ -33,7 +33,10 @@ def _declared(name: str) -> ast.FunctionDef:
 class TestCreateVmTakesACpuType:
     def test_cpu_type_is_an_optional_keyword(self):
         fn = _declared("create_vm_from_cloud_init")
-        kwonly = {arg.arg: default for arg, default in zip(fn.args.kwonlyargs, fn.args.kw_defaults)}
+        kwonly = {
+            arg.arg: default
+            for arg, default in zip(fn.args.kwonlyargs, fn.args.kw_defaults, strict=True)
+        }
         assert "cpu_type" in kwonly
         default = kwonly["cpu_type"]
         assert isinstance(default, ast.Constant) and default.value is None
@@ -52,7 +55,7 @@ class TestCpuTypeListing:
         assert get_type_hints(VMCpuTypeDict) == {
             "name": str,
             "description": str,
-            "features": List[str],
+            "features": list[str],
             "available": bool,
             "default": bool,
         }

tests/test_vm_identifiers.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 from typing import get_type_hints
 
 import pytest
+
 from napalm_device_types.models import (
     VMConfigDict,
     VMDict,