diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index c547f57..9f03dbf 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -275,7 +275,7 @@ async def nb_problem(log_status, msg: str): def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ - Optional[str], Optional[int], Optional[int], Optional[str], Optional[Any], Optional[Any], Optional[Any], Optional[Any], Optional[Any] + Optional[str], Optional[int], Optional[int], Optional[Any], Optional[Any], Optional[Any], Optional[Any], Optional[Any] ]: """ Resolve MAC -> (hostname, iface_id, device_id, action_next, action_last, action_next_timestamp, action_state, sot_ts) @@ -284,7 +284,7 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ """ mac_norm = normalize_mac(mac) if not mac_norm: - return None, None, None, None, None, None, None, None, None + return None, None, None, None, None, None, None, None # Step 0: short TTL cache (avoid repeated NetBox GETs for chatty devices) if NB_LOOKUP_CACHE_TTL > 0: @@ -314,12 +314,12 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ data, code = http_get_json(f"{base}/api/dcim/mac-addresses/", params={"mac_address": mac_norm, "limit": "2"}, headers=h) if code != 200 or not data: asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac query http={code} mac={mac_norm}")) - return None, None, None, None, None, None, None, None, None + return None, None, None, None, None, None, None, None results = (data or {}).get("results") or [] if not results: asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac not found mac={mac_norm}")) - return None, None, None, None, None, None, None, None, None + return None, None, None, None, None, None, None, None rec = results[0] aot = (rec.get("assigned_object_type") or "").strip() @@ -330,7 +330,7 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ if not aot or aoid is None: asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac unassigned mac={mac_norm}")) - return None, None, None, None, None, None, None, None, None + return None, None, None, None, None, None, None, None if aot != "dcim.interface": asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm} iface_id={aoid}")) return None, aoid, None, None, None, None, None, None, None @@ -356,7 +356,7 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ device, code3 = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", headers=h) if code3 != 200 or not device: # treat as "no extra info" - return host, aoid, dev_id, None, None, None, None, None, None + return host, aoid, dev_id, None, None, None, None, None cf = device.get("custom_fields") or {} action_next = cf.get("action_next")