hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 00:50:46 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2

Split ZTE L2VPN MAC outgoing into pw/AC/SID fields and skip separator rows.

Browse source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-21 14:34:43 +08:00
parent 4b38427679
commit 0dc63bb96a
4 changed files with 173 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

118
netx_api/biz_state/parsers/zte/l2vpn_mac.py
View file

@ -16,6 +16,84 @@ _MAC_RE = re.compile(
r"(?P<vpn>\S+)\s+(?P<vlan>\S+)\s+(?P<out>.+?)\s+(?P<attr>\S+)\s*$"
)
_SID_OUT_RE = re.compile(
r"(?i)^SID\s+(?P<vpn_sid>\S+)\s*,\s*(?P<neighbor_sid>\S+)\s*$"
)
_AC_OUT_RE = re.compile(
r"^(?P<ac_port>\S+)\s*,\s*[EeIi]:(?P<exter_vlan>\S+)\s*$"
)
_PW_OUT_RE = re.compile(r"^(?P<pw>\S+)\s*,\s*(?P<neighbor>\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

8
netx_api/biz_state/profiles.py
View file

@ -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"),
]

3
netx_api/cli_templates/zte/zte_zxros_show_mac_l2vpn.textfsm
View file

@ -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

66
tests/test_zte_extended_parsers.py
View file

@ -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,