NAPALM/napalm-hpe-aruba-procurve
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
napalm-hpe-aruba-procurve/tests/unit/test_driver.py
T
Christian Manivong 7f07da6857 fix(cli): the trunk gets its own row, as over REST 
Since f35b59e a CLI poll reports trunk members as `3` with
trunk_group `Trk3`, but `show interfaces brief` has no line for Trk3
itself, so the trunk was missing from the interface list. The REST path
built that row on its own.

Both paths now use napalm_device_types.add_lag_interfaces, and the REST
path's copy is gone. One visible change there: `lag_members` is now in
port order (7, 10) instead of API order; the description already was.

The CLI path does not know the mode yet, so its trunks carry no
lag_mode. That needs `show trunks`, whose output is not at hand.

Needs napalm-device-types f3fa75b.

Closes #5
2026-09-25 10:44:11 +02:00

942 lines
35 KiB
Python
Raw Permalink 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()
# ===========================================================================
# LAG rows (#5)
# ===========================================================================
class TestApiGetInterfacesLag:
"""The REST path's trunk rows, kept as they were when it moved to add_lag_interfaces."""
def _client(self, ports):
from napalm_procurve.api_client import ProcurveApiClient
client = ProcurveApiClient(hostname="192.168.0.1", username="manager", password="secret")
responses = {
"ports": {"port_element": ports},
"port-statistics": {"port_statistics_element": [
{"id": p["id"], "port_speed_mbps": 1000} for p in ports
]},
"system/status/switch": {},
}
client.get = MagicMock(side_effect=lambda endpoint, **kw: responses[endpoint])
return client
def test_trunk_rows_with_mode(self):
client = self._client([
{"id": "1", "is_port_up": True, "is_port_enabled": True, "trunk_group": ""},
{"id": "3", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk3", "trunk_mode": "PTT_LACP"},
{"id": "4", "is_port_up": True, "is_port_enabled": True, "trunk_group": "trk3", "trunk_mode": "PTT_LACP"},
{"id": "10", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"},
{"id": "7", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"},
])
ifaces = client.get_interfaces()
assert ifaces["trk3"]["lag_members"] == ["3", "4"]
assert ifaces["trk3"]["lag_mode"] == "lacp"
assert ifaces["trk3"]["is_up"] is True
assert ifaces["trk3"]["speed"] == 2000.0
assert ifaces["trk6"]["lag_members"] == ["7", "10"]
assert ifaces["trk6"]["lag_mode"] == "trunk"
assert ifaces["trk6"]["description"] == "LAG (7, 10)"
assert ifaces["3"]["trunk_group"] == "trk3"
class TestCliGetInterfacesLag:
def test_trunk_rows_from_member_ports(self, driver):
driver._send_command = MagicMock(
side_effect=lambda cmd: SHOW_INTERFACES_BRIEF_INTRUSION if cmd == "show interfaces brief" else ""
)
ifaces = driver.get_interfaces()
assert ifaces["Trk3"]["lag_members"] == ["3", "4"]
assert ifaces["Trk6"]["lag_members"] == ["6", "7"]
assert ifaces["Trk6"]["is_up"] is True
assert ifaces["Trk6"]["is_enabled"] is True
Reference in New Issue View Git Blame Copy Permalink