NAPALM/napalm-hpe-aruba-procurve
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f35b59ec2d4a7bf8b8eeb1db3c57c0fc71081d6e
napalm-hpe-aruba-procurve/tests/unit/test_driver.py
T
Christian Manivong f35b59ec2d fix(cli): read J.15 port status, and say why each transport failed 
A 2800-series switch on J.15.09 polled over CLI came back with no
interfaces and an empty OS version, while VLANs and ARP parsed fine.

`show interfaces brief` on that firmware has an Intrusion Alert column
between the `|` and Enabled, and puts Mode before MDI rather than after:

      Port   Type      | Alert     Enabled Status Mode       Mode ...
      3-Trk3 100/1000T | No        Yes     Down   1000FDx    MDI  ...

The regex read Alert as Enabled, then failed on Yes where it wanted
Up|Down, so no line matched. It now skips the Alert column where there is
one and takes the speed from either position. Trunk members are listed as
`<port>-Trk<n>`; the port is `<port>` and the suffix becomes its
trunk_group, so the per-port `show interfaces 3` is a command the switch
knows.

The empty OS version was the alternatives list in _send_command. It moved
to the next command on "% Invalid" or "Error", but ProCurve rejects an
unknown command with "Invalid input: system-information" -- so that line
was parsed as system information. "Invalid input" now counts as failure.

Getting there took longer than it should have, because the first symptom
was "Authentication failed: Login failed". That was Telnet's error, the
last transport tried; the REST probe and both SSH attempts had failed
before it and said nothing above debug level. In fact the switch had run
out of CLI sessions and closed SSH straight after the password. open()
now records why each transport failed, and both the auth error and the
final "Cannot connect" carry that list.

Fixtures are that switch's output, with hostname, serial and MAC
replaced.

Closes #2
Closes #3
Closes #4
2026-09-25 08:55:11 +02:00

887 lines
32 KiB
Python
Raw Blame History

"""Unit tests for ProcurveDriver and parsers — no real device required."""
import pytest
from unittest.mock import MagicMock, patch
from napalm.base.exceptions import ConnectionException
from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, 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
"""
# 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
"""
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
"""
# Layout with the Intrusion Alert column; trunk members are "<port>-Trk<n>" (#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
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"
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):
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
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:
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
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):
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()
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
# ===========================================================================
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()
Reference in New Issue View Git Blame Copy Permalink