From 42d5f9c5a74f74a55c3f089349885ed21478293d Mon Sep 17 00:00:00 2001 From: oliver Date: Fri, 16 Oct 2026 22:14:34 +0800 Subject: [PATCH] fix(ntc_parse): strip prompt/command echo for ntc-templates 9.x strict templates ntc-templates 9.x ends template states with '^. -> Error', so an echoed 'R2#show lldp neighbors' line or trailing prompt aborts the whole TextFSM parse (Cisco LLDP brief returned 0 rows). Strip edge echo/prompt lines before parsing in parse_cli/apply_rule; covers Cisco/H3C/Juniper/AOS community templates too. Co-Authored-By: Claude Opus 5.5 --- netx_api/ntc_parse.py | 28 +++++++++++++++++++++++++++- tests/test_ntc_parse.py | 33 ++++++++++++++++++++++++++++++++- 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/netx_api/ntc_parse.py b/netx_api/ntc_parse.py index 6aa8f55..2352e38 100644 --- a/netx_api/ntc_parse.py +++ b/netx_api/ntc_parse.py @@ -95,6 +95,31 @@ def resolve_cli_platform(*, vendor: str = "", device_type: str = "", vendor_key: return "" +# Device prompt: , [~HUAWEI], R2#, user@host>, RP/0/RSP0/CPU0:xr# ... +_PROMPT = r"(?:<[^<>\s]+>|\[[^\[\]\s]+\]|[\w.\-@:/()~]+[#>])" +_PROMPT_ECHO_RE = re.compile(rf"^\s*{_PROMPT}\s*\S.*$") +_PROMPT_ONLY_RE = re.compile(rf"^\s*{_PROMPT}\s*$") + + +def strip_cli_echo(text: str) -> str: + """Drop leading ``prompt+command`` echo and trailing bare prompt lines. + + Newer ntc-templates (9.x) end Start/body states with ``^. -> Error``, so an + echoed ``R2#show lldp neighbors`` or a trailing ``R2#`` aborts the whole parse. + Only the edges are touched; table body lines are left as-is. + """ + lines = str(text or "").splitlines() + start = 0 + while start < len(lines) and not lines[start].strip(): + start += 1 + if start < len(lines) and _PROMPT_ECHO_RE.match(lines[start]): + start += 1 + end = len(lines) + while end > start and (not lines[end - 1].strip() or _PROMPT_ONLY_RE.match(lines[end - 1])): + end -= 1 + return "\n".join(lines[start:end]) + "\n" + + def row_get(row: dict[str, Any], *names: str) -> str: if not row: return "" @@ -286,6 +311,7 @@ def apply_rule(*, platform: str, rule_key: str, text: str, command: str = "") -> wrap_cmd = str(command or "").strip() or key.replace("_", " ") if plat: raw = apply_cli_wrap(raw, platform=plat, command=wrap_cmd) + raw = strip_cli_echo(raw) entries = _find_entries_for_stem(key) if plat: @@ -350,7 +376,7 @@ def parse_cli( """Parse CLI text: wrap-join → custom templates → community ntc-templates.""" plat = str(platform or "").strip() cmd = str(command or "").strip() - raw = apply_cli_wrap(str(text or ""), platform=plat, command=cmd) + raw = strip_cli_echo(apply_cli_wrap(str(text or ""), platform=plat, command=cmd)) if not plat or not cmd or not raw.strip(): return [] diff --git a/tests/test_ntc_parse.py b/tests/test_ntc_parse.py index a184016..a3c601e 100644 --- a/tests/test_ntc_parse.py +++ b/tests/test_ntc_parse.py @@ -4,7 +4,7 @@ from __future__ import annotations import unittest -from netx_api.ntc_parse import parse_cli, resolve_cli_platform +from netx_api.ntc_parse import parse_cli, resolve_cli_platform, strip_cli_echo from netx_api.port_traffic_parsers import parse_interface_brief, parse_interface_detail from netx_api import topology_lldp as lldp @@ -172,5 +172,36 @@ ge-0/0/1.0 ge-0/0/1 00:11:22:33:44:55 ge-0/0/0.0 r1 self.assertEqual(jhits[0].local_port, "ge-0/0/1.0") +class CliEchoTests(unittest.TestCase): + """ntc-templates 9.x ends states with ``^. -> Error``; echo/prompt must be stripped.""" + + def test_strip_edges_only(self) -> None: + text = "\nR2#show lldp neighbors\nDevice ID Local Intf\nr1 Gi0/1\nR2#\n" + self.assertEqual(strip_cli_echo(text), "Device ID Local Intf\nr1 Gi0/1\n") + for prompt in ("display lldp neighbor", "[~HUAWEI]display lldp neighbor", + "user@r2> show lldp neighbors", "ZXR10#show lldp neighbor brief"): + self.assertEqual(strip_cli_echo(prompt + "\nbody\n"), "body\n") + self.assertEqual(strip_cli_echo("Device ID Local Intf\n"), "Device ID Local Intf\n") + + def test_lldp_with_echo_and_prompt(self) -> None: + h3c = """\ +display lldp neighbor-information list +Local Interface Chassis ID Port ID System Name +GE1/0/1 0000-5e00-0101 GE1/0/2 SW-A + +""" + self.assertEqual(len(lldp.parse_h3c_lldp(h3c)), 1) + junos = """\ +user@r2> show lldp neighbors +Local Interface Parent Interface Chassis Id Port info System Name +ge-0/0/1.0 ge-0/0/1 00:11:22:33:44:55 ge-0/0/0.0 r1 + +user@r2> +""" + self.assertEqual(len(lldp.parse_juniper_lldp(junos)), 1) + cisco = "R2#show lldp neighbors detail\n" + CISCO_LLDP_DETAIL + "R2#\n" + self.assertEqual(len(lldp.parse_cisco_lldp(cisco)), 1) + + if __name__ == "__main__": unittest.main()