mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 05:30:46 +08:00
Decouple cutover HF from portrait and tighten ops UI.
Portrait stays on biz-monitor; cutover create/ensure owns purpose=cutover_hf slots with project collect override, multi-interval bindings, window pause, and industrial detail UX. Also expand ZTE collect/config parsers and TextFSM for route/BGP/L2VPN/VRRP coverage. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
a65c9fc420
commit
285c19e5a3
67 changed files with 9510 additions and 723 deletions
248
tests/test_zte_config_real_dump.py
Normal file
248
tests/test_zte_config_real_dump.py
Normal file
|
|
@ -0,0 +1,248 @@
|
|||
"""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*$", line)
|
||||
if m:
|
||||
a, mask = m.group(1), m.group(2) or ""
|
||||
block_ips.append(f"{a}/{mask}" if mask else a)
|
||||
continue
|
||||
m = re.match(r"(?i)^\s*ipv6\s+address\s+(\S+)\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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue