fix: what bandit and the remaining lint rules found
- Long lines are wrapped: the get_environment awk command is split into adjacent string literals, the rollback message and the system mixin's docstring are wrapped; the command and the text are unchanged. - The two except Exception: pass (the snmpd community in get_snmp_config, the network VLANs in get_ssids) now log at debug level (B110), so a swallowed failure leaves a trace. - tests: raise ... from exc, unused loop variable.
This commit is contained in:
1 parent
a212bc8715
commit
8b8905f0c7
5 files changed
+19
-7
No files matched your search
@@ -91,7 +91,8 @@ class OpenWrtConfigMixin(ConfigLifecycleMixin):
|
|||||||
"""
|
"""
|
||||||
if self._backup_config is None:
|
if self._backup_config is None:
|
||||||
raise CommandErrorException(
|
raise CommandErrorException(
|
||||||
"No backup configuration available – commit_config has not been called in this session."
|
"No backup configuration available – "
|
||||||
|
"commit_config has not been called in this session."
|
||||||
)
|
)
|
||||||
|
|
||||||
escaped = self._backup_config.replace("'", "'\\''")
|
escaped = self._backup_config.replace("'", "'\\''")
|
||||||
|
|||||||
@@ -12,9 +12,12 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
import re
|
import re
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
#: Printed after a command so its exit status survives the shell session.
|
#: Printed after a command so its exit status survives the shell session.
|
||||||
_RC_MARKER = "__OWRT_RC="
|
_RC_MARKER = "__OWRT_RC="
|
||||||
_RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
|
_RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
|
||||||
@@ -30,7 +33,10 @@ def _split_status(raw: str) -> tuple[str, int | None]:
|
|||||||
|
|
||||||
|
|
||||||
class OpenWrtSystemMixin:
|
class OpenWrtSystemMixin:
|
||||||
"""Mixin providing system-level NAPALM getters (environment, NTP, SNMP, users, services, updates, device actions)."""
|
"""Mixin providing system-level NAPALM getters.
|
||||||
|
|
||||||
|
Environment, NTP, SNMP, users, services, updates and device actions.
|
||||||
|
"""
|
||||||
|
|
||||||
def get_environment(self) -> dict[str, Any]:
|
def get_environment(self) -> dict[str, Any]:
|
||||||
"""Return device environment data (CPU, memory).
|
"""Return device environment data (CPU, memory).
|
||||||
@@ -39,7 +45,8 @@ class OpenWrtSystemMixin:
|
|||||||
Memory from ``/proc/meminfo``.
|
Memory from ``/proc/meminfo``.
|
||||||
"""
|
"""
|
||||||
cpu_out = self._send_command(
|
cpu_out = self._send_command(
|
||||||
"awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} END{print (1-(idle1/total1))*100}' /proc/stat"
|
"awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} "
|
||||||
|
"END{print (1-(idle1/total1))*100}' /proc/stat"
|
||||||
)
|
)
|
||||||
mem_out = self._send_command("cat /proc/meminfo")
|
mem_out = self._send_command("cat /proc/meminfo")
|
||||||
|
|
||||||
@@ -250,7 +257,7 @@ class OpenWrtSystemMixin:
|
|||||||
if uci_comm:
|
if uci_comm:
|
||||||
community = uci_comm
|
community = uci_comm
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
logger.debug("snmpd community: ignored", exc_info=True)
|
||||||
|
|
||||||
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
|
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
|
||||||
|
|
||||||
|
|||||||
@@ -12,9 +12,12 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
import re
|
import re
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class OpenWrtWirelessMixin:
|
class OpenWrtWirelessMixin:
|
||||||
"""Mixin providing wireless/radio NAPALM getters."""
|
"""Mixin providing wireless/radio NAPALM getters."""
|
||||||
@@ -155,7 +158,8 @@ class OpenWrtWirelessMixin:
|
|||||||
if vlan is not None:
|
if vlan is not None:
|
||||||
net_vlan[iface] = vlan
|
net_vlan[iface] = vlan
|
||||||
except Exception:
|
except Exception:
|
||||||
pass # Non-fatal: VLAN info is optional enrichment
|
# Non-fatal: VLAN info is optional enrichment.
|
||||||
|
logger.debug("network VLANs: ignored", exc_info=True)
|
||||||
|
|
||||||
# Build result; group entries with the same SSID name, merging bands
|
# Build result; group entries with the same SSID name, merging bands
|
||||||
result: dict[str, Any] = {}
|
result: dict[str, Any] = {}
|
||||||
|
|||||||
@@ -911,7 +911,7 @@ class TestGetRouteTo:
|
|||||||
|
|
||||||
def test_required_keys(self, driver):
|
def test_required_keys(self, driver):
|
||||||
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
||||||
for prefix, routes in driver.get_route_to().items():
|
for _prefix, routes in driver.get_route_to().items():
|
||||||
for route in routes:
|
for route in routes:
|
||||||
assert set(route.keys()) >= {
|
assert set(route.keys()) >= {
|
||||||
"protocol",
|
"protocol",
|
||||||
|
|||||||
@@ -76,7 +76,7 @@ def test_a_changed_key_is_refused():
|
|||||||
try:
|
try:
|
||||||
raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
|
raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
|
||||||
except paramiko.SSHException as exc: # what netmiko does with it
|
except paramiko.SSHException as exc: # what netmiko does with it
|
||||||
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
|
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc
|
||||||
|
|
||||||
driver = _driver(seen, SshHostKey.of(OTHER_KEY))
|
driver = _driver(seen, SshHostKey.of(OTHER_KEY))
|
||||||
with patch(HANDLER, side_effect=refuse):
|
with patch(HANDLER, side_effect=refuse):
|
||||||
|
|||||||
Reference in new issue
Block a user