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

feat(ume): denormalize host_name onto alarms for display and grouping

Browse source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-05-20 00:05:10 +08:00
parent 95db646403
commit 21f7cfba41
5 changed files with 136 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

20
tests/test_ume_sync.py
View file

@ -342,6 +342,16 @@ class UmeSyncServiceTests(unittest.TestCase):
return rows, _D()
self.db.add(
UmeInventoryNE(
ne_id="NE-1",
ne_name="ne1",
user_label="网元1",
host_name="host-ne-1",
)
)
self.db.commit()
job1, _ = sync_alarms_current(self.db, _C(), trigger_mode="manual")
self.assertEqual(job1.status, "done")
self.assertEqual(job1.inserted_count, 1)
@ -353,6 +363,7 @@ class UmeSyncServiceTests(unittest.TestCase):
row = self.db.get(UmeAlarmCurrent, "AK-1")
self.assertIsNotNone(row)
self.assertEqual(row.perceived_severity, "critical")
self.assertEqual(row.host_name, "host-ne-1")
def test_sync_current_alarms_marker_pagination(self):
class _C:
@ -547,12 +558,17 @@ class UmeSyncServiceTests(unittest.TestCase):
self.assertIn("ne_exists", set(data["selectable_fields"]))
def test_extract_ume_raw_group_field(self):
alarm = UmeAlarmCurrent(alarm_key="AK-X", perceived_severity="major")
ne = UmeInventoryNE(ne_id="NE-X", user_label="site-x")
alarm = UmeAlarmCurrent(alarm_key="AK-X", perceived_severity="major", host_name="host-x")
ne = UmeInventoryNE(ne_id="NE-X", user_label="site-x", host_name="inv-host")
self.assertEqual(_extract_ume_raw_group_field(alarm, ne, "alarm_perceived_severity"), "major")
self.assertEqual(_extract_ume_raw_group_field(alarm, ne, "ne_user_label"), "site-x")
self.assertEqual(_extract_ume_raw_group_field(alarm, ne, "ne_host_name"), "host-x")
self.assertEqual(_extract_ume_raw_group_field(alarm, None, "ne_exists"), "0")
def test_ume_alarms_fields_includes_alarm_host_name(self):
data = ume_alarms_fields()
self.assertIn("alarm_host_name", set(data["selectable_fields"]))
if __name__ == "__main__":
unittest.main()