Fix ZTE LLDP Scope table parse and skip discover without vendor.

Match ZXROS Scope/Holdtime columns, disable Netmiko cmd_verify on show commands, and do not SSH LLDP when vendor/device_type cannot be resolved.

Co-authored-by: Cursor <cursoragent@cursor.com>

netx_api/cli_templates/zte/zte_zxros_show_lldp_neighbor_brief.textfsm

@@ -1,15 +1,32 @@
 Value LOCAL_INTERFACE (\S+)
+Value SCOPE (\S+)
 Value CHASSIS_ID (\S+)
 Value NEIGHBOR_PORT_ID (\S+)
-Value NEIGHBOR_NAME (\S+.*)
+Value HOLDTIME (\d+)
+Value NEIGHBOR_NAME (\S.*?)
 
 Start
-  ^Local\s+Interface\s+Chassis\s+ID -> Neighbors
+  ^Local\s+Interface\s+Scope\s+Chassis\s+ID -> Neighbors
+  ^Local\s+Interface\s+Chassis\s+ID -> NeighborsSimple
+  ^Total\s+neighbors
+  ^-+
   ^\s*$$
   ^.* -> Start
 
 Neighbors
-  ^${LOCAL_INTERFACE}\s+${CHASSIS_ID}\s+${NEIGHBOR_PORT_ID}\s+${NEIGHBOR_NAME}\s*$$ -> Record
+  ^${LOCAL_INTERFACE}\s+${SCOPE}\s+${CHASSIS_ID}\s+${NEIGHBOR_PORT_ID}\s+${HOLDTIME}\s+${NEIGHBOR_NAME}\s*$$ -> Record
+  ^-+
+  ^Total\s+neighbors
   ^\s*$$
-  ^Local\s+Interface -> Neighbors
+  ^\S+\s*#\s*$$ -> End
-  ^.* -> Start
+  ^.*#\s*$$ -> End
+  ^.* -> Neighbors
+
+NeighborsSimple
+  ^${LOCAL_INTERFACE}\s+${CHASSIS_ID}\s+${NEIGHBOR_PORT_ID}\s+${NEIGHBOR_NAME}\s*$$ -> Record
+  ^-+
+  ^Total\s+neighbors
+  ^\s*$$
+  ^\S+\s*#\s*$$ -> End
+  ^.*#\s*$$ -> End
+  ^.* -> NeighborsSimple

netx_api/ne_netmiko.py

@@ -38,8 +38,22 @@ def send_show_command(conn: Any, command: str, *, read_timeout: int = 120) -> st
 
     Do not use ``send_command_timing`` for Cisco config collection: long idle during
     ``Building configuration...`` is treated as end-of-output and truncates the config.
+
+    ``cmd_verify=False``: Netmiko's default echo check often raises
+    ``Pattern not detected: 'show\\ lldp\\ ...'`` on IOSv / hop / slow echo paths.
     """
     cmd = str(command or "").strip()
     if not cmd:
         return ""
-    return str(conn.send_command(command_string=cmd, read_timeout=read_timeout) or "")
+    try:
+        return str(
+            conn.send_command(
+                command_string=cmd,
+                read_timeout=read_timeout,
+                cmd_verify=False,
+            )
+            or ""
+        )
+    except TypeError:
+        # Older Netmiko without cmd_verify kwarg.
+        return str(conn.send_command(command_string=cmd, read_timeout=read_timeout) or "")

netx_api/topology_discover_scan.py

@@ -23,7 +23,13 @@ from .topology_fabric import (
     ensure_fabric_node_for_ume,
     upsert_fabric_edge,
 )
-from .topology_lldp import NeighborHit, parse_neighbor_output, parser_meta, pick_neighbor_command
+from .topology_lldp import (
+    NeighborHit,
+    can_discover_lldp,
+    parse_neighbor_output,
+    parser_meta,
+    pick_neighbor_command,
+)
 
 
 def _discover_one_target(
@@ -70,10 +76,21 @@ def _discover_one_target(
         # Release unique-index locks before slow SSH.
         db.commit()
 
-        cmd, _proto = pick_neighbor_command(
+        vendor = target.get("vendor") or ""
-            vendor=target.get("vendor") or "",
+        device_type = target.get("device_type") or ""
-            device_type=target.get("device_type") or "",
+        pkey, is_stub = parser_meta(vendor=vendor, device_type=device_type)
-        )
+        if not can_discover_lldp(vendor=vendor, device_type=device_type):
+            return {
+                **base,
+                "ok": False,
+                "command": "",
+                "parser_key": pkey,
+                "parser_stub": True,
+                "error": "vendor_or_device_type_required",
+                "raw_preview": "",
+            }
+
+        cmd, _proto = pick_neighbor_command(vendor=vendor, device_type=device_type)
         exec_kwargs: dict[str, Any] = {"read_timeout_sec": 60}
         if target.get("ume_ne_id") and not db.get(ManagedNE, target["ne_id"]):
             exec_kwargs["ume_ne_id"] = target["ume_ne_id"]
@@ -102,14 +119,12 @@ def _discover_one_target(
             }
 
         raw = str(exec_out.get("output") or "")
-        pkey, is_stub = parser_meta(
-            vendor=target.get("vendor") or "", device_type=target.get("device_type") or ""
-        )
         hits = parse_neighbor_output(
             raw,
             protocol="lldp",
-            vendor=target.get("vendor") or "",
+            vendor=vendor,
-            device_type=target.get("device_type") or "",
+            device_type=device_type,
+            command=cmd,
         )
         stub_flag = bool(is_stub and raw.strip() and not hits)
 

netx_api/topology_lldp.py

@@ -159,6 +159,11 @@ def pick_neighbor_command(
     return lldp_command_for_vendor(vendor, device_type), "lldp"
 
 
+def can_discover_lldp(*, vendor: str = "", device_type: str = "") -> bool:
+    """False when inventory has no usable vendor/device_type (do not SSH a guess command)."""
+    return resolve_vendor_key(vendor, device_type) != "generic"
+
+
 def parser_meta(*, vendor: str = "", device_type: str = "") -> tuple[str, bool]:
     """Return (parser_key, is_stub)."""
     key = resolve_vendor_key(vendor, device_type)
@@ -202,6 +207,17 @@ def _map_lldp_rows(rows: list[dict[str, Any]]) -> list[NeighborHit]:
     return hits
 
 
+def _strip_collect_wrapper(text: str) -> str:
+    """Drop SecureCRT-style headers injected by ``_collect_on_device``."""
+    lines: list[str] = []
+    for ln in str(text or "").splitlines():
+        s = ln.strip()
+        if s.startswith(">>>") and '"String"' in s:
+            continue
+        lines.append(ln)
+    return "\n".join(lines)
+
+
 def _parse_lldp_via_ntc(
     text: str,
     *,
@@ -226,15 +242,18 @@ def parse_neighbor_output(
     command: str = "",
 ) -> list[NeighborHit]:
     """Parse neighbor CLI via TextFSM only (custom then community)."""
-    raw = str(text or "")
+    raw = _strip_collect_wrapper(text)
     if not raw.strip():
         return []
     _ = protocol  # CDP discovery removed; always parse as LLDP
+    if not can_discover_lldp(vendor=vendor, device_type=device_type):
+        return []
+    cmd = (command or "").strip() or lldp_command_for_vendor(vendor, device_type)
     return _parse_lldp_via_ntc(
         raw,
         vendor=vendor,
         device_type=device_type,
-        command=command or lldp_command_for_vendor(vendor, device_type),
+        command=cmd,
     )
 
 

tests/test_ntc_parse.py

@@ -14,6 +14,16 @@ Local Interface      Chassis ID         Port ID              System Name
 gei-0/1/0/1          0011.2233.4455     gei-0/1/0/2          R1
 """
 
+ZTE_LLDP_SCOPE = """\
+Total neighbors: 12
+Local Interface   Scope  Chassis ID      Port ID           Holdtime  System Name
+-----------------------------------------------------------------------------------------
+xgei-1/1/0/1     NB     026e.8219.bc57  xgei-1/1/0/26     100       CSR1_6120HSC
+xgei-1/1/0/3     NB     00d0.0000.081f  gei-1/2/0/3       113       OLT/CPE_6180H
+xxvgei-1/1/0/16  NB     026e.8219.bc57  xxvgei-1/1/0/32   100       CSR1_6120HSC
+cgei-1/1/0/34    NB     0022.9354.6e60  cgei-0/3/0/34     110       AG5
+"""
+
 ZTE_BRIEF = """\
 Interface               Attribute  Mode         BW    Admin Phy   Prot  Description
 xgei-1/1/0/1            optical    Duplex/full  1G    up    up    up    C2930L100-EQ2
@@ -67,6 +77,19 @@ class CustomTemplateTests(unittest.TestCase):
         self.assertEqual(rows[0].get("local_interface"), "gei-0/1/0/1")
         self.assertEqual(rows[0].get("neighbor_name"), "R1")
 
+    def test_zte_lldp_scope_holdtime(self) -> None:
+        rows = parse_cli(
+            platform="zte_zxros",
+            command="show lldp neighbor brief",
+            text=ZTE_LLDP_SCOPE,
+        )
+        self.assertEqual(len(rows), 4)
+        by_local = {r.get("local_interface"): r for r in rows}
+        self.assertEqual(by_local["xgei-1/1/0/1"].get("neighbor_name"), "CSR1_6120HSC")
+        self.assertEqual(by_local["xgei-1/1/0/3"].get("neighbor_name"), "OLT/CPE_6180H")
+        self.assertEqual(by_local["xgei-1/1/0/3"].get("neighbor_port_id"), "gei-1/2/0/3")
+        self.assertEqual(by_local["cgei-1/1/0/34"].get("scope"), "NB")
+
     def test_zte_brief_custom(self) -> None:
         rows = parse_cli(
             platform="zte_zxros",
@@ -94,6 +117,13 @@ class WiredParserTests(unittest.TestCase):
         self.assertEqual(hits[0].remote_name, "R1")
         self.assertEqual(hits[0].local_port, "gei-0/1/0/1")
 
+    def test_zte_lldp_scope_parser(self) -> None:
+        hits = lldp.parse_zte_lldp(ZTE_LLDP_SCOPE)
+        self.assertEqual(len(hits), 4)
+        by_local = {h.local_port: h for h in hits}
+        self.assertEqual(by_local["xgei-1/1/0/3"].remote_name, "OLT/CPE_6180H")
+        self.assertEqual(by_local["xxvgei-1/1/0/16"].remote_port, "xxvgei-1/1/0/32")
+
     def test_cisco_lldp_community(self) -> None:
         hits = lldp.parse_cisco_lldp(CISCO_LLDP_DETAIL)
         self.assertEqual(len(hits), 1)

tests/test_topology.py

@@ -74,6 +74,15 @@ Local Interface      Chassis ID         Port ID              System Name
 gei-0/1/0/1          0011.2233.4455     gei-0/1/0/2          R1
 """
 
+ZTE_LLDP_SCOPE = """
+Total neighbors: 12
+Local Interface   Scope  Chassis ID      Port ID           Holdtime  System Name
+-----------------------------------------------------------------------------------------
+xgei-1/1/0/1     NB     026e.8219.bc57  xgei-1/1/0/26     100       CSR1_6120HSC
+xgei-1/1/0/3     NB     00d0.0000.081f  gei-1/2/0/3       113       OLT/CPE_6180H
+cgei-1/1/0/34    NB     0022.9354.6e60  cgei-0/3/0/34     110       AG5
+"""
+
 
 class LldpParserTests(unittest.TestCase):
     def test_cisco_brief(self) -> None:
@@ -89,6 +98,18 @@ class LldpParserTests(unittest.TestCase):
         self.assertEqual(hits[0].remote_name.lower(), "r1")
         self.assertEqual(hits[0].remote_ip, "192.168.0.1")
 
+    def test_no_vendor_skips_discover(self) -> None:
+        self.assertFalse(lldp.can_discover_lldp(vendor="", device_type=""))
+        self.assertFalse(lldp.can_discover_lldp(vendor="", device_type="generic"))
+        self.assertTrue(lldp.can_discover_lldp(vendor="Cisco", device_type="cisco_ios"))
+        hits = lldp.parse_neighbor_output(
+            CISCO_LLDP_BRIEF,
+            vendor="",
+            device_type="generic",
+            command="show lldp neighbors",
+        )
+        self.assertEqual(hits, [])
+
     def test_huawei(self) -> None:
         hits = lldp.parse_huawei_lldp(HUAWEI_LLDP)
         self.assertGreaterEqual(len(hits), 1)
@@ -100,6 +121,13 @@ class LldpParserTests(unittest.TestCase):
         self.assertEqual(len(hits), 1)
         self.assertEqual(hits[0].remote_name, "R1")
 
+    def test_zte_scope_holdtime(self) -> None:
+        hits = lldp.parse_zte_lldp(ZTE_LLDP_SCOPE)
+        self.assertEqual(len(hits), 3)
+        self.assertEqual(hits[0].remote_name, "CSR1_6120HSC")
+        self.assertEqual(hits[1].remote_name, "OLT/CPE_6180H")
+        self.assertEqual(hits[1].remote_port, "gei-1/2/0/3")
+
     def test_pick_command_lldp_only(self) -> None:
         cmd, tag = lldp.pick_neighbor_command(protocol="cdp", vendor="Cisco", device_type="cisco_ios")
         self.assertEqual(tag, "lldp")

web/src/i18n/en.ts

@@ -1395,6 +1395,7 @@ const en = {
     discoverNeFail: "Collect failed",
     discoverUnmatched: "unmatched {{count}}",
     discoverUnmatchedMore: "{{count}} more not listed",
+    discoverVendorRequired: "Vendor/device type missing; skipped LLDP command",
     discoverParserStub: "Parser \"{{parser}}\" is still a stub; output was collected but neighbors were not parsed",
     discoverNoTargets:
       "No scannable NEs on the map. Add managed NEs, or UME NEs with a CLI connect profile.",

web/src/i18n/zh.ts

@@ -1389,6 +1389,7 @@ const zh = {
     discoverNeFail: "采集失败",
     discoverUnmatched: "未匹配 {{count}}",
     discoverUnmatchedMore: "另有 {{count}} 条未列出",
+    discoverVendorRequired: "未设置厂商/设备类型，已跳过下发 LLDP 命令",
     discoverParserStub: "厂商解析器「{{parser}}」仍为 stub，已采集到输出但未能解析邻居",
     discoverNoTargets: "画布上没有可扫描网元。请添加托管网元，或添加已配置 CLI 凭据的 UME 网元。",
     discoverFilter: "结果筛选",

web/src/pages/network/LldpLinksPage.tsx

@@ -685,7 +685,9 @@ export function LldpLinksPage() {
 
             {!itemDetail.ok ? (
               <p className="topo-discover__error">
-                {itemDetail.error || t("topology.discoverNeFail")}
+                {itemDetail.error === "vendor_or_device_type_required"
+                  ? t("topology.discoverVendorRequired")
+                  : itemDetail.error || t("topology.discoverNeFail")}
               </p>
             ) : null}
             {itemDetail.parser_stub ? (