From 0dc63bb96a53184f101ea583a0e3616f94e63d65 Mon Sep 17 00:00:00 2001 From: oliver Date: Mon, 21 Sep 2026 14:34:43 +0800 Subject: [PATCH] Split ZTE L2VPN MAC outgoing into pw/AC/SID fields and skip separator rows. Co-authored-by: Cursor --- netx_api/biz_state/parsers/zte/l2vpn_mac.py | 118 +++++++++++++++--- netx_api/biz_state/profiles.py | 8 +- .../zte/zte_zxros_show_mac_l2vpn.textfsm | 3 +- tests/test_zte_extended_parsers.py | 66 ++++++++++ 4 files changed, 173 insertions(+), 22 deletions(-) diff --git a/netx_api/biz_state/parsers/zte/l2vpn_mac.py b/netx_api/biz_state/parsers/zte/l2vpn_mac.py index 4820155..b6dc07e 100644 --- a/netx_api/biz_state/parsers/zte/l2vpn_mac.py +++ b/netx_api/biz_state/parsers/zte/l2vpn_mac.py @@ -16,6 +16,84 @@ _MAC_RE = re.compile( r"(?P\S+)\s+(?P\S+)\s+(?P.+?)\s+(?P\S+)\s*$" ) +_SID_OUT_RE = re.compile( + r"(?i)^SID\s+(?P\S+)\s*,\s*(?P\S+)\s*$" +) +_AC_OUT_RE = re.compile( + r"^(?P\S+)\s*,\s*[EeIi]:(?P\S+)\s*$" +) +_PW_OUT_RE = re.compile(r"^(?P\S+)\s*,\s*(?P\S+)\s*$") + + +def _empty_outgoing(raw: str = "") -> dict[str, str]: + text = str(raw or "").strip() + return { + "outgoing": text[:256], + "pw": "", + "neighbor": "", + "ac_port": "", + "exter_vlan": "", + "vpn_sid": "", + "neighbor_sid": "", + } + + +def _parse_outgoing(raw: str) -> dict[str, str]: + """Split Outgoing Information into typed fields; keep raw in outgoing.""" + out = _empty_outgoing(raw) + text = out["outgoing"] + if not text: + return out + + m = _SID_OUT_RE.match(text) + if m: + out["vpn_sid"] = m.group("vpn_sid")[:128] + out["neighbor_sid"] = m.group("neighbor_sid")[:128] + return out + + m = _AC_OUT_RE.match(text) + if m: + out["ac_port"] = m.group("ac_port")[:128] + out["exter_vlan"] = m.group("exter_vlan")[:32] + return out + + # ESI / other opaque forms stay in outgoing only. + if text.upper().startswith("ESI:"): + return out + + m = _PW_OUT_RE.match(text) + if m: + out["pw"] = m.group("pw")[:128] + out["neighbor"] = m.group("neighbor")[:128] + return out + + return out + + +def _is_mac(token: str) -> bool: + t = str(token or "").strip() + return bool(re.fullmatch(r"[0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}", t)) + + +def _row( + *, + mac: str, + vpn: str, + vlan: str, + outgoing: str, + attribute: str, +) -> dict[str, Any] | None: + if not _is_mac(mac) or not vpn or vpn.lower() == "vpn": + return None + row = { + "mac": mac[:64], + "vpn": vpn[:128], + "vlan": str(vlan or "").strip()[:32], + "attribute": str(attribute or "").strip()[:64], + } + row.update(_parse_outgoing(outgoing)) + return row + def _map_fsm_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] @@ -23,21 +101,20 @@ def _map_fsm_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: for r in rows: mac = row_get(r, "MAC", "mac") vpn = row_get(r, "VPN", "vpn") - if not mac or mac.lower() == "mac": - continue key = (mac, vpn) if key in seen: continue - seen.add(key) - out.append( - { - "mac": mac[:64], - "vpn": vpn[:128], - "vlan": row_get(r, "VLAN", "vlan")[:32], - "outgoing": row_get(r, "OUTGOING", "outgoing")[:256], - "attribute": row_get(r, "ATTRIBUTE", "attribute")[:64], - } + mapped = _row( + mac=mac, + vpn=vpn, + vlan=row_get(r, "VLAN", "vlan"), + outgoing=row_get(r, "OUTGOING", "outgoing"), + attribute=row_get(r, "ATTRIBUTE", "attribute"), ) + if not mapped: + continue + seen.add(key) + out.append(mapped) return out @@ -54,16 +131,17 @@ def _hand_parse(*, raw_text: str, **_kw: Any) -> list[dict[str, Any]]: key = (m.group("mac"), m.group("vpn")) if key in seen: continue - seen.add(key) - out.append( - { - "mac": m.group("mac")[:64], - "vpn": m.group("vpn")[:128], - "vlan": m.group("vlan")[:32], - "outgoing": m.group("out").strip()[:256], - "attribute": m.group("attr")[:64], - } + mapped = _row( + mac=m.group("mac"), + vpn=m.group("vpn"), + vlan=m.group("vlan"), + outgoing=m.group("out").strip(), + attribute=m.group("attr"), ) + if not mapped: + continue + seen.add(key) + out.append(mapped) return out diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index b9d8c18..3012ffa 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -404,7 +404,13 @@ _L2VPN_MAC_FIELDS: list[FieldDef] = [ FieldDef("mac", length=64, indexed=True, is_key=True, display_name="MAC"), FieldDef("vpn", length=128, indexed=True, is_key=True, display_name="VPN"), FieldDef("vlan", length=32, role="meta", display_name="VLAN"), - FieldDef("outgoing", length=256, role="state", display_name="Outgoing"), + FieldDef("pw", length=128, role="state", display_name="PW"), + FieldDef("neighbor", length=128, role="state", display_name="Neighbor"), + FieldDef("ac_port", length=128, role="state", display_name="AC Port"), + FieldDef("exter_vlan", length=32, role="state", display_name="Exter VLAN"), + FieldDef("vpn_sid", length=128, role="state", display_name="VPN SID"), + FieldDef("neighbor_sid", length=128, role="state", display_name="Neighbor SID"), + FieldDef("outgoing", length=256, role="meta", display_name="Outgoing"), FieldDef("attribute", length=64, role="meta", display_name="Attribute"), ] diff --git a/netx_api/cli_templates/zte/zte_zxros_show_mac_l2vpn.textfsm b/netx_api/cli_templates/zte/zte_zxros_show_mac_l2vpn.textfsm index 11ab13c..10b404c 100644 --- a/netx_api/cli_templates/zte/zte_zxros_show_mac_l2vpn.textfsm +++ b/netx_api/cli_templates/zte/zte_zxros_show_mac_l2vpn.textfsm @@ -1,4 +1,4 @@ -Value MAC (\S+) +Value MAC ([0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}) Value VPN (\S+) Value VLAN (\S+) Value OUTGOING (.*) @@ -13,6 +13,7 @@ Start Entries ^${MAC}\s+${VPN}\s+${VLAN}\s+${OUTGOING}\s+${ATTRIBUTE}\s*$$ -> Record + ^-+(\s+-+)*\s*$$ ^\s*$$ ^\S+\s*#\s*$$ -> End ^.*#\s*$$ -> End diff --git a/tests/test_zte_extended_parsers.py b/tests/test_zte_extended_parsers.py index 9596cbc..481c810 100644 --- a/tests/test_zte_extended_parsers.py +++ b/tests/test_zte_extended_parsers.py @@ -14,6 +14,7 @@ from netx_api.biz_state.parsers.zte import ( normalize_bgp_route, normalize_ip_route, normalize_ipv6_route, + normalize_l2vpn_mac, normalize_l2vpn_pw, normalize_l2vpn_pw_detail, normalize_optical_brief, @@ -129,6 +130,23 @@ pw2 10.0.0.2 128 Ethernet S DOWN - - L:VPN_B pw3 10.0.0.3 128 Ethernet UP 101 201 W:100 """ +_L2VPN_MAC = """ +Total MAC Entries: 502 + +Headers: Src--Source filter, Dst--Destination filter + E--Exter-VLAN ID I--Inter-VLAN ID + +MAC VPN VLAN Outgoing Information Attribute +-------------- -------------- ---- -------------------------------- ------------ +0000.9999.0002 EVPNvpls501 0 SID fc00:a000:1005::567b:0, 2400:9800:7000:1005:: Static +0011.2232.42e3 mxy-bvi-ag4-ag3 0 SID fc00:a000:1002::3469:0, 2400:9800:7000:1002:: Dynamic +0099.2232.7788 css-irb-ag3-1 0 ESI:0:109501000010000001 Static +004b.0100.0001 css-irb-ag3-1 0 smartgroup11.31, E:31 Dynamic +0011.2232.42e4 mxy-kompella-2 0 auto_pw3, 24.11.100.5 Dynamic +0011.2232.42e4 mxy-kompella-1 0 PW1,1.1.1.1 Dynamic +0000.2200.0001 EVPNvpls501 0 smartgroup2345.2501, E:2501 Dynamic +""" + _L2VPN_PW_DETAIL = """ Service type and instance name:[VPLS qualified demo-vpls-1] Peer IP address : 10.0.0.6 VCID : 19001 @@ -276,6 +294,54 @@ class ZteExtendedParserTests(unittest.TestCase): self.assertEqual(by["cgei-0/1/0/2:1"]["tx_power"], "2.3,2.4,2.3,2.2") self.assertEqual(by["xgei-0/2/0/1"]["rx_power"], "-2.4") + def test_l2vpn_mac_outgoing_split(self) -> None: + fsm_rows = apply_rule( + platform="zte_zxros", + rule_key="zte_zxros_show_mac_l2vpn", + text=_L2VPN_MAC, + command="show mac l2vpn", + ) + self.assertEqual(len(fsm_rows), 7) + self.assertTrue(all("----" not in r.get("MAC", "") for r in fsm_rows)) + + rows = normalize_l2vpn_mac( + raw_text=_L2VPN_MAC, + vendor="zte", + device_type="zte_zxros", + command="show mac l2vpn | one-line", + ) + self.assertEqual(len(rows), 7) + by = {(r["mac"], r["vpn"]): r for r in rows} + + sid = by[("0000.9999.0002", "EVPNvpls501")] + self.assertEqual(sid["vpn_sid"], "fc00:a000:1005::567b:0") + self.assertEqual(sid["neighbor_sid"], "2400:9800:7000:1005::") + self.assertEqual(sid["pw"], "") + self.assertEqual(sid["ac_port"], "") + self.assertTrue(sid["outgoing"].startswith("SID ")) + + ac = by[("004b.0100.0001", "css-irb-ag3-1")] + self.assertEqual(ac["ac_port"], "smartgroup11.31") + self.assertEqual(ac["exter_vlan"], "31") + self.assertEqual(ac["pw"], "") + + pw = by[("0011.2232.42e4", "mxy-kompella-2")] + self.assertEqual(pw["pw"], "auto_pw3") + self.assertEqual(pw["neighbor"], "24.11.100.5") + self.assertEqual(pw["ac_port"], "") + + pw2 = by[("0011.2232.42e4", "mxy-kompella-1")] + self.assertEqual(pw2["pw"], "PW1") + self.assertEqual(pw2["neighbor"], "1.1.1.1") + + esi = by[("0099.2232.7788", "css-irb-ag3-1")] + self.assertEqual(esi["outgoing"], "ESI:0:109501000010000001") + self.assertEqual(esi["pw"], "") + self.assertEqual(esi["vpn_sid"], "") + + fields = {f.name for f in metric_field_map()["l2vpn_mac"]} + self.assertTrue({"pw", "neighbor", "ac_port", "exter_vlan", "vpn_sid", "neighbor_sid"} <= fields) + def test_bgp_route_and_aux_render(self) -> None: routes = normalize_bgp_route( raw_text=_BGP_ROUTE_IN,