diff --git a/napalm_device_types/access_point.py b/napalm_device_types/access_point.py index 94a1146..2e8a9d6 100644 --- a/napalm_device_types/access_point.py +++ b/napalm_device_types/access_point.py @@ -190,7 +190,8 @@ class AccessPointDriver( Keys are SSID names. Each value contains: * ssid (string) - SSID name (repeated for convenience) - * bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``) + * bridge (string) - bridge interface the VAP is attached to + (e.g. ``"br-lan"``, ``"br-guest"``) * vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation) * tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port * client_isolation (bool) - whether clients on this SSID are isolated from each other diff --git a/napalm_device_types/host_firewall.py b/napalm_device_types/host_firewall.py index 259f94b..3b03deb 100644 --- a/napalm_device_types/host_firewall.py +++ b/napalm_device_types/host_firewall.py @@ -166,6 +166,9 @@ def _sections(lines: list[str]) -> dict[str, tuple[list[str], int | None]]: # ── iptables-save ──────────────────────────────────────────────────────────── +#: The token that negates the option after it (``! -s 10.0.0.0/8``). +_NEGATION = "!" + #: Options that start a new part of a rule, outside any ``-m`` module. _TOP_LEVEL = frozenset( { @@ -244,13 +247,13 @@ def _module_options(module: str, rt: _RuleTokens, matches: list[NetfilterMatchDi unknown = module not in _KNOWN_MODULES while rt.more(): token = rt.peek() - if token in _TOP_LEVEL or (token == "!" and rt.peek(1) in _TOP_LEVEL): + if token in _TOP_LEVEL or (token == _NEGATION and rt.peek(1) in _TOP_LEVEL): break negated = False - if token == "!": + if token == _NEGATION: rt.take() negated = True - text.append("!") + text.append(_NEGATION) option = rt.take() value = rt.take() if rt.more() and not str(rt.peek()).startswith("-") else "" text += [option] + ([value] if value else []) @@ -266,7 +269,10 @@ def _module_options(module: str, rt: _RuleTokens, matches: list[NetfilterMatchDi def _module_match(module: str, option: str, value: str, negated: bool) -> Any: - """The match one module option makes; False for one that matches nothing new; None if not read.""" + """The match one module option makes. + + False for one that matches nothing new; None if not read. + """ if module == "comment": return False if module in ("tcp", "udp", "sctp", "udplite"): @@ -323,7 +329,7 @@ def _iptables_rule(line: str, chains: set[str]) -> tuple[str, NetfilterRuleDict] } while rt.more(): token = rt.take() - if token == "!": + if token == _NEGATION: negated = True continue if token in simple: diff --git a/napalm_device_types/hypervisor.py b/napalm_device_types/hypervisor.py index 8cec25b..5406f1d 100644 --- a/napalm_device_types/hypervisor.py +++ b/napalm_device_types/hypervisor.py @@ -330,7 +330,8 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri :param name: VM name or its ``vmid``. :param snapshot: Name of the snapshot to delete. :raises ValueError: If the VM or snapshot does not exist. - :raises RuntimeError: If other snapshots depend on this one (must delete children first). + :raises RuntimeError: If other snapshots depend on this one + (must delete children first). Example:: diff --git a/napalm_device_types/kernel.py b/napalm_device_types/kernel.py index bf31cca..629ddab 100644 --- a/napalm_device_types/kernel.py +++ b/napalm_device_types/kernel.py @@ -41,7 +41,8 @@ KERNEL_FACTS_COMMAND = ( "if [ -r $m/modules.builtin ]; then echo '[builtin]'; cat $m/modules.builtin; fi; " "if [ -r $m/modules.dep ]; then echo '[available]'; cut -d: -f1 $m/modules.dep; fi; " "if [ -r /boot/config-$r ]; then echo '[config]'; grep '^CONFIG_' /boot/config-$r; " - "elif [ -r /proc/config.gz ]; then echo '[config]'; zcat /proc/config.gz | grep '^CONFIG_'; fi; " + "elif [ -r /proc/config.gz ]; then echo '[config]'; " + "zcat /proc/config.gz | grep '^CONFIG_'; fi; " "} 2>/dev/null | gzip -c | base64; " "printf '%s%s\\n' KFACTS_ END" ) diff --git a/napalm_device_types/mac_acl.py b/napalm_device_types/mac_acl.py index a72e5d5..b1d20f9 100644 --- a/napalm_device_types/mac_acl.py +++ b/napalm_device_types/mac_acl.py @@ -43,15 +43,18 @@ class MacAclMixin: "name": "CorpWiFi", "policy": "allow", "entries": [ - {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"}, - {"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"}, + {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", + "description": "CEO-Laptop"}, + {"mac": "AA:BB:CC:DD:EE:02", "action": "allow", + "description": "CFO-Laptop"}, ], }, "GuestNet": { "name": "GuestNet", "policy": "deny", "entries": [ - {"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"}, + {"mac": "DE:AD:BE:EF:00:01", "action": "deny", + "description": "blocked device"}, ], }, } @@ -72,7 +75,8 @@ class MacAclMixin: Example:: - driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"]) + macs = ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"] + driver.push_mac_acl("CorpWiFi", "whitelist", macs) driver.push_mac_acl("GuestNet", "off", []) """ ... diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index e2ae9d0..6b4effc 100644 --- a/napalm_device_types/models.py +++ b/napalm_device_types/models.py @@ -1493,7 +1493,9 @@ class ManagementProtocolsDict(TypedDict): class SystemIdentityDict(TypedDict): - """A host's SMBIOS identity, what ties it to its BMC (:mod:`napalm_device_types.system_identity`). + """A host's SMBIOS identity, what ties it to its BMC. + + See :mod:`napalm_device_types.system_identity`. Each value is empty when unknown or a vendor placeholder.""" diff --git a/napalm_device_types/ssh_host_keys.py b/napalm_device_types/ssh_host_keys.py index c1ee214..a9d2df4 100644 --- a/napalm_device_types/ssh_host_keys.py +++ b/napalm_device_types/ssh_host_keys.py @@ -212,7 +212,7 @@ class HostKeyCheck: if self.expected is None: client.set_missing_host_key_policy(_AcceptFirst()) return - with self._known_hosts_file() as path: + with self._known_hosts_file(self.expected) as path: client.get_host_keys().load(path) client.set_missing_host_key_policy(_RefuseUnknown()) @@ -243,7 +243,7 @@ class HostKeyCheck: with self.refusing(): yield kwargs return - with self._known_hosts_file() as path: + with self._known_hosts_file(self.expected) as path: kwargs.update(ssh_strict=True, alt_host_keys=True, alt_key_file=path) with self.refusing(): yield kwargs @@ -285,13 +285,12 @@ class HostKeyCheck: ) @contextmanager - def _known_hosts_file(self) -> Iterator[str]: - """A private ``known_hosts`` file holding only the expected key.""" - assert self.expected is not None + def _known_hosts_file(self, expected: SshHostKey) -> Iterator[str]: + """A private ``known_hosts`` file holding only *expected*.""" fd, path = tempfile.mkstemp(prefix="known_hosts-") try: with os.fdopen(fd, "w") as f: - f.write(self.expected.known_hosts_line(self.host, self.port) + "\n") + f.write(expected.known_hosts_line(self.host, self.port) + "\n") yield path finally: try: diff --git a/napalm_device_types/ssh_server.py b/napalm_device_types/ssh_server.py index e8e33f7..a20c9db 100644 --- a/napalm_device_types/ssh_server.py +++ b/napalm_device_types/ssh_server.py @@ -369,7 +369,8 @@ class OpenSshServerMixin: stdin=content.encode(), ) tmp = staged.stdout.strip() - if staged.exit_code != 0 or not tmp.startswith("/tmp/netork-ssh."): + # A path on the device, made by mktemp under umask 077: not a local temp file. + if staged.exit_code != 0 or not tmp.startswith("/tmp/netork-ssh."): # nosec B108 raise RuntimeError( f"could not stage {path}: {(staged.stderr or staged.stdout).strip()}" ) @@ -378,7 +379,8 @@ class OpenSshServerMixin: # GNU stat, then BSD's; 644 for a file that does not exist yet. f'r={quote(mode_of or path)}; m=$(stat -c %a -- "$r" 2>/dev/null' f' || stat -f %Lp -- "$r" 2>/dev/null) || m=644;' - f' install -m "$m" -- {quote(tmp)} {quote(part)} && mv -f -- {quote(part)} {quote(path)};' + f' install -m "$m" -- {quote(tmp)} {quote(part)}' + f" && mv -f -- {quote(part)} {quote(path)};" f" s=$?; rm -f -- {quote(tmp)} {quote(part)}; exit $s" ) if result.exit_code != 0: diff --git a/napalm_device_types/storage.py b/napalm_device_types/storage.py index 1455221..74c62c4 100644 --- a/napalm_device_types/storage.py +++ b/napalm_device_types/storage.py @@ -63,7 +63,8 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver): * rpm (int) - rotational speed; ``0`` for SSD/NVMe * temperature (int) - current temperature in Celsius; ``-1`` if unavailable * health (string) - ``"healthy"``, ``"warning"``, ``"failed"``, or ``"unknown"`` - * pool (string) - name of the pool this disk belongs to; empty string if unassigned/spare + * pool (string) - name of the pool this disk belongs to; + empty string if unassigned/spare Example:: @@ -107,12 +108,14 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver): * type (string) - ``"zfs"``, ``"lvm"``, ``"md"``, ``"hardware-raid"``, ``"btrfs"`` * level (string) - RAID/redundancy level: ``"mirror"``, ``"raidz1"``, ``"raidz2"``, ``"raidz3"``, ``"stripe"``, ``"raid0"`` … ``"raid60"``, ``"single"`` - * status (string) - ``"online"``, ``"degraded"``, ``"faulted"``, ``"offline"``, ``"unknown"`` + * status (string) - ``"online"``, ``"degraded"``, ``"faulted"``, + ``"offline"``, ``"unknown"`` * total (int) - usable capacity in bytes * used (int) - used space in bytes * available (int) - free space in bytes * disks (list of strings) - slot identifiers of member drives - * auto_expand (bool) - whether the pool grows automatically when disks are replaced with larger ones + * auto_expand (bool) - whether the pool grows automatically when disks + are replaced with larger ones * dedup (bool) - whether deduplication is enabled * compression (string) - pool-level compression algorithm: ``"off"``, ``"lz4"``, ``"gzip"``, ``"zstd"``, etc. @@ -155,7 +158,8 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver): * used (int) - space currently used in bytes * available (int) - space available in bytes * mountpoint (string) - local mount path; empty string for block volumes / LUNs - * compression (string) - active compression algorithm (``"off"``, ``"lz4"``, ``"zstd"``, etc.) + * compression (string) - active compression algorithm + (``"off"``, ``"lz4"``, ``"zstd"``, etc.) * dedup (bool) - whether deduplication is active on this volume * readonly (bool) - whether the volume is mounted read-only * snapshots (int) - number of snapshots currently held @@ -254,7 +258,8 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver): Each entry contains: - * name (string) - snapshot name (e.g. ``"auto-2026-05-11"`` or full ``"tank/media@auto-2026-05-11"``) + * name (string) - snapshot name (e.g. ``"auto-2026-05-11"`` + or full ``"tank/media@auto-2026-05-11"``) * volume (string) - parent volume / dataset path * created (float) - creation timestamp (Unix epoch) * size (int) - bytes of unique data referenced only by this snapshot diff --git a/pyproject.toml b/pyproject.toml index 6de333c..e6e83da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,6 +71,10 @@ target-version = "py310" [tool.ruff.lint] select = ["E", "F", "I", "UP", "B"] +[tool.ruff.lint.per-file-ignores] +# Recorded device output keeps its lines. +"tests/**" = ["E501"] + [tool.mypy] python_version = "3.10" strict = true diff --git a/tests/test_ssh_host_keys.py b/tests/test_ssh_host_keys.py index f2d91a8..6470349 100644 --- a/tests/test_ssh_host_keys.py +++ b/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) diff --git a/tests/test_vm_cpu_types.py b/tests/test_vm_cpu_types.py index 14ffd39..61b8516 100644 --- a/tests/test_vm_cpu_types.py +++ b/tests/test_vm_cpu_types.py @@ -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