diff --git a/napalm_procurve/parsers.py b/napalm_procurve/parsers.py index ac9b835..c520294 100644 --- a/napalm_procurve/parsers.py +++ b/napalm_procurve/parsers.py @@ -234,12 +234,13 @@ def parse_model_from_version(output: str) -> tuple[str, str]: # 1 100/1000T | Yes Down Auto Unknown off 0 _INTF_BRIEF_RE = re.compile( - r"^\s*(\S+)\s+" # Port + r"^\s*([^\s-]+)" # Port + r"(?:-(Trk\d+))?\s+" # Trunk group of a member port ("3-Trk3") r"(\S+)\s+\|" # Type | - r"\s+(Yes|No)\s+" # Enabled + r"\s+(?:(?:Yes|No)\s+)?" # Intrusion Alert, on firmware that has the column + r"(Yes|No)\s+" # Enabled r"(Up|Down)\s+" # Link - r"\S+\s+" # MDI (ignored) - r"(\S+)", # Mode (speed/duplex) + r"(\S+)\s+(\S+)", # Mode and MDI, in either order depending on firmware re.IGNORECASE, ) @@ -255,13 +256,13 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]: m = _INTF_BRIEF_RE.match(line) if not m: continue - port, itype, enabled, link, mode = ( - m.group(1), m.group(2), m.group(3), m.group(4), m.group(5) - ) + port, trunk_group, itype, enabled, link, *mode_or_mdi = m.groups() speed = 0.0 duplex = "" - # Mode examples: "1000FDx", "100HDx", "Unknown", "Auto" - sm = re.match(r"(\d+)(FDx|HDx)?", mode, re.I) + # Mode examples: "1000FDx", "100HDx", "Unknown"; MDI is "Auto", "MDI", "MDIX", "NA" + sm = next( + filter(None, (re.match(r"(\d+)(FDx|HDx)?", t, re.I) for t in mode_or_mdi)), None + ) if sm: speed = float(sm.group(1)) duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half" @@ -276,6 +277,8 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]: "mtu": -1, "mac_address": "", } + if trunk_group: + interfaces[port]["trunk_group"] = trunk_group return interfaces diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py index 313c1f5..c8057ae 100644 --- a/napalm_procurve/procurve.py +++ b/napalm_procurve/procurve.py @@ -138,6 +138,8 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): self._device: Optional[ConnectHandler] = None # REST API backend self._api: Optional[ProcurveApiClient] = None + # Why each transport failed during the current open(), as ": " + self._attempts: List[str] = [] # Config management state (CLI only) self._candidate_config: Optional[str] = None @@ -159,38 +161,45 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): 4. Telnet """ forced = self.force_transport - tried: List[str] = [] + self._attempts = [] # --- 1. REST API --- if not forced or forced == "api": if self._try_api(): return - tried.append("api") # --- 2. SSH standard --- if not forced or forced == "ssh": if self._try_ssh(legacy=False): return - tried.append("ssh") # --- 3. SSH legacy KEX --- if not forced or forced == "ssh_legacy": if self._try_ssh(legacy=True): return - tried.append("ssh_legacy") # --- 4. Telnet --- if not forced or forced == "telnet": if self._try_telnet(): return - tried.append("telnet") raise ConnectionException( f"Cannot connect to {self.hostname}. " - f"Tried transports: {', '.join(tried)}. " + f"Tried transports: {'; '.join(self._attempts)}. " "Check connectivity, credentials and whether SSH/Telnet/API is enabled." ) + def _auth_failure(self, transport: str, exc: Exception) -> ConnectionException: + """An authentication error that also says why the earlier transports failed. + + Without them, a Telnet "Login failed" reads as a wrong password when SSH + was merely refused and Telnet was the only transport left to answer (#2). + """ + msg = f"Authentication failed for {self.hostname} via {transport}: {exc}" + if self._attempts: + msg += f" (earlier: {'; '.join(self._attempts)})" + return ConnectionException(msg) + def close(self) -> None: """Close the active connection.""" if self._api: @@ -234,10 +243,12 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): if not ver: logger.warning("REST API not detected on %s — falling back to CLI", self.hostname) + self._attempts.append("api: not detected") return False # Try connecting with the requested SSL setting first; if it fails due to a # self-signed certificate (ssl_verify=True), transparently retry unverified. + error: Optional[Exception] = None for ssl_verify in ([self.ssl_verify] if not self.ssl_verify else [True, False]): client = ProcurveApiClient( hostname=self.hostname, @@ -252,6 +263,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): client.connect() except Exception as exc: logger.debug("REST API connect failed (ssl_verify=%s): %s", ssl_verify, exc) + error = exc continue self._api = client self._transport = "api" @@ -260,6 +272,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): return True logger.warning("REST API connect failed for %s — falling back to CLI", self.hostname) + self._attempts.append(f"api: {error}") return False def _netmiko_kwargs(self, legacy: bool = False) -> dict: @@ -285,22 +298,23 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): def _try_ssh(self, legacy: bool = False) -> bool: """Probe and connect via SSH. Returns True on success.""" label = "SSH-legacy" if legacy else "SSH" + transport = "ssh_legacy" if legacy else "ssh" logger.debug("Trying %s for %s", label, self.hostname) try: conn = ConnectHandler(**self._netmiko_kwargs(legacy)) self._device = conn - self._transport = "ssh_legacy" if legacy else "ssh" + self._transport = transport logger.info("Connected to %s via %s", self.hostname, label) return True except NetmikoAuthenticationException as exc: - raise ConnectionException( - f"Authentication failed for {self.hostname}: {exc}" - ) from exc + raise self._auth_failure(transport, exc) from exc except NetmikoTimeoutException: logger.debug("%s timeout for %s", label, self.hostname) + self._attempts.append(f"{transport}: timed out") return False except Exception as exc: logger.debug("%s failed for %s: %s", label, self.hostname, exc) + self._attempts.append(f"{transport}: {exc}") return False def _try_telnet(self) -> bool: @@ -321,11 +335,10 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): logger.info("Connected to %s via Telnet", self.hostname) return True except NetmikoAuthenticationException as exc: - raise ConnectionException( - f"Authentication failed for {self.hostname}: {exc}" - ) from exc + raise self._auth_failure("telnet", exc) from exc except Exception as exc: logger.debug("Telnet failed for %s: %s", self.hostname, exc) + self._attempts.append(f"telnet: {exc}") return False # ------------------------------------------------------------------ @@ -355,7 +368,8 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): last = "" for cmd in command: last = _do(cmd) - if "% Invalid" not in last and "Error" not in last: + # ProCurve says "Invalid input: …"; other firmware "% Invalid …" + if "Invalid input" not in last and "% Invalid" not in last and "Error" not in last: return last return last return _do(command) diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 950fb09..02b29c7 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -3,7 +3,9 @@ import pytest from unittest.mock import MagicMock, patch -from napalm_procurve.procurve import ProcurveDriver, _parse_ping_output +from napalm.base.exceptions import ConnectionException + +from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, ProcurveDriver, _parse_ping_output from napalm_procurve import parsers @@ -43,6 +45,30 @@ Status and Counters - General System Information Base MAC Addr : aabbcc-ddeeff """ +# J.15.09 firmware (#3/#4): no model anywhere, uptime in minutes. +SHOW_SYSTEM_J15 = """\ + Status and Counters - General System Information + + System Name : myswitch + System Contact : + System Location : + + MAC Age Time (sec) : 300 + + Time Zone : 120 + Daylight Time Rule : None + + + Software revision : J.15.09.0028 Base MAC Addr : a1b2c3-d4e5f6 + ROM Version : J.14.05 Serial Number : SG12345678 + Allow V1 Modules : Yes + + Up Time : 5 mins Memory - Total : 58,720,256 + CPU Util (%) : 98 Free : 39,550,272 +""" + +SHOW_SYSTEM_INFORMATION_INVALID = "Invalid input: system-information" + SHOW_VERSION_2520G = """\ HP J9565A 2520G-8-PoE Switch Software revision : R.11.27 @@ -59,6 +85,25 @@ Status and Counters - Port Status 3 100/1000T | No Down Auto Unknown off 0 """ +# Layout with the Intrusion Alert column; trunk members are "-Trk" (#3). +SHOW_INTERFACES_BRIEF_INTRUSION = """\ + Status and Counters - Port Status + + | Intrusion MDI Flow Bcast + Port Type | Alert Enabled Status Mode Mode Ctrl Limit + ------ --------- + --------- ------- ------ ---------- ---- ---- ----- + 1 100/1000T | No Yes Up 1000FDx MDI on 0 + 2 100/1000T | No Yes Down 1000FDx MDI off 0 + 3-Trk3 100/1000T | No Yes Down 1000FDx MDI off 0 + 4-Trk3 100/1000T | No Yes Down 1000FDx MDI off 0 + 5 100/1000T | No Yes Up 1000FDx MDI on 0 + 6-Trk6 100/1000T | No No Down 1000FDx NA off 0 + 7-Trk6 100/1000T | No Yes Up 1000FDx MDI off 0 + 8 100/1000T | No Yes Up 1000FDx MDIX off 0 + 9 100/1000T | No Yes Down 1000FDx MDIX off 0 + 10 1000SX | No Yes Up 1000FDx NA off 0 +""" + SHOW_INTERFACES_PORT = """\ Status and Counters - Port Counters for port 1 @@ -212,6 +257,13 @@ class TestParseSystemInfo: assert info["os_version"] == "YA.16.04.0006" assert info["serial_number"] == "SG87654321" + def test_j15(self): + info = parsers.parse_system_info(SHOW_SYSTEM_J15) + assert info["hostname"] == "myswitch" + assert info["os_version"] == "J.15.09.0028" + assert info["serial_number"] == "SG12345678" + assert info["base_mac"] == "a1:b2:c3:d4:e5:f6" + class TestParseInterfacesBrief: def test_parses_ports(self): @@ -234,6 +286,30 @@ class TestParseInterfacesBrief: def test_port_3_disabled(self): ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) assert ifaces["3"]["is_enabled"] is False + assert "trunk_group" not in ifaces["3"] + + +class TestParseInterfacesBriefIntrusionAlert: + """The layout with an Intrusion Alert column before Enabled (#3).""" + + def test_parses_every_port(self): + ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION) + assert sorted(ifaces, key=int) == [str(n) for n in range(1, 11)] + + def test_alert_column_is_not_read_as_enabled(self): + ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION) + assert ifaces["1"]["is_enabled"] is True + assert ifaces["1"]["is_up"] is True + assert ifaces["1"]["speed"] == 1000.0 + assert ifaces["2"]["is_up"] is False + assert ifaces["6"]["is_enabled"] is False + + def test_trunk_suffix_becomes_trunk_group(self): + ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION) + assert ifaces["3"]["trunk_group"] == "Trk3" + assert ifaces["4"]["trunk_group"] == "Trk3" + assert ifaces["7"]["trunk_group"] == "Trk6" + assert "trunk_group" not in ifaces["1"] class TestParseArpTable: @@ -349,6 +425,40 @@ class TestDriverGetInterfaces: assert "1" in ifaces assert ifaces["1"]["is_up"] is True + def test_trunk_member_detail_uses_bare_port_name(self, driver): + """`show interfaces 3-Trk3` is not a command; the port is `3` (#3).""" + driver._send_command = MagicMock(return_value=SHOW_INTERFACES_BRIEF_INTRUSION) + driver.get_interfaces() + sent = [c.args[0] for c in driver._send_command.call_args_list] + assert "show interfaces 3" in sent + assert not any("Trk" in c for c in sent) + + +class TestSendCommandAlternatives: + """A list of commands falls through to the next on a CLI error (#4).""" + + def test_invalid_input_tries_next_command(self, driver): + outputs = { + "show system-information": SHOW_SYSTEM_INFORMATION_INVALID, + "show system information": SHOW_SYSTEM_J15, + } + driver._device.send_command.side_effect = lambda cmd, **kw: outputs[cmd] + out = driver._send_command(["show system-information", "show system information"]) + assert out == SHOW_SYSTEM_J15.strip() + + def test_facts_on_j15_firmware(self, driver): + outputs = { + "show system-information": SHOW_SYSTEM_INFORMATION_INVALID, + "show system information": SHOW_SYSTEM_J15, + "show interfaces brief": SHOW_INTERFACES_BRIEF_INTRUSION, + "show version": "Image stamp: /ws/swbuildm/J_rel/code/build\n J.15.09.0028", + } + driver._device.send_command.side_effect = lambda cmd, **kw: outputs[cmd] + facts = driver.get_facts() + assert facts["os_version"] == "J.15.09.0028" + assert facts["serial_number"] == "SG12345678" + assert len(facts["interface_list"]) == 10 + class TestDriverGetArpTable: def test_arp_from_cli(self, driver): @@ -425,6 +535,60 @@ class TestDriverTransportDetection: with pytest.raises(Exception): drv.open() + def test_auth_failure_names_why_earlier_transports_failed(self): + """A Telnet login failure reports the transport and the earlier failures. + + Otherwise "Login failed" reads as a wrong password when SSH was merely + refused and Telnet was the only transport left to answer (#2). + """ + from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException + + def connect(**kwargs): + if kwargs["device_type"] == ProcurveDriver.NETMIKO_DEVICE_TYPE_TELNET: + raise NetmikoAuthenticationException("Login failed: 192.168.0.1") + if kwargs["disabled_algorithms"] == _SSH_DISABLED_STANDARD: + raise ConnectionRefusedError("[Errno 111] Connection refused") + raise NetmikoTimeoutException("TCP connection to device failed") + + with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)): + with patch("napalm_procurve.procurve.ConnectHandler", side_effect=connect): + drv = ProcurveDriver("192.168.0.1", "manager", "secret") + with pytest.raises(ConnectionException) as exc_info: + drv.open() + + msg = str(exc_info.value) + assert "Authentication failed for 192.168.0.1 via telnet: Login failed: 192.168.0.1" in msg + assert "api: not detected" in msg + assert "ssh: [Errno 111] Connection refused" in msg + assert "ssh_legacy: timed out" in msg + + def test_all_transports_fail_names_each_reason(self): + """The final error gives a reason per transport, not just its name.""" + with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)): + with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("no route to host")): + drv = ProcurveDriver("192.168.0.1", "manager", "secret") + with pytest.raises(ConnectionException) as exc_info: + drv.open() + + msg = str(exc_info.value) + assert "api: not detected" in msg + assert "ssh: no route to host" in msg + assert "ssh_legacy: no route to host" in msg + assert "telnet: no route to host" in msg + + def test_reopen_does_not_carry_earlier_attempts(self): + """Each open() reports only its own attempts.""" + with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)): + with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("first")): + drv = ProcurveDriver("192.168.0.1", "manager", "secret") + with pytest.raises(ConnectionException): + drv.open() + with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("second")): + with pytest.raises(ConnectionException) as exc_info: + drv.open() + + assert "first" not in str(exc_info.value) + # =========================================================================== # VLAN parser tests