diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index 2e0f215..bd3c058 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -706,7 +706,7 @@ def _load_metric_rows(db: Session, *, batch_id: str, metric_id: str) -> list[dic if rows: out = [dict(r.data_json or {}) for r in rows] if metric_id == "arp": - from .parsers.zte_status import is_valid_arp_age + from .parsers.zte.arp import is_valid_arp_age # Compare only dynamic ARP (Age is HH:MM:SS); drop static H / incomplete flags out = [ diff --git a/netx_api/biz_state/parsers/__init__.py b/netx_api/biz_state/parsers/__init__.py index 11521b3..d68b75b 100644 --- a/netx_api/biz_state/parsers/__init__.py +++ b/netx_api/biz_state/parsers/__init__.py @@ -1,62 +1,27 @@ -"""Parser callbacks keyed by parser_id.""" +"""Biz-state CLI parsers: vendor folders + one module per command/metric.""" from __future__ import annotations from typing import Any, Callable -from ...lldp_shared import NeighborHit, parse_neighbor_output -from .vrf import normalize_vrf_list, normalize_vrf_route_summary -from .zte_status import ( - normalize_arp, - normalize_bgp_peer, - normalize_interface_brief, - normalize_isis_adjacency, - normalize_nd6_cache, -) +from .common.lldp_neighbors import normalize_lldp_neighbors +from .common.vrf_list import normalize_vrf_list +from .common.vrf_route_summary import normalize_vrf_route_summary +from .zte import PARSERS as _ZTE_PARSERS NormalizeFn = Callable[..., list[dict[str, Any]]] - -def normalize_lldp_neighbors( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - hits: list[NeighborHit] = parse_neighbor_output( - raw_text, - vendor=vendor, - device_type=device_type, - command=command, - ) - _ = params - rows: list[dict[str, Any]] = [] - for h in hits: - rows.append( - { - "local_if": str(h.local_port or "").strip(), - "remote_sys": str(h.remote_name or "").strip(), - "remote_if": str(h.remote_port or "").strip(), - "remote_ip": str(h.remote_ip or "").strip(), - "protocol": str(h.protocol or "lldp").strip() or "lldp", - } - ) - return rows - - _REGISTRY: dict[str, NormalizeFn] = { "lldp_neighbors": normalize_lldp_neighbors, "vrf_list": normalize_vrf_list, "vrf_route_summary": normalize_vrf_route_summary, - "isis_adjacency": normalize_isis_adjacency, - "interface_brief": normalize_interface_brief, - "arp": normalize_arp, - "nd6_cache": normalize_nd6_cache, - "bgp_peer": normalize_bgp_peer, + **_ZTE_PARSERS, } def get_parser(parser_id: str) -> NormalizeFn | None: return _REGISTRY.get(str(parser_id or "").strip()) + + +def registered_parser_ids() -> list[str]: + return sorted(_REGISTRY.keys()) diff --git a/netx_api/biz_state/parsers/common/__init__.py b/netx_api/biz_state/parsers/common/__init__.py new file mode 100644 index 0000000..7da73de --- /dev/null +++ b/netx_api/biz_state/parsers/common/__init__.py @@ -0,0 +1 @@ +"""Cross-vendor / shared parsers.""" diff --git a/netx_api/biz_state/parsers/common/lldp_neighbors.py b/netx_api/biz_state/parsers/common/lldp_neighbors.py new file mode 100644 index 0000000..a1d56d2 --- /dev/null +++ b/netx_api/biz_state/parsers/common/lldp_neighbors.py @@ -0,0 +1,36 @@ +"""LLDP neighbor normalize (multi-vendor via lldp_shared / TextFSM).""" + +from __future__ import annotations + +from typing import Any + +from ....lldp_shared import NeighborHit, parse_neighbor_output + + +def normalize_lldp_neighbors( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + hits: list[NeighborHit] = parse_neighbor_output( + raw_text, + vendor=vendor, + device_type=device_type, + command=command, + ) + _ = params + rows: list[dict[str, Any]] = [] + for h in hits: + rows.append( + { + "local_if": str(h.local_port or "").strip(), + "remote_sys": str(h.remote_name or "").strip(), + "remote_if": str(h.remote_port or "").strip(), + "remote_ip": str(h.remote_ip or "").strip(), + "protocol": str(h.protocol or "lldp").strip() or "lldp", + } + ) + return rows diff --git a/netx_api/biz_state/parsers/common/vrf_list.py b/netx_api/biz_state/parsers/common/vrf_list.py new file mode 100644 index 0000000..48f943c --- /dev/null +++ b/netx_api/biz_state/parsers/common/vrf_list.py @@ -0,0 +1,57 @@ +"""VRF / VPN-instance list discovery parser (multi-vendor TextFSM + fallback).""" + +from __future__ import annotations + +import re +from typing import Any + +from ....lldp_shared import resolve_vendor_key +from ....ntc_parse import parse_cli, resolve_cli_platform, row_get + + +def normalize_vrf_list( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = params + platform = resolve_cli_platform( + vendor=vendor, + device_type=device_type, + vendor_key=resolve_vendor_key(vendor, device_type), + ) + cmd = str(command or "").strip() + rows = parse_cli(platform=platform, command=cmd, text=raw_text) if platform and cmd else [] + out: list[dict[str, Any]] = [] + seen: set[str] = set() + for r in rows: + name = row_get(r, "NAME", "VPN_INSTANCE", "VRF", "vrf_name") + if not name or name.lower() in ("name", "vrf", "vpn-instance"): + continue + if name in seen: + continue + seen.add(name) + out.append( + { + "vrf_name": name[:128], + "rd": row_get(r, "DEFAULT_RD", "RD", "rd")[:64], + "protocols": row_get(r, "PROTOCOLS", "ADDRESS_FAMILY", "protocols")[:64], + } + ) + if out: + return out + for line in str(raw_text or "").splitlines(): + m = re.match(r"^\s*([A-Za-z0-9_./:-]+)\s+(\d+:\d+||\S+:\S+)\s*", line) + if not m: + continue + name = m.group(1) + if name.lower() in ("name", "vrf", "vpn-instance", "total"): + continue + if name in seen: + continue + seen.add(name) + out.append({"vrf_name": name[:128], "rd": m.group(2)[:64], "protocols": ""}) + return out diff --git a/netx_api/biz_state/parsers/vrf.py b/netx_api/biz_state/parsers/common/vrf_route_summary.py similarity index 50% rename from netx_api/biz_state/parsers/vrf.py rename to netx_api/biz_state/parsers/common/vrf_route_summary.py index d9abb8d..5fe7261 100644 --- a/netx_api/biz_state/parsers/vrf.py +++ b/netx_api/biz_state/parsers/common/vrf_route_summary.py @@ -1,63 +1,10 @@ -"""VRF list + per-VRF route summary parsers.""" +"""Per-VRF route summary counts (multi-vendor line scrape).""" from __future__ import annotations import re from typing import Any -from ...lldp_shared import resolve_vendor_key -from ...ntc_parse import parse_cli, resolve_cli_platform, row_get - - -def normalize_vrf_list( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = params - platform = resolve_cli_platform( - vendor=vendor, - device_type=device_type, - vendor_key=resolve_vendor_key(vendor, device_type), - ) - cmd = str(command or "").strip() - rows = parse_cli(platform=platform, command=cmd, text=raw_text) if platform and cmd else [] - out: list[dict[str, Any]] = [] - seen: set[str] = set() - for r in rows: - name = row_get(r, "NAME", "VPN_INSTANCE", "VRF", "vrf_name") - if not name or name.lower() in ("name", "vrf", "vpn-instance"): - continue - if name in seen: - continue - seen.add(name) - out.append( - { - "vrf_name": name[:128], - "rd": row_get(r, "DEFAULT_RD", "RD", "rd")[:64], - "protocols": row_get(r, "PROTOCOLS", "ADDRESS_FAMILY", "protocols")[:64], - } - ) - if out: - return out - # Fallback: loose line scrape for labs without TextFSM hit - for line in str(raw_text or "").splitlines(): - m = re.match(r"^\s*([A-Za-z0-9_./:-]+)\s+(\d+:\d+||\S+:\S+)\s*", line) - if not m: - continue - name = m.group(1) - if name.lower() in ("name", "vrf", "vpn-instance", "total"): - continue - if name in seen: - continue - seen.add(name) - out.append({"vrf_name": name[:128], "rd": m.group(2)[:64], "protocols": ""}) - return out - - _SOURCE_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ ("connected", re.compile(r"(?i)^\s*connected\s+(\d+)\s*$")), ("static", re.compile(r"(?i)^\s*static\s+(\d+)\s*$")), @@ -90,7 +37,6 @@ def normalize_vrf_route_summary( break if not found: - # Count non-empty data-ish lines as opaque "routes" bucket n = 0 for line in text.splitlines(): s = line.strip() diff --git a/netx_api/biz_state/parsers/zte/__init__.py b/netx_api/biz_state/parsers/zte/__init__.py new file mode 100644 index 0000000..59d75b9 --- /dev/null +++ b/netx_api/biz_state/parsers/zte/__init__.py @@ -0,0 +1,31 @@ +"""ZTE ZXROS status parsers — one module per show command.""" + +from __future__ import annotations + +from typing import Any, Callable + +from .arp import is_valid_arp_age, normalize_arp +from .bgp_peer import normalize_bgp_peer +from .interface_brief import normalize_interface_brief +from .isis_adjacency import normalize_isis_adjacency +from .nd6_cache import normalize_nd6_cache + +NormalizeFn = Callable[..., list[dict[str, Any]]] + +PARSERS: dict[str, NormalizeFn] = { + "isis_adjacency": normalize_isis_adjacency, + "interface_brief": normalize_interface_brief, + "arp": normalize_arp, + "nd6_cache": normalize_nd6_cache, + "bgp_peer": normalize_bgp_peer, +} + +__all__ = [ + "PARSERS", + "is_valid_arp_age", + "normalize_arp", + "normalize_bgp_peer", + "normalize_interface_brief", + "normalize_isis_adjacency", + "normalize_nd6_cache", +] diff --git a/netx_api/biz_state/parsers/zte/arp.py b/netx_api/biz_state/parsers/zte/arp.py new file mode 100644 index 0000000..37a195c --- /dev/null +++ b/netx_api/biz_state/parsers/zte/arp.py @@ -0,0 +1,63 @@ +"""ZTE: show arp [| one-line].""" + +from __future__ import annotations + +import re +from typing import Any + +_ARP_AGE_TIME_RE = re.compile(r"^\d{1,2}:\d{2}:\d{2}$") + + +def is_valid_arp_age(age: str) -> bool: + """True when Age looks like a dynamic timer (HH:MM:SS), not static flags like H.""" + return bool(_ARP_AGE_TIME_RE.match(str(age or "").strip())) + + +def normalize_arp( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = (vendor, device_type, command, params) + 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"): + continue + if line.lower().startswith("the count"): + continue + 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]): + 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 "" + 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", + } + ) + return out diff --git a/netx_api/biz_state/parsers/zte/bgp_peer.py b/netx_api/biz_state/parsers/zte/bgp_peer.py new file mode 100644 index 0000000..1a29eee --- /dev/null +++ b/netx_api/biz_state/parsers/zte/bgp_peer.py @@ -0,0 +1,79 @@ +"""ZTE: show bgp {vpnv4|ipv4|vpnv6} unicast summary.""" + +from __future__ import annotations + +import re +from typing import Any + +_BGP_PEER_RE = re.compile( + r"^(?P\d{1,3}(?:\.\d{1,3}){3})\s+(?P\d+)\s+(?P\d+)\s+" + r"(?P\d+)\s+(?P\d+)\s+(?P\S+)\s+(?P\S+)\s*$", + re.I, +) + + +def _detect_bgp_afi(command: str, params: dict[str, str] | None) -> str: + if params and params.get("afi"): + return str(params.get("afi") or "").strip().lower() + low = str(command or "").lower() + if "vpnv6" in low: + return "vpnv6" + if "vpnv4" in low: + return "vpnv4" + if "ipv6" in low: + return "ipv6" + if "ipv4" in low: + return "ipv4" + return "unknown" + + +def normalize_bgp_peer( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = (vendor, device_type) + afi = _detect_bgp_afi(command, params) + out: list[dict[str, Any]] = [] + seen: set[str] = set() + for raw in str(raw_text or "").splitlines(): + line = raw.strip() + if not line: + continue + low = line.lower() + if low.startswith("bgp router") or low.startswith("local as") or low.startswith("all "): + continue + if low.startswith("neighbor") and "msg" in low: + continue + m = _BGP_PEER_RE.match(line) + if not m: + continue + nei = m.group("nei") + if nei in seen: + continue + seen.add(nei) + state_raw = m.group("state") + if state_raw.isdigit(): + state = "Established" + pfx = state_raw + else: + state = state_raw + pfx = "" + out.append( + { + "afi": afi[:32], + "neighbor": nei[:64], + "ver": m.group("ver")[:8], + "as_num": m.group("asn")[:16], + "msg_rcvd": m.group("rx")[:32], + "msg_send": m.group("tx")[:32], + "up_down": m.group("up")[:32], + "state": state[:64], + "pfx_rcd": pfx[:32], + "state_or_pfx": state_raw[:64], + } + ) + return out diff --git a/netx_api/biz_state/parsers/zte/interface_brief.py b/netx_api/biz_state/parsers/zte/interface_brief.py new file mode 100644 index 0000000..b9c9989 --- /dev/null +++ b/netx_api/biz_state/parsers/zte/interface_brief.py @@ -0,0 +1,81 @@ +"""ZTE: show interface brief.""" + +from __future__ import annotations + +import re +from typing import Any + +from ....lldp_shared import resolve_vendor_key +from ....ntc_parse import parse_cli, resolve_cli_platform, row_get + +_IFACE_RE = re.compile( + r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)" + r"(?:\s+(?P\S+))?\s+(?Pup|down)\s+(?Pup|down)\s+(?Pup|down)" + r"(?:\s+(?P.*))?$", + re.I, +) + + +def normalize_interface_brief( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = params + platform = resolve_cli_platform( + vendor=vendor, + device_type=device_type, + vendor_key=resolve_vendor_key(vendor, device_type), + ) + cmd = str(command or "show interface brief").strip() or "show interface brief" + rows = parse_cli(platform=platform, command=cmd, text=raw_text) if platform else [] + out: list[dict[str, Any]] = [] + seen: set[str] = set() + for r in rows: + iface = row_get(r, "INTERFACE", "interface") + if not iface or iface.lower() == "interface": + continue + if iface in seen: + continue + seen.add(iface) + out.append( + { + "interface": iface[:128], + "attribute": row_get(r, "ATTRIBUTE", "attribute")[:64], + "mode": row_get(r, "MODE", "mode")[:64], + "bw": row_get(r, "BW", "bw")[:32], + "admin": row_get(r, "ADMIN", "admin")[:16], + "phy": row_get(r, "PHY", "phy")[:16], + "prot": row_get(r, "PROT", "prot")[:16], + "description": row_get(r, "DESCRIPTION", "description")[:256], + } + ) + if out: + return out + for raw in str(raw_text or "").splitlines(): + line = raw.strip() + if not line or line.lower().startswith("interface"): + continue + m = _IFACE_RE.match(line) + if not m: + continue + iface = m.group("iface") + if iface in seen: + continue + seen.add(iface) + out.append( + { + "interface": iface[:128], + "attribute": (m.group("attr") or "")[:64], + "mode": (m.group("mode") or "")[:64], + "bw": (m.group("bw") or "")[:32], + "admin": (m.group("admin") or "")[:16], + "phy": (m.group("phy") or "")[:16], + "prot": (m.group("prot") or "")[:16], + "description": (m.group("desc") or "").strip()[:256], + } + ) + return out diff --git a/netx_api/biz_state/parsers/zte/isis_adjacency.py b/netx_api/biz_state/parsers/zte/isis_adjacency.py new file mode 100644 index 0000000..fd0822c --- /dev/null +++ b/netx_api/biz_state/parsers/zte/isis_adjacency.py @@ -0,0 +1,56 @@ +"""ZTE: show isis adjacency [| one-line].""" + +from __future__ import annotations + +import re +from typing import Any + +_ISIS_ROW_RE = re.compile( + r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" + r"(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" + r"(?P\S+)\s+(?P\S+)\s*$", + re.I, +) +_PROCESS_RE = re.compile(r"(?i)^\s*Process\s+ID\s*:\s*(\d+)\s*$") + + +def normalize_isis_adjacency( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = (vendor, device_type, command, params) + out: list[dict[str, Any]] = [] + process_id = "" + for raw in str(raw_text or "").splitlines(): + line = raw.strip() + if not line: + continue + pm = _PROCESS_RE.match(line) + if pm: + process_id = pm.group(1) + continue + if line.lower().startswith("interface") and "system" in line.lower(): + continue + m = _ISIS_ROW_RE.match(line) + if not m: + continue + out.append( + { + "process_id": process_id, + "interface": m.group("iface")[:128], + "system_id": m.group("sys")[:128], + "state": m.group("state")[:32], + "lev": m.group("lev")[:16], + "holds": m.group("holds")[:32], + "snpa": m.group("snpa")[:64], + "pri": m.group("pri")[:16], + "mt": m.group("mt")[:16], + "nsf": m.group("nsf")[:32], + "af": m.group("af")[:64], + } + ) + return out diff --git a/netx_api/biz_state/parsers/zte/nd6_cache.py b/netx_api/biz_state/parsers/zte/nd6_cache.py new file mode 100644 index 0000000..3871cc8 --- /dev/null +++ b/netx_api/biz_state/parsers/zte/nd6_cache.py @@ -0,0 +1,54 @@ +"""ZTE: show nd6 cache [| one-line].""" + +from __future__ import annotations + +import re +from typing import Any + +_ND6_ROW_RE = re.compile( + r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" + r"(?P\S+)\s+(?P\S+)\s*$", + re.I, +) + + +def normalize_nd6_cache( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = (vendor, device_type, command, params) + out: list[dict[str, Any]] = [] + seen: set[tuple[str, str]] = set() + for raw in str(raw_text or "").splitlines(): + line = raw.strip() + if not line: + continue + low = line.lower() + if low.startswith("s-static") or low.startswith("total cache") or low.startswith("only current"): + continue + if low.startswith("address") and "link" in low: + continue + m = _ND6_ROW_RE.match(line) + if not m: + continue + addr = m.group("addr") + iface = m.group("iface") + key = (addr, iface) + if key in seen: + continue + seen.add(key) + out.append( + { + "address": addr[:128], + "link_address": m.group("link")[:64], + "age": m.group("age")[:64], + "status": m.group("status")[:32], + "interface": iface[:128], + "type": m.group("type")[:32], + } + ) + return out diff --git a/netx_api/biz_state/parsers/zte_status.py b/netx_api/biz_state/parsers/zte_status.py deleted file mode 100644 index a066d40..0000000 --- a/netx_api/biz_state/parsers/zte_status.py +++ /dev/null @@ -1,331 +0,0 @@ -"""ZTE ZXROS status table parsers (ISIS / interface / ARP / ND6 / BGP).""" - -from __future__ import annotations - -import re -from typing import Any - -from ...lldp_shared import resolve_vendor_key -from ...ntc_parse import parse_cli, resolve_cli_platform, row_get - -_IFACE_RE = re.compile( - r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)" - r"(?:\s+(?P\S+))?\s+(?Pup|down)\s+(?Pup|down)\s+(?Pup|down)" - r"(?:\s+(?P.*))?$", - re.I, -) - -_ISIS_ROW_RE = re.compile( - r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" - r"(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" - r"(?P\S+)\s+(?P\S+)\s*$", - re.I, -) - -_ARP_ROW_RE = re.compile( - r"^(?P\d{1,3}(?:\.\d{1,3}){3})\s+(?P\S+)\s+(?P\S+)\s+" - r"(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)\s*$", - re.I, -) - -_ND6_ROW_RE = re.compile( - r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" - r"(?P\S+)\s+(?P\S+)\s*$", - re.I, -) - -_BGP_PEER_RE = re.compile( - r"^(?P\d{1,3}(?:\.\d{1,3}){3})\s+(?P\d+)\s+(?P\d+)\s+" - r"(?P\d+)\s+(?P\d+)\s+(?P\S+)\s+(?P\S+)\s*$", - re.I, -) - -_PROCESS_RE = re.compile(r"(?i)^\s*Process\s+ID\s*:\s*(\d+)\s*$") -_HEADER_HINTS = ( - "interface", - "system id", - "address", - "neighbor", - "ip", - "hardware", - "link-address", -) - - -def _is_header(line: str) -> bool: - low = line.lower() - return any(h in low for h in _HEADER_HINTS) and ("---" in low or " " in line) - - -def normalize_isis_adjacency( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = (vendor, device_type, command, params) - out: list[dict[str, Any]] = [] - process_id = "" - for raw in str(raw_text or "").splitlines(): - line = raw.strip() - if not line: - continue - pm = _PROCESS_RE.match(line) - if pm: - process_id = pm.group(1) - continue - if line.lower().startswith("interface") and "system" in line.lower(): - continue - m = _ISIS_ROW_RE.match(line) - if not m: - continue - out.append( - { - "process_id": process_id, - "interface": m.group("iface")[:128], - "system_id": m.group("sys")[:128], - "state": m.group("state")[:32], - "lev": m.group("lev")[:16], - "holds": m.group("holds")[:32], - "snpa": m.group("snpa")[:64], - "pri": m.group("pri")[:16], - "mt": m.group("mt")[:16], - "nsf": m.group("nsf")[:32], - "af": m.group("af")[:64], - } - ) - return out - - -def normalize_interface_brief( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = params - platform = resolve_cli_platform( - vendor=vendor, - device_type=device_type, - vendor_key=resolve_vendor_key(vendor, device_type), - ) - cmd = str(command or "show interface brief").strip() or "show interface brief" - rows = parse_cli(platform=platform, command=cmd, text=raw_text) if platform else [] - out: list[dict[str, Any]] = [] - seen: set[str] = set() - for r in rows: - iface = row_get(r, "INTERFACE", "interface") - if not iface or iface.lower() == "interface": - continue - if iface in seen: - continue - seen.add(iface) - out.append( - { - "interface": iface[:128], - "attribute": row_get(r, "ATTRIBUTE", "attribute")[:64], - "mode": row_get(r, "MODE", "mode")[:64], - "bw": row_get(r, "BW", "bw")[:32], - "admin": row_get(r, "ADMIN", "admin")[:16], - "phy": row_get(r, "PHY", "phy")[:16], - "prot": row_get(r, "PROT", "prot")[:16], - "description": row_get(r, "DESCRIPTION", "description")[:256], - } - ) - if out: - return out - for raw in str(raw_text or "").splitlines(): - line = raw.strip() - if not line or line.lower().startswith("interface"): - continue - m = _IFACE_RE.match(line) - if not m: - continue - iface = m.group("iface") - if iface in seen: - continue - seen.add(iface) - out.append( - { - "interface": iface[:128], - "attribute": (m.group("attr") or "")[:64], - "mode": (m.group("mode") or "")[:64], - "bw": (m.group("bw") or "")[:32], - "admin": (m.group("admin") or "")[:16], - "phy": (m.group("phy") or "")[:16], - "prot": (m.group("prot") or "")[:16], - "description": (m.group("desc") or "").strip()[:256], - } - ) - return out - - -_ARP_AGE_TIME_RE = re.compile(r"^\d{1,2}:\d{2}:\d{2}$") - - -def is_valid_arp_age(age: str) -> bool: - """True when Age looks like a dynamic timer (HH:MM:SS), not static flags like H.""" - return bool(_ARP_AGE_TIME_RE.match(str(age or "").strip())) - - -def normalize_arp( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = (vendor, device_type, command, params) - 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"): - continue - if line.lower().startswith("the count"): - continue - 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]): - 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 "" - 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", - } - ) - return out - - -def normalize_nd6_cache( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = (vendor, device_type, command, params) - out: list[dict[str, Any]] = [] - seen: set[tuple[str, str]] = set() - for raw in str(raw_text or "").splitlines(): - line = raw.strip() - if not line: - continue - low = line.lower() - if low.startswith("s-static") or low.startswith("total cache") or low.startswith("only current"): - continue - if low.startswith("address") and "link" in low: - continue - m = _ND6_ROW_RE.match(line) - if not m: - continue - addr = m.group("addr") - iface = m.group("iface") - key = (addr, iface) - if key in seen: - continue - seen.add(key) - out.append( - { - "address": addr[:128], - "link_address": m.group("link")[:64], - "age": m.group("age")[:64], - "status": m.group("status")[:32], - "interface": iface[:128], - "type": m.group("type")[:32], - } - ) - return out - - -def _detect_bgp_afi(command: str, params: dict[str, str] | None) -> str: - if params and params.get("afi"): - return str(params.get("afi") or "").strip().lower() - low = str(command or "").lower() - if "vpnv6" in low: - return "vpnv6" - if "vpnv4" in low: - return "vpnv4" - if "ipv6" in low: - return "ipv6" - if "ipv4" in low: - return "ipv4" - return "unknown" - - -def normalize_bgp_peer( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = (vendor, device_type) - afi = _detect_bgp_afi(command, params) - out: list[dict[str, Any]] = [] - seen: set[str] = set() - for raw in str(raw_text or "").splitlines(): - line = raw.strip() - if not line: - continue - low = line.lower() - if low.startswith("bgp router") or low.startswith("local as") or low.startswith("all "): - continue - if low.startswith("neighbor") and "msg" in low: - continue - m = _BGP_PEER_RE.match(line) - if not m: - continue - nei = m.group("nei") - if nei in seen: - continue - seen.add(nei) - state_raw = m.group("state") - if state_raw.isdigit(): - state = "Established" - pfx = state_raw - else: - state = state_raw - pfx = "" - out.append( - { - "afi": afi[:32], - "neighbor": nei[:64], - "ver": m.group("ver")[:8], - "as_num": m.group("asn")[:16], - "msg_rcvd": m.group("rx")[:32], - "msg_send": m.group("tx")[:32], - "up_down": m.group("up")[:32], - "state": state[:64], - "pfx_rcd": pfx[:32], - "state_or_pfx": state_raw[:64], - } - ) - return out diff --git a/tests/test_zte_status_parsers.py b/tests/test_zte_status_parsers.py index 0c3c97f..6231103 100644 --- a/tests/test_zte_status_parsers.py +++ b/tests/test_zte_status_parsers.py @@ -5,7 +5,8 @@ from __future__ import annotations import unittest from pathlib import Path -from netx_api.biz_state.parsers.zte_status import ( +from netx_api.biz_state.parsers.zte import ( + is_valid_arp_age, normalize_arp, normalize_bgp_peer, normalize_interface_brief, @@ -95,7 +96,7 @@ class ZteStatusParserTests(unittest.TestCase): 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_status import is_valid_arp_age + 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"))