diff --git a/netx_api/biz_state/parsers/zte/arp.py b/netx_api/biz_state/parsers/zte/arp.py index 25a37e2..c347f6e 100644 --- a/netx_api/biz_state/parsers/zte/arp.py +++ b/netx_api/biz_state/parsers/zte/arp.py @@ -1,13 +1,21 @@ -"""ZTE: show arp [| one-line].""" +"""ZTE: show arp [| one-line]. + +Pipeline: TextFSM rule ``zte_zxros_show_arp`` → map rows → hand fallback. +""" from __future__ import annotations import re -from typing import Any +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_arp",) _ARP_AGE_TIME_RE = re.compile(r"^\d{1,2}:\d{2}:\d{2}$") - -RULE_KEYS: tuple[str, ...] = () +_IP_RE = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$") def is_valid_arp_age(age: str) -> bool: @@ -15,19 +23,62 @@ def is_valid_arp_age(age: str) -> bool: return bool(_ARP_AGE_TIME_RE.match(str(age or "").strip())) -def normalize_arp( +def _row_from_fields( *, - raw_text: str, - fsm_tables=None, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = (vendor, device_type, command, params, fsm_tables) + ip: str, + age: str, + mac: str, + iface: str, + exter: str = "", + inter: str = "", + sub: str = "", +) -> dict[str, Any] | None: + ip = str(ip or "").strip() + iface = str(iface or "").strip() + if not ip or not iface: + return None + age = str(age or "").strip() + dynamic = is_valid_arp_age(age) + return { + "ip": ip[:64], + "age": age[:32], + "mac": str(mac or "").strip()[:64], + "interface": iface[:128], + "exter_vlan": str(exter or "").strip()[:32], + "inter_vlan": str(inter or "").strip()[:32], + "sub_interface": str(sub or "").strip()[:128], + "entry_type": "dynamic" if dynamic else "static", + } + + +def _map_fsm_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + seen: set[tuple[str, str]] = set() + for r in rows: + ip = row_get(r, "IP", "ip") + iface = row_get(r, "INTERFACE", "interface") + key = (ip, iface) + if not ip or not iface or key in seen: + continue + mapped = _row_from_fields( + ip=ip, + age=row_get(r, "AGE", "age"), + mac=row_get(r, "MAC", "mac", "HARDWARE", "hardware"), + iface=iface, + exter=row_get(r, "EXTER_VLAN", "exter_vlan"), + inter=row_get(r, "INTER_VLAN", "inter_vlan"), + sub=row_get(r, "SUB_INTERFACE", "sub_interface"), + ) + if not mapped: + continue + seen.add(key) + out.append(mapped) + return out + + +def _hand_parse(*, raw_text: str, **_kw: Any) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] seen: set[tuple[str, str]] = set() - ip_re = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$") for raw in str(raw_text or "").splitlines(): line = raw.strip() if not line or line.startswith("---") or line.lower().startswith("arp protect"): @@ -37,32 +88,60 @@ def normalize_arp( if "hardware" in line.lower() and "address" in line.lower(): continue parts = line.split() - if len(parts) < 4 or not ip_re.match(parts[0]): + if len(parts) < 4 or not _IP_RE.match(parts[0]): continue - ip = parts[0] - age = parts[1] - mac = parts[2] - iface = parts[3] - exter = parts[4] if len(parts) > 4 else "" - inter = parts[5] if len(parts) > 5 else "" - sub = parts[6] if len(parts) > 6 else "" + ip, age, mac, iface = parts[0], parts[1], parts[2], parts[3] key = (ip, iface) if key in seen: continue - seen.add(key) - dynamic = is_valid_arp_age(age) - out.append( - { - "ip": ip[:64], - "age": age[:32], - "mac": mac[:64], - "interface": iface[:128], - "exter_vlan": exter[:32], - "inter_vlan": inter[:32], - "sub_interface": sub[:128], - "entry_type": "dynamic" if dynamic else "static", - } + mapped = _row_from_fields( + ip=ip, + age=age, + mac=mac, + iface=iface, + exter=parts[4] if len(parts) > 4 else "", + inter=parts[5] if len(parts) > 5 else "", + sub=parts[6] if len(parts) > 6 else "", ) + if not mapped: + continue + seen.add(key) + out.append(mapped) return out + +def normalize_arp( + *, + 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 arp").strip() or "show arp" + if platform: + 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, + vendor=vendor, + device_type=device_type, + command=command, + ) + + normalize_arp.RULE_KEYS = RULE_KEYS diff --git a/netx_api/cli_templates/index b/netx_api/cli_templates/index index 37c292a..387eea6 100644 --- a/netx_api/cli_templates/index +++ b/netx_api/cli_templates/index @@ -7,6 +7,7 @@ zte/zte_zxros_show_lldp_neighbor_brief.textfsm, .*, zte_zxros, sh[[ow]] lld[[p]] zte/zte_zxros_show_interface_brief.textfsm, .*, zte_zxros, sh[[ow]] int[[erface]] br[[ief]] zte/zte_zxros_show_interface.textfsm, .*, zte_zxros, sh[[ow]] int[[erface]] .+ zte/zte_zxros_show_ip_vrf.textfsm, .*, zte_zxros, sh[[ow]] ip vrf$ +zte/zte_zxros_show_arp.textfsm, .*, zte_zxros, sh[[ow]] arp # --- cisco --- cisco/cisco_ios_show_interfaces.textfsm, .*, cisco_ios, sh[[ow]] int[[erfaces]] .+ diff --git a/netx_api/cli_templates/zte/zte_zxros_show_arp.textfsm b/netx_api/cli_templates/zte/zte_zxros_show_arp.textfsm new file mode 100644 index 0000000..ee81650 --- /dev/null +++ b/netx_api/cli_templates/zte/zte_zxros_show_arp.textfsm @@ -0,0 +1,32 @@ +Value IP (\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) +Value AGE (\S+) +Value MAC ([0-9a-fA-F]{4}\.[0-9a-fA-F]{4}\.[0-9a-fA-F]{4}) +Value INTERFACE (\S+) +Value EXTER_VLAN (\S+) +Value INTER_VLAN (\S+) +Value SUB_INTERFACE (\S+) + +Start + ^IP\s+Hardware -> Header + ^Address\s+Age\s+Address -> Entries + ^-+ -> Entries + ^\s*$$ + ^.* -> Start + +Header + ^Address\s+Age\s+Address -> Entries + ^-+ -> Entries + ^\s*$$ + ^.* -> Header + +Entries + ^${IP}\s+${AGE}\s+${MAC}\s+${INTERFACE}\s+${EXTER_VLAN}\s+${INTER_VLAN}\s+${SUB_INTERFACE}\s*$$ -> Record + ^${IP}\s+${AGE}\s+${MAC}\s+${INTERFACE}\s+${EXTER_VLAN}\s+${INTER_VLAN}\s*$$ -> Record + ^${IP}\s+${AGE}\s+${MAC}\s+${INTERFACE}\s*$$ -> Record + ^-+ + ^The\s+count + ^Arp\s+protect + ^\s*$$ + ^\S+\s*#\s*$$ -> End + ^.*#\s*$$ -> End + ^.* -> Entries diff --git a/tests/test_fsm_parser_pipeline.py b/tests/test_fsm_parser_pipeline.py index 2b057b6..d308f38 100644 --- a/tests/test_fsm_parser_pipeline.py +++ b/tests/test_fsm_parser_pipeline.py @@ -15,9 +15,11 @@ xgei-1/1/0/2 optical Duplex/full 1G up down down """ ZTE_ARP = """\ -IP Age Hardware address Interface Exter Interface VPN name -192.168.1.1 03:22:07 0011.2233.4455 gei-0/1/0/1 N/A --- -192.168.1.2 H 00aa.bbcc.ddee gei-0/1/0/1 N/A --- +IP Hardware Exter Inter Sub +Address Age Address Interface VlanID VlanID Interface +-------------------------------------------------------------------------------- +192.168.1.1 03:22:07 0011.2233.4455 gei-0/1/0/1 N/A N/A gei-0/1/0/1 +192.168.1.2 H 00aa.bbcc.ddee gei-0/1/0/1 N/A N/A N/A """ @@ -59,11 +61,12 @@ class ParserMetaTests(unittest.TestCase): assert meta is not None self.assertIn("zte_zxros_show_interface_brief", meta["rule_keys"]) - def test_arp_empty_rule_keys(self) -> None: + def test_arp_rule_keys(self) -> None: meta = get_parser_meta("arp") self.assertIsNotNone(meta) assert meta is not None - self.assertEqual(meta["rule_keys"], ()) + self.assertEqual(meta["rule_keys"], ("zte_zxros_show_arp",)) + self.assertTrue(meta["rule_keys_declared"]) def test_run_parser_interface_brief(self) -> None: records, tables, keys = run_parser( @@ -78,19 +81,22 @@ class ParserMetaTests(unittest.TestCase): self.assertGreaterEqual(len(records), 2) self.assertEqual(records[0]["interface"], "xgei-1/1/0/1") - def test_run_parser_arp_hand_only(self) -> None: - # Declared empty RULE_KEYS → no index auto-bind; hand parse only. + def test_run_parser_arp_via_fsm(self) -> None: records, tables, keys = run_parser( "arp", raw_text=ZTE_ARP, vendor="zte", device_type="zte_zxros", - command="show arp", + command="show arp | one-line", + textfsm_command="show arp", ) - self.assertEqual(keys, []) - self.assertEqual(tables, {}) - self.assertGreaterEqual(len(records), 1) - self.assertTrue(any(r["ip"] == "192.168.1.1" for r in records)) + self.assertEqual(keys, ["zte_zxros_show_arp"]) + self.assertEqual(len(tables.get("zte_zxros_show_arp") or []), 2) + self.assertEqual(len(records), 2) + by_ip = {r["ip"]: r for r in records} + self.assertEqual(by_ip["192.168.1.1"]["entry_type"], "dynamic") + self.assertEqual(by_ip["192.168.1.2"]["entry_type"], "static") + self.assertEqual(by_ip["192.168.1.1"]["mac"], "0011.2233.4455") if __name__ == "__main__": diff --git a/tests/test_zte_status_parsers.py b/tests/test_zte_status_parsers.py index 6231103..fafe0c6 100644 --- a/tests/test_zte_status_parsers.py +++ b/tests/test_zte_status_parsers.py @@ -86,7 +86,32 @@ class ZteStatusParserTests(unittest.TestCase): def test_arp(self) -> None: text = _section(self.log, "show arp", ("show nd6", "PAG3_", "M6000-4SE-3#show nd6")) - rows = normalize_arp(raw_text=text) + from netx_api.biz_state.parsers import run_parser + from netx_api.ntc_parse import apply_rule + + fsm = apply_rule( + platform="zte_zxros", + rule_key="zte_zxros_show_arp", + text=text, + command="show arp", + ) + self.assertEqual(len(fsm), 13) + + records, tables, keys = run_parser( + "arp", + raw_text=text, + vendor="zte", + device_type="zte_zxros", + command="show arp | one-line", + textfsm_command="show arp", + ) + self.assertEqual(keys, ["zte_zxros_show_arp"]) + self.assertEqual(len(tables["zte_zxros_show_arp"]), 13) + + rows = normalize_arp( + raw_text=text, vendor="zte", device_type="zte_zxros", command="show arp" + ) + self.assertEqual(rows, records) self.assertGreaterEqual(len(rows), 10) ips = {r["ip"] for r in rows} self.assertIn("192.166.1.65", ips) @@ -95,12 +120,13 @@ class ZteStatusParserTests(unittest.TestCase): dynamics = [r for r in rows if r["entry_type"] == "dynamic"] self.assertGreaterEqual(len(statics), 5) self.assertGreaterEqual(len(dynamics), 5) - self.assertTrue(all(r["age"] == "H" or not r["age"][0].isdigit() for r in statics) or True) - from netx_api.biz_state.parsers.zte.arp import is_valid_arp_age - self.assertTrue(all(is_valid_arp_age(r["age"]) for r in dynamics)) self.assertFalse(is_valid_arp_age("H")) self.assertTrue(is_valid_arp_age("03:22:07")) + # VLAN-subif row from TextFSM + sub = next(r for r in rows if r["ip"] == "192.166.1.73") + self.assertEqual(sub["interface"], "cgei-0/2/0/1.30") + self.assertEqual(sub["exter_vlan"], "30") def test_nd6(self) -> None: text = _section(self.log, "show nd6 cache", ("PAG3_", "show bgp"))