diff --git a/netx_api/biz_state/collect_runner.py b/netx_api/biz_state/collect_runner.py index b17569a..5476b31 100644 --- a/netx_api/biz_state/collect_runner.py +++ b/netx_api/biz_state/collect_runner.py @@ -187,6 +187,7 @@ _GENERIC_METRICS = { "ip_route", "ipv6_route", "l2vpn_pw", + "l2vpn_pw_detail", "l2vpn_mac", "evpn_mac", "config_vrf", diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index 29fab80..26636a5 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -499,6 +499,7 @@ def _default_zte_status_sheets() -> list[dict[str, Any]]: _default_sheet_for_metric("optical_brief", compare_roles=("state",)), *_bgp_afi_sheets(), _default_sheet_for_metric("l2vpn_pw", compare_roles=("state",)), + _default_sheet_for_metric("l2vpn_pw_detail", compare_roles=("state",)), _default_lldp_sheet(), ] diff --git a/netx_api/biz_state/parsers/zte/__init__.py b/netx_api/biz_state/parsers/zte/__init__.py index 3ef653c..18a3794 100644 --- a/netx_api/biz_state/parsers/zte/__init__.py +++ b/netx_api/biz_state/parsers/zte/__init__.py @@ -27,6 +27,7 @@ from .ipv6_route import normalize_ipv6_route from .isis_adjacency import normalize_isis_adjacency from .l2vpn_mac import normalize_l2vpn_mac from .l2vpn_pw import normalize_l2vpn_pw +from .l2vpn_pw_detail import normalize_l2vpn_pw_detail from .nd6_cache import normalize_nd6_cache from .optical_brief import normalize_optical_brief from .ospf_neighbor import normalize_ospf_neighbor @@ -49,6 +50,7 @@ PARSERS: dict[str, NormalizeFn] = { "ip_route": normalize_ip_route, "ipv6_route": normalize_ipv6_route, "l2vpn_pw": normalize_l2vpn_pw, + "l2vpn_pw_detail": normalize_l2vpn_pw_detail, "l2vpn_mac": normalize_l2vpn_mac, "evpn_mac": normalize_evpn_mac, "config_vrf": normalize_config_vrf, @@ -82,6 +84,7 @@ __all__ = [ "normalize_isis_adjacency", "normalize_l2vpn_mac", "normalize_l2vpn_pw", + "normalize_l2vpn_pw_detail", "normalize_nd6_cache", "normalize_optical_brief", "normalize_ospf_neighbor", diff --git a/netx_api/biz_state/parsers/zte/l2vpn_pw_detail.py b/netx_api/biz_state/parsers/zte/l2vpn_pw_detail.py new file mode 100644 index 0000000..aed71f9 --- /dev/null +++ b/netx_api/biz_state/parsers/zte/l2vpn_pw_detail.py @@ -0,0 +1,216 @@ +"""ZTE: show l2vpn forwardinfo detail (per-PW VPLS/VPWS block).""" + +from __future__ import annotations + +import re +from typing import Any, Mapping + +from ....lldp_shared import resolve_vendor_key +from ....ntc_parse import apply_rules, resolve_cli_platform, row_get +from ..common.pipeline import prefer_fsm + +RULE_KEYS = ("zte_zxros_show_l2vpn_forwardinfo_detail",) + +_SERVICE_RE = re.compile(r"^Service type and instance name:\[([^\]]+)\]\s*$", re.I) +# Two-column "Label : value Label : value" or single "Label : value" +_KV_PAIR_RE = re.compile( + r"([A-Za-z][A-Za-z0-9 /|_.-]*)\s*:\s*" + r"(\S.*?)" + r"(?=\s{2,}[A-Za-z][A-Za-z0-9 /|_.-]*\s*:|\s*$)" +) + + +def _norm_label(label: str) -> str: + return re.sub(r"\s+", " ", str(label or "").strip()).lower() + + +_LABEL_MAP = { + "peer ip address": "peer", + "vcid": "vcid", + "connection mode": "conn_mode", + "vcid extend": "vcid_ext", + "signaling protocol": "signaling", + "vc type": "vc_type", + "last status change time": "last_change", + "create time": "create_time", + "mpls vc local label": "local_label", + "remote label": "remote_label", + "pw name": "pw_name", + "control word": "control_word", + "activation status": "activation_status", + "band width": "bandwidth", + "tunnel destination": "tunnel_dest", + "related interface name": "related_if", + "frr type": "frr_type", + "vc status": "vc_status", + "remote status": "remote_status", + "mc selection": "mc_selection", + "vccv cc type": "vccv_cc", + "vccv cv type": "vccv_cv", +} + + +def _empty() -> dict[str, str]: + return { + "service_instance": "", + "pw_name": "", + "peer": "", + "vcid": "", + "conn_mode": "", + "vcid_ext": "", + "signaling": "", + "vc_type": "", + "last_change": "", + "create_time": "", + "local_label": "", + "remote_label": "", + "control_word": "", + "activation_status": "", + "bandwidth": "", + "tunnel_dest": "", + "related_if": "", + "frr_type": "", + "vc_status": "", + "remote_status": "", + "mc_selection": "", + "vccv_cc": "", + "vccv_cv": "", + } + + +def _row_out(cur: dict[str, str]) -> dict[str, Any] | None: + pw = str(cur.get("pw_name") or "").strip() + if not pw: + return None + return { + "service_instance": str(cur.get("service_instance") or "")[:256], + "pw_name": pw[:128], + "peer": str(cur.get("peer") or "")[:64], + "vcid": str(cur.get("vcid") or "")[:64], + "conn_mode": str(cur.get("conn_mode") or "")[:32], + "vcid_ext": str(cur.get("vcid_ext") or "")[:32], + "signaling": str(cur.get("signaling") or "")[:32], + "vc_type": str(cur.get("vc_type") or "")[:32], + "last_change": str(cur.get("last_change") or "")[:64], + "create_time": str(cur.get("create_time") or "")[:64], + "local_label": str(cur.get("local_label") or "")[:32], + "remote_label": str(cur.get("remote_label") or "")[:32], + "control_word": str(cur.get("control_word") or "")[:32], + "activation_status": str(cur.get("activation_status") or "")[:32], + "bandwidth": str(cur.get("bandwidth") or "")[:64], + "tunnel_dest": str(cur.get("tunnel_dest") or "")[:64], + "related_if": str(cur.get("related_if") or "")[:128], + "frr_type": str(cur.get("frr_type") or "")[:32], + "vc_status": str(cur.get("vc_status") or "")[:32], + "remote_status": str(cur.get("remote_status") or "")[:32], + "mc_selection": str(cur.get("mc_selection") or "")[:32], + "vccv_cc": str(cur.get("vccv_cc") or "")[:64], + "vccv_cv": str(cur.get("vccv_cv") or "")[:64], + } + + +def _map_fsm_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + seen: set[str] = set() + for r in rows: + cur = _empty() + cur["service_instance"] = row_get(r, "SERVICE_INSTANCE", "service_instance") + cur["pw_name"] = row_get(r, "PW_NAME", "pw_name") + cur["peer"] = row_get(r, "PEER", "peer") + cur["vcid"] = row_get(r, "VCID", "vcid") + cur["conn_mode"] = row_get(r, "CONN_MODE", "conn_mode") + cur["vcid_ext"] = row_get(r, "VCID_EXT", "vcid_ext") + cur["signaling"] = row_get(r, "SIGNALING", "signaling") + cur["vc_type"] = row_get(r, "VC_TYPE", "vc_type") + cur["last_change"] = row_get(r, "LAST_CHANGE", "last_change") + cur["create_time"] = row_get(r, "CREATE_TIME", "create_time") + cur["local_label"] = row_get(r, "LOCAL_LABEL", "local_label") + cur["remote_label"] = row_get(r, "REMOTE_LABEL", "remote_label") + cur["control_word"] = row_get(r, "CONTROL_WORD", "control_word") + cur["activation_status"] = row_get(r, "ACTIVATION", "activation_status") + cur["bandwidth"] = row_get(r, "BANDWIDTH", "bandwidth") + cur["tunnel_dest"] = row_get(r, "TUNNEL_DEST", "tunnel_dest") + cur["related_if"] = row_get(r, "RELATED_IF", "related_if") + cur["frr_type"] = row_get(r, "FRR_TYPE", "frr_type") + cur["vc_status"] = row_get(r, "VC_STATUS", "vc_status") + cur["remote_status"] = row_get(r, "REMOTE_STATUS", "remote_status") + cur["mc_selection"] = row_get(r, "MC_SELECTION", "mc_selection") + cur["vccv_cc"] = row_get(r, "VCCV_CC", "vccv_cc") + cur["vccv_cv"] = row_get(r, "VCCV_CV", "vccv_cv") + row = _row_out(cur) + if not row or not str(row.get("vc_status") or "").strip(): + continue + key = f"{row['pw_name']}|{row['peer']}|{row['vcid']}" + if key in seen: + continue + seen.add(key) + out.append(row) + return out + + +def _hand_parse(*, raw_text: str, **_kw: Any) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + seen: set[str] = set() + cur: dict[str, str] | None = None + + def _flush() -> None: + nonlocal cur + if not cur: + return + row = _row_out(cur) + cur = None + if not row: + return + key = f"{row['pw_name']}|{row['peer']}|{row['vcid']}" + if key in seen: + return + seen.add(key) + out.append(row) + + for raw in str(raw_text or "").splitlines(): + line = raw.rstrip() + if not line.strip(): + continue + sm = _SERVICE_RE.match(line.strip()) + if sm: + _flush() + cur = _empty() + cur["service_instance"] = sm.group(1).strip() + continue + if cur is None: + continue + for m in _KV_PAIR_RE.finditer(line): + field = _LABEL_MAP.get(_norm_label(m.group(1))) + if not field: + continue + cur[field] = m.group(2).strip() + _flush() + return out + + +def normalize_l2vpn_pw_detail( + *, + raw_text: str, + fsm_tables: Mapping[str, list[dict[str, Any]]] | None = None, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = params + tables = dict(fsm_tables or {}) + if not any(tables.get(k) for k in RULE_KEYS): + platform = resolve_cli_platform( + vendor=vendor, + device_type=device_type, + vendor_key=resolve_vendor_key(vendor, device_type), + ) + cmd = str(command or "show l2vpn forwardinfo detail").strip() + if platform and cmd: + tables = apply_rules( + platform=platform, text=raw_text, rule_keys=RULE_KEYS, command=cmd + ) + return prefer_fsm(tables, RULE_KEYS, _map_fsm_rows, _hand_parse, raw_text=raw_text) + + +normalize_l2vpn_pw_detail.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index 066b4a8..b9d743e 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -375,6 +375,30 @@ _L2VPN_PW_FIELDS: list[FieldDef] = [ FieldDef("vpn_owner", length=256, role="meta", display_name="VPNOwner"), ] +_L2VPN_PW_DETAIL_FIELDS: list[FieldDef] = [ + FieldDef("pw_name", length=128, indexed=True, is_key=True, display_name="PW"), + FieldDef("peer", length=64, indexed=True, is_key=True, display_name="Peer"), + FieldDef("vcid", length=64, indexed=True, is_key=True, display_name="VCID"), + FieldDef("vc_status", length=32, role="state", display_name="VC Status"), + FieldDef("remote_status", length=32, role="state", display_name="Remote"), + FieldDef("activation_status", length=32, role="state", display_name="Activation"), + FieldDef("service_instance", length=256, role="meta", display_name="Service"), + FieldDef("conn_mode", length=32, role="meta", display_name="Mode"), + FieldDef("signaling", length=32, role="meta", display_name="Signaling"), + FieldDef("vc_type", length=32, role="meta", display_name="VC Type"), + FieldDef("control_word", length=32, role="meta", display_name="CW"), + FieldDef("local_label", length=32, role="meta", display_name="Local Label"), + FieldDef("remote_label", length=32, role="meta", display_name="Remote Label"), + FieldDef("tunnel_dest", length=64, role="meta", display_name="Tunnel Dest"), + FieldDef("related_if", length=128, is_interface=True, role="meta", display_name="Related IF"), + FieldDef("frr_type", length=32, role="meta", display_name="FRR"), + FieldDef("vccv_cc", length=64, role="meta", display_name="VCCV CC"), + FieldDef("vccv_cv", length=64, role="meta", display_name="VCCV CV"), + FieldDef("bandwidth", length=64, role="meta", display_name="Bandwidth"), + FieldDef("create_time", length=64, role="meta", display_name="Create"), + FieldDef("last_change", length=64, role="meta", display_name="Last Change"), +] + _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"), @@ -1085,15 +1109,31 @@ def _zte_status_profiles() -> list[ParseProfile]: parser_id="l2vpn_pw", title="L2VPN PW State", command_template="show l2vpn forwardinfo | one-line", - match=r"(?i)^\s*show\s+l2vpn\s+forwardinfo(?:\s*\|\s*one-line)?\s*$", + match=r"(?i)^\s*show\s+l2vpn\s+forwardinfo(?!\s+detail)(?:\s*\|\s*one-line)?\s*$", textfsm_command="show l2vpn forwardinfo", - description="L2VPN pseudowire forwardinfo state.", + description="L2VPN pseudowire forwardinfo state (brief table).", fields=list(_L2VPN_PW_FIELDS), tags=["l2vpn", "pw", "status"], sort_order=470, enabled=True, kind="collect", ), + ParseProfile( + profile_id="zte.l2vpn_pw_detail", + vendor_key="zte", + metric_id="l2vpn_pw_detail", + parser_id="l2vpn_pw_detail", + title="L2VPN PW Detail", + command_template="show l2vpn forwardinfo detail", + match=r"(?i)^\s*show\s+l2vpn\s+forwardinfo\s+detail(?:\s*\|.*)?\s*$", + textfsm_command="show l2vpn forwardinfo detail", + description="L2VPN PW forwardinfo detail (VPLS/VPWS block: VC/remote/activation).", + fields=list(_L2VPN_PW_DETAIL_FIELDS), + tags=["l2vpn", "pw", "status", "detail"], + sort_order=475, + enabled=True, + kind="collect", + ), ParseProfile( profile_id="zte.l2vpn_mac", vendor_key="zte", diff --git a/netx_api/cli_templates/index b/netx_api/cli_templates/index index 71aa45c..ae00425 100644 --- a/netx_api/cli_templates/index +++ b/netx_api/cli_templates/index @@ -16,6 +16,7 @@ zte/zte_zxros_show_opticalinfo_brief.textfsm, .*, zte_zxros, sh[[ow]] opticalinf zte/zte_zxros_show_bgp_neighbor_routes.textfsm, .*, zte_zxros, sh[[ow]] bgp .+ nei[[ghbor]] (in|out) zte/zte_zxros_show_ip_forwarding_route.textfsm, .*, zte_zxros, sh[[ow]] ip forw[[arding]] rou[[te]] zte/zte_zxros_show_ipv6_forwarding_route.textfsm, .*, zte_zxros, sh[[ow]] ipv6 forw[[arding]] rou[[te]] +zte/zte_zxros_show_l2vpn_forwardinfo_detail.textfsm, .*, zte_zxros, sh[[ow]] l2vpn forw[[ardinfo]] det[[ail]] zte/zte_zxros_show_l2vpn_forwardinfo.textfsm, .*, zte_zxros, sh[[ow]] l2vpn forw[[ardinfo]] zte/zte_zxros_show_mac_l2vpn.textfsm, .*, zte_zxros, sh[[ow]] mac l2vpn zte/zte_zxros_show_bgp_evpn_mac.textfsm, .*, zte_zxros, sh[[ow]] bgp evpn mac diff --git a/netx_api/cli_templates/zte/zte_zxros_show_l2vpn_forwardinfo_detail.textfsm b/netx_api/cli_templates/zte/zte_zxros_show_l2vpn_forwardinfo_detail.textfsm new file mode 100644 index 0000000..5075c2d --- /dev/null +++ b/netx_api/cli_templates/zte/zte_zxros_show_l2vpn_forwardinfo_detail.textfsm @@ -0,0 +1,50 @@ +Value Required,Filldown SERVICE_INSTANCE (.+?) +Value Filldown PEER (\S+) +Value Filldown VCID (\S+) +Value Filldown CONN_MODE (\S+) +Value Filldown VCID_EXT (\S+) +Value Filldown SIGNALING (\S+) +Value Filldown VC_TYPE (\S+) +Value Filldown LAST_CHANGE (\S+) +Value Filldown CREATE_TIME (.+?) +Value Filldown LOCAL_LABEL (\S+) +Value Filldown REMOTE_LABEL (\S+) +Value Filldown PW_NAME (\S+) +Value Filldown CONTROL_WORD (\S+) +Value Filldown ACTIVATION (\S+) +Value Filldown BANDWIDTH (.+?) +Value Filldown TUNNEL_DEST (\S+) +Value Filldown RELATED_IF (\S+) +Value Filldown FRR_TYPE (\S+) +Value Filldown VC_STATUS (\S+) +Value Filldown REMOTE_STATUS (\S+) +Value Filldown MC_SELECTION (\S+) +Value Filldown VCCV_CC (\S+) +Value VCCV_CV (\S+) + +Start + ^Service type and instance name:\[${SERVICE_INSTANCE}\] + ^\s+Peer IP address\s+:\s+${PEER}\s+VCID\s+:\s+${VCID}\s*$$ + ^\s+Connection mode\s+:\s+${CONN_MODE}\s+VCID Extend\s+:\s+${VCID_EXT}\s*$$ + ^\s+Signaling protocol\s+:\s+${SIGNALING}\s+VC type\s+:\s+${VC_TYPE}\s*$$ + ^\s+Last status change time\s+:\s+${LAST_CHANGE}\s+Create time\s+:\s+${CREATE_TIME}\s*$$ + ^\s+MPLS VC local label\s+:\s+${LOCAL_LABEL}\s+Remote label\s+:\s+${REMOTE_LABEL}\s*$$ + ^\s+PW name\s+:\s+${PW_NAME}\s+Control Word\s+:\s+${CONTROL_WORD}\s*$$ + ^\s+Activation status\s+:\s+${ACTIVATION}\s*$$ + ^\s+Band Width\s+:\s+${BANDWIDTH}\s*$$ + ^\s+Tunnel destination\s+:\s+${TUNNEL_DEST}\s*$$ + ^\s+Related interface name\s+:\s+${RELATED_IF}\s*$$ + ^\s+FRR type\s+:\s+${FRR_TYPE}\s*$$ + ^\s+VC status\s+:\s+${VC_STATUS}\s*$$ + ^\s+Remote status\s+:\s+${REMOTE_STATUS}\s*$$ + ^\s+MC selection\s+:\s+${MC_SELECTION}\s*$$ + ^\s+VCCV CC type\s+:\s+${VCCV_CC}\s*$$ + ^\s+VCCV CV type\s+:\s+${VCCV_CV}\s*$$ -> Record + ^Headers: + ^Codes + ^PW count + ^-+ + ^\s*$$ + ^\S+\s*#\s*$$ -> End + ^.*#\s*$$ -> End + ^.* -> Start diff --git a/tests/test_zte_extended_parsers.py b/tests/test_zte_extended_parsers.py index 5aadc99..2fb6c97 100644 --- a/tests/test_zte_extended_parsers.py +++ b/tests/test_zte_extended_parsers.py @@ -3,6 +3,7 @@ from __future__ import annotations import unittest +from pathlib import Path from netx_api.biz_state.collect_session import resolve_aux_command from netx_api.biz_state.command_match import expand_from_bindings, match_command @@ -14,6 +15,7 @@ from netx_api.biz_state.parsers.zte import ( normalize_ip_route, normalize_ipv6_route, normalize_l2vpn_pw, + normalize_l2vpn_pw_detail, normalize_optical_brief, normalize_ospf_neighbor, normalize_vrrp, @@ -127,6 +129,48 @@ 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_PW_DETAIL = """ +Service type and instance name:[VPLS qualified demo-vpls-1] + Peer IP address : 10.0.0.6 VCID : 19001 + Connection mode : HUB VCID Extend : 0 + Signaling protocol : LDP VC type : VLAN + Last status change time : 09:46:46 Create time : 3d 07:08:42 + MPLS VC local label : 165840 Remote label : 167978 + PW name : pw19001 Control Word : ENABLE + Activation status : ENABLE + Band Width : 0 kbps + Tunnel destination : 10.0.0.6 + Related interface name : - + FRR type : NULL + VC status : UP + Remote status : ALLOK + MC selection : - + VCCV CC type : CWORD + VCCV CV type : LSP + +Service type and instance name:[VPLS demo-vpls-2] + Peer IP address : 10.0.0.6 VCID : 19011 + Connection mode : HUB VCID Extend : 0 + Signaling protocol : LDP VC type : Ethernet + Last status change time : 09:46:48 Create time : 1d 01:00:00 + MPLS VC local label : 165841 Remote label : 167981 + PW name : pw19011 Control Word : DISABLE + Activation status : ENABLE + Band Width : 0 kbps + Tunnel destination : 10.0.0.6 + Related interface name : - + FRR type : NULL + VC status : DOWN + Remote status : PWSA + MC selection : - + VCCV CC type : ALERT_LABEL + VCCV CV type : LSP|BFD_BASIC_HEAD +""" + +_L2VPN_PW_DETAIL_SAMPLE = ( + Path(__file__).resolve().parents[2] / "test" / "show-zte" / "show-l2vpn-forwarding-detail" +) + class ZteExtendedParserTests(unittest.TestCase): @classmethod @@ -284,6 +328,58 @@ class ZteExtendedParserTests(unittest.TestCase): self.assertEqual(pw[0]["state"].upper(), "UP") self.assertEqual(pw[1]["state"].upper(), "DOWN") + def test_l2vpn_pw_detail(self) -> None: + rows = normalize_l2vpn_pw_detail( + raw_text=_L2VPN_PW_DETAIL, + vendor="zte", + device_type="zte_zxros", + command="show l2vpn forwardinfo detail", + ) + self.assertEqual(len(rows), 2) + by = {r["pw_name"]: r for r in rows} + self.assertEqual(by["pw19001"]["vc_status"].upper(), "UP") + self.assertEqual(by["pw19001"]["remote_status"], "ALLOK") + self.assertEqual(by["pw19001"]["vcid"], "19001") + self.assertEqual(by["pw19001"]["peer"], "10.0.0.6") + self.assertIn("demo-vpls-1", by["pw19001"]["service_instance"]) + self.assertEqual(by["pw19011"]["vc_status"].upper(), "DOWN") + self.assertEqual(by["pw19011"]["vccv_cv"], "LSP|BFD_BASIC_HEAD") + + fsm_rows = apply_rule( + platform="zte_zxros", + rule_key="zte_zxros_show_l2vpn_forwardinfo_detail", + text=_L2VPN_PW_DETAIL, + ) + self.assertGreaterEqual(len(fsm_rows), 2) + + brief = match_command( + vendor_key="zte", command="show l2vpn forwardinfo detail | exclude Tunnel" + ) + self.assertIsNotNone(brief) + assert brief is not None + self.assertEqual(brief.profile.profile_id, "zte.l2vpn_pw_detail") + no_detail = match_command(vendor_key="zte", command="show l2vpn forwardinfo | one-line") + self.assertIsNotNone(no_detail) + assert no_detail is not None + self.assertEqual(no_detail.profile.profile_id, "zte.l2vpn_pw") + + @unittest.skipUnless( + _L2VPN_PW_DETAIL_SAMPLE.is_file(), + "test/show-zte/show-l2vpn-forwarding-detail not present", + ) + def test_l2vpn_pw_detail_real_sample(self) -> None: + raw = _L2VPN_PW_DETAIL_SAMPLE.read_text(encoding="utf-8", errors="replace") + rows = normalize_l2vpn_pw_detail( + raw_text=raw, + vendor="zte", + device_type="zte_zxros", + command="show l2vpn forwardinfo detail", + ) + self.assertEqual(len(rows), 6) + self.assertTrue(all(r["vc_status"].upper() == "UP" for r in rows)) + self.assertEqual(rows[0]["pw_name"], "pw19001") + self.assertEqual(rows[-1]["pw_name"], "pw1230002") + def test_profiles_and_expand(self) -> None: for mid in ( "ospf_neighbor", @@ -293,6 +389,7 @@ class ZteExtendedParserTests(unittest.TestCase): "ip_route", "ipv6_route", "l2vpn_pw", + "l2vpn_pw_detail", "l2vpn_mac", "evpn_mac", "interface_detail", @@ -303,6 +400,7 @@ class ZteExtendedParserTests(unittest.TestCase): self.assertIsNotNone(get_profile("zte.bgp_vpnv4_vrf_summary")) self.assertIsNotNone(get_profile("zte.ospf_neighbor")) self.assertIsNotNone(get_profile("zte.interface_detail")) + self.assertIsNotNone(get_profile("zte.l2vpn_pw_detail")) self.assertIsNotNone(get_profile("zte.bgp_vpnv6_neighbor_in")) self.assertIsNotNone(get_profile("zte.bgp_vpnv6_neighbor_out"))