style: ruff format and the lint rules of the driver standard

Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM
driver uses (E, F, I, UP, B at line length 100). typing.List/Dict/Optional
become builtin generics and X | None, imports are sorted, the coding cookies
are gone. Two tests that read the source text follow the new spelling.

What ruff cannot fix comes in the next commit.
This commit is contained in:
christianmanivong committed 2026-10-22 19:03:21 +02:00
1 parent ac8e7c4c11
commit 08296d0d2e
77 files changed
+1329 -1074

No files matched your search

+1 -3
View File
@@ -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):
+6 -2
View File
@@ -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")
+8 -7
View File
@@ -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}
-1
View File
@@ -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
+9 -10
View File
@@ -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)
+1 -1
View File
@@ -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__
+8 -3
View File
@@ -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"
+8 -7
View File
@@ -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}
+18 -5
View File
@@ -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
+4 -1
View File
@@ -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]"),
]
+3 -1
View File
@@ -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,
+2 -2
View File
@@ -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):
+6 -2
View File
@@ -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",
}
+1 -1
View File
@@ -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():
+1
View File
@@ -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,
-1
View File
@@ -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
+1
View File
@@ -9,6 +9,7 @@ sibling base.
"""
import pytest
from napalm_device_types import FirewallDriver
+146 -41
View File
@@ -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"
+3 -1
View File
@@ -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):
+79 -28
View File
@@ -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
)
+24 -6
View File
@@ -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():
+1 -1
View File
@@ -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)
+1 -1
View File
@@ -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"),
+8 -5
View File
@@ -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():
+9 -7
View File
@@ -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"),
]
+2 -2
View File
@@ -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
@@ -52,7 +52,7 @@ class TestCpuTypeListing:
assert get_type_hints(VMCpuTypeDict) == {
"name": str,
"description": str,
"features": List[str],
"features": list[str],
"available": bool,
"default": bool,
}
+1
View File
@@ -12,6 +12,7 @@ from __future__ import annotations
from typing import get_type_hints
import pytest
from napalm_device_types.models import (
VMConfigDict,
VMDict,