diff --git a/netx_api/biz_state/collect_runner.py b/netx_api/biz_state/collect_runner.py index c35d487..d2ac5be 100644 --- a/netx_api/biz_state/collect_runner.py +++ b/netx_api/biz_state/collect_runner.py @@ -235,7 +235,8 @@ def _run_primary_parse_job(job: Any) -> tuple[bool, bool]: any_fail = False def _declared_total(raw: str) -> int: - m = re.search(r"(?i)total\s+number\s+of\s+routes\s*:\s*(\d+)", raw or "") + text = (raw or "").replace("\u00a0", " ") + m = re.search(r"(?i)total\s+number\s+of\s+routes\s*:\s*(\d+)", text) return int(m.group(1)) if m else 0 def _flush_item(item: SpooledCommand, *, records: list[dict[str, Any]] | None = None) -> None: diff --git a/netx_api/biz_state/parsers/common/pipeline.py b/netx_api/biz_state/parsers/common/pipeline.py index cd001fe..c79a61e 100644 --- a/netx_api/biz_state/parsers/common/pipeline.py +++ b/netx_api/biz_state/parsers/common/pipeline.py @@ -29,3 +29,34 @@ def prefer_fsm( if mapped: return mapped return hand_fn(raw_text=raw_text, **kw) + + +def prefer_hand( + fsm_tables: Mapping[str, list[dict[str, Any]]] | None, + keys: Sequence[str], + map_rows: MapRowsFn, + hand_fn: HandFn, + *, + raw_text: str, + **kw: Any, +) -> list[dict[str, Any]]: + """Hand-first dual path: callback when non-empty; else first mapped FSM table. + + Use when one TextFSM cannot cover divergent CLI layouts (e.g. BGP neighbor + in vs out, status vs plain, heavy wraps). Empty ``keys`` / ``RULE_KEYS=()`` + makes this hand-only. + """ + hand = hand_fn(raw_text=raw_text, **kw) + if hand: + return hand + tables = fsm_tables or {} + for key in keys or (): + k = str(key or "").strip() + if not k: + continue + rows = tables.get(k) or [] + if rows: + mapped = map_rows(rows) + if mapped: + return mapped + return hand diff --git a/netx_api/biz_state/parsers/zte/bgp_route.py b/netx_api/biz_state/parsers/zte/bgp_route.py index f753266..fac0c15 100644 --- a/netx_api/biz_state/parsers/zte/bgp_route.py +++ b/netx_api/biz_state/parsers/zte/bgp_route.py @@ -1,34 +1,38 @@ -"""ZTE: show bgp … neighbor {in|out} [vrf] routes.""" +"""ZTE: show bgp … neighbor {in|out} [vrf] routes. + +Hand-only parser. IN/OUT, status vs plain, From column, RD+VRF annotations and +heavy IPv6 wraps diverge too much for one TextFSM; dual-path policy is hand-first +(``RULE_KEYS=()`` skips FSM in ``run_parser``). +""" from __future__ import annotations import ipaddress import re -from typing import Any, Mapping +from typing import Any -from ....lldp_shared import resolve_vendor_key -from ....ntc_parse import apply_rules, resolve_cli_platform, row_get -from ..common.pipeline import prefer_fsm from .bgp_peer import _detect_bgp_afi, _detect_local_as, _detect_vrf -RULE_KEYS = ("zte_zxros_show_bgp_neighbor_routes",) +# Declared empty → run_parser will not apply TextFSM for this metric. +RULE_KEYS: tuple[str, ...] = () -# Status codes may be glued or spaced: "*", "* i", "*>i", "* s", "< *" -_FLAGS = r"(?P(?:[*<>isd]+(?:\s+[*<>isd]+)*)?)" -# Single-line IPv4-style: * i 10.1.0.0/24 10.0.0.1 … path -_ROUTE_ONE_LINE_RE = re.compile( - rf"^\s*{_FLAGS}\s+" - r"(?P\S+)\s+" - r"(?P\S+)\s+" - r"(?P.*)$" +# Status must stay glued: "* i" / "*i" / "*>i" — never leave "i" as NETWORK. +_STATUS_LEAD_RE = re.compile( + r"^\s*(?P\*(?:\s*[<>isd]+)?|[<>isd]+)\s+" ) -# Network alone (often IPv6 wrap): * i 2407::1/128 -_NET_ONLY_RE = re.compile(rf"^\s*{_FLAGS}\s*(?P\S+)\s*$") +# After optional status: network + next-hop + remainder +_NET_NH_RE = re.compile( + r"^\s*(?P\S+)\s+(?P\S+)(?:\s+(?P.*))?$" +) +_NET_ONLY_RE = re.compile(r"^\s*(?P\S+)\s*$") _DIR_RE = re.compile(r"(?i)\bneighbor\s+(in|out)\s+") -# Neighbor may be IPv4 or IPv6 (consume until EOL / pipe) _NEI_RE = re.compile(r"(?i)\bneighbor\s+(?:in|out)\s+(\S+)") _TOTAL_RE = re.compile(r"(?i)total\s+number\s+of\s+routes\s*:\s*(\d+)") -_RD_RE = re.compile(r"(?i)^Route\s+Distinguisher\s*:\s*(\S+)") +# Route Distinguisher:2:111 (default for vrf css-srv6-mpls-1) +_RD_RE = re.compile( + r"(?i)^Route\s+Distinguisher\s*:\s*(?P\S+)" + r"(?:\s*\(\s*default\s+for\s+vrf\s+(?P\S+)\s*\))?" +) _HEADER_NETS = frozenset( { "network", @@ -46,6 +50,12 @@ _HEADER_NETS = frozenset( "from", } ) +_PATH_ENDS = frozenset({"?", "i", "e", "incomplete"}) + + +def _normalize_cli_text(raw_text: str) -> str: + """NBSP / odd spaces from paste or terminals → regular space.""" + return str(raw_text or "").replace("\u00a0", " ").replace("\u2007", " ") def _detect_direction(command: str, params: dict[str, str] | None) -> str: @@ -78,11 +88,11 @@ def _looks_like_ip_or_prefix(tok: str) -> bool: def _looks_like_prefix(net: str) -> bool: - return _looks_like_ip_or_prefix(net) + return _looks_like_ip_or_prefix(net) and "/" in str(net or "") def _declared_total(raw_text: str) -> int | None: - m = _TOTAL_RE.search(str(raw_text or "")) + m = _TOTAL_RE.search(_normalize_cli_text(raw_text)) if not m: return None try: @@ -97,23 +107,39 @@ def _route_dedupe_key(*, rd: str, net: str, nh: str) -> str: def _split_rest(rest: str, *, path_continuation: bool = False) -> tuple[str, str, str, str]: - """Parse trailing Metric LocPrf Tag/RtPrf Path columns (some may be blank). + """Parse trailing [From] Metric LocPrf Tag/RtPrf Path columns. - ``path_continuation``: indented wrap line after next-hop (often ``20 65254 ?`` - or ``4761 ?``) — prefer path/tag over inventing a metric. + OUT tables include a From column (IP) between Next Hop and Metric. + ``path_continuation``: indented wrap after network/nh (often ``0 100 283610 ?``). """ - parts = str(rest or "").split() + parts = [p for p in str(rest or "").split() if p] if not parts: return "", "", "", "" - if path_continuation and parts[-1] in ("?", "i", "e", "incomplete"): + + # Drop leading From IP when present (OUT neighbor tables). + if _looks_like_ip_or_prefix(parts[0]) and "/" not in parts[0]: + parts = parts[1:] + if not parts: + return "", "", "", "" + + if path_continuation and parts[-1] in _PATH_ENDS: if len(parts) == 1: return "", "", "", parts[0] if len(parts) == 2 and parts[0].isdigit(): - # ``4761 ?`` → path + # ``100 ?`` → loc_prf + origin; ``4761 ?`` → AS-path + origin + n = int(parts[0]) + if n <= 255: + return "", parts[0], "", parts[1] return "", "", "", " ".join(parts) + if len(parts) == 3 and parts[0].isdigit() and parts[1].isdigit(): + # ``0 100 ?`` → metric + loc + path + return parts[0], parts[1], "", parts[2] + if len(parts) >= 4 and all(parts[i].isdigit() for i in range(3)): + # ``0 100 283610 ?`` → metric loc tag path + return parts[0], parts[1], parts[2], " ".join(parts[3:]) if len(parts) >= 3 and parts[0].isdigit(): - # ``20 65254 ?`` → rtprf + path return "", "", parts[0], " ".join(parts[1:]) + metric = loc = tag = "" nums: list[str] = [] path_parts: list[str] = [] @@ -122,18 +148,33 @@ def _split_rest(rest: str, *, path_continuation: bool = False) -> tuple[str, str nums.append(p) else: path_parts.append(p) - if len(nums) >= 1: + # One number before origin → LocPrf (OUT often omits Metric). + if len(nums) == 1 and path_parts: + loc = nums[0] + elif len(nums) >= 1: metric = nums[0] - if len(nums) >= 2: - loc = nums[1] - if len(nums) >= 3: - tag = nums[2] + if len(nums) >= 2: + loc = nums[1] + if len(nums) >= 3: + tag = nums[2] path = " ".join(path_parts) return metric, loc, tag, path +def _rest_looks_complete(rest: str) -> bool: + """True when remainder has a path origin — safe to emit without waiting for wrap.""" + parts = str(rest or "").split() + if not parts: + return False + if parts[-1] in _PATH_ENDS: + return True + # From-only (single IP) → wait for metric/path wrap + if len(parts) == 1 and _looks_like_ip_or_prefix(parts[0]) and "/" not in parts[0]: + return False + return len(parts) >= 2 + + def _empty_if_total_zero(raw_text: str) -> bool: - """True when device reports Total number of routes: 0.""" total = _declared_total(raw_text) return total == 0 @@ -152,13 +193,14 @@ def _skip_noise_line(line: str) -> bool: return True if low.startswith("valid ") or low.startswith("invalid "): return True - # Banner / clock lines (e.g. "09:50:02 Indonesia Sat Sep 19 2026") if re.match(r"^\d{1,2}:\d{2}:\d{2}\b", low): return True if low.endswith("#") or "#'" in low: return True if re.search(r"\S+\s*#\s*$", line): return True + if low.startswith("show "): + return True return False @@ -181,7 +223,6 @@ def _emit_route( if not _looks_like_prefix(net): return if nh and not _looks_like_ip_or_prefix(nh): - # Path/metric-only continuation without a real next-hop — keep empty nh if re.search(r"[A-Za-z]", nh): return nh = "" @@ -212,57 +253,6 @@ def _emit_route( ) -def _map_fsm_rows( - rows: list[dict[str, Any]], - *, - local_as: str, - afi: str, - vrf: str, - neighbor: str, - direction: str, -) -> list[dict[str, Any]]: - out: list[dict[str, Any]] = [] - seen: set[str] = set() - for r in rows: - net = row_get(r, "NETWORK", "network") - if not _looks_like_prefix(net): - continue - nh = row_get(r, "NEXT_HOP", "next_hop") - if str(nh or "").strip().lower() in _HEADER_NETS: - continue - if nh and not _looks_like_ip_or_prefix(nh): - # Reject FSM false hits like NETWORK=20 NEXT_HOP=65254 - continue - rd = row_get(r, "RD", "rd") - key = _route_dedupe_key(rd=rd, net=net, nh=nh or "") - if key in seen: - continue - seen.add(key) - path = row_get(r, "PATH", "path") - flags = row_get(r, "STATUS", "status_codes", "FLAGS") - out.append( - { - "local_as": local_as[:16], - "afi": afi[:32], - "vrf": vrf[:128], - "neighbor": neighbor[:128], - "direction": direction[:8], - "rd": (rd or "")[:64], - "network": net[:128], - "next_hop": nh[:128], - "metric": row_get(r, "METRIC", "metric")[:32], - "loc_prf": row_get(r, "LOC_PRF", "loc_prf")[:32], - "tag": row_get(r, "TAG", "RT_PRF", "tag")[:32], - "path": path[:256], - "status_codes": re.sub(r"\s+", "", (flags or "").strip())[:16], - "as_num": "", - "state": "", - "pfx_rcd": "", - } - ) - return out - - def _hand_parse( *, raw_text: str, @@ -273,100 +263,122 @@ def _hand_parse( direction: str = "", **_kw: Any, ) -> list[dict[str, Any]]: - """Parse neighbor in/out tables; join IPv6 network / next-hop / path wraps.""" + """Parse neighbor in/out tables; join heavy IPv6 / From / metric wraps.""" out: list[dict[str, Any]] = [] seen: set[str] = set() pending_net = "" pending_flags = "" pending_nh = "" + pending_rest = "" current_rd = "" + current_vrf = vrf def _flush_pending(*, rest: str = "", path_continuation: bool = False) -> None: - nonlocal pending_net, pending_flags, pending_nh + nonlocal pending_net, pending_flags, pending_nh, pending_rest if not pending_net: return + merged = " ".join(x for x in (pending_rest, rest) if x).strip() _emit_route( out, seen, local_as=local_as, afi=afi, - vrf=vrf, + vrf=current_vrf, neighbor=neighbor, direction=direction, rd=current_rd, net=pending_net, nh=pending_nh, - rest=rest, + rest=merged, flags=pending_flags, path_continuation=path_continuation, ) pending_net = "" pending_flags = "" pending_nh = "" + pending_rest = "" - for raw in str(raw_text or "").splitlines(): + def _start_pending(*, net: str, nh: str, flags: str, rest: str) -> None: + nonlocal pending_net, pending_flags, pending_nh, pending_rest + _flush_pending() + pending_net = net + pending_flags = flags + pending_nh = nh + pending_rest = rest + + for raw in _normalize_cli_text(raw_text).splitlines(): line = raw.rstrip() rd_m = _RD_RE.match(line.strip()) if rd_m: _flush_pending() - current_rd = rd_m.group(1).strip() + current_rd = (rd_m.group("rd") or "").strip() + vrf_from_rd = (rd_m.group("vrf") or "").strip() + if vrf_from_rd: + current_vrf = vrf_from_rd continue if _skip_noise_line(line): continue - # Continuation: indented next-hop after network-only line. - # RR vpnv6 often puts NH + LocPrf/Path on the same indented line - # ("24.11.0.8 100 0 ?") — take first token - # as NH or ECMP legs collapse under empty next_hop dedupe (~half count). - if pending_net and not pending_nh and line[:1].isspace(): + # Indented continuation while a route is open + if pending_net and line[:1].isspace(): tok = line.strip() + if not tok: + continue parts = tok.split() first = parts[0] if parts else "" - if _looks_like_ip_or_prefix(first) and "/" not in first: - pending_nh = first - rest = " ".join(parts[1:]) - if rest: - _flush_pending(rest=rest, path_continuation=True) - continue - # Metrics/path without explicit next-hop (rare) - if tok and not _looks_like_prefix(first): + if not pending_nh: + if _looks_like_ip_or_prefix(first) and "/" not in first: + pending_nh = first + more = " ".join(parts[1:]) + if more and _rest_looks_complete(more): + _flush_pending(rest=more, path_continuation=True) + elif more: + pending_rest = " ".join(x for x in (pending_rest, more) if x) + continue + if not _looks_like_prefix(first): + _flush_pending(rest=tok, path_continuation=True) + continue + else: _flush_pending(rest=tok, path_continuation=True) continue - # Continuation: indented path/metric after network+nh - if pending_net and pending_nh and line[:1].isspace(): - tok = line.strip() - if tok: - _flush_pending(rest=tok, path_continuation=True) - continue + # Peel optional status codes (* i / *i / > …) + flags = "" + body = line + st = _STATUS_LEAD_RE.match(line) + if st: + flags = (st.group("flags") or "").strip() + body = line[st.end() :] - # Full one-liner (typical IPv4 / RR "* i prefix …") - m = _ROUTE_ONE_LINE_RE.match(line) + # Full / partial: network + next-hop (+ rest) + m = _NET_NH_RE.match(body) if m and _looks_like_prefix(m.group("net")) and _looks_like_ip_or_prefix(m.group("nh")): - _flush_pending() - _emit_route( - out, - seen, - local_as=local_as, - afi=afi, - vrf=vrf, - neighbor=neighbor, - direction=direction, - rd=current_rd, - net=m.group("net"), - nh=m.group("nh"), - rest=m.group("rest"), - flags=m.group("flags") or "", - ) + rest = (m.group("rest") or "").strip() + if _rest_looks_complete(rest): + _flush_pending() + _emit_route( + out, + seen, + local_as=local_as, + afi=afi, + vrf=current_vrf, + neighbor=neighbor, + direction=direction, + rd=current_rd, + net=m.group("net"), + nh=m.group("nh"), + rest=rest, + flags=flags, + ) + else: + # Empty rest or From-only → wait for metric/path wrap + _start_pending(net=m.group("net"), nh=m.group("nh"), flags=flags, rest=rest) continue - # Network alone → wait for next-hop / path wraps (IPv6) - m_net = _NET_ONLY_RE.match(line) + # Network alone → wait for next-hop wrap + m_net = _NET_ONLY_RE.match(body) if m_net and _looks_like_prefix(m_net.group("net")): - _flush_pending() - pending_net = m_net.group("net") - pending_flags = m_net.group("flags") or "" - pending_nh = "" + _start_pending(net=m_net.group("net"), nh="", flags=flags, rest="") continue _flush_pending() @@ -376,62 +388,22 @@ def _hand_parse( def normalize_bgp_route( *, 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, + **_kw: Any, ) -> list[dict[str, Any]]: - local_as = _detect_local_as(command, params) - afi = _detect_bgp_afi(command, params) - vrf = _detect_vrf(command, params) - neighbor = _detect_neighbor(command, params) - direction = _detect_direction(command, params) + """Hand-only normalize (see module docstring).""" + raw_text = _normalize_cli_text(raw_text) if _empty_if_total_zero(raw_text): return [] - - 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 bgp neighbor routes").strip() - if platform and cmd: - tables = apply_rules( - platform=platform, text=raw_text, rule_keys=RULE_KEYS, command=cmd - ) - - def _map(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: - return _map_fsm_rows( - rows, - local_as=local_as, - afi=afi, - vrf=vrf, - neighbor=neighbor, - direction=direction, - ) - - def _hand(*, raw_text: str, **kw: Any) -> list[dict[str, Any]]: - return _hand_parse( - raw_text=raw_text, - local_as=local_as, - afi=afi, - vrf=vrf, - neighbor=neighbor, - direction=direction, - **kw, - ) - - rows = prefer_fsm(tables, RULE_KEYS, _map, _hand, raw_text=raw_text) - # Prefer hand when FSM under-parses vs device-declared total (common on vpnv6 wraps). - declared = _declared_total(raw_text) - if declared is not None and declared > 0 and len(rows) < int(declared * 0.9): - hand_rows = _hand(raw_text=raw_text) - if len(hand_rows) > len(rows): - return hand_rows - return rows + return _hand_parse( + raw_text=raw_text, + local_as=_detect_local_as(command, params), + afi=_detect_bgp_afi(command, params), + vrf=_detect_vrf(command, params), + neighbor=_detect_neighbor(command, params), + direction=_detect_direction(command, params), + ) normalize_bgp_route.RULE_KEYS = RULE_KEYS diff --git a/netx_api/cli_templates/index b/netx_api/cli_templates/index index ae00425..7856b71 100644 --- a/netx_api/cli_templates/index +++ b/netx_api/cli_templates/index @@ -13,7 +13,8 @@ zte/zte_zxros_show_bgp_summary.textfsm, .*, zte_zxros, sh[[ow]] bgp .+ sum[[mary zte/zte_zxros_show_ospf_neighbor.textfsm, .*, zte_zxros, sh[[ow]] ip ospf nei[[ghbor]] zte/zte_zxros_show_vrrp_brief.textfsm, .*, zte_zxros, sh[[ow]] vrrp .+ br[[ief]] zte/zte_zxros_show_opticalinfo_brief.textfsm, .*, zte_zxros, sh[[ow]] opticalinfo br[[ief]] -zte/zte_zxros_show_bgp_neighbor_routes.textfsm, .*, zte_zxros, sh[[ow]] bgp .+ nei[[ghbor]] (in|out) +# bgp neighbor in/out routes: hand-only (parsers/zte/bgp_route.py); template kept as archive +# 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]] diff --git a/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm b/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm index 1b6f2a7..95a87f6 100644 --- a/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm +++ b/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm @@ -1,7 +1,12 @@ +# ARCHIVE — not registered in index. bgp_route uses hand-only normalize +# (IN/OUT / status / From / RD+VRF / wraps diverge too much for one template). Value Filldown RD (\S+) -Value STATUS ((?:\*[<>isd]*(?:\s+[isd]+)?|[<>isd]+)?) +Value Filldown VRF (\S+) +Value STATUS (\*(?:\s*[<>isd]+)?|[<>isd]+) Value NETWORK (\d{1,3}(?:\.\d{1,3}){3}/\d{1,2}|[0-9A-Fa-f:]+/\d{1,3}) -Value NEXT_HOP (\d{1,3}(?:\.\d{1,3}){3}|[0-9A-Fa-f:]+) +# IPv4, IPv6, or IPv4-mapped (::FFFF:a.b.c.d) +Value NEXT_HOP (\d{1,3}(?:\.\d{1,3}){3}|::[Ff]{4}:\d{1,3}(?:\.\d{1,3}){3}|[0-9A-Fa-f:]+) +Value FROM (\d{1,3}(?:\.\d{1,3}){3}|::[Ff]{4}:\d{1,3}(?:\.\d{1,3}){3}|[0-9A-Fa-f:]+) Value METRIC (\S*) Value LOC_PRF (\S*) Value TAG (\S*) @@ -12,13 +17,12 @@ Start ^\s*\*?\s*Network\s+Next -> Routes ^Network\s+Next -> Routes ^\s*Dest\s+Next -> Routes - # RD may appear before header on some snippets; keep Filldown ready - ^Route\s+Distinguisher\s*:?\s*${RD} + ^Route\s+Distinguisher\s*:?\s*${RD}(?:\s*\(\s*default\s+for\s+vrf\s+${VRF}\s*\))? ^\s*$$ ^.* -> Start Routes - ^Route\s+Distinguisher\s*:?\s*${RD} + ^Route\s+Distinguisher\s*:?\s*${RD}(?:\s*\(\s*default\s+for\s+vrf\s+${VRF}\s*\))? ^Status\s+codes ^Origin\s+codes ^Current\s+AS @@ -30,11 +34,28 @@ Routes ^\s*\*?\s*Network\s+Next ^Network\s+Next ^\s*Dest\s+Next - # One-line IPv4 / RR "* i prefix nh …" + # IN one-line with status: * i prefix nh metric loc tag path ^\s*${STATUS}\s+${NETWORK}\s+${NEXT_HOP}\s+${METRIC}\s+${LOC_PRF}\s+${TAG}\s+${PATH}\s*$$ -> Record + ^\s*${STATUS}\s+${NETWORK}\s+${NEXT_HOP}\s+${METRIC}\s+${LOC_PRF}\s+${PATH}\s*$$ -> Record ^\s*${STATUS}\s+${NETWORK}\s+${NEXT_HOP}\s+${PATH}\s*$$ -> Record - # IPv6 wrap: network alone, then next-hop (+ optional metric/path) lines + # OUT one-line without status: net nh [from] metric locprf [tag] path + ^\s*${NETWORK}\s+${NEXT_HOP}\s+${FROM}\s+${METRIC}\s+${LOC_PRF}\s+${TAG}\s+${PATH}\s*$$ -> Record + ^\s*${NETWORK}\s+${NEXT_HOP}\s+${FROM}\s+${METRIC}\s+${LOC_PRF}\s+${PATH}\s*$$ -> Record + ^\s*${NETWORK}\s+${NEXT_HOP}\s+${FROM}\s+${LOC_PRF}\s+${PATH}\s*$$ -> Record + ^\s*${NETWORK}\s+${NEXT_HOP}\s+${LOC_PRF}\s+${PATH}\s*$$ -> Record + # IN wrap: status+net+nh then indented metrics + ^\s*${STATUS}\s+${NETWORK}\s+${NEXT_HOP}\s*$$ + ^\s+${METRIC}\s+${LOC_PRF}\s+${TAG}\s+${PATH}\s*$$ -> Record + ^\s+${METRIC}\s+${LOC_PRF}\s+${PATH}\s*$$ -> Record + ^\s+${PATH}\s*$$ -> Record + # OUT wrap: net+nh+from then indented metrics (From may be long IPv6) + ^\s*${NETWORK}\s+${NEXT_HOP}\s+${FROM}\s*$$ + ^\s+${METRIC}\s+${LOC_PRF}\s+${TAG}\s+${PATH}\s*$$ -> Record + ^\s+${METRIC}\s+${LOC_PRF}\s+${PATH}\s*$$ -> Record + ^\s+${PATH}\s*$$ -> Record + # Classic wrap: network alone, then next-hop (+ optional metric/path) ^\s*${STATUS}\s+${NETWORK}\s*$$ + ^\s*${NETWORK}\s*$$ ^\s+${NEXT_HOP}\s+${METRIC}\s+${LOC_PRF}\s+${TAG}\s+${PATH}\s*$$ -> Record ^\s+${NEXT_HOP}\s+${METRIC}\s+${LOC_PRF}\s+${PATH}\s*$$ -> Record ^\s+${NEXT_HOP}\s+${PATH}\s*$$ -> Record diff --git a/tests/test_zte_extended_parsers.py b/tests/test_zte_extended_parsers.py index bde8530..f705455 100644 --- a/tests/test_zte_extended_parsers.py +++ b/tests/test_zte_extended_parsers.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re import unittest from pathlib import Path @@ -1235,5 +1236,138 @@ Route Distinguisher:10.0.0.1:100 self.assertTrue(all(r.get("status_codes") == "*i" for r in routes)) self.assertTrue(all(r.get("rd") == "10.0.0.1:100" for r in routes)) + def test_bgp_route_prod_snippets_rd_vrf_and_wraps(self) -> None: + """Live RR snippets: OUT without status, RD+VRF, vpnv6 *i + NH wrap.""" + from pathlib import Path + + fixture = ( + Path(__file__).resolve().parents[2] + / "test" + / "show-zte" + / "Untitled-1.ini" + ) + if not fixture.is_file(): + fixture = ( + Path(__file__).resolve().parents[3] + / "test" + / "show-zte" + / "Untitled-1.ini" + ) + raw_all = fixture.read_text(encoding="utf-8", errors="replace") + + def _section(cmd_prefix: str) -> tuple[str, str]: + text = raw_all.replace("\u00a0", " ") + parts = re.split(r"(?=^show bgp )", text, flags=re.M) + for part in parts: + part = part.strip() + if not part.startswith("show bgp"): + continue + head = part.splitlines()[0].strip() + if head.startswith(cmd_prefix): + body = "\n".join(part.splitlines()[1:]) + return head, body + self.fail(f"section not found: {cmd_prefix}") + + # ipv4 out — sample truncated vs declared 34; parse all pasted routes + cmd, body = _section("show bgp ipv4 unicast neighbor out") + ipv4_out = normalize_bgp_route( + raw_text=body, + command=cmd, + vendor="ZTE", + device_type="zte_zxros", + params={ + "neighbor": "24.11.0.8", + "direction": "out", + "afi": "ipv4", + "local_as": "24208", + }, + ) + self.assertEqual(len(ipv4_out), 21) + self.assertTrue(all(r["next_hop"] for r in ipv4_out)) + + # vpnv6 out — RD + VRF + ::FFFF NH + From wraps + cmd, body = _section("show bgp vpnv6 unicast neighbor out") + v6_out = normalize_bgp_route( + raw_text=body, + command=cmd, + vendor="ZTE", + device_type="zte_zxros", + params={ + "neighbor": "24.11.0.8", + "direction": "out", + "afi": "vpnv6", + "local_as": "24208", + }, + ) + self.assertEqual(len(v6_out), 559) + self.assertTrue(all(r.get("rd") for r in v6_out)) + self.assertTrue(all(r.get("vrf") for r in v6_out)) + self.assertEqual(v6_out[0]["rd"], "2:111") + self.assertEqual(v6_out[0]["vrf"], "css-srv6-mpls-1") + self.assertTrue(v6_out[0]["next_hop"].upper().startswith("::FFFF:")) + self.assertGreaterEqual( + sum(1 for r in v6_out if r["network"] == "60::/64"), 2, "ECMP under RD" + ) + + # vpnv4 out — RD + VRF + cmd, body = _section("show bgp vpnv4 unicast neighbor out") + v4_out = normalize_bgp_route( + raw_text=body, + command=cmd, + vendor="ZTE", + device_type="zte_zxros", + params={ + "neighbor": "24.11.0.8", + "direction": "out", + "afi": "vpnv4", + "local_as": "24208", + }, + ) + self.assertEqual(len(v4_out), 54) + self.assertTrue(all(r.get("rd") and r.get("vrf") for r in v4_out)) + + # vpnv6 in — * i + IPv4-mapped NH wrap onto metric line + cmd, body = _section("show bgp vpnv6 unicast neighbor in") + v6_in = normalize_bgp_route( + raw_text=body, + command=cmd, + vendor="ZTE", + device_type="zte_zxros", + params={ + "neighbor": "24.11.0.8", + "direction": "in", + "afi": "vpnv6", + "local_as": "24208", + }, + ) + self.assertEqual(len(v6_in), 22) + self.assertTrue(all(r.get("rd") for r in v6_in)) + self.assertTrue(all(r.get("status_codes") == "*i" for r in v6_in)) + self.assertTrue(all(":" in r["network"] for r in v6_in)) + self.assertTrue(all(r["next_hop"] for r in v6_in)) + self.assertEqual(v6_in[0]["network"], "100:0:17::/64") + self.assertEqual(v6_in[0]["next_hop"].upper(), "::FFFF:24.11.0.5") + + # vpnv4 in — RD fill-down + ECMP next-hops + cmd, body = _section("show bgp vpnv4 unicast neighbor in") + v4_in = normalize_bgp_route( + raw_text=body, + command=cmd, + vendor="ZTE", + device_type="zte_zxros", + params={ + "neighbor": "24.11.0.8", + "direction": "in", + "afi": "vpnv4", + "local_as": "24208", + }, + ) + self.assertEqual(len(v4_in), 21) + self.assertTrue(all(r.get("rd") for r in v4_in)) + multi = [r for r in v4_in if r["network"] == "1.0.0.1/32"] + self.assertEqual(len(multi), 3) + self.assertEqual(len({r["next_hop"] for r in multi}), 3) + + if __name__ == "__main__": unittest.main()