"""Unit tests for ProcurveDriver and parsers — no real device required.""" import pytest from unittest.mock import MagicMock, patch from napalm_procurve.procurve import ProcurveDriver, _parse_ping_output from napalm_procurve import parsers # =========================================================================== # Sample CLI output fixtures (captured from real ProCurve devices) # =========================================================================== SHOW_SYSTEM_INFO_2520G = """\ Status and Counters - General System Information System Name : myswitch System Contact : System Location : MAC Age Time (sec) : 300 Time Zone : 0 Daylight Time Rule : None Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6 ROM Version : R.10.06 Serial Number : SG12345678 Active Image : Primary Up Time : 5 days 3:14 CPU Util (%) : 2 Memory - Total : 65011712 IP Mgmt - Pkts Rx : 12345 Free : 37056512 """ SHOW_SYSTEM_INFO_2530 = """\ Status and Counters - General System Information System Name : aruba2530 System Contact : System Location : Firmware revision: YA.16.04.0006 Serial Number : SG87654321 Up Time : 2 days 1:30 Base MAC Addr : aabbcc-ddeeff """ SHOW_VERSION_2520G = """\ HP J9565A 2520G-8-PoE Switch Software revision : R.11.27 """ SHOW_INTERFACES_BRIEF = """\ Status and Counters - Port Status Flow Bcast Port Type | Enabled Link MDI Mode Ctrl Limit ------- ------------ + ------- ------ ----- ----------- ---- ----- 1 100/1000T | Yes Up Auto 1000FDx off 0 2 100/1000T | Yes Down Auto 10HDx off 0 3 100/1000T | No Down Auto Unknown off 0 """ SHOW_INTERFACES_PORT = """\ Status and Counters - Port Counters for port 1 Name : Link status : Up MAC address : a1b2c3-d4e5f6 Totals (since last boot): Bytes Rx : 123456 Bytes Tx : 654321 """ SHOW_ARP = """\ IP ARP table IP Address MAC Address Type Port --------------- ----------------- ------- ---- 192.168.0.1 a1b2c3-d4e5f6 dynamic 1 192.168.0.254 112233-445566 dynamic 2 """ SHOW_MAC = """\ Status and Counters - Port Address Table - All Ports MAC Address Port VLAN Type ---------------- ------ ---- ------- a1b2c3-d4e5f6 1 1 dynamic 112233-445566 2 1 static """ SHOW_LLDP_REMOTE = """\ Status and Counters - Local Device Info LLDP Remote Devices Information LocalPort | ChassisId PortId SysName Capabilities TTL --------- + --------------------- ------------- -------------- ------------ ------- 1 | a1 b2 c3 d4 e5 f6 Gi 0/1 other-switch B 120 """ SHOW_LLDP_DETAIL = """\ LLDP Remote Device Information Detail for Port 1 ChassisType : mac-address ChassisId : a1 b2 c3 d4 e5 f6 PortType : locally-assigned PortId : Gi0/1 Port Description : GigabitEthernet0/1 System Name : other-switch System Description : Cisco IOS Software System Capabilities : B R Enabled Capabilities : B """ SHOW_IP = """\ Internet (IP) Service Default Gateway : 192.168.0.254 IP Configuration for VLAN 1: IP Config : Manual IP Address : 192.168.0.1 Subnet Mask : 255.255.255.0 """ SHOW_NTP = """\ NTP associations Address Ref-clock St When Poll Reach Delay Offset Jitter --------------- --------- -- ----- ---- ----- ------ ------ ------ *10.0.0.10 10.0.0.1 2 64 1024 377 0.620 0.000 0.000 10.0.0.20 .INIT. 16 - 1024 0 0.000 0.000 0.000 """ SHOW_RUNNING_CONFIG = """\ ; J9565A Configuration Editor; Created on release #R.11.27 ; Ver #01:14.3f.f8 hostname "myswitch" module 1 type J9565A vlan 1 name "DEFAULT_VLAN" untagged 1-8 ip address 192.168.0.1 255.255.255.0 exit ip default-gateway 192.168.0.254 """ PING_OUTPUT_SUCCESS = """\ PING 10.0.0.1: 100 data bytes !!!!! ----10.0.0.1 PING Statistics---- 5 packets transmitted, 5 packets received, 0% packet loss round-trip (ms) min/avg/max = 0/1/2 """ PING_OUTPUT_FAILURE = """\ PING 10.99.99.99: 100 data bytes ..... ----10.99.99.99 PING Statistics---- 5 packets transmitted, 0 packets received, 100% packet loss """ # =========================================================================== # Fixtures # =========================================================================== @pytest.fixture def driver(): """Return a ProcurveDriver with mocked SSH connection (no real device).""" with patch("napalm_procurve.procurve.ConnectHandler"): drv = ProcurveDriver( hostname="192.168.0.1", username="manager", password="secret", ) drv._transport = "ssh" drv._device = MagicMock() drv._device.base_prompt = "HP-2520G" yield drv # =========================================================================== # Parser unit tests (pure functions — no mock needed) # =========================================================================== class TestMacConversion: def test_standard_conversion(self): assert parsers.procurve_mac_to_std("a1b2c3-d4e5f6") == "a1:b2:c3:d4:e5:f6" def test_already_colon(self): assert parsers.procurve_mac_to_std("a1:b2:c3:d4:e5:f6") == "a1:b2:c3:d4:e5:f6" def test_uppercase_normalised(self): assert parsers.procurve_mac_to_std("A1B2C3-D4E5F6") == "a1:b2:c3:d4:e5:f6" class TestParseSystemInfo: def test_2520g(self): info = parsers.parse_system_info(SHOW_SYSTEM_INFO_2520G) assert info["hostname"] == "myswitch" assert info["os_version"] == "R.11.27" assert info["serial_number"] == "SG12345678" assert info["base_mac"] == "a1:b2:c3:d4:e5:f6" assert info["uptime_seconds"] == 5 * 86400 + 3 * 3600 + 14 * 60 def test_2530(self): info = parsers.parse_system_info(SHOW_SYSTEM_INFO_2530) assert info["hostname"] == "aruba2530" assert info["os_version"] == "YA.16.04.0006" assert info["serial_number"] == "SG87654321" class TestParseInterfacesBrief: def test_parses_ports(self): ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) assert "1" in ifaces assert "2" in ifaces assert "3" in ifaces def test_port_1_up(self): ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) assert ifaces["1"]["is_up"] is True assert ifaces["1"]["is_enabled"] is True assert ifaces["1"]["speed"] == 1000.0 def test_port_2_down(self): ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) assert ifaces["2"]["is_up"] is False assert ifaces["2"]["is_enabled"] is True def test_port_3_disabled(self): ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) assert ifaces["3"]["is_enabled"] is False class TestParseArpTable: def test_two_entries(self): table = parsers.parse_arp_table(SHOW_ARP) assert len(table) == 2 def test_first_entry(self): table = parsers.parse_arp_table(SHOW_ARP) entry = next(e for e in table if e["ip"] == "192.168.0.1") assert entry["mac"] == "a1:b2:c3:d4:e5:f6" assert entry["interface"] == "1" class TestParseMacTable: def test_two_entries(self): table = parsers.parse_mac_table(SHOW_MAC) assert len(table) == 2 def test_static_flag(self): table = parsers.parse_mac_table(SHOW_MAC) static_entry = next(e for e in table if e["mac"] == "11:22:33:44:55:66") assert static_entry["static"] is True def test_dynamic_flag(self): table = parsers.parse_mac_table(SHOW_MAC) dyn_entry = next(e for e in table if e["mac"] == "a1:b2:c3:d4:e5:f6") assert dyn_entry["static"] is False class TestParseLldpNeighbors: def test_one_neighbor(self): nbrs = parsers.parse_lldp_neighbors(SHOW_LLDP_REMOTE) assert "1" in nbrs assert nbrs["1"][0]["hostname"] == "other-switch" def test_port_id(self): nbrs = parsers.parse_lldp_neighbors(SHOW_LLDP_REMOTE) assert "Gi" in nbrs["1"][0]["port"] or "0/1" in nbrs["1"][0]["port"] class TestParseLldpDetail: def test_system_name(self): detail = parsers.parse_lldp_neighbors_detail(SHOW_LLDP_DETAIL) assert "1" in detail assert detail["1"][0]["remote_system_name"] == "other-switch" def test_port_description(self): detail = parsers.parse_lldp_neighbors_detail(SHOW_LLDP_DETAIL) assert "GigabitEthernet0/1" in detail["1"][0]["remote_port_description"] class TestParseIpInterfaces: def test_vlan1(self): result = parsers.parse_ip_interfaces(SHOW_IP) assert "VLAN1" in result assert "192.168.0.1" in result["VLAN1"]["ipv4"] assert result["VLAN1"]["ipv4"]["192.168.0.1"]["prefix_length"] == 24 class TestParseNtpServers: def test_two_servers(self): servers = parsers.parse_ntp_servers(SHOW_NTP) assert "10.0.0.10" in servers assert "10.0.0.20" in servers class TestParsePingOutput: def test_success(self): result = _parse_ping_output(PING_OUTPUT_SUCCESS, 5) assert result["success"]["probes_sent"] == 5 assert result["success"]["packet_loss"] == 0 assert result["success"]["rtt_avg"] == 1.0 def test_failure(self): result = _parse_ping_output(PING_OUTPUT_FAILURE, 5) assert result["success"]["probes_sent"] == 5 assert result["success"]["packet_loss"] == 5 # =========================================================================== # Driver integration tests (mocked netmiko) # =========================================================================== class TestDriverGetFacts: def test_facts_from_cli(self, driver): def side_effect(cmd, **kw): cmd_str = cmd if isinstance(cmd, str) else cmd[0] if "system-information" in cmd_str or "system information" in cmd_str: return SHOW_SYSTEM_INFO_2520G if "interfaces brief" in cmd_str: return SHOW_INTERFACES_BRIEF if "version" in cmd_str: return SHOW_VERSION_2520G return "" driver._send_command = MagicMock(side_effect=side_effect) facts = driver.get_facts() assert facts["vendor"] == "HPE" assert facts["hostname"] == "myswitch" assert facts["os_version"] == "R.11.27" assert facts["serial_number"] == "SG12345678" assert isinstance(facts["interface_list"], list) assert "1" in facts["interface_list"] class TestDriverGetInterfaces: def test_interfaces_from_cli(self, driver): driver._send_command = MagicMock(return_value=SHOW_INTERFACES_BRIEF) ifaces = driver.get_interfaces() assert "1" in ifaces assert ifaces["1"]["is_up"] is True class TestDriverGetArpTable: def test_arp_from_cli(self, driver): driver._send_command = MagicMock(return_value=SHOW_ARP) table = driver.get_arp_table() assert len(table) == 2 assert any(e["ip"] == "192.168.0.1" for e in table) class TestDriverGetMacTable: def test_mac_from_cli(self, driver): driver._send_command = MagicMock(return_value=SHOW_MAC) table = driver.get_mac_address_table() assert len(table) == 2 class TestDriverGetConfig: def test_running_config(self, driver): driver._send_command = MagicMock(return_value=SHOW_RUNNING_CONFIG) cfg = driver.get_config(retrieve="running") assert "hostname" in cfg["running"] assert cfg["startup"] == "" class TestDriverConfigManagement: def test_load_and_discard(self, driver): driver.load_merge_candidate(config="hostname newname") assert driver._candidate_config == "hostname newname" driver.discard_config() assert driver._candidate_config is None def test_compare_config(self, driver): driver._send_command = MagicMock(return_value=SHOW_RUNNING_CONFIG) driver.load_merge_candidate(config='hostname "differentname"') diff = driver.compare_config() assert "differentname" in diff or "myswitch" in diff def test_has_pending_commit_false(self, driver): assert driver.has_pending_commit() is False def test_has_pending_commit_true(self, driver): driver.load_merge_candidate(config="no lldp run") assert driver.has_pending_commit() is True class TestDriverTransportDetection: def test_force_ssh_transport(self): """Forcing transport=ssh skips API probe.""" with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn: mock_conn.return_value = MagicMock() drv = ProcurveDriver( "192.168.0.1", "manager", "secret", optional_args={"transport": "ssh"}, ) drv.open() assert drv._transport == "ssh" def test_force_telnet_transport(self): """Forcing transport=telnet skips API and SSH probes.""" with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn: mock_conn.return_value = MagicMock() drv = ProcurveDriver( "192.168.0.1", "manager", "secret", optional_args={"transport": "telnet"}, ) drv.open() assert drv._transport == "telnet" def test_all_transports_fail_raises(self): """ConnectionException raised when all transports fail.""" with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)): with patch("napalm_procurve.procurve.ConnectHandler", side_effect=Exception("no conn")): drv = ProcurveDriver("192.168.0.1", "manager", "secret") with pytest.raises(Exception): drv.open() # =========================================================================== # VLAN parser tests # =========================================================================== RUNNING_CONFIG_VLANS = """\ ; J9565A Configuration Editor; Created on release #R.11.27 ; Ver #01:14.3f.f8 hostname "myswitch" module 1 type J9565A vlan 1 name "DEFAULT_VLAN" untagged 1-4,6-8 ip address 192.168.0.1 255.255.255.0 exit vlan 10 name "MGMT" tagged 1,2,5 exit vlan 20 name "IoT" untagged 5 tagged 1-3 exit """ SHOW_VLANS_BRIEF = """\ Status and Counters - VLAN Information VLAN ID Name | Status Voice Jumbo ------- ---------------- + ---------- ----- ----- 1 DEFAULT_VLAN | Port-based No No 10 MANAGEMENT | Port-based No No 20 IoT | Port-based No No """ SHOW_VLAN_DETAIL_10 = """\ Status and Counters - VLAN Information - VLAN 10 VLAN ID : 10 Name : MANAGEMENT Port Information Mode Unknown VLAN Status ---------------- -------- ------------ ------- 1 Tagged Learn Up 2 Tagged Learn Up 5 Tagged Learn Up """ SHOW_VLAN_DETAIL_20 = """\ Status and Counters - VLAN Information - VLAN 20 VLAN ID : 20 Name : IoT Port Information Mode Unknown VLAN Status ---------------- -------- ------------ ------- 1 Tagged Learn Up 2 Tagged Learn Up 3 Tagged Learn Up 5 Untagged Learn Up """ class TestExpandProcurvePorts: def test_numeric_range(self): assert parsers.expand_procurve_ports("1-4") == ["1", "2", "3", "4"] def test_mixed_list(self): assert parsers.expand_procurve_ports("1-3,6,8") == ["1", "2", "3", "6", "8"] def test_trunk_port(self): assert parsers.expand_procurve_ports("Trk1,Trk2") == ["Trk1", "Trk2"] def test_alpha_range(self): assert parsers.expand_procurve_ports("A1-A3") == ["A1", "A2", "A3"] def test_single_port(self): assert parsers.expand_procurve_ports("5") == ["5"] def test_mixed_all(self): result = parsers.expand_procurve_ports("1-2,Trk1,A1-A2") assert result == ["1", "2", "Trk1", "A1", "A2"] class TestParseVlansFromRunningConfig: def test_vlan_count(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) assert len(vlans) == 3 assert 1 in vlans assert 10 in vlans assert 20 in vlans def test_vlan_names(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) assert vlans[1]["name"] == "DEFAULT_VLAN" assert vlans[10]["name"] == "MGMT" assert vlans[20]["name"] == "IoT" def test_vlan1_untagged_ports(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) # "1-4,6-8" → 1,2,3,4,6,7,8 assert vlans[1]["untagged"] == ["1", "2", "3", "4", "6", "7", "8"] assert vlans[1]["tagged"] == [] def test_vlan10_tagged_ports(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) assert vlans[10]["tagged"] == ["1", "2", "5"] assert vlans[10]["untagged"] == [] def test_vlan20_mixed(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) assert vlans[20]["untagged"] == ["5"] assert vlans[20]["tagged"] == ["1", "2", "3"] def test_interfaces_union_untagged_first(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) # For VLAN 20: untagged ["5"] + tagged ["1","2","3"] → interfaces ordered untagged first assert vlans[20]["interfaces"][0] == "5" assert set(vlans[20]["interfaces"]) == {"1", "2", "3", "5"} def test_interfaces_napalm_key_present(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) for vdata in vlans.values(): assert "interfaces" in vdata class TestParseVlansBrief: def test_detects_all_vlans(self): result = parsers.parse_vlans_brief(SHOW_VLANS_BRIEF) assert set(result.keys()) == {1, 10, 20} def test_names(self): result = parsers.parse_vlans_brief(SHOW_VLANS_BRIEF) assert result[1] == "DEFAULT_VLAN" assert result[20] == "IoT" class TestParseVlanPortsDetail: def test_all_tagged(self): ports = parsers.parse_vlan_ports_detail(SHOW_VLAN_DETAIL_10) assert ports["tagged"] == ["1", "2", "5"] assert ports["untagged"] == [] def test_mixed_tagged_untagged(self): ports = parsers.parse_vlan_ports_detail(SHOW_VLAN_DETAIL_20) assert ports["tagged"] == ["1", "2", "3"] assert ports["untagged"] == ["5"] class TestDerivePvids: def test_basic_pvid_derivation(self): vlans = { 1: {"untagged": ["1", "2", "3"], "tagged": []}, 10: {"untagged": ["4"], "tagged": ["5"]}, } pvids = parsers.derive_port_pvids(vlans) assert pvids["1"] == 1 assert pvids["2"] == 1 assert pvids["3"] == 1 assert pvids["4"] == 10 def test_tagged_only_port_gets_pvid1(self): vlans = { 1: {"untagged": [], "tagged": []}, 10: {"untagged": [], "tagged": ["5"]}, } pvids = parsers.derive_port_pvids(vlans) assert pvids["5"] == 1 def test_pvid_from_running_config(self): vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) pvids = parsers.derive_port_pvids(vlans) # Port 1 is untagged in VLAN 1 assert pvids["1"] == 1 # Port 5 is untagged in VLAN 20 assert pvids["5"] == 20 # Port 2 is untagged in VLAN 1 but tagged in VLAN 20 — PVID is 1 assert pvids["2"] == 1 class TestDriverGetVlans: def test_get_vlans_from_cli(self, driver): driver._send_command = MagicMock(return_value=RUNNING_CONFIG_VLANS) vlans = driver.get_vlans() assert 1 in vlans assert 10 in vlans assert vlans[1]["name"] == "DEFAULT_VLAN" assert vlans[10]["tagged"] == ["1", "2", "5"] assert vlans[20]["untagged"] == ["5"] # NAPALM compat assert "interfaces" in vlans[1] def test_get_vlans_fallback_to_show_vlans(self, driver): """When running-config has no VLANs, fall back to show vlans.""" call_count = [0] def side_effect(cmd, **kw): call_count[0] += 1 if "running-config" in (cmd if isinstance(cmd, str) else cmd[0]): return "hostname myswitch\n" # no vlan blocks if "show vlans" == (cmd if isinstance(cmd, str) else cmd[0]): return SHOW_VLANS_BRIEF if "show vlans 10" in (cmd if isinstance(cmd, str) else cmd[0]): return SHOW_VLAN_DETAIL_10 if "show vlans 20" in (cmd if isinstance(cmd, str) else cmd[0]): return SHOW_VLAN_DETAIL_20 return "" driver._send_command = MagicMock(side_effect=side_effect) vlans = driver.get_vlans() assert 10 in vlans assert vlans[10]["tagged"] == ["1", "2", "5"] def test_get_port_pvids(self, driver): driver._send_command = MagicMock(return_value=RUNNING_CONFIG_VLANS) pvids = driver.get_port_pvids() # Port 5 is untagged in VLAN 20 assert pvids["5"] == 20 # Ports 1-4 are untagged in VLAN 1 assert pvids["1"] == 1 assert pvids["4"] == 1 # =========================================================================== # set_interface over the REST API — changing an existing VLAN membership # =========================================================================== def _api_driver(post_status: int = 400, post_body: str = '{"message":"Association exists"}'): """Driver on the REST transport with a scripted API client.""" with patch("napalm_procurve.procurve.ConnectHandler"): drv = ProcurveDriver(hostname="192.168.0.1", username="manager", password="secret") drv._transport = "api" api = MagicMock() post_resp = MagicMock(ok=post_status < 400, status_code=post_status, text=post_body) api.post.return_value = post_resp api.put.return_value = MagicMock(ok=True, status_code=200, text="{}") drv._api = api return drv, api class TestApiSetInterfaceExistingMembership: """A port that already carries the VLAN needs its mode changed, not a new row. Reproduced on a 2530-24G-PoEP (REST v7): port 10 holds VLAN 10 untagged alongside 30/40/50 tagged. Posting the tagged membership answers `400 {"message":"Association exists"}` — only `409` was treated as "already there". The resource is `{vlan_id}-{port_id}` and takes a PUT. """ def test_trunk_falls_back_to_put_on_association_exists(self): drv, api = _api_driver(post_status=400) drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]}) api.put.assert_called_once() path = api.put.call_args[0][0] assert path == "vlans-ports/10-10" assert api.put.call_args[1]["json"]["port_mode"] == "POM_TAGGED_STATIC" def test_access_falls_back_to_put_on_association_exists(self): drv, api = _api_driver(post_status=400) drv.set_interface("10", {"mode": "access", "access_vlan": 20}) assert api.put.call_args[0][0] == "vlans-ports/20-10" assert api.put.call_args[1]["json"]["port_mode"] == "POM_UNTAGGED" def test_conflict_status_also_falls_back(self): """Other firmware answers 409 for the same situation.""" drv, api = _api_driver(post_status=409) drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]}) api.put.assert_called_once() def test_a_new_membership_still_uses_post_alone(self): drv, api = _api_driver(post_status=200) drv.set_interface("11", {"mode": "trunk", "trunk_vlans": [30]}) api.post.assert_called_once() api.put.assert_not_called() def test_a_failing_put_reports_the_response_body(self): """The message used to drop it, so "Association exists" never surfaced.""" drv, api = _api_driver(post_status=400) api.put.return_value = MagicMock(ok=False, status_code=403, text='{"message":"denied"}') with pytest.raises(Exception) as exc: drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]}) assert "denied" in str(exc.value) def test_an_unrelated_post_failure_reports_the_response_body(self): drv, api = _api_driver(post_status=500, post_body='{"message":"boom"}') with pytest.raises(Exception) as exc: drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]}) assert "boom" in str(exc.value) api.put.assert_not_called()