mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 00:50:46 +08:00
Wire ZTE show arp through TextFSM dual-input pipeline.
Add zte_zxros_show_arp rule + prefer_fsm mapping using the lab one-line sample. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
02b8a7073b
commit
0f235133ed
5 changed files with 195 additions and 51 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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]] .+
|
||||
|
|
|
|||
32
netx_api/cli_templates/zte/zte_zxros_show_arp.textfsm
Normal file
32
netx_api/cli_templates/zte/zte_zxros_show_arp.textfsm
Normal file
|
|
@ -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
|
||||
|
|
@ -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__":
|
||||
|
|
|
|||
|
|
@ -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"))
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue