From 3adf8d8b75b3145769734d0536d29f1e1af52b3f Mon Sep 17 00:00:00 2001 From: oliver Date: Sat, 29 Aug 2026 00:24:06 +0800 Subject: [PATCH] Prefer vendor label for LLDP command when it conflicts with device_type. Huawei NEs still tagged zte_zxros were getting show lldp neighbor brief and timing out; use display lldp neighbor and align UME inventory vendor before the jump. Co-authored-by: Cursor --- netx_api/topology_discover_common.py | 22 ++++++++++- netx_api/topology_discover_scan.py | 8 +++- netx_api/topology_lldp.py | 58 +++++++++++++++++++++------- tests/test_topology.py | 12 ++++++ 4 files changed, 82 insertions(+), 18 deletions(-) diff --git a/netx_api/topology_discover_common.py b/netx_api/topology_discover_common.py index cf6b82f..9558318 100644 --- a/netx_api/topology_discover_common.py +++ b/netx_api/topology_discover_common.py @@ -175,17 +175,35 @@ def _ume_target_dict(db: Session, uid: str, default_profile: Any) -> dict[str, s ume = db.query(UmeInventoryNE).filter(UmeInventoryNE.ne_id == uid).one_or_none() if ume is None: return None + inv_vendor = str(ume.vendor or "").strip() if default_profile is not None: dtype, vendor = infer_device_type_vendor(str(ume.ne_type or ""), default_profile) else: - dtype, vendor = "zte_zxros", (ume.vendor or "ZTE") + dtype, vendor = "zte_zxros", (inv_vendor or "ZTE") + # Prefer UME inventory vendor over CLI-profile defaults so Huawei is not + # treated as ZTE (show lldp neighbor brief) when ne_type rules miss. + if inv_vendor: + vendor = inv_vendor + from .topology_lldp import _key_from_device_type, _key_from_vendor_label + + label_key = _key_from_vendor_label(inv_vendor) + dtype_key = _key_from_device_type(str(dtype or "")) + if label_key and label_key != dtype_key: + dtype = { + "huawei": "huawei", + "cisco": "cisco_ios", + "zte": "zte_zxros", + "h3c": "hp_comware", + "juniper": "juniper_junos", + "nokia": "nokia_sros", + }.get(label_key, dtype) name = (ume.host_name or ume.ne_name or ume.user_label or ume.ip_address or uid).strip() return { "ne_id": uid, "ume_ne_id": uid, "ne_name": name, "ne_ip": ume.ip_address or "", - "vendor": vendor or (ume.vendor or "ZTE"), + "vendor": vendor or (inv_vendor or "ZTE"), "device_type": dtype or "zte_zxros", } diff --git a/netx_api/topology_discover_scan.py b/netx_api/topology_discover_scan.py index 91926ab..8d8d624 100644 --- a/netx_api/topology_discover_scan.py +++ b/netx_api/topology_discover_scan.py @@ -92,7 +92,6 @@ def _discover_one_target( "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"] @@ -108,9 +107,14 @@ def _discover_one_target( return { **base, "ok": False, - "command": cmd, + "command": "", "error": str(exc.detail or "resolve_failed")[:500], } + # Prefer resolved session identity (overrides / profile) for the LLDP verb. + vendor = str(creds.get("vendor") or vendor or "") + device_type = str(creds.get("device_type") or device_type or "") + pkey, is_stub = parser_meta(vendor=vendor, device_type=device_type) + cmd, _proto = pick_neighbor_command(vendor=vendor, device_type=device_type) skip = cli_creds_skip_reason(creds, interactive=False) if skip: return {**base, "ok": False, "command": cmd, "error": skip} diff --git a/netx_api/topology_lldp.py b/netx_api/topology_lldp.py index 5db008b..ce0e66c 100644 --- a/netx_api/topology_lldp.py +++ b/netx_api/topology_lldp.py @@ -52,8 +52,11 @@ _DEVICE_TYPE_PREFIX_RULES: tuple[tuple[str, str], ...] = ( _VENDOR_LABEL_TO_KEY: dict[str, str] = { "cisco": "cisco", "huawei": "huawei", + "华为": "huawei", + "hw": "huawei", "h3c": "h3c", "zte": "zte", + "中兴": "zte", "juniper": "juniper", "nokia": "nokia", "ericsson": "ericsson", @@ -110,23 +113,49 @@ VENDOR_LLDP_PROFILES: dict[str, VendorLldpProfile] = { STUB_PARSER_KEYS = frozenset({"nokia", "ericsson", "generic"}) -def resolve_vendor_key(vendor: str = "", device_type: str = "") -> str: - """Map inventory device_type (preferred) or vendor label -> profile key.""" +def _key_from_device_type(device_type: str) -> str: dtype = str(device_type or "").strip().lower() - if dtype: - for prefix, key in _DEVICE_TYPE_PREFIX_RULES: - if dtype == prefix.rstrip("_") or dtype.startswith(prefix): - return key - if dtype in VENDOR_LLDP_PROFILES and dtype != "generic": - return dtype + if not dtype: + return "" + for prefix, key in _DEVICE_TYPE_PREFIX_RULES: + if dtype == prefix.rstrip("_") or dtype.startswith(prefix): + return key + if dtype in VENDOR_LLDP_PROFILES and dtype != "generic": + return dtype + return "" + +def _key_from_vendor_label(vendor: str) -> str: label = str(vendor or "").strip().lower() - if label: - if label in _VENDOR_LABEL_TO_KEY: - return _VENDOR_LABEL_TO_KEY[label] - for token, key in _VENDOR_LABEL_TO_KEY.items(): - if label == token or label.startswith(f"{token} ") or label.startswith(f"{token}-"): - return key + if not label: + return "" + if label in _VENDOR_LABEL_TO_KEY: + return _VENDOR_LABEL_TO_KEY[label] + for token, key in _VENDOR_LABEL_TO_KEY.items(): + if label == token or label.startswith(f"{token} ") or label.startswith(f"{token}-"): + return key + # Substring for values like "Huawei Technologies" + if len(token) >= 3 and token in label: + return key + return "" + + +def resolve_vendor_key(vendor: str = "", device_type: str = "") -> str: + """Map inventory vendor / device_type -> LLDP profile key. + + Prefer Netmiko ``device_type`` when it is specific. When vendor label and + device_type disagree (e.g. vendor=Huawei but dtype still zte_zxros from a + default profile), prefer the vendor label so Huawei gets ``display lldp`` + instead of ZTE's ``show lldp neighbor brief``. + """ + dtype_key = _key_from_device_type(device_type) + label_key = _key_from_vendor_label(vendor) + if label_key and dtype_key and label_key != dtype_key: + return label_key + if dtype_key: + return dtype_key + if label_key: + return label_key return "generic" @@ -138,6 +167,7 @@ def get_vendor_profile(vendor: str = "", device_type: str = "") -> VendorLldpPro def lldp_command_for_vendor(vendor: str = "", device_type: str = "") -> str: dtype = str(device_type or "").strip().lower() # AOS has a community template; SROS profile command differs. + # Keep AOS command even when vendor label says Nokia/Alcatel. if dtype.startswith("alcatel_aos"): return "show lldp remote-system" return get_vendor_profile(vendor, device_type).lldp_command diff --git a/tests/test_topology.py b/tests/test_topology.py index 690bb27..fe86736 100644 --- a/tests/test_topology.py +++ b/tests/test_topology.py @@ -157,6 +157,18 @@ class LldpParserTests(unittest.TestCase): key, stub = lldp.parser_meta(vendor="Ericsson", device_type="ericsson_ipos") self.assertTrue(stub) + def test_pick_command_vendor_wins_on_dtype_conflict(self) -> None: + # Stale/default zte_zxros must not force ZTE LLDP on a Huawei-labeled NE. + self.assertEqual( + lldp.pick_neighbor_command(vendor="Huawei", device_type="zte_zxros")[0], + "display lldp neighbor", + ) + self.assertEqual(lldp.resolve_vendor_key("Huawei", "zte_zxros"), "huawei") + self.assertEqual( + lldp.pick_neighbor_command(vendor="Huawei", device_type="huawei")[0], + "display lldp neighbor", + ) + class DeadlockHelperTests(unittest.TestCase): def test_is_deadlock_error_detects_pg_message(self) -> None: