mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 04:20:45 +08:00
oliver
393fd0e0b4
Use custom cli_templates plus community ntc-templates for LLDP and port traffic across ZTE/Huawei/Cisco/H3C/Juniper/Nokia, drop regex parsers, and surface Admin/Phy/Prot on port discover. Also batch-add topology NEs and tighten workbench branding. Co-authored-by: Cursor <cursoragent@cursor.com>
325 lines
13 KiB
Python
325 lines
13 KiB
Python
"""Unit tests for ZTE / Huawei port traffic brief/detail parsers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from netx_api.port_traffic_commands import commands_for_vendor, detail_command
|
|
from netx_api.port_traffic_parsers import (
|
|
parse_bw_to_bps,
|
|
parse_cisco_interface_brief,
|
|
parse_cisco_interface_detail,
|
|
parse_huawei_interface_brief,
|
|
parse_huawei_interface_detail,
|
|
parse_interface_brief,
|
|
parse_interface_detail,
|
|
parse_zte_interface_brief,
|
|
parse_zte_interface_detail,
|
|
resolve_util_pct,
|
|
)
|
|
|
|
SAMPLE_DETAIL = """\
|
|
AL5458-ACC-6120HS#show interface xgei-1/1/0/1
|
|
13:47:08 Africa/Algiers Fri Jul 31 2026
|
|
xgei-1/1/0/1 is up, ifindex: 8194
|
|
Description: C2930L100-EQ2
|
|
Line protocol is up, IPv4 protocol is up, IPv6 protocol is down,
|
|
detected status is RX-OK/TX-OK
|
|
Last line protocol up time : 2026-05-26 15:14:49
|
|
Hardware is XGigabit Ethernet, address is 00d0.0000.088f
|
|
Internet address is 192.168.1.1/30
|
|
BW 1 Gbit/s
|
|
IP MTU 1500 bytes
|
|
MTU 1562 bytes
|
|
MPLS MTU 1548 bytes
|
|
|
|
Fec-eth : N/A
|
|
Fec-bypass : N/A
|
|
ARP type ARP
|
|
ARP Timeout 04:00:00
|
|
Last Clear Time : 2026-05-26 15:13:48 Last Refresh Time: 2026-07-31 13:47:00
|
|
Rate period : 30 s
|
|
Input : 824 bit/s 1 packet/s
|
|
Output : 824 bit/s 1 packet/s
|
|
Peak rate:
|
|
Input : 3536 bit/s peak time 2026-05-26 15:17:10
|
|
Output : 6896 bit/s peak time 2026-05-26 15:17:10
|
|
Intf utilization: input 0% output 0%
|
|
HardwareCounters:
|
|
In_Bytes 462427066 In_Packets 6529765
|
|
"""
|
|
|
|
SAMPLE_BRIEF = """\
|
|
AL5458-ACC-6120HS#show interface brief
|
|
13:43:00 Africa/Algiers Fri Jul 31 2026
|
|
Interface Attribute Mode BW Admin Phy Prot Description
|
|
xgei-1/1/0/1 optical Duplex/full 1G up up up C2930L100-EQ2
|
|
xgei-1/1/0/2 optical Duplex/full 1G up down down
|
|
xgei-1/1/0/4 optical Duplex/full 10G up down down 123456test11255
|
|
cgei-1/1/0/33 optical Duplex/full 100G up down down
|
|
xxvgei-1/1/0/11 optical Duplex/full 25G up down down
|
|
smartgroup1 N/A N/A up up down
|
|
smartgroup10 N/A N/A 1G up up up
|
|
bvi2 N/A N/A up up up
|
|
"""
|
|
|
|
HUAWEI_BRIEF = """\
|
|
<r1>display interface brief
|
|
PHY: Physical
|
|
*down: administratively down
|
|
InUti/OutUti: input utility/output utility
|
|
Interface PHY Protocol InUti OutUti inErrors outErrors
|
|
Ethernet1/0/0 up up 0.01% 0% 0 0
|
|
Ethernet1/0/1 up down 0% 0% 0 0
|
|
GigabitEthernet0/0/0 up down 0% 0% 0 0
|
|
NULL0 up up(s) 0% 0% 0 0
|
|
<r1>
|
|
"""
|
|
|
|
HUAWEI_DETAIL = """\
|
|
<r1>display interface Ethernet1/0/0
|
|
Ethernet1/0/0 current state : UP (ifindex: 5)
|
|
Line protocol current state : UP
|
|
Description:
|
|
Route Port,The Maximum Transmit Unit is 1500
|
|
Last 300 seconds input rate: 0 bits/sec, 1 packets/sec
|
|
Last 300 seconds output rate: 0 bits/sec, 0 packets/sec
|
|
Input peak rate 0 bits/sec, Record time: -
|
|
Output peak rate 0 bits/sec, Record time: -
|
|
Last 300 seconds input utility rate: 0.01%
|
|
Last 300 seconds output utility rate: 0.00%
|
|
<r1>
|
|
"""
|
|
|
|
CISCO_BRIEF = """\
|
|
R2#show ip interface brief
|
|
Interface IP-Address OK? Method Status Protocol
|
|
GigabitEthernet0/0 192.168.0.128 YES NVRAM up up
|
|
GigabitEthernet0/1 172.16.0.2 YES manual up up
|
|
GigabitEthernet0/2 unassigned YES NVRAM administratively down down
|
|
GigabitEthernet0/3 unassigned YES NVRAM administratively down down
|
|
R2#
|
|
"""
|
|
|
|
CISCO_DETAIL = """\
|
|
R2#show interfaces GigabitEthernet0/1
|
|
GigabitEthernet0/1 is up, line protocol is up
|
|
Hardware is iGbE, address is 5000.0003.0001 (bia 5000.0003.0001)
|
|
Internet address is 172.16.0.2/30
|
|
MTU 1500 bytes, BW 1000000 Kbit/sec, DLY 10 usec,
|
|
reliability 255/255, txload 1/255, rxload 1/255
|
|
Encapsulation ARPA, loopback not set
|
|
Last clearing of "show interface" counters never
|
|
5 minute input rate 0 bits/sec, 0 packets/sec
|
|
5 minute output rate 0 bits/sec, 0 packets/sec
|
|
524 packets input, 158824 bytes, 0 no buffer
|
|
R2#
|
|
"""
|
|
|
|
|
|
class BwParseTests(unittest.TestCase):
|
|
def test_compact(self):
|
|
self.assertEqual(parse_bw_to_bps("1G"), 1_000_000_000)
|
|
self.assertEqual(parse_bw_to_bps("10G"), 10_000_000_000)
|
|
self.assertEqual(parse_bw_to_bps("100G"), 100_000_000_000)
|
|
self.assertEqual(parse_bw_to_bps("25G"), 25_000_000_000)
|
|
self.assertEqual(parse_bw_to_bps("1M"), 1_000_000)
|
|
|
|
def test_detail_line(self):
|
|
self.assertEqual(parse_bw_to_bps("BW 1 Gbit/s"), 1_000_000_000)
|
|
|
|
def test_cisco_kbit_sec(self):
|
|
self.assertEqual(parse_bw_to_bps("MTU 1500 bytes, BW 1000000 Kbit/sec, DLY 10 usec,"), 1_000_000_000)
|
|
|
|
|
|
class BriefParserTests(unittest.TestCase):
|
|
def test_sample_brief(self):
|
|
rows = parse_zte_interface_brief(SAMPLE_BRIEF)
|
|
by_name = {r.ifname: r for r in rows}
|
|
self.assertIn("xgei-1/1/0/1", by_name)
|
|
r1 = by_name["xgei-1/1/0/1"]
|
|
self.assertEqual(r1.bw_bps, 1_000_000_000)
|
|
self.assertEqual(r1.admin, "up")
|
|
self.assertEqual(r1.phy, "up")
|
|
self.assertEqual(r1.prot, "up")
|
|
self.assertEqual(r1.description, "C2930L100-EQ2")
|
|
self.assertEqual(by_name["xgei-1/1/0/4"].bw_bps, 10_000_000_000)
|
|
self.assertEqual(by_name["cgei-1/1/0/33"].bw_bps, 100_000_000_000)
|
|
self.assertEqual(by_name["xxvgei-1/1/0/11"].bw_bps, 25_000_000_000)
|
|
self.assertEqual(by_name["smartgroup1"].bw_bps, 0)
|
|
self.assertEqual(by_name["smartgroup10"].bw_bps, 1_000_000_000)
|
|
self.assertEqual(by_name["xgei-1/1/0/2"].phy, "down")
|
|
|
|
def test_rejects_trailing_hostname_prompt(self):
|
|
text = SAMPLE_BRIEF + "\nAL5458-ACC-6120HS#\n"
|
|
rows = parse_zte_interface_brief(text)
|
|
names = [r.ifname for r in rows]
|
|
self.assertNotIn("AL5458-ACC-6120HS#", names)
|
|
self.assertNotIn("AL5458-ACC-6120HS", names)
|
|
self.assertTrue(any(n.startswith("xgei-") for n in names))
|
|
|
|
def test_rejects_prompt_without_hash(self):
|
|
text = (
|
|
SAMPLE_BRIEF
|
|
+ "\nAL5458-ACC-6120HS\n"
|
|
+ "AL5458-ACC-6120HS#show something\n"
|
|
)
|
|
rows = parse_zte_interface_brief(text)
|
|
names = [r.ifname for r in rows]
|
|
self.assertNotIn("AL5458-ACC-6120HS", names)
|
|
self.assertNotIn("AL5458-ACC-6120HS#show", names)
|
|
|
|
|
|
class DetailParserTests(unittest.TestCase):
|
|
def test_sample_detail(self):
|
|
d = parse_zte_interface_detail(SAMPLE_DETAIL)
|
|
self.assertEqual(d.ifname, "xgei-1/1/0/1")
|
|
self.assertEqual(d.bw_bps, 1_000_000_000)
|
|
self.assertEqual(d.rate_period_sec, 30)
|
|
self.assertEqual(d.in_bps, 824.0)
|
|
self.assertEqual(d.out_bps, 824.0)
|
|
self.assertEqual(d.in_util_pct, 0.0)
|
|
self.assertEqual(d.out_util_pct, 0.0)
|
|
self.assertEqual(d.description, "C2930L100-EQ2")
|
|
self.assertNotEqual(d.in_bps, 3536.0)
|
|
|
|
|
|
class HuaweiBriefParserTests(unittest.TestCase):
|
|
def test_sample_brief(self):
|
|
rows = parse_huawei_interface_brief(HUAWEI_BRIEF)
|
|
by_name = {r.ifname: r for r in rows}
|
|
self.assertEqual(len(rows), 4)
|
|
self.assertEqual(by_name["Ethernet1/0/0"].phy, "up")
|
|
self.assertEqual(by_name["Ethernet1/0/0"].prot, "up")
|
|
self.assertEqual(by_name["Ethernet1/0/0"].bw_bps, 0)
|
|
self.assertEqual(by_name["Ethernet1/0/1"].prot, "down")
|
|
self.assertEqual(by_name["NULL0"].prot, "up") # up(s)
|
|
self.assertEqual(by_name["GigabitEthernet0/0/0"].ifname, "GigabitEthernet0/0/0")
|
|
|
|
def test_dispatch(self):
|
|
rows = parse_interface_brief(HUAWEI_BRIEF, "huawei")
|
|
self.assertTrue(any(r.ifname == "Ethernet1/0/0" for r in rows))
|
|
|
|
|
|
class HuaweiDetailParserTests(unittest.TestCase):
|
|
def test_sample_detail(self):
|
|
d = parse_huawei_interface_detail(HUAWEI_DETAIL)
|
|
self.assertEqual(d.ifname, "Ethernet1/0/0")
|
|
self.assertEqual(d.admin_oper, "up")
|
|
self.assertEqual(d.bw_bps, 0)
|
|
self.assertEqual(d.rate_period_sec, 300)
|
|
self.assertEqual(d.in_bps, 0.0)
|
|
self.assertEqual(d.out_bps, 0.0)
|
|
self.assertEqual(d.in_util_pct, 0.01)
|
|
self.assertEqual(d.out_util_pct, 0.0)
|
|
|
|
def test_dispatch(self):
|
|
d = parse_interface_detail(HUAWEI_DETAIL, "huawei")
|
|
self.assertEqual(d.in_util_pct, 0.01)
|
|
|
|
|
|
class CiscoBriefParserTests(unittest.TestCase):
|
|
def test_sample_brief(self):
|
|
rows = parse_cisco_interface_brief(CISCO_BRIEF)
|
|
by_name = {r.ifname: r for r in rows}
|
|
self.assertEqual(len(rows), 4)
|
|
self.assertEqual(by_name["GigabitEthernet0/0"].phy, "up")
|
|
self.assertEqual(by_name["GigabitEthernet0/0"].prot, "up")
|
|
self.assertEqual(by_name["GigabitEthernet0/1"].admin, "up")
|
|
self.assertEqual(by_name["GigabitEthernet0/2"].admin, "down")
|
|
self.assertEqual(by_name["GigabitEthernet0/2"].phy, "down")
|
|
self.assertEqual(by_name["GigabitEthernet0/2"].prot, "down")
|
|
|
|
def test_dispatch(self):
|
|
rows = parse_interface_brief(CISCO_BRIEF, "cisco")
|
|
self.assertTrue(any(r.ifname == "GigabitEthernet0/1" for r in rows))
|
|
|
|
|
|
class CiscoDetailParserTests(unittest.TestCase):
|
|
def test_sample_detail(self):
|
|
d = parse_cisco_interface_detail(CISCO_DETAIL)
|
|
self.assertEqual(d.ifname, "GigabitEthernet0/1")
|
|
self.assertEqual(d.admin_oper, "up")
|
|
self.assertEqual(d.bw_bps, 1_000_000_000)
|
|
self.assertEqual(d.rate_period_sec, 300)
|
|
self.assertEqual(d.in_bps, 0.0)
|
|
self.assertEqual(d.out_bps, 0.0)
|
|
self.assertEqual(d.in_util_pct, 0.0)
|
|
self.assertEqual(d.out_util_pct, 0.0)
|
|
|
|
def test_dispatch(self):
|
|
d = parse_interface_detail(CISCO_DETAIL, "cisco")
|
|
self.assertEqual(d.bw_bps, 1_000_000_000)
|
|
|
|
|
|
class CommandsTests(unittest.TestCase):
|
|
def test_zte_matrix(self):
|
|
cmds = commands_for_vendor("ZTE", "zxros")
|
|
assert cmds is not None
|
|
self.assertEqual(cmds.brief, "show interface brief")
|
|
self.assertEqual(detail_command(cmds, "xgei-1/1/0/1"), "show interface xgei-1/1/0/1")
|
|
|
|
def test_huawei_matrix(self):
|
|
cmds = commands_for_vendor("Huawei", "huawei_vrp")
|
|
assert cmds is not None
|
|
self.assertEqual(cmds.vendor_key, "huawei")
|
|
self.assertEqual(cmds.brief, "display interface brief")
|
|
self.assertEqual(
|
|
detail_command(cmds, "Ethernet1/0/0"),
|
|
"display interface Ethernet1/0/0",
|
|
)
|
|
|
|
def test_cisco_matrix(self):
|
|
cmds = commands_for_vendor("Cisco", "cisco_ios")
|
|
assert cmds is not None
|
|
self.assertEqual(cmds.vendor_key, "cisco")
|
|
self.assertEqual(cmds.brief, "show ip interface brief")
|
|
self.assertEqual(
|
|
detail_command(cmds, "GigabitEthernet0/1"),
|
|
"show interfaces GigabitEthernet0/1",
|
|
)
|
|
|
|
def test_h3c_juniper_nokia_mikrotik_matrix(self):
|
|
h3c = commands_for_vendor("H3C", "hp_comware")
|
|
assert h3c is not None
|
|
self.assertEqual(h3c.vendor_key, "h3c")
|
|
self.assertEqual(h3c.brief, "display interface brief")
|
|
self.assertEqual(detail_command(h3c, "GE1/0/1"), "display interface GE1/0/1")
|
|
|
|
junos = commands_for_vendor("Juniper", "juniper_junos")
|
|
assert junos is not None
|
|
self.assertEqual(junos.vendor_key, "juniper")
|
|
self.assertEqual(junos.brief, "show interfaces")
|
|
self.assertEqual(detail_command(junos, "ge-0/0/0"), "show interfaces ge-0/0/0")
|
|
|
|
nokia = commands_for_vendor("Nokia", "nokia_sros")
|
|
assert nokia is not None
|
|
self.assertEqual(nokia.vendor_key, "nokia")
|
|
self.assertEqual(nokia.brief, "show port")
|
|
self.assertEqual(detail_command(nokia, "1/1/1"), "show port 1/1/1")
|
|
|
|
mikrotik = commands_for_vendor("MikroTik", "mikrotik_routeros")
|
|
assert mikrotik is not None
|
|
self.assertEqual(mikrotik.vendor_key, "mikrotik")
|
|
self.assertTrue(mikrotik.brief.startswith("/interface"))
|
|
|
|
self.assertIsNone(commands_for_vendor("Ericsson", "ericsson_ipos"))
|
|
|
|
|
|
class ResolveUtilTests(unittest.TestCase):
|
|
def test_keeps_vendor_util(self):
|
|
self.assertEqual(resolve_util_pct(0.01, 0.0, 1_000_000_000), 0.01)
|
|
|
|
def test_derives_when_vendor_zero(self):
|
|
# 98 Kbit/s on 1 Gbit/s ≈ 0.0098%
|
|
self.assertAlmostEqual(resolve_util_pct(0.0, 98_000.0, 1_000_000_000), 0.0098)
|
|
|
|
def test_zero_when_no_traffic(self):
|
|
self.assertEqual(resolve_util_pct(0.0, 0.0, 1_000_000_000), 0.0)
|
|
|
|
def test_zero_when_no_bw(self):
|
|
self.assertEqual(resolve_util_pct(0.0, 98_000.0, 0), 0.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|