mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 03:10: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>
253 lines
10 KiB
Python
253 lines
10 KiB
Python
"""Regression: config-intent parsers vs real ``test/show-zte/show-config``.
|
|
|
|
Skipped when the dump is absent. Validates counts and field fidelity against
|
|
raw MIM section lines (not synthetic fixtures).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from netx_api.biz_state.parsers.zte.config_bgp_peer import normalize_config_bgp_peer
|
|
from netx_api.biz_state.parsers.zte.config_common import extract_mim_section
|
|
from netx_api.biz_state.parsers.zte.config_interface import normalize_config_interface
|
|
from netx_api.biz_state.parsers.zte.config_isis import normalize_config_isis
|
|
from netx_api.biz_state.parsers.zte.config_l2vpn_pw import normalize_config_l2vpn_pw
|
|
from netx_api.biz_state.parsers.zte.config_ospf import normalize_config_ospf
|
|
from netx_api.biz_state.parsers.zte.config_static_route import (
|
|
_V4_HEAD,
|
|
_parse_v4,
|
|
_parse_v4_rest,
|
|
normalize_config_static_route,
|
|
)
|
|
from netx_api.biz_state.parsers.zte.config_vrf import normalize_config_vrf
|
|
|
|
_CONFIG = Path(__file__).resolve().parents[2] / "test" / "show-zte" / "show-config"
|
|
|
|
|
|
@unittest.skipUnless(_CONFIG.is_file(), "test/show-zte/show-config not present")
|
|
class ZteConfigRealDumpTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.text = _CONFIG.read_text(encoding="utf-8", errors="replace").replace("\r\n", "\n")
|
|
|
|
def test_vrf_count_and_rd(self) -> None:
|
|
body = extract_mim_section(self.text, "vrf")
|
|
raw = re.findall(r"(?im)^\s*ip\s+vrf\s+(\S+)\s*$", body)
|
|
rows = normalize_config_vrf(raw_text=self.text, command="show running-config vrf")
|
|
self.assertEqual(len(rows), len(raw))
|
|
by = {r["vrf_name"]: r for r in rows}
|
|
cur = None
|
|
expect_rd: dict[str, str] = {}
|
|
for line in body.splitlines():
|
|
m = re.match(r"(?i)^\s*ip\s+vrf\s+(\S+)\s*$", line)
|
|
if m:
|
|
cur = m.group(1)
|
|
continue
|
|
m = re.match(r"(?i)^\s*rd\s+(\S+)\s*$", line)
|
|
if m and cur:
|
|
expect_rd[cur] = m.group(1)
|
|
for name, rd in expect_rd.items():
|
|
self.assertEqual(by[name]["rd"], rd, msg=name)
|
|
|
|
def test_interface_merged_fields(self) -> None:
|
|
body = extract_mim_section(self.text, "if-intf")
|
|
rows = normalize_config_interface(
|
|
raw_text=self.text, command="show running-config if-intf"
|
|
)
|
|
by = {r["interface"]: r for r in rows}
|
|
exp: dict[str, dict] = {}
|
|
cur = None
|
|
block_ips: list[str] = []
|
|
block_ip6s: list[str] = []
|
|
block_admin = None
|
|
block_vrf = ""
|
|
block_mtu = ""
|
|
|
|
def commit(name: str) -> None:
|
|
nonlocal block_admin, block_vrf, block_mtu, block_ips, block_ip6s
|
|
e = exp[name]
|
|
if block_admin is not None:
|
|
e["admin"] = block_admin
|
|
if block_vrf:
|
|
e["vrf"] = block_vrf
|
|
if block_mtu:
|
|
e["mtu"] = block_mtu
|
|
for ip in block_ips:
|
|
if ip not in e["ips"]:
|
|
e["ips"].append(ip)
|
|
for ip in block_ip6s:
|
|
if ip not in e["ip6s"]:
|
|
e["ip6s"].append(ip)
|
|
|
|
for line in body.splitlines():
|
|
m = re.match(r"(?i)^\s*interface\s+(\S+)\s*$", line)
|
|
if m:
|
|
if cur:
|
|
commit(cur)
|
|
cur = m.group(1)
|
|
exp.setdefault(cur, {"vrf": "", "admin": "up", "ips": [], "ip6s": [], "mtu": ""})
|
|
block_ips, block_ip6s = [], []
|
|
block_admin, block_vrf, block_mtu = None, "", ""
|
|
continue
|
|
if not cur:
|
|
continue
|
|
if re.match(r"^\$\s*$", line):
|
|
commit(cur)
|
|
cur = None
|
|
continue
|
|
low = line.strip().lower()
|
|
if low == "shutdown":
|
|
block_admin = "down"
|
|
elif low == "no shutdown":
|
|
block_admin = "up"
|
|
m = re.match(r"(?i)^\s*ip\s+vrf\s+forwarding\s+(\S+)\s*$", line)
|
|
if m:
|
|
block_vrf = m.group(1)
|
|
continue
|
|
m = re.match(
|
|
r"(?i)^\s*ip\s+address\s+(\S+)(?:\s+(\S+))?(?:\s+secondary)?\s*$",
|
|
line,
|
|
)
|
|
if m:
|
|
a, mask = m.group(1), m.group(2) or ""
|
|
if mask.lower() == "secondary":
|
|
mask = ""
|
|
block_ips.append(f"{a}/{mask}" if mask else a)
|
|
continue
|
|
m = re.match(r"(?i)^\s*ipv6\s+address\s+(\S+)(?:\s+secondary)?\s*$", line)
|
|
if m:
|
|
block_ip6s.append(m.group(1))
|
|
continue
|
|
m = re.match(r"(?i)^\s*mtu\s+(\d+)\s*$", line)
|
|
if m:
|
|
block_mtu = m.group(1)
|
|
if cur:
|
|
commit(cur)
|
|
|
|
self.assertEqual(len(rows), len(exp))
|
|
for name, e in exp.items():
|
|
g = by[name]
|
|
self.assertEqual(g["vrf"], e["vrf"], msg=name)
|
|
self.assertEqual(g["admin"], e["admin"], msg=name)
|
|
self.assertEqual(
|
|
[x for x in (g["ip_address"] or "").split(",") if x],
|
|
e["ips"],
|
|
msg=name,
|
|
)
|
|
|
|
def test_static_route_line_coverage(self) -> None:
|
|
body = extract_mim_section(self.text, "static")
|
|
raw = [ln for ln in body.splitlines() if re.match(r"(?i)^\s*ip\s+route\s+", ln)]
|
|
rows = normalize_config_static_route(
|
|
raw_text=self.text, command="show running-config static"
|
|
)
|
|
self.assertEqual(len(rows), len(raw))
|
|
self.assertEqual(rows, _parse_v4(body))
|
|
for ln in raw:
|
|
m = _V4_HEAD.match(ln)
|
|
self.assertIsNotNone(m, msg=ln[:120])
|
|
assert m is not None
|
|
extra = _parse_v4_rest(m.group("rest") or "")
|
|
hit = any(
|
|
r["vrf"] == (m.group("vrf") or "").strip()
|
|
and r["prefix"] == m.group("prefix")
|
|
and r["mask"] == m.group("mask")
|
|
and r["next_hop"] == extra["next_hop"]
|
|
and r["interface"] == extra["interface"]
|
|
and r["nexthop_vrf"] == extra["nexthop_vrf"]
|
|
for r in rows
|
|
)
|
|
self.assertTrue(hit, msg=ln[:140])
|
|
bfd_raw = sum(1 for ln in raw if re.search(r"(?i)\bbfd\b", ln))
|
|
self.assertEqual(sum(1 for r in rows if r["bfd"]), bfd_raw)
|
|
|
|
def test_static_route_v6(self) -> None:
|
|
body = extract_mim_section(self.text, "ipv6-static-route")
|
|
raw = [ln for ln in body.splitlines() if re.match(r"(?i)^\s*ipv6\s+route\s+", ln)]
|
|
rows = normalize_config_static_route(
|
|
raw_text=self.text, command="show running-config ipv6-static-route"
|
|
)
|
|
self.assertEqual(len(rows), len(raw))
|
|
|
|
def test_bgp_activate_and_no_secrets(self) -> None:
|
|
body = extract_mim_section(self.text, "bgp")
|
|
activates = {
|
|
m.group(1)
|
|
for m in re.finditer(
|
|
r"(?im)^\s*neighbor\s+(\S+)\s+activate(?:\s+disable)?\s*$", body
|
|
)
|
|
}
|
|
rows = normalize_config_bgp_peer(
|
|
raw_text=self.text, command="show running-config bgp"
|
|
)
|
|
blob = " ".join(str(v) for r in rows for v in r.values()).lower()
|
|
self.assertNotIn("password", blob)
|
|
self.assertNotIn("cipher", blob)
|
|
# Every activate token appears as neighbor (IP) or peer_group (name)
|
|
covered = {r["neighbor"] for r in rows if r["neighbor"]} | {
|
|
r["peer_group"] for r in rows if r["peer_group"]
|
|
}
|
|
self.assertTrue(activates <= covered)
|
|
# neighbor column is IP-only
|
|
for r in rows:
|
|
if r["neighbor"]:
|
|
self.assertTrue(
|
|
re.match(r"^\d{1,3}(?:\.\d{1,3}){3}$", r["neighbor"])
|
|
or ":" in r["neighbor"],
|
|
msg=r["neighbor"],
|
|
)
|
|
|
|
def test_l2vpn_neighbour_coverage(self) -> None:
|
|
body = extract_mim_section(self.text, "l2vpn")
|
|
neis = set(
|
|
re.findall(r"(?im)^\s*neighbour\s+(\S+)\s+vcid\s+(\S+)\s*$", body)
|
|
)
|
|
rows = normalize_config_l2vpn_pw(
|
|
raw_text=self.text, command="show running-config l2vpn"
|
|
)
|
|
parsed = {(r["peer"], r["vcid"]) for r in rows if r["peer"] and r["vcid"]}
|
|
self.assertEqual(parsed & neis, neis)
|
|
for r in rows:
|
|
if r["peer"] and r["vcid"]:
|
|
self.assertIn((r["peer"], r["vcid"]), neis)
|
|
|
|
def test_ospf_v2_interface_coverage(self) -> None:
|
|
body = extract_mim_section(self.text, "ospfv2")
|
|
procs = re.findall(
|
|
r"(?im)^\s*router\s+ospf\s+(\S+)(?:\s+vrf\s+(\S+))?\s*$", body
|
|
)
|
|
ifaces = re.findall(r"(?im)^\s*interface\s+(\S+)\s*$", body)
|
|
rows = normalize_config_ospf(
|
|
raw_text=self.text, command="show running-config ospfv2"
|
|
)
|
|
self.assertTrue(all(r["af"] == "ipv4" for r in rows))
|
|
parsed_if = {r["interface"] for r in rows if r["interface"]}
|
|
self.assertEqual(parsed_if, set(ifaces))
|
|
self.assertEqual(len({(r["process_id"], r["vrf"]) for r in rows}), len(procs))
|
|
blob = " ".join(str(v) for r in rows for v in r.values()).lower()
|
|
self.assertNotIn("encrypted", blob)
|
|
self.assertNotIn("message-digest-key", blob)
|
|
|
|
def test_ospf_v3_and_isis(self) -> None:
|
|
v3_body = extract_mim_section(self.text, "ospfv3")
|
|
v3_if = re.findall(r"(?im)^\s*interface\s+(\S+)\s*$", v3_body)
|
|
v3 = normalize_config_ospf(
|
|
raw_text=self.text, command="show running-config ospfv3"
|
|
)
|
|
self.assertEqual({r["interface"] for r in v3 if r["interface"]}, set(v3_if))
|
|
|
|
isis_body = extract_mim_section(self.text, "isis")
|
|
isis_if = re.findall(r"(?im)^\s*interface\s+(\S+)\s*$", isis_body)
|
|
rows = normalize_config_isis(
|
|
raw_text=self.text, command="show running-config isis"
|
|
)
|
|
self.assertEqual({r["interface"] for r in rows}, set(isis_if))
|
|
blob = " ".join(str(v) for r in rows for v in r.values()).lower()
|
|
self.assertNotIn("encrypted", blob)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|