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

fix(ume): show WSS connection state separately from alarm activity

Browse source 
Track ws_connection for UI pills, stop overwriting runtime last_error on alarm events, and wire pause/resume to reconnect WSS.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-05-23 11:19:00 +08:00
parent 5bd83c58dc
commit 31e745ef95
5 changed files with 142 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

13
tests/test_ume_sync.py
View file

@ -20,6 +20,7 @@ from netx_api.ume_alarm_ws import (
cancel_alarm_subscription_manual,
establish_alarm_subscription_manual,
get_subscription_status,
get_ws_connection_status,
get_ws_logs,
load_persisted_subscription,
parse_subscription_id_from_already_exists_error,
@ -738,6 +739,18 @@ class UmeAlarmNotificationTests(unittest.TestCase):
self.assertTrue(changed)
self.assertIsNone(self.db.get(UmeAlarmCurrent, "AK-CLEARED-1"))
def test_ws_connection_status_not_alarm_action(self):
from netx_api import ume_alarm_ws as ws_mod
with ws_mod._ws_connection_lock:
ws_mod._ws_connection_state = "init"
ws_mod._ws_connection_detail = ""
ws_mod._set_ws_connection_state("connected", detail="ws connected")
st = get_ws_connection_status()
self.assertEqual(st["state"], "connected")
self.assertEqual(st["label"], "已连接")
self.assertEqual(st["detail"], "ws connected")
def test_ws_log_ring_buffer(self):
from netx_api import ume_alarm_ws as ws_mod