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 <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-29 00:24:06 +08:00
parent b36174bfa4
commit 3adf8d8b75
4 changed files with 82 additions and 18 deletions

View file

@ -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",
}

View file

@ -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}

View file

@ -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

View file

@ -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: