mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 02:00:46 +08:00
oliver
ac4f115ee8
Append secondary IPs with M/S tags, skip ND6 clock banners, split optical power vs thresholds including QSFP lanes, broaden static-route egress ifnames, and add | one-line to ZTE collect templates except interface detail. Co-authored-by: Cursor <cursoragent@cursor.com>
223 lines
9.5 KiB
Python
223 lines
9.5 KiB
Python
"""Unit tests for ZTE ZXROS status parsers (samples from test/log)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from netx_api.biz_state.parsers.zte import (
|
|
is_valid_arp_age,
|
|
normalize_arp,
|
|
normalize_bgp_peer,
|
|
normalize_interface_brief,
|
|
normalize_isis_adjacency,
|
|
normalize_nd6_cache,
|
|
)
|
|
from netx_api.biz_state.profiles import get_profile, metric_field_map, profiles_for_vendor
|
|
|
|
|
|
def _log_text() -> str:
|
|
p = Path(__file__).resolve().parents[2] / "test" / "log"
|
|
if not p.is_file():
|
|
# Fallback: relative to monorepo root when tests run from netx/
|
|
p = Path(__file__).resolve().parents[3] / "test" / "log"
|
|
return p.read_text(encoding="utf-8", errors="ignore") if p.is_file() else ""
|
|
|
|
|
|
def _section(blob: str, start_marker: str, end_markers: tuple[str, ...]) -> str:
|
|
i = blob.find(start_marker)
|
|
if i < 0:
|
|
return ""
|
|
rest = blob[i:]
|
|
cut = len(rest)
|
|
for em in end_markers:
|
|
j = rest.find(em, len(start_marker))
|
|
if j >= 0:
|
|
cut = min(cut, j)
|
|
return rest[:cut]
|
|
|
|
|
|
class ZteStatusParserTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.log = _log_text()
|
|
|
|
def test_profiles_registered(self) -> None:
|
|
zte = {p.profile_id for p in profiles_for_vendor("zte", kind="collect")}
|
|
self.assertIn("zte.isis_adjacency", zte)
|
|
self.assertIn("zte.interface_brief", zte)
|
|
self.assertIn("zte.arp", zte)
|
|
self.assertIn("zte.nd6_cache", zte)
|
|
self.assertIn("zte.bgp_vpnv4_summary", zte)
|
|
self.assertIn("zte.bgp_ipv4_summary", zte)
|
|
self.assertIn("zte.bgp_vpnv6_summary", zte)
|
|
for mid in ("isis_adjacency", "interface_brief", "arp", "nd6_cache", "bgp_peer", "if_intf"):
|
|
self.assertIn(mid, metric_field_map())
|
|
self.assertIsNotNone(get_profile("zte.isis_adjacency"))
|
|
ii = get_profile("zte.if_intf")
|
|
self.assertIsNotNone(ii)
|
|
assert ii is not None
|
|
self.assertFalse(ii.enabled)
|
|
self.assertNotIn("zte.if_intf", zte)
|
|
self.assertIn("zte.config_interface", zte)
|
|
|
|
def test_isis_adjacency(self) -> None:
|
|
text = _section(
|
|
self.log,
|
|
"show isis adjacency",
|
|
("show interface brief", "show arp", "M6000-4SE-3#show"),
|
|
)
|
|
rows = normalize_isis_adjacency(raw_text=text)
|
|
self.assertGreaterEqual(len(rows), 5)
|
|
procs = {r["process_id"] for r in rows}
|
|
self.assertIn("1", procs)
|
|
self.assertIn("20", procs)
|
|
up = [r for r in rows if r["state"].upper() == "UP"]
|
|
self.assertEqual(len(up), len(rows))
|
|
|
|
def test_isis_adjacency_empty_mt_ipv4(self) -> None:
|
|
"""IPv4 rows often leave MT blank; must still parse NSF/AF."""
|
|
text = """
|
|
Process ID: 0
|
|
|
|
Process ID: 1
|
|
Interface System id State Lev Holds SNPA(802.2) Pri MT NSF AF
|
|
smartgroup107 PAG7_6120HS UP L1 26 PPP - Disable IPv4
|
|
smartgroup32766.1 AG4 UP/UP L1L2 27/27 PPP -/- Disable IPv4
|
|
smartgroup4501 C3 UP/UP L1L2 30/30 PPP -/- Disable IPv4
|
|
smartgroup11 MDN-SGI-P-ZM8SP UP L2 24 PPP - M Disable IPv6
|
|
"""
|
|
rows = normalize_isis_adjacency(raw_text=text)
|
|
self.assertEqual(len(rows), 4)
|
|
self.assertTrue(all(r["process_id"] == "1" for r in rows))
|
|
v4 = [r for r in rows if r["af"] == "IPv4"]
|
|
self.assertEqual(len(v4), 3)
|
|
self.assertTrue(all(r["mt"] == "" for r in v4))
|
|
self.assertEqual(v4[0]["interface"], "smartgroup107")
|
|
self.assertEqual(v4[0]["system_id"], "PAG7_6120HS")
|
|
self.assertEqual(v4[0]["state"], "UP")
|
|
self.assertEqual(v4[0]["nsf"], "Disable")
|
|
self.assertEqual(v4[1]["state"], "UP/UP")
|
|
self.assertEqual(v4[1]["pri"], "-/-")
|
|
v6 = next(r for r in rows if r["af"] == "IPv6")
|
|
self.assertEqual(v6["mt"], "M")
|
|
self.assertEqual(v6["nsf"], "Disable")
|
|
|
|
def test_interface_brief(self) -> None:
|
|
text = _section(
|
|
self.log,
|
|
"show interface brief",
|
|
("show arp", "show nd6", "M6000-4SE-3#show arp"),
|
|
)
|
|
rows = normalize_interface_brief(
|
|
raw_text=text, vendor="zte", device_type="zte_zxros", command="show interface brief"
|
|
)
|
|
self.assertGreaterEqual(len(rows), 10)
|
|
by_if = {r["interface"]: r for r in rows}
|
|
self.assertIn("cgei-0/1/0/1", by_if)
|
|
self.assertEqual(by_if["cgei-0/1/0/1"]["admin"].lower(), "up")
|
|
self.assertEqual(by_if["cgei-0/1/0/3"]["admin"].lower(), "down")
|
|
|
|
def test_arp(self) -> None:
|
|
text = _section(self.log, "show arp", ("show nd6", "PAG3_", "M6000-4SE-3#show nd6"))
|
|
from netx_api.biz_state.parsers import run_parser
|
|
from netx_api.ntc_parse import apply_rule
|
|
|
|
fsm = apply_rule(
|
|
platform="zte_zxros",
|
|
rule_key="zte_zxros_show_arp",
|
|
text=text,
|
|
command="show arp",
|
|
)
|
|
self.assertEqual(len(fsm), 13)
|
|
|
|
records, tables, keys = run_parser(
|
|
"arp",
|
|
raw_text=text,
|
|
vendor="zte",
|
|
device_type="zte_zxros",
|
|
command="show arp | one-line",
|
|
textfsm_command="show arp",
|
|
)
|
|
self.assertEqual(keys, ["zte_zxros_show_arp"])
|
|
self.assertEqual(len(tables["zte_zxros_show_arp"]), 13)
|
|
|
|
rows = normalize_arp(
|
|
raw_text=text, vendor="zte", device_type="zte_zxros", command="show arp"
|
|
)
|
|
self.assertEqual(rows, records)
|
|
self.assertGreaterEqual(len(rows), 10)
|
|
ips = {r["ip"] for r in rows}
|
|
self.assertIn("192.166.1.65", ips)
|
|
self.assertIn("10.229.234.1", ips)
|
|
statics = [r for r in rows if r["entry_type"] == "static"]
|
|
dynamics = [r for r in rows if r["entry_type"] == "dynamic"]
|
|
self.assertGreaterEqual(len(statics), 5)
|
|
self.assertGreaterEqual(len(dynamics), 5)
|
|
self.assertTrue(all(is_valid_arp_age(r["age"]) for r in dynamics))
|
|
self.assertFalse(is_valid_arp_age("H"))
|
|
self.assertTrue(is_valid_arp_age("03:22:07"))
|
|
# VLAN-subif row from TextFSM
|
|
sub = next(r for r in rows if r["ip"] == "192.166.1.73")
|
|
self.assertEqual(sub["interface"], "cgei-0/2/0/1.30")
|
|
self.assertEqual(sub["exter_vlan"], "30")
|
|
|
|
def test_nd6(self) -> None:
|
|
text = _section(self.log, "show nd6 cache", ("PAG3_", "show bgp"))
|
|
rows = normalize_nd6_cache(raw_text=text)
|
|
self.assertGreaterEqual(len(rows), 5)
|
|
addrs = {r["address"] for r in rows}
|
|
self.assertTrue(any("fe80::" in a for a in addrs))
|
|
|
|
def test_nd6_skips_clock_banner(self) -> None:
|
|
text = """
|
|
14:06:53 Beijing Sun Sep 20 2026
|
|
S-static NP-near packet RP-remote packet B-bgp HB-hotback D-dhcp
|
|
Total Cache Number Is:5398
|
|
Only Current Valid Items Are Shown Below:
|
|
Address Link-Address Age Status Interface Type
|
|
10:460::2 (incomplete) 1s Incomplete smartgroup2345.1120 NP
|
|
10:794::2 (incomplete) 1s Incomplete smartgroup2345.1940 NP
|
|
fe80::1 aaaa.bbbb.cccc 03:22:07 Reachable gei-0/0/0/1 S
|
|
"""
|
|
rows = normalize_nd6_cache(raw_text=text)
|
|
self.assertEqual(len(rows), 3)
|
|
self.assertNotIn("14:06:53", {r["address"] for r in rows})
|
|
self.assertEqual(rows[0]["address"], "10:460::2")
|
|
self.assertEqual(rows[0]["status"], "Incomplete")
|
|
self.assertEqual(rows[0]["type"], "NP")
|
|
self.assertEqual(rows[2]["age"], "03:22:07")
|
|
|
|
def test_bgp_peers(self) -> None:
|
|
v4 = _section(self.log, "show bgp vpnv4 unicast summary", ("show bgp ipv4",))
|
|
rows = normalize_bgp_peer(raw_text=v4, command="show bgp vpnv4 unicast summary")
|
|
self.assertGreaterEqual(len(rows), 8)
|
|
self.assertTrue(all(r["afi"] == "vpnv4" for r in rows))
|
|
est = [r for r in rows if r["state"] == "Established"]
|
|
conn = [r for r in rows if r["state"] == "Connect"]
|
|
self.assertGreaterEqual(len(est), 3)
|
|
self.assertGreaterEqual(len(conn), 3)
|
|
self.assertTrue(all("state_or_pfx" not in r for r in rows))
|
|
self.assertTrue(all(r["pfx_rcd"].isdigit() for r in est))
|
|
self.assertTrue(all(r["pfx_rcd"] == "" for r in conn))
|
|
one = next(r for r in rows if r["neighbor"] == "10.206.63.205")
|
|
self.assertEqual(one["state"], "Established")
|
|
self.assertEqual(one["pfx_rcd"], "642")
|
|
|
|
ipv4 = _section(self.log, "show bgp ipv4 unicast summary", ("show bgp vpnv6",))
|
|
rows2 = normalize_bgp_peer(raw_text=ipv4, command="show bgp ipv4 unicast summary")
|
|
self.assertGreaterEqual(len(rows2), 8)
|
|
self.assertTrue(all(r["afi"] == "ipv4" for r in rows2))
|
|
|
|
v6 = _section(self.log, "show bgp vpnv6 unicast summary", ("END",))
|
|
# file ends after vpnv6 — take remainder
|
|
if not v6.strip():
|
|
i = self.log.find("show bgp vpnv6 unicast summary")
|
|
v6 = self.log[i:] if i >= 0 else ""
|
|
rows3 = normalize_bgp_peer(raw_text=v6, command="show bgp vpnv6 unicast summary")
|
|
self.assertGreaterEqual(len(rows3), 5)
|
|
self.assertTrue(all(r["afi"] == "vpnv6" for r in rows3))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|