NAPALM/napalm-netgear
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(get_config): use paged send for show running-config/startup-config

Browse Source 
"v7" CLI firmware has no "terminal length 0" equivalent (see
_send_paged_command's docstring), so a config long enough to paginate
emits "--More--" prompts that _send_command's expect_string=base_prompt
match never sees. On real hardware (GS110TPv3, 7 VLANs, 10 interfaces)
this hung every scheduled config-backup poll for 30s and failed with
"Pattern not detected: '<prompt>[>#]' in output.", so the config was
never actually backed up. get_mac_address_table()/get_vlans() already
use _send_paged_command() for the same reason on other long outputs;
get_config() now does too.

Also fixed tests/unit/test_driver.py's import/patch target
(napalm_netgear_plus -> napalm_netgear, the actual package name) —
the whole file has been uncollectable since its initial commit, no CI
was wired up here to catch it. And removed a dead unreachable
`return {"success": ..., "output": ...}` line after get_health_metrics's
real return (undefined names, ruff F821), unrelated leftover found
while fixing the above.
This commit is contained in:
Christian Manivong
2026-07-10 09:54:29 +02:00
parent 383563d714
commit 46e419d232
2 changed files with 118 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_netgear/netgear_smart.py
+58 -58
View File
@@ -113,7 +113,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
# Config management state
self._candidate_config: Optional[str] = None
self._candidate_mode: Optional[str] = None # 'merge' or 'replace'
self._candidate_mode: Optional[str] = None # 'merge' or 'replace'
self._backup_config: Optional[str] = None
# CLI style detection cache: True = "v7" (Cisco-like, GigabitEthernet/lag
@@ -141,13 +141,9 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
if not self.force_no_enable:
self.device.enable()
except NetmikoTimeoutException as exc:
raise ConnectionException(
f"Cannot connect to {self.hostname}: {exc}"
) from exc
raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
except NetmikoAuthenticationException as exc:
raise ConnectionException(
f"Authentication failed for {self.hostname}: {exc}"
) from exc
raise ConnectionException(f"Authentication failed for {self.hostname}: {exc}") from exc
except Exception as exc:
raise ConnectionException(
f"Verbindungsaufbau fehlgeschlagen (ConnectionException): "
@@ -422,9 +418,8 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
desc = self._parse_key_value(sysinfo, "System Description")
model = desc.split()[0] if desc else ""
os_version = (
self._parse_key_value(version, "Software Version")
or self._parse_key_value(sysinfo, "Software Version")
os_version = self._parse_key_value(version, "Software Version") or self._parse_key_value(
sysinfo, "Software Version"
)
uptime_str = self._parse_key_value(sysinfo, "System Uptime")
@@ -613,9 +608,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
n = self._get_port_count_v7()
interfaces: Dict[str, Dict] = {}
if n:
status_out = self._send_command(
f"show interfaces GigabitEthernet 1-{n} status"
)
status_out = self._send_command(f"show interfaces GigabitEthernet 1-{n} status")
interfaces.update(self._parse_interface_status_v7(status_out))
self._add_lag_info_v7(interfaces)
@@ -695,9 +688,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
"lag_mode": "lacp" if "lacp" in type_s.lower() else "trunk",
}
def _parse_interfaces(
self, port_output: str, intf_output: str = ""
) -> Dict[str, Dict]:
def _parse_interfaces(self, port_output: str, intf_output: str = "") -> Dict[str, Dict]:
"""Parse ``show port all`` and ``show interface all`` into NAPALM format."""
interfaces: Dict[str, Dict] = {}
@@ -730,9 +721,9 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
if len(parts) < 5 + offset:
continue
port = parts[0]
admin = parts[1 + offset] # Enable / Disable
phys_status = parts[3 + offset] # e.g. "100M/Full", "1G/Full", or "-"
link_status = parts[4 + offset] # Up / Down
admin = parts[1 + offset] # Enable / Disable
phys_status = parts[3 + offset] # e.g. "100M/Full", "1G/Full", or "-"
link_status = parts[4 + offset] # Up / Down
speed_val = 0.0
speed_m = re.search(r"(\d+)([MG])", phys_status)
@@ -845,10 +836,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
configs = {"running": "", "startup": "", "candidate": ""}
if retrieve in ("all", "running"):
configs["running"] = self._send_command("show running-config")
configs["running"] = self._send_paged_command("show running-config")
if retrieve in ("all", "startup"):
configs["startup"] = self._send_command("show startup-config")
configs["startup"] = self._send_paged_command("show startup-config")
if sanitized:
configs = napalm_helpers.sanitize_configs(configs, C.CISCO_SANITIZE_FILTERS)
@@ -1080,9 +1071,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
if interface and row["local_port"] != interface:
continue
detail_out = self._send_command(
f"show lldp remote-device detail {row['local_port']}"
)
detail_out = self._send_command(f"show lldp remote-device detail {row['local_port']}")
parsed = self._parse_lldp_detail(detail_out)
# Fill in from summary table if the detail command returned nothing
@@ -1271,9 +1260,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
raise MergeConfigException("No candidate configuration is staged.")
ex_cls = (
ReplaceConfigException
if self._candidate_mode == "replace"
else MergeConfigException
ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException
)
self._backup_config = self._send_command("show running-config")
@@ -1449,9 +1436,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
Parses ``show interface counters`` (or ``show interface ethernet 0/1``).
Each block starts with ``Interface................................ 0/1``.
"""
output = self._send_command(
["show interface counters", "show interface ethernet all"]
)
output = self._send_command(["show interface counters", "show interface ethernet all"])
def _int(s: str) -> int:
return int(s.replace(",", "")) if s.strip().replace(",", "").isdigit() else 0
@@ -1461,12 +1446,18 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
current: Optional[Dict] = None
_zero: Dict = {
"tx_errors": 0, "rx_errors": 0,
"tx_discards": 0, "rx_discards": 0,
"tx_octets": 0, "rx_octets": 0,
"tx_unicast_packets": 0, "rx_unicast_packets": 0,
"tx_multicast_packets": 0, "rx_multicast_packets": 0,
"tx_broadcast_packets": 0, "rx_broadcast_packets": 0,
"tx_errors": 0,
"rx_errors": 0,
"tx_discards": 0,
"rx_discards": 0,
"tx_octets": 0,
"rx_octets": 0,
"tx_unicast_packets": 0,
"rx_unicast_packets": 0,
"tx_multicast_packets": 0,
"rx_multicast_packets": 0,
"tx_broadcast_packets": 0,
"rx_broadcast_packets": 0,
}
_mapping = {
@@ -1750,9 +1741,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
) -> Dict[str, Union[str, Dict[str, Any]]]:
"""Execute a list of CLI commands and return their raw output."""
if encoding != "text":
raise NotImplementedError(
f"Encoding '{encoding}' is not supported by this driver."
)
raise NotImplementedError(f"Encoding '{encoding}' is not supported by this driver.")
result: Dict[str, Union[str, Dict[str, Any]]] = {}
for cmd in commands:
result[cmd] = self._send_command(cmd)
@@ -1916,7 +1905,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
if "poe_priority" in config:
priority = self._POE_PRIORITY_MAP_REV.get(config["poe_priority"], "low")
lines.append(f"power inline priority {priority}")
if "allocated_power_in_watts" in config and config.get("poe_allocation_method") == "PPAM_VALUE":
if (
"allocated_power_in_watts" in config
and config.get("poe_allocation_method") == "PPAM_VALUE"
):
limit_mw = int(round(float(config["allocated_power_in_watts"]) * 1000))
lines.append(f"power inline limit {limit_mw}")
lines.append("exit")
@@ -1986,19 +1978,29 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
import re
from napalm_device_types._ucd_metrics import build_if_metrics, ticks_to_seconds
_OID_SNMP_UPTIME = "1.3.6.1.2.1.1.3.0"
_NETGEAR_MEM_FREE = "1.3.6.1.4.1.4526.100.1.1.4.1.0"
_OID_SNMP_UPTIME = "1.3.6.1.2.1.1.3.0"
_NETGEAR_MEM_FREE = "1.3.6.1.4.1.4526.100.1.1.4.1.0"
_NETGEAR_MEM_TOTAL = "1.3.6.1.4.1.4526.100.1.1.4.2.0"
_NETGEAR_CPU_STR = "1.3.6.1.4.1.4526.100.1.1.4.9.0"
_OID_IF_DESCR = "1.3.6.1.2.1.2.2.1.2"
_OID_IF_SPEED = "1.3.6.1.2.1.2.2.1.5"
_OID_IF_IN_OCT = "1.3.6.1.2.1.2.2.1.10"
_OID_IF_OUT_OCT = "1.3.6.1.2.1.2.2.1.16"
_OID_IF_IN_ERR = "1.3.6.1.2.1.2.2.1.14"
_OID_IF_OUT_ERR = "1.3.6.1.2.1.2.2.1.20"
_NETGEAR_CPU_STR = "1.3.6.1.4.1.4526.100.1.1.4.9.0"
_OID_IF_DESCR = "1.3.6.1.2.1.2.2.1.2"
_OID_IF_SPEED = "1.3.6.1.2.1.2.2.1.5"
_OID_IF_IN_OCT = "1.3.6.1.2.1.2.2.1.10"
_OID_IF_OUT_OCT = "1.3.6.1.2.1.2.2.1.16"
_OID_IF_IN_ERR = "1.3.6.1.2.1.2.2.1.14"
_OID_IF_OUT_ERR = "1.3.6.1.2.1.2.2.1.20"
(uptime_raw, free_s, total_s, cpu_s,
descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather(
(
uptime_raw,
free_s,
total_s,
cpu_s,
descr,
speed,
in_oct,
out_oct,
in_err,
out_err,
) = await asyncio.gather(
snmp_get(_OID_SNMP_UPTIME),
snmp_get(_NETGEAR_MEM_FREE),
snmp_get(_NETGEAR_MEM_TOTAL),
@@ -2019,17 +2021,17 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
if free_s and total_s:
try:
free_kb = int(free_s)
free_kb = int(free_s)
total_kb = int(total_s)
used_kb = max(0, total_kb - free_kb)
used_kb = max(0, total_kb - free_kb)
metrics["memory_total_bytes"] = total_kb * 1024
metrics["memory_used_bytes"] = used_kb * 1024
metrics["memory_percent"] = round(used_kb / total_kb * 100, 1) if total_kb else 0.0
metrics["memory_used_bytes"] = used_kb * 1024
metrics["memory_percent"] = round(used_kb / total_kb * 100, 1) if total_kb else 0.0
except ValueError:
pass
if cpu_s:
m = re.search(r'5\s+Secs\s*\(\s*([\d.]+)\s*%\)', cpu_s)
m = re.search(r"5\s+Secs\s*\(\s*([\d.]+)\s*%\)", cpu_s)
if m:
try:
metrics["cpu_percent"] = float(m.group(1))
@@ -2038,5 +2040,3 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err)
return metrics
return {"success": success, "output": "\n".join(lines)}
tests/unit/test_driver.py
+60 -40
View File
@@ -3,7 +3,7 @@
import pytest
from unittest.mock import MagicMock, patch
from napalm_netgear_plus.netgear_smart import NetgearSmartDriver
from napalm_netgear.netgear_smart import NetgearSmartDriver
# ---------------------------------------------------------------------------
@@ -14,7 +14,7 @@ from napalm_netgear_plus.netgear_smart import NetgearSmartDriver
@pytest.fixture
def driver():
"""Return a driver instance with a mocked Netmiko connection."""
with patch("napalm_netgear_plus.netgear_smart.ConnectHandler"):
with patch("napalm_netgear.netgear_smart.ConnectHandler"):
drv = NetgearSmartDriver(
hostname="192.168.0.239",
username="admin",
@@ -153,54 +153,50 @@ class TestGetFacts:
driver._send_command = lambda cmd: SHOW_SYSINFO
driver._get_interface_list = lambda: []
facts = driver.get_facts()
for key in ("vendor", "model", "hostname", "os_version", "serial_number",
"uptime", "interface_list", "fqdn"):
for key in (
"vendor",
"model",
"hostname",
"os_version",
"serial_number",
"uptime",
"interface_list",
"fqdn",
):
assert key in facts
def test_vendor(self, driver):
driver._send_command = lambda cmd: (
SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
)
driver._send_command = lambda cmd: SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
driver._get_interface_list = lambda: []
facts = driver.get_facts()
assert facts["vendor"] == "Netgear"
def test_model(self, driver):
driver._send_command = lambda cmd: (
SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
)
driver._send_command = lambda cmd: SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
driver._get_interface_list = lambda: []
facts = driver.get_facts()
assert facts["model"] == "GS110TP"
def test_hostname(self, driver):
driver._send_command = lambda cmd: (
SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
)
driver._send_command = lambda cmd: SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
driver._get_interface_list = lambda: []
facts = driver.get_facts()
assert facts["hostname"] == "myswitch"
def test_os_version(self, driver):
driver._send_command = lambda cmd: (
SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
)
driver._send_command = lambda cmd: SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
driver._get_interface_list = lambda: []
facts = driver.get_facts()
assert facts["os_version"] == "6.6.3"
def test_serial_number(self, driver):
driver._send_command = lambda cmd: (
SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
)
driver._send_command = lambda cmd: SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
driver._get_interface_list = lambda: []
facts = driver.get_facts()
assert facts["serial_number"] == "1FE2A0B1C2"
def test_uptime_parsing(self, driver):
driver._send_command = lambda cmd: (
SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
)
driver._send_command = lambda cmd: SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION
driver._get_interface_list = lambda: []
facts = driver.get_facts()
expected = 2 * 86400 + 3 * 3600 + 15 * 60 + 42
@@ -214,14 +210,12 @@ class TestGetFacts:
class TestParseUptimeSeconds:
def test_full(self):
assert NetgearSmartDriver._parse_uptime_seconds(
"2 days 3 hrs 15 mins 42 secs"
) == float(2 * 86400 + 3 * 3600 + 15 * 60 + 42)
assert NetgearSmartDriver._parse_uptime_seconds("2 days 3 hrs 15 mins 42 secs") == float(
2 * 86400 + 3 * 3600 + 15 * 60 + 42
)
def test_zero(self):
assert NetgearSmartDriver._parse_uptime_seconds(
"0 days 0 hrs 0 mins 0 secs"
) == 0.0
assert NetgearSmartDriver._parse_uptime_seconds("0 days 0 hrs 0 mins 0 secs") == 0.0
def test_hours_only(self):
assert NetgearSmartDriver._parse_uptime_seconds("1 hrs 0 mins 0 secs") == 3600.0
@@ -256,6 +250,7 @@ class TestGetInterfaces:
if "port" in cmd:
return SHOW_PORT_ALL
return SHOW_INTERFACE_ALL
driver._send_command = _send
def test_all_ports_present(self, driver):
@@ -381,14 +376,10 @@ class TestParseVlanPorts:
assert NetgearSmartDriver._parse_vlan_ports("0/1") == ["0/1"]
def test_range(self):
assert NetgearSmartDriver._parse_vlan_ports("0/1-0/4") == [
"0/1", "0/2", "0/3", "0/4"
]
assert NetgearSmartDriver._parse_vlan_ports("0/1-0/4") == ["0/1", "0/2", "0/3", "0/4"]
def test_short_range(self):
assert NetgearSmartDriver._parse_vlan_ports("0/1-4") == [
"0/1", "0/2", "0/3", "0/4"
]
assert NetgearSmartDriver._parse_vlan_ports("0/1-4") == ["0/1", "0/2", "0/3", "0/4"]
def test_comma_separated(self):
result = NetgearSmartDriver._parse_vlan_ports("0/1, 0/3, 0/5")
@@ -447,6 +438,7 @@ class TestGetSnmpInformation:
if "sysinfo" in cmd:
return SHOW_SYSINFO
return SHOW_SNMP
driver._send_command = _send
snmp = driver.get_snmp_information()
assert "public" in snmp["community"]
@@ -458,6 +450,7 @@ class TestGetSnmpInformation:
if "sysinfo" in cmd:
return SHOW_SYSINFO
return SHOW_SNMP
driver._send_command = _send
snmp = driver.get_snmp_information()
assert snmp["location"] == "Server Room"
@@ -540,16 +533,19 @@ class TestConfigManagement:
def test_load_merge_raises_without_input(self, driver):
from napalm.base.exceptions import MergeConfigException
with pytest.raises(MergeConfigException):
driver.load_merge_candidate()
def test_load_replace_raises_without_input(self, driver):
from napalm.base.exceptions import ReplaceConfigException
with pytest.raises(ReplaceConfigException):
driver.load_replace_candidate()
def test_rollback_raises_without_backup(self, driver):
from napalm.base.exceptions import CommandErrorException
with pytest.raises(CommandErrorException):
driver.rollback()
@@ -580,9 +576,7 @@ exit
"""
def test_diff_detects_change(self):
cmds = NetgearSmartDriver._diff_to_commands(
TestConfigDiff.BACKUP, TestConfigDiff.CURRENT
)
cmds = NetgearSmartDriver._diff_to_commands(TestConfigDiff.BACKUP, TestConfigDiff.CURRENT)
# Should contain a "no description" and "description uplink" line
joined = " ".join(cmds)
assert "description" in joined
@@ -595,19 +589,45 @@ exit
class TestGetConfig:
def test_running_retrieved(self, driver):
driver._send_command = lambda cmd: "! running config"
driver._send_paged_command = lambda cmd: "! running config"
cfg = driver.get_config(retrieve="running")
assert cfg["running"] == "! running config"
assert cfg["startup"] == ""
def test_startup_retrieved(self, driver):
driver._send_command = lambda cmd: "! startup config"
driver._send_paged_command = lambda cmd: "! startup config"
cfg = driver.get_config(retrieve="startup")
assert cfg["startup"] == "! startup config"
assert cfg["running"] == ""
def test_uses_paged_command_not_plain_send_command(self, driver):
"""Regression: get_config() called _send_command() (Netmiko
send_command with expect_string=<base_prompt>[>#]) for
"show running-config"/"show startup-config". On "v7" CLI firmware
there is no "terminal length 0" equivalent (see
_send_paged_command's docstring), so a config long enough to
paginate emits "--More--" prompts that expect_string never matches
— send_command hangs until read_timeout and fails with
"Pattern not detected: '<prompt>[>#]' in output.". Real device:
Netgear GS110TPv3 (7 VLANs, 10 interfaces) hit this on every
scheduled config-backup poll, 30s timeout, no config ever saved.
get_mac_address_table()/get_vlans() already use
_send_paged_command() for exactly this reason — get_config() must
too."""
driver._send_command = MagicMock(
side_effect=AssertionError(
"get_config() must use _send_paged_command(), not _send_command()"
)
)
driver._send_paged_command = MagicMock(return_value="! config")
driver.get_config()
assert driver._send_paged_command.call_count == 2
driver._send_command.assert_not_called()
def test_candidate_always_empty(self, driver):
driver._send_command = lambda cmd: ""
driver._send_paged_command = lambda cmd: ""
cfg = driver.get_config()
assert cfg["candidate"] == ""