hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-08 23:33:21 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2

fix(netx): derive ne_id from objectName ME{net_id}

Browse source 
When alarmkey does not contain a net_id, fall back to extracting ME{net_id} from objectName so alarms can join inventory correctly. Add regression coverage for the ME{} objectName pattern.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-05-13 17:02:46 +08:00
parent 626bb2b02e
commit 15d9cb5db3
2 changed files with 13 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

8
netx_api/ume_sync_service.py
View file

@ -3,6 +3,7 @@ from __future__ import annotations
import json
import logging
import hashlib
import re
from datetime import datetime, timezone
from typing import Any
@ -81,6 +82,13 @@ def _derive_ne_id_from_alarm(alarm: dict[str, Any]) -> str:
parts = [p for p in alarm_key.split() if p]
if len(parts) >= 2:
return _s(parts[0])
# Fallback: some alarms encode net_id as ME{<net_id>} in objectName.
object_name = _s(_pick(alarm, "objectName", "object-name"))
if object_name:
m = re.search(r"ME\{([^}]+)\}", object_name, flags=re.IGNORECASE)
if m:
return _s(m.group(1))
return ""

5
tests/test_ume_sync.py
View file

@ -487,6 +487,7 @@ class UmeSyncServiceTests(unittest.TestCase):
alarm_hash = {"alarmkey": "00ceb960-1b62-478e-8303-0935ffea1d28#99010"}
alarm_csv = {"alarmkey": "00ceb960-1b62-478e-8303-0935ffea1d28, 4237, 79"}
alarm_space = {"alarmkey": "00ceb960-1b62-478e-8303-0935ffea1d28 4205 3588"}
alarm_object_name_me = {"alarmkey": "no-net-id-here", "objectName": "OLT/ME{00ceb960-1b62-478e-8303-0935ffea1d28}/PON-1"}
self.assertEqual(
_derive_ne_id_from_alarm(alarm_hash),
@ -500,6 +501,10 @@ class UmeSyncServiceTests(unittest.TestCase):
_derive_ne_id_from_alarm(alarm_space),
"00ceb960-1b62-478e-8303-0935ffea1d28",
)
self.assertEqual(
_derive_ne_id_from_alarm(alarm_object_name_me),
"00ceb960-1b62-478e-8303-0935ffea1d28",
)
def test_serialize_ume_alarm_raw_row_select_fields(self):
alarm = UmeAlarmCurrent(