From ac4f115ee8d8df2e54da8b6c48ff7f2769de8869 Mon Sep 17 00:00:00 2001 From: oliver Date: Sun, 20 Sep 2026 14:49:41 +0800 Subject: [PATCH] Harden ZTE biz-state parsers and default | one-line collect CLIs. Append secondary IPs with M/S tags, skip ND6 clock banners, split optical power vs thresholds including QSFP lanes, broaden static-route egress ifnames, and add | one-line to ZTE collect templates except interface detail. Co-authored-by: Cursor --- .../biz_state/parsers/zte/config_interface.py | 48 +++-- .../parsers/zte/config_static_route.py | 42 +++- netx_api/biz_state/parsers/zte/nd6_cache.py | 18 ++ .../biz_state/parsers/zte/optical_brief.py | 192 ++++++++++++++---- netx_api/biz_state/profiles.py | 84 ++++---- .../zte_zxros_show_opticalinfo_brief.textfsm | 8 +- netx_api/topology_lldp.py | 2 +- tests/test_biz_state_profiles.py | 2 +- tests/test_enrich_framework.py | 2 +- tests/test_zte_config_intent.py | 26 ++- tests/test_zte_config_real_dump.py | 9 +- tests/test_zte_extended_parsers.py | 30 ++- tests/test_zte_status_parsers.py | 19 ++ 13 files changed, 373 insertions(+), 109 deletions(-) diff --git a/netx_api/biz_state/parsers/zte/config_interface.py b/netx_api/biz_state/parsers/zte/config_interface.py index 6f2b394..c3ef054 100644 --- a/netx_api/biz_state/parsers/zte/config_interface.py +++ b/netx_api/biz_state/parsers/zte/config_interface.py @@ -2,6 +2,10 @@ Duplicate ``interface`` blocks (common for tunnels) are merged: later values override empty ones; ``admin`` always takes the last explicit setting. + +IPv4/IPv6 addresses (including ``secondary``) are collected in order into +``ip_address`` / ``ipv6_address``; ``secondary_tag`` / ``ipv6_secondary_tag`` +mark the first as ``M`` and later ones as ``S``. """ from __future__ import annotations @@ -15,12 +19,29 @@ RULE_KEYS: tuple[str, ...] = () _IFACE_RE = re.compile(r"^\s*interface\s+(\S+)\s*$", re.I) _VRF_RE = re.compile(r"^\s*ip\s+vrf\s+forwarding\s+(\S+)\s*$", re.I) -_IP_RE = re.compile(r"^\s*ip\s+address\s+(\S+)(?:\s+(\S+))?\s*$", re.I) -_IP6_RE = re.compile(r"^\s*ipv6\s+address\s+(\S+)\s*$", re.I) +_IP_RE = re.compile( + r"^\s*ip\s+address\s+(\S+)(?:\s+(\S+))?(?:\s+secondary)?\s*$", + re.I, +) +_IP6_RE = re.compile(r"^\s*ipv6\s+address\s+(\S+)(?:\s+secondary)?\s*$", re.I) _DESC_RE = re.compile(r"^\s*description\s+(.*?)\s*$", re.I) _MTU_RE = re.compile(r"^\s*mtu\s+(\d+)\s*$", re.I) +def _ms_tags(count: int) -> str: + if count <= 0: + return "" + return ",".join("M" if i == 0 else "S" for i in range(count)) + + +def _merge_addr_list(prev: str, new: str, *, limit: int = 256) -> str: + existing = [x for x in (prev or "").split(",") if x] + for ip in (new or "").split(","): + if ip and ip not in existing: + existing.append(ip) + return ",".join(existing)[:limit] + + def _row_from_cur(cur: dict[str, Any]) -> dict[str, Any]: ips = cur.get("_ips") or [] ip6s = cur.get("_ip6s") or [] @@ -30,7 +51,9 @@ def _row_from_cur(cur: dict[str, Any]) -> dict[str, Any]: "description": str(cur.get("description") or "")[:256], "admin": str(cur.get("admin") or "up")[:16], "ip_address": ",".join(ips)[:256], + "secondary_tag": _ms_tags(len(ips))[:64], "ipv6_address": ",".join(ip6s)[:256], + "ipv6_secondary_tag": _ms_tags(len(ip6s))[:64], "mtu": str(cur.get("mtu") or "")[:16], "_admin_set": bool(cur.get("_admin_set")), } @@ -57,17 +80,15 @@ def _flush( if row[key]: prev[key] = row[key] if row["ip_address"]: - existing = [x for x in (prev["ip_address"] or "").split(",") if x] - for ip in row["ip_address"].split(","): - if ip and ip not in existing: - existing.append(ip) - prev["ip_address"] = ",".join(existing)[:256] + merged = _merge_addr_list(prev["ip_address"], row["ip_address"]) + prev["ip_address"] = merged + prev["secondary_tag"] = _ms_tags(len([x for x in merged.split(",") if x]))[:64] if row["ipv6_address"]: - existing = [x for x in (prev["ipv6_address"] or "").split(",") if x] - for ip in row["ipv6_address"].split(","): - if ip and ip not in existing: - existing.append(ip) - prev["ipv6_address"] = ",".join(existing)[:256] + merged = _merge_addr_list(prev["ipv6_address"], row["ipv6_address"]) + prev["ipv6_address"] = merged + prev["ipv6_secondary_tag"] = _ms_tags( + len([x for x in merged.split(",") if x]) + )[:64] return by_iface[iface] = len(out) out.append(row) @@ -132,6 +153,9 @@ def normalize_config_interface( if m: addr = m.group(1).strip() mask = (m.group(2) or "").strip() + # Avoid treating the literal word "secondary" as a mask when only one token follows. + if mask.lower() == "secondary": + mask = "" cur["_ips"].append(f"{addr}/{mask}" if mask else addr) continue m = _IP6_RE.match(line) diff --git a/netx_api/biz_state/parsers/zte/config_static_route.py b/netx_api/biz_state/parsers/zte/config_static_route.py index 47b13fe..a5a7043 100644 --- a/netx_api/biz_state/parsers/zte/config_static_route.py +++ b/netx_api/biz_state/parsers/zte/config_static_route.py @@ -27,9 +27,18 @@ _V6_HEAD = re.compile( ) _IPV4_RE = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$") +# Broad ZTE if-name match: whitelist prefixes + generic if-like tokens. _IFACE_RE = re.compile( - r"(?i)^(null\d*|gei-\S+|xgei-\S+|cgei-\S+|bvi\S*|vlan\S*|smartgroup\S*|" - r"loopback\S*|mgmt\S*|pos\S*|eth\S*|tunnel\S*|irb\S*|ve\S*)$" + r"(?i)^(?:" + r"null\d*" + r"|(?:xxv|xl|x|c|v6|q|fe)?gei-\S+" + r"|bvi\S*|vlan\S*|smartgroup\S*|loopback\S*|mgmt_?\S*" + r"|pos\S*|(?:fast)?eth\S*" + r"|(?:te_|gre_|ip_|v6)?tunnel\S*" + r"|irb\S*|ve\S*|dialer\S*|serial\S*|ppp\S*|bundle\S*" + r"|fei-\S*|qi-\S*" + r"|[a-z][a-z0-9]*[-_][a-z0-9_./:-]+" # e.g. te_tunnel36, gre_tunnel1 + r")$" ) _KW = frozenset( { @@ -52,8 +61,21 @@ def _is_ipv4(tok: str) -> bool: return bool(_IPV4_RE.match(tok or "")) +def _is_ipv6_tok(tok: str) -> bool: + t = str(tok or "") + return ":" in t and not _is_iface(t) + + def _is_iface(tok: str) -> bool: - return bool(_IFACE_RE.match(tok or "")) + t = str(tok or "").strip() + if not t or t.lower() in _KW: + return False + if _is_ipv4(t): + return False + # IPv6 next-hop (has colon) is not an interface name. + if ":" in t and re.match(r"(?i)^[0-9a-f:]+(/\d+)?$", t): + return False + return bool(_IFACE_RE.match(t)) def _take_kw(tokens: list[str], i: int, row: dict[str, str]) -> int: @@ -112,14 +134,19 @@ def _parse_v4_rest(rest: str) -> dict[str, str]: _take_kw(tokens, 2, row) return row if _is_iface(tokens[0]): + # interface [next-hop] row["interface"] = tokens[0] i = 1 - if i < len(tokens) and _is_ipv4(tokens[i]) and tokens[i].lower() not in _KW: + if i < len(tokens) and _is_ipv4(tokens[i]): row["next_hop"] = tokens[i] i += 1 elif _is_ipv4(tokens[0]): + # next-hop [interface] row["next_hop"] = tokens[0] i = 1 + if i < len(tokens) and _is_iface(tokens[i]): + row["interface"] = tokens[i] + i += 1 _take_kw(tokens, i, row) return row @@ -147,12 +174,15 @@ def _parse_v6_rest(rest: str) -> dict[str, str]: if _is_iface(tokens[0]): row["interface"] = tokens[0] i = 1 - if i < len(tokens) and ":" in tokens[i] and tokens[i].lower() not in _KW: + if i < len(tokens) and _is_ipv6_tok(tokens[i]): row["next_hop"] = tokens[i] i += 1 - elif ":" in tokens[0]: + elif _is_ipv6_tok(tokens[0]): row["next_hop"] = tokens[0] i = 1 + if i < len(tokens) and _is_iface(tokens[i]): + row["interface"] = tokens[i] + i += 1 _take_kw(tokens, i, row) return row diff --git a/netx_api/biz_state/parsers/zte/nd6_cache.py b/netx_api/biz_state/parsers/zte/nd6_cache.py index 91c3fcc..e9c97d3 100644 --- a/netx_api/biz_state/parsers/zte/nd6_cache.py +++ b/netx_api/biz_state/parsers/zte/nd6_cache.py @@ -10,10 +10,23 @@ _ND6_ROW_RE = re.compile( r"(?P\S+)\s+(?P\S+)\s*$", re.I, ) +# CLI banner / wall-clock line (e.g. "14:06:53 Beijing Sun Sep 20 2026") +_CLOCK_LINE_RE = re.compile(r"^\d{1,2}:\d{2}:\d{2}\b") +_TIME_OF_DAY_RE = re.compile(r"^\d{1,2}:\d{2}:\d{2}$") RULE_KEYS: tuple[str, ...] = () +def _looks_like_ipv6(addr: str) -> bool: + a = str(addr or "").strip() + if not a or ":" not in a: + return False + # HH:MM:SS is not an IPv6 address + if _TIME_OF_DAY_RE.match(a): + return False + return bool(re.search(r"[0-9a-fA-F]", a)) + + def normalize_nd6_cache( *, raw_text: str, @@ -30,6 +43,8 @@ def normalize_nd6_cache( line = raw.strip() if not line: continue + if _CLOCK_LINE_RE.match(line): + continue low = line.lower() if low.startswith("s-static") or low.startswith("total cache") or low.startswith("only current"): continue @@ -39,6 +54,8 @@ def normalize_nd6_cache( if not m: continue addr = m.group("addr") + if not _looks_like_ipv6(addr): + continue iface = m.group("iface") key = (addr, iface) if key in seen: @@ -56,4 +73,5 @@ def normalize_nd6_cache( ) return out + normalize_nd6_cache.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/zte/optical_brief.py b/netx_api/biz_state/parsers/zte/optical_brief.py index 3f78a6e..d49bef3 100644 --- a/netx_api/biz_state/parsers/zte/optical_brief.py +++ b/netx_api/biz_state/parsers/zte/optical_brief.py @@ -1,4 +1,9 @@ -"""ZTE: show opticalinfo brief.""" +"""ZTE: show opticalinfo brief [| one-line]. + +Power cells look like ``-6.0/[-20.0,-3.0]`` (value / [low,high] threshold). +50G+ (QSFP) modules print one lane per line; continuation lines are merged +into comma-separated ``rx_power`` / ``tx_power`` (and matching thresholds). +""" from __future__ import annotations @@ -11,57 +16,172 @@ from ..common.pipeline import prefer_fsm RULE_KEYS = ("zte_zxros_show_opticalinfo_brief",) -_OPT_RE = re.compile( - r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" - r"(?P\S+)\s+(?P\S+)\s+(?P\S+)", +# value/[low,high] or bare N/A|Unsupported|number (no threshold) +_CELL_RE = re.compile( + r"^(?PN/A|Unsupported|-?\d+(?:\.\d+)?)(?:/(?P\[[^\]]+\]))?$", + re.I, +) +_HDR_RE = re.compile( + r"^(?P\S+)\s+(?Poffline)\s*$|" + r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)\s+" + r"(?P\S+)\s+(?P\S+)\s+(?P\S+)(?:\s+(?P\S+))?\s*$", + re.I, +) +# Lane continuation (indented); require a status token so bare "7.5 7.7" summary is skipped. +_CONT_RE = re.compile( + r"^\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)(?:\s+(?P\S+))?\s*$", + re.I, ) +def _parse_cell(raw: str) -> tuple[str, str] | None: + s = str(raw or "").strip() + if not s: + return None + m = _CELL_RE.match(s) + if not m: + return None + return (m.group("val"), m.group("th") or "") + + +def _join(parts: list[str], *, limit: int = 128) -> str: + return ",".join(p for p in parts if p != "")[:limit] + + +def _dedupe_join(parts: list[str], *, limit: int = 128) -> str: + """Join thresholds; collapse when every lane shares the same bracket.""" + vals = [p for p in parts if p] + if not vals: + return "" + if len(set(vals)) == 1: + return vals[0][:limit] + return _join(vals, limit=limit) + + +def _empty_row(iface: str, optic_type: str = "", wavelength: str = "") -> dict[str, Any]: + return { + "interface": iface[:128], + "optic_type": optic_type[:64], + "wavelength": wavelength[:32], + "rx_power": "", + "rx_threshold": "", + "tx_power": "", + "tx_threshold": "", + "status": "", + "_rx": [], + "_rx_th": [], + "_tx": [], + "_tx_th": [], + "_status": [], + } + + +def _append_lane(row: dict[str, Any], rx_raw: str, tx_raw: str, status: str = "") -> bool: + rx = _parse_cell(rx_raw) + tx = _parse_cell(tx_raw) + if rx is None or tx is None: + return False + # Skip bare numeric summary lines (no threshold on either side, no status). + if not rx[1] and not tx[1] and not status: + return False + row["_rx"].append(rx[0]) + row["_rx_th"].append(rx[1]) + row["_tx"].append(tx[0]) + row["_tx_th"].append(tx[1]) + if status: + row["_status"].append(status) + return True + + +def _finalize(row: dict[str, Any]) -> dict[str, Any]: + statuses = row.pop("_status", []) + rx = row.pop("_rx", []) + rx_th = row.pop("_rx_th", []) + tx = row.pop("_tx", []) + tx_th = row.pop("_tx_th", []) + row["rx_power"] = _join(rx) + row["rx_threshold"] = _dedupe_join(rx_th) + row["tx_power"] = _join(tx) + row["tx_threshold"] = _dedupe_join(tx_th) + if statuses: + # Prefer first Abnormal/Unknown if any, else first status. + bad = next((s for s in statuses if s.lower() not in ("normal",)), None) + row["status"] = (bad or statuses[0])[:32] + return row + + def _map_fsm_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: - out: list[dict[str, Any]] = [] - seen: set[str] = set() + """Aggregate Filldown lane rows from TextFSM into one row per interface.""" + by: dict[str, dict[str, Any]] = {} + order: list[str] = [] for r in rows: iface = row_get(r, "INTERFACE", "interface") - if not iface or iface.lower() == "interface" or iface in seen: + if not iface or iface.lower() == "interface": continue - seen.add(iface) - out.append( - { - "interface": iface[:128], - "optic_type": row_get(r, "OPTIC_TYPE", "optic_type")[:64], - "wavelength": row_get(r, "WAVELENGTH", "wavelength")[:32], - "rx_power": row_get(r, "RX_POWER", "rx_power")[:64], - "tx_power": row_get(r, "TX_POWER", "tx_power")[:64], - "status": row_get(r, "STATUS", "status")[:32], - } + rx_raw = row_get(r, "RX_POWER", "rx_power") + tx_raw = row_get(r, "TX_POWER", "tx_power") + otype = row_get(r, "OPTIC_TYPE", "optic_type") + wave = row_get(r, "WAVELENGTH", "wavelength") + # Offline lines: only INTERFACE is new; Filldown may still hold prior type/wave. + if (not rx_raw and not tx_raw) or otype.lower() == "offline": + if iface not in by: + by[iface] = _empty_row(iface, "offline") + by[iface]["status"] = "offline" + order.append(iface) + continue + if iface not in by: + by[iface] = _empty_row(iface, otype, wave) + order.append(iface) + else: + cur = by[iface] + if otype and cur["optic_type"] in ("", "offline"): + cur["optic_type"] = otype[:64] + if wave and not cur["wavelength"]: + cur["wavelength"] = wave[:32] + _append_lane( + by[iface], + rx_raw, + tx_raw, + row_get(r, "STATUS", "status"), ) - return out + return [_finalize(by[i]) for i in order] def _hand_parse(*, raw_text: str, **_kw: Any) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] - seen: set[str] = set() + cur: dict[str, Any] | None = None + + def _flush() -> None: + nonlocal cur + if cur: + out.append(_finalize(cur)) + cur = None + for raw in str(raw_text or "").splitlines(): - line = raw.strip() - if not line or line.lower().startswith("interface"): + if not raw.strip(): continue - m = _OPT_RE.match(line) + low = raw.strip().lower() + if low.startswith("interface") and "type" in low: + continue + m = _HDR_RE.match(raw.rstrip()) + if m: + _flush() + if m.group("otype") and m.group("otype").lower() == "offline": + cur = _empty_row(m.group("iface"), "offline") + cur["status"] = "offline" + _flush() + continue + iface = m.group("iface2") or "" + cur = _empty_row(iface, m.group("otype2") or "", m.group("wave") or "") + _append_lane(cur, m.group("rx") or "", m.group("tx") or "", m.group("status") or "") + continue + if cur is None: + continue + m = _CONT_RE.match(raw) if not m: continue - iface = m.group("iface") - if iface in seen: - continue - seen.add(iface) - out.append( - { - "interface": iface[:128], - "optic_type": m.group("otype")[:64], - "wavelength": m.group("wave")[:32], - "rx_power": m.group("rx")[:64], - "tx_power": m.group("tx")[:64], - "status": m.group("status")[:32], - } - ) + _append_lane(cur, m.group("rx"), m.group("tx"), m.group("status") or "") + _flush() return out diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index 4cfa076..066b4a8 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -5,6 +5,7 @@ Display fields may be overridden at runtime via biz_state_command_override (hot from __future__ import annotations +import re from dataclasses import dataclass, field from typing import Any @@ -88,10 +89,9 @@ def _lldp_profiles() -> list[ParseProfile]: # Still register so UI can show; parse may return empty until templates exist. pass cmd = vp.lldp_command - # Escape for regex: match exact command ignoring extra whitespace flexibility - escaped = r"\s+".join( - __import__("re").escape(p) for p in cmd.split() if p - ) + # Escape for regex; ZTE (and others) may append "| one-line" on the template. + base_cmd = re.sub(r"\s*\|\s*one-line\s*$", "", cmd, flags=re.I).strip() or cmd + escaped = r"\s+".join(re.escape(p) for p in base_cmd.split() if p) out.append( ParseProfile( profile_id=f"{key}.lldp_neighbors", @@ -100,8 +100,8 @@ def _lldp_profiles() -> list[ParseProfile]: parser_id="lldp_neighbors", title="LLDP Neighbors", command_template=cmd, - match=rf"(?i)^\s*{escaped}\s*$", - textfsm_command=cmd, + match=rf"(?i)^\s*{escaped}(?:\s*\|\s*one-line)?\s*$", + textfsm_command=base_cmd, description=vp.notes or "LLDP neighbor table snapshot for cutover compare.", sample_output="", placeholders=[], @@ -318,8 +318,10 @@ _OPTICAL_FIELDS: list[FieldDef] = [ FieldDef("optic_type", length=64, role="meta", display_name="Type"), FieldDef("wavelength", length=32, role="meta", display_name="Wavelength"), # Optical power jitters β€” display only; compare status for cutover. - FieldDef("rx_power", length=64, role="counter", display_name="RxPower"), - FieldDef("tx_power", length=64, role="counter", display_name="TxPower"), + FieldDef("rx_power", length=128, role="counter", display_name="RxPower"), + FieldDef("rx_threshold", length=128, role="meta", display_name="Rx Threshold"), + FieldDef("tx_power", length=128, role="counter", display_name="TxPower"), + FieldDef("tx_threshold", length=128, role="meta", display_name="Tx Threshold"), FieldDef("status", length=32, role="state", display_name="Status"), ] @@ -405,7 +407,9 @@ _CONFIG_IFACE_FIELDS: list[FieldDef] = [ FieldDef("vrf", length=128, indexed=True, is_key=True, display_name="VRF"), FieldDef("admin", length=16, role="state", display_name="Admin"), FieldDef("ip_address", length=256, role="state", display_name="IPv4"), + FieldDef("secondary_tag", length=64, role="state", display_name="IPv4 M/S"), FieldDef("ipv6_address", length=256, role="state", display_name="IPv6"), + FieldDef("ipv6_secondary_tag", length=64, role="state", display_name="IPv6 M/S"), FieldDef("mtu", length=16, role="meta", display_name="MTU"), FieldDef("description", length=256, role="meta", display_name="描述"), ] @@ -510,8 +514,8 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="interface_brief", parser_id="interface_brief", title="Interface Brief", - command_template="show interface brief", - match=r"(?i)^\s*show\s+interface\s+brief\s*$", + command_template="show interface brief | one-line", + match=r"(?i)^\s*show\s+interface\s+brief(?:\s*\|\s*one-line)?\s*$", textfsm_command="show interface brief", description="Interface admin/phy/prot status brief.", fields=list(_IFACE_BRIEF_FIELDS), @@ -570,7 +574,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="if_intf", parser_id="if_intf", title="IF VRF (if-intf)", - command_template="show running-config if-intf", + command_template="show running-config if-intf | one-line", match=r"(?i)^\s*show\s+running-config\s+if-intf\s*$", textfsm_command="show running-config if-intf", description=( @@ -605,7 +609,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP VPNv4 Summary", - command_template="show bgp vpnv4 unicast summary", + command_template="show bgp vpnv4 unicast summary | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv4\s+unicast\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv4 unicast summary", description="BGP VPNv4 peer summary (afi=vpnv4).", @@ -621,7 +625,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP IPv4 Summary", - command_template="show bgp ipv4 unicast summary", + command_template="show bgp ipv4 unicast summary | one-line", match=r"(?i)^\s*show\s+bgp\s+ipv4\s+unicast\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp ipv4 unicast summary", description="BGP IPv4 unicast peer summary (afi=ipv4).", @@ -637,7 +641,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP VPNv6 Summary", - command_template="show bgp vpnv6 unicast summary", + command_template="show bgp vpnv6 unicast summary | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv6\s+unicast\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv6 unicast summary", description="BGP VPNv6 peer summary (afi=vpnv6).", @@ -653,7 +657,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP IPv6 Summary", - command_template="show bgp ipv6 unicast summary", + command_template="show bgp ipv6 unicast summary | one-line", match=r"(?i)^\s*show\s+bgp\s+ipv6\s+unicast\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp ipv6 unicast summary", description="BGP IPv6 unicast peer summary (afi=ipv6).", @@ -669,7 +673,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP VPNv4 VRF Summary", - command_template="show bgp vpnv4 unicast vrf summary", + command_template="show bgp vpnv4 unicast vrf summary | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv4\s+unicast\s+vrf\s+(?P\S+)\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv4 unicast summary", description="Per-VRF BGP VPNv4 peer summary.", @@ -686,7 +690,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP VPNv6 VRF Summary", - command_template="show bgp vpnv6 unicast vrf summary", + command_template="show bgp vpnv6 unicast vrf summary | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv6\s+unicast\s+vrf\s+(?P\S+)\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv6 unicast summary", description="Per-VRF BGP VPNv6 peer summary.", @@ -703,7 +707,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP L2VPN EVPN Summary", - command_template="show bgp l2vpn evpn summary", + command_template="show bgp l2vpn evpn summary | one-line", match=r"(?i)^\s*show\s+bgp\s+l2vpn\s+evpn\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp l2vpn evpn summary", description="BGP L2VPN EVPN peer summary (afi=evpn).", @@ -719,7 +723,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_peer", parser_id="bgp_peer", title="BGP L2VPN VPLS Summary", - command_template="show bgp l2vpn vpls summary", + command_template="show bgp l2vpn vpls summary | one-line", match=r"(?i)^\s*show\s+bgp\s+l2vpn\s+vpls\s+summary(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp l2vpn vpls summary", description="BGP L2VPN VPLS peer summary (afi=vpls).", @@ -801,7 +805,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv4 Neighbor In", - command_template="show bgp vpnv4 unicast neighbor in ", + command_template="show bgp vpnv4 unicast neighbor in | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv4\s+unicast\s+neighbor\s+in\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv4 unicast neighbor in", description="Routes learned from VPNv4 neighbor; summary aux for peer state.", @@ -826,7 +830,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv4 Neighbor Out", - command_template="show bgp vpnv4 unicast neighbor out ", + command_template="show bgp vpnv4 unicast neighbor out | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv4\s+unicast\s+neighbor\s+out\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv4 unicast neighbor out", description="Routes advertised to VPNv4 neighbor (large; bind neighbor).", @@ -851,7 +855,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv6 Neighbor In", - command_template="show bgp vpnv6 unicast neighbor in ", + command_template="show bgp vpnv6 unicast neighbor in | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv6\s+unicast\s+neighbor\s+in\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv6 unicast neighbor in", description="Routes learned from VPNv6 neighbor; summary aux for peer state.", @@ -876,7 +880,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv6 Neighbor Out", - command_template="show bgp vpnv6 unicast neighbor out ", + command_template="show bgp vpnv6 unicast neighbor out | one-line", match=r"(?i)^\s*show\s+bgp\s+vpnv6\s+unicast\s+neighbor\s+out\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$", textfsm_command="show bgp vpnv6 unicast neighbor out", description="Routes advertised to VPNv6 neighbor (large; bind neighbor).", @@ -901,7 +905,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv4 VRF Neighbor In", - command_template="show bgp vpnv4 unicast vrf neighbor in ", + command_template="show bgp vpnv4 unicast vrf neighbor in | one-line", match=( r"(?i)^\s*show\s+bgp\s+vpnv4\s+unicast\s+vrf\s+(?P\S+)\s+" r"neighbor\s+in\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" @@ -929,7 +933,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv4 VRF Neighbor Out", - command_template="show bgp vpnv4 unicast vrf neighbor out ", + command_template="show bgp vpnv4 unicast vrf neighbor out | one-line", match=( r"(?i)^\s*show\s+bgp\s+vpnv4\s+unicast\s+vrf\s+(?P\S+)\s+" r"neighbor\s+out\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" @@ -957,7 +961,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv6 VRF Neighbor In", - command_template="show bgp vpnv6 unicast vrf neighbor in ", + command_template="show bgp vpnv6 unicast vrf neighbor in | one-line", match=( r"(?i)^\s*show\s+bgp\s+vpnv6\s+unicast\s+vrf\s+(?P\S+)\s+" r"neighbor\s+in\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" @@ -985,7 +989,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="bgp_route", parser_id="bgp_route", title="BGP VPNv6 VRF Neighbor Out", - command_template="show bgp vpnv6 unicast vrf neighbor out ", + command_template="show bgp vpnv6 unicast vrf neighbor out | one-line", match=( r"(?i)^\s*show\s+bgp\s+vpnv6\s+unicast\s+vrf\s+(?P\S+)\s+" r"neighbor\s+out\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$" @@ -1014,7 +1018,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="ip_route", parser_id="ip_route", title="IPv4 Forwarding VRF", - command_template="show ip forwarding route vrf ", + command_template="show ip forwarding route vrf | one-line", match=r"(?i)^\s*show\s+ip\s+forwarding\s+route\s+vrf\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$", textfsm_command="show ip forwarding route", description="IPv4 FIB for one VRF.", @@ -1031,7 +1035,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="ip_route", parser_id="ip_route", title="IPv4 Forwarding Global", - command_template="show ip forwarding route", + command_template="show ip forwarding route | one-line", match=r"(?i)^\s*show\s+ip\s+forwarding\s+route(?:\s*\|\s*one-line)?\s*$", textfsm_command="show ip forwarding route", description="Global IPv4 FIB (large; disabled by default).", @@ -1047,7 +1051,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="ipv6_route", parser_id="ipv6_route", title="IPv6 Forwarding VRF", - command_template="show ipv6 forwarding route vrf ", + command_template="show ipv6 forwarding route vrf | one-line", match=r"(?i)^\s*show\s+ipv6\s+forwarding\s+route\s+vrf\s+(?P\S+)(?:\s*\|\s*one-line)?\s*$", textfsm_command="show ipv6 forwarding route", description="IPv6 FIB for one VRF.", @@ -1064,7 +1068,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="ipv6_route", parser_id="ipv6_route", title="IPv6 Forwarding Global", - command_template="show ipv6 forwarding route", + command_template="show ipv6 forwarding route | one-line", match=r"(?i)^\s*show\s+ipv6\s+forwarding\s+route(?:\s*\|\s*one-line)?\s*$", textfsm_command="show ipv6 forwarding route", description="Global IPv6 FIB (large; disabled by default).", @@ -1129,7 +1133,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_vrf", parser_id="config_vrf", title="Config VRF Intent", - command_template="show running-config vrf", + command_template="show running-config vrf | one-line", match=r"(?i)^\s*show\s+running-config\s+vrf(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config vrf", description="VRF RD/RT/AF intent from running-config vrf (no secrets).", @@ -1145,7 +1149,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_interface", parser_id="config_interface", title="Config Interface Intent", - command_template="show running-config if-intf", + command_template="show running-config if-intf | one-line", match=r"(?i)^\s*show\s+running-config\s+if-intf(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config if-intf", description=( @@ -1164,7 +1168,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_bgp_peer", parser_id="config_bgp_peer", title="Config BGP Peer Intent", - command_template="show running-config bgp", + command_template="show running-config bgp | one-line", match=r"(?i)^\s*show\s+running-config\s+bgp(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config bgp", description="BGP neighbor AF activate / remote-as intent (passwords skipped).", @@ -1180,7 +1184,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_l2vpn_pw", parser_id="config_l2vpn_pw", title="Config L2VPN PW Intent", - command_template="show running-config l2vpn", + command_template="show running-config l2vpn | one-line", match=r"(?i)^\s*show\s+running-config\s+l2vpn(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config l2vpn", description="VPWS/VPLS pseudo-wire peer/vcid intent.", @@ -1196,7 +1200,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_static_route", parser_id="config_static_route", title="Config Static Route (IPv4)", - command_template="show running-config static", + command_template="show running-config static | one-line", match=r"(?i)^\s*show\s+running-config\s+static(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config static", description="IPv4 static route intent from running-config static.", @@ -1212,7 +1216,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_static_route", parser_id="config_static_route", title="Config Static Route (IPv6)", - command_template="show running-config ipv6-static-route", + command_template="show running-config ipv6-static-route | one-line", match=r"(?i)^\s*show\s+running-config\s+ipv6-static-route(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config ipv6-static-route", description="IPv6 static route intent from running-config ipv6-static-route.", @@ -1228,7 +1232,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_ospf", parser_id="config_ospf", title="Config OSPF Intent (v2)", - command_template="show running-config ospfv2", + command_template="show running-config ospfv2 | one-line", match=r"(?i)^\s*show\s+running-config\s+ospfv2(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config ospfv2", description="OSPFv2 process/area/interface intent (auth secrets skipped).", @@ -1244,7 +1248,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_ospf", parser_id="config_ospf", title="Config OSPF Intent (v3)", - command_template="show running-config ospfv3", + command_template="show running-config ospfv3 | one-line", match=r"(?i)^\s*show\s+running-config\s+ospfv3(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config ospfv3", description="OSPFv3 process/area/interface intent.", @@ -1260,7 +1264,7 @@ def _zte_status_profiles() -> list[ParseProfile]: metric_id="config_isis", parser_id="config_isis", title="Config ISIS Intent", - command_template="show running-config isis", + command_template="show running-config isis | one-line", match=r"(?i)^\s*show\s+running-config\s+isis(?:\s*\|\s*one-line)?\s*$", textfsm_command="show running-config isis", description="IS-IS process/interface intent (auth secrets skipped).", diff --git a/netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm b/netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm index ca843b7..a7b88fa 100644 --- a/netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm +++ b/netx_api/cli_templates/zte/zte_zxros_show_opticalinfo_brief.textfsm @@ -1,6 +1,6 @@ -Value INTERFACE (\S+) -Value OPTIC_TYPE (\S+) -Value WAVELENGTH (\S+) +Value Filldown INTERFACE (\S+) +Value Filldown OPTIC_TYPE (\S+) +Value Filldown WAVELENGTH (\S+) Value RX_POWER (\S+) Value TX_POWER (\S+) Value STATUS (\S+) @@ -11,7 +11,9 @@ Start ^.* -> Start Entries + ^${INTERFACE}\s+offline\s*$$ -> Record ^${INTERFACE}\s+${OPTIC_TYPE}\s+${WAVELENGTH}\s+${RX_POWER}\s+${TX_POWER}\s+${STATUS}(?:\s+\S+)?\s*$$ -> Record + ^\s+${RX_POWER}\s+${TX_POWER}\s+${STATUS}(?:\s+\S+)?\s*$$ -> Record ^\s*$$ ^\S+\s*#\s*$$ -> End ^.*#\s*$$ -> End diff --git a/netx_api/topology_lldp.py b/netx_api/topology_lldp.py index ce0e66c..4aec09e 100644 --- a/netx_api/topology_lldp.py +++ b/netx_api/topology_lldp.py @@ -84,7 +84,7 @@ VENDOR_LLDP_PROFILES: dict[str, VendorLldpProfile] = { ), "zte": VendorLldpProfile( key="zte", - lldp_command="show lldp neighbor brief", + lldp_command="show lldp neighbor brief | one-line", notes="device_type zte_*; NetX custom TextFSM.", ), "juniper": VendorLldpProfile( diff --git a/tests/test_biz_state_profiles.py b/tests/test_biz_state_profiles.py index 7cef8d0..6a904e2 100644 --- a/tests/test_biz_state_profiles.py +++ b/tests/test_biz_state_profiles.py @@ -36,7 +36,7 @@ class BizStateProfileTests(unittest.TestCase): assert p is not None pairs = expand_from_bindings(profile=p) self.assertEqual(len(pairs), 1) - self.assertEqual(pairs[0][0], "show lldp neighbor brief") + self.assertEqual(pairs[0][0], "show lldp neighbor brief | one-line") def test_preview_custom_raw(self) -> None: prev = preview_task_item( diff --git a/tests/test_enrich_framework.py b/tests/test_enrich_framework.py index aa21aae..dc3da32 100644 --- a/tests/test_enrich_framework.py +++ b/tests/test_enrich_framework.py @@ -60,7 +60,7 @@ class AuxResolveTests(unittest.TestCase): def test_resolve_from_profile_id(self) -> None: ra = resolve_aux_command(AuxCommand(key="if_intf", profile_id="zte.config_interface")) - self.assertEqual(ra.command, "show running-config if-intf") + self.assertEqual(ra.command, "show running-config if-intf | one-line") self.assertEqual(ra.parser_id, "config_interface") def test_arp_profile_slim(self) -> None: diff --git a/tests/test_zte_config_intent.py b/tests/test_zte_config_intent.py index 7a3a220..a24acf5 100644 --- a/tests/test_zte_config_intent.py +++ b/tests/test_zte_config_intent.py @@ -48,7 +48,9 @@ interface gei-0/0/0/1 description uplink ip vrf forwarding CUST_A ip address 10.0.0.1 255.255.255.0 + ip address 10.0.0.2 255.255.255.0 secondary ipv6 address 2001:db8::1/64 + ipv6 address 2001:db8::2/64 secondary mtu 9000 $ interface gei-0/0/0/2 @@ -116,6 +118,9 @@ ip route vrf CUST_A 10.1.1.0 255.255.255.0 bvi12.100 10.1.1.1 ip route vrf CUST_A 10.2.2.0 255.255.255.0 gei-0/0/0/1.10 10.2.2.1 bfd enable ip route vrf CUST_B 10.3.3.0 255.255.255.0 null1 metric 255 ip route vrf CUST_A 10.4.4.0 255.255.255.0 smartgroup1.1 10.4.4.1 track LINK_A +ip route 10.5.5.0 255.255.255.0 xxvgei-0/2/0/1 +ip route 10.6.6.0 255.255.255.0 te_tunnel36 +ip route 10.7.7.0 255.255.255.0 10.7.7.1 xxvgei-0/2/0/2 name nh_then_if ! """ @@ -256,10 +261,20 @@ class ZteConfigIntentTests(unittest.TestCase): by = {r["interface"]: r for r in rows} self.assertEqual(by["gei-0/0/0/1"]["vrf"], "CUST_A") self.assertEqual(by["gei-0/0/0/1"]["admin"], "up") - self.assertIn("10.0.0.1", by["gei-0/0/0/1"]["ip_address"]) - self.assertIn("2001:db8::1/64", by["gei-0/0/0/1"]["ipv6_address"]) + self.assertEqual( + by["gei-0/0/0/1"]["ip_address"], + "10.0.0.1/255.255.255.0,10.0.0.2/255.255.255.0", + ) + self.assertEqual(by["gei-0/0/0/1"]["secondary_tag"], "M,S") + self.assertEqual( + by["gei-0/0/0/1"]["ipv6_address"], + "2001:db8::1/64,2001:db8::2/64", + ) + self.assertEqual(by["gei-0/0/0/1"]["ipv6_secondary_tag"], "M,S") self.assertEqual(by["gei-0/0/0/1"]["mtu"], "9000") self.assertEqual(by["gei-0/0/0/2"]["admin"], "down") + self.assertEqual(by["gei-0/0/0/2"]["secondary_tag"], "M") + self.assertEqual(by["gei-0/0/0/2"]["ipv6_secondary_tag"], "") def test_config_bgp_peer_skips_password(self) -> None: rows = normalize_config_bgp_peer( @@ -355,6 +370,13 @@ $ self.assertEqual(by[("CUST_A", "10.2.2.0", "gei-0/0/0/1.10", "10.2.2.1")]["bfd"], "enable") self.assertEqual(by[("CUST_B", "10.3.3.0", "null1", "")]["metric"], "255") self.assertEqual(by[("CUST_A", "10.4.4.0", "smartgroup1.1", "10.4.4.1")]["track"], "LINK_A") + # Interface-only (no next-hop address) β€” including xxvgei / te_tunnel. + self.assertEqual(by[("", "10.5.5.0", "xxvgei-0/2/0/1", "")]["mask"], "255.255.255.0") + self.assertEqual(by[("", "10.6.6.0", "te_tunnel36", "")]["interface"], "te_tunnel36") + # next-hop then interface + self.assertEqual( + by[("", "10.7.7.0", "xxvgei-0/2/0/2", "10.7.7.1")]["route_name"], "nh_then_if" + ) def test_config_static_route_v6(self) -> None: rows = normalize_config_static_route( diff --git a/tests/test_zte_config_real_dump.py b/tests/test_zte_config_real_dump.py index a23f569..c8824d0 100644 --- a/tests/test_zte_config_real_dump.py +++ b/tests/test_zte_config_real_dump.py @@ -107,12 +107,17 @@ class ZteConfigRealDumpTests(unittest.TestCase): if m: block_vrf = m.group(1) continue - m = re.match(r"(?i)^\s*ip\s+address\s+(\S+)(?:\s+(\S+))?\s*$", line) + m = re.match( + r"(?i)^\s*ip\s+address\s+(\S+)(?:\s+(\S+))?(?:\s+secondary)?\s*$", + line, + ) if m: a, mask = m.group(1), m.group(2) or "" + if mask.lower() == "secondary": + mask = "" block_ips.append(f"{a}/{mask}" if mask else a) continue - m = re.match(r"(?i)^\s*ipv6\s+address\s+(\S+)\s*$", line) + m = re.match(r"(?i)^\s*ipv6\s+address\s+(\S+)(?:\s+secondary)?\s*$", line) if m: block_ip6s.append(m.group(1)) continue diff --git a/tests/test_zte_extended_parsers.py b/tests/test_zte_extended_parsers.py index 1430ff1..cb1eb77 100644 --- a/tests/test_zte_extended_parsers.py +++ b/tests/test_zte_extended_parsers.py @@ -66,7 +66,12 @@ smartgroup1.5 5 110 1000 P Init 0.0.0.0 10.0.2.1 _OPTICAL_SAMPLE = """ Interface Type Wavelength RxPower(dBm) TxPower(dBm) Status Intensity(Rx) gei-0/0/0/1 1G-10km-SFP 1310nm -6.0/[-20.0,-3.0] -5.9/[-9.0,-3.0] Normal Normal -cgei-0/1/0/1 100G-LR4 1310nm -2.1/[-8.0,2.0] -1.5/[-4.0,2.0] Normal Normal +cgei-0/1/0/1 100G-10km-QSFP28 1310nm 1.5/[-10.6,4.5] 0.9/[-4.3,5.5] Normal Normal + 2.1/[-10.6,4.5] 1.5/[-4.3,5.5] Normal Normal + 1.6/[-10.6,4.5] 2.1/[-4.3,5.5] Normal Normal + 0.9/[-10.6,4.5] 2.2/[-4.3,5.5] Normal Normal + 7.5 7.7 +gei-0/0/0/11 offline """ _BGP_ROUTE_IN = """ @@ -162,9 +167,24 @@ class ZteExtendedParserTests(unittest.TestCase): states = {r["state"] for r in vrrp} self.assertIn("Master", states) - opt = normalize_optical_brief(raw_text=_OPTICAL_SAMPLE, command="show opticalinfo brief") - self.assertEqual(len(opt), 2) - self.assertEqual(opt[0]["status"], "Normal") + opt = normalize_optical_brief( + raw_text=_OPTICAL_SAMPLE, + vendor="zte", + device_type="zte_zxros", + command="show opticalinfo brief", + ) + self.assertEqual(len(opt), 3) + by = {r["interface"]: r for r in opt} + self.assertEqual(by["gei-0/0/0/1"]["rx_power"], "-6.0") + self.assertEqual(by["gei-0/0/0/1"]["rx_threshold"], "[-20.0,-3.0]") + self.assertEqual(by["gei-0/0/0/1"]["tx_power"], "-5.9") + self.assertEqual(by["gei-0/0/0/1"]["tx_threshold"], "[-9.0,-3.0]") + self.assertEqual(by["gei-0/0/0/1"]["status"], "Normal") + self.assertEqual(by["cgei-0/1/0/1"]["rx_power"], "1.5,2.1,1.6,0.9") + self.assertEqual(by["cgei-0/1/0/1"]["tx_power"], "0.9,1.5,2.1,2.2") + self.assertEqual(by["cgei-0/1/0/1"]["rx_threshold"], "[-10.6,4.5]") + self.assertEqual(by["cgei-0/1/0/1"]["tx_threshold"], "[-4.3,5.5]") + self.assertEqual(by["gei-0/0/0/11"]["optic_type"], "offline") def test_bgp_route_and_aux_render(self) -> None: routes = normalize_bgp_route( @@ -193,7 +213,7 @@ class ZteExtendedParserTests(unittest.TestCase): AuxCommand(key="bgp_summary", profile_id="zte.bgp_vpnv4_vrf_summary"), params={"vrf": "CUST_A"}, ) - self.assertEqual(ra.command, "show bgp vpnv4 unicast vrf CUST_A summary") + self.assertEqual(ra.command, "show bgp vpnv4 unicast vrf CUST_A summary | one-line") def test_ip_ipv6_route_and_pw(self) -> None: ip = normalize_ip_route( diff --git a/tests/test_zte_status_parsers.py b/tests/test_zte_status_parsers.py index e396f1f..21518f5 100644 --- a/tests/test_zte_status_parsers.py +++ b/tests/test_zte_status_parsers.py @@ -169,6 +169,25 @@ smartgroup11 MDN-SGI-P-ZM8SP UP L2 24 PPP addrs = {r["address"] for r in rows} self.assertTrue(any("fe80::" in a for a in addrs)) + def test_nd6_skips_clock_banner(self) -> None: + text = """ +14:06:53 Beijing Sun Sep 20 2026 +S-static NP-near packet RP-remote packet B-bgp HB-hotback D-dhcp +Total Cache Number Is:5398 +Only Current Valid Items Are Shown Below: +Address Link-Address Age Status Interface Type +10:460::2 (incomplete) 1s Incomplete smartgroup2345.1120 NP +10:794::2 (incomplete) 1s Incomplete smartgroup2345.1940 NP +fe80::1 aaaa.bbbb.cccc 03:22:07 Reachable gei-0/0/0/1 S +""" + rows = normalize_nd6_cache(raw_text=text) + self.assertEqual(len(rows), 3) + self.assertNotIn("14:06:53", {r["address"] for r in rows}) + self.assertEqual(rows[0]["address"], "10:460::2") + self.assertEqual(rows[0]["status"], "Incomplete") + self.assertEqual(rows[0]["type"], "NP") + self.assertEqual(rows[2]["age"], "03:22:07") + def test_bgp_peers(self) -> None: v4 = _section(self.log, "show bgp vpnv4 unicast summary", ("show bgp ipv4",)) rows = normalize_bgp_peer(raw_text=v4, command="show bgp vpnv4 unicast summary")