From 46e419d232ef1c3141c33934e540507776146f21 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 10 Jul 2026 09:54:29 +0200 Subject: [PATCH] fix(get_config): use paged send for show running-config/startup-config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit "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: '[>#]' 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. --- napalm_netgear/netgear_smart.py | 116 ++++++++++++++++---------------- tests/unit/test_driver.py | 100 ++++++++++++++++----------- 2 files changed, 118 insertions(+), 98 deletions(-) diff --git a/napalm_netgear/netgear_smart.py b/napalm_netgear/netgear_smart.py index 5d72bc9..b989a8f 100644 --- a/napalm_netgear/netgear_smart.py +++ b/napalm_netgear/netgear_smart.py @@ -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)} diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 8c93e31..f6abb14 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -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=[>#]) 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: '[>#]' 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"] == ""