UserAlert wip

Author: wk9874

simvue/api/objects/alert/user.py

@@ -83,8 +83,8 @@ def get_status(self, run_id: str) -> typing.Literal["ok", "critical"] | None:
 
     def on_reconnect(self, id_mapping: dict[str, str]) -> None:
         """Set status update on reconnect"""
-        for run_id, status in self._staging.get("status", self._local_status).items():
-            self.set_status(run_id, status)
+        for offline_id, status in self._staging.get("status", {}).items():
+            self.set_status(id_mapping.get(offline_id), status)
 
     @pydantic.validate_call
     def set_status(self, run_id: str, status: typing.Literal["ok", "critical"]) -> None:

tests/unit/test_user_alert.py

@@ -172,4 +172,33 @@ def test_user_alert_status() -> None:
     _run.delete()
     _folder.delete(recursive=True, runs_only=False, delete_runs=True)
     _alert.delete()
+    
+    
+@pytest.mark.api
+@pytest.mark.offline
+def test_user_alert_status_offline() -> None:
+    _uuid: str = f"{uuid.uuid4()}".split("-")[0]
+    _alert = UserAlert.new(
+        name=f"users_alert_{_uuid}",
+        notification="none",
+        description=None,
+        offline=True
+    )
+    _alert.commit()
+    _folder = Folder.new(path=f"/simvue_unit_tests/{_uuid}", offline=True)
+    _run = Run.new(folder=f"/simvue_unit_tests/{_uuid}", offline=True)
+    _folder.commit()
+    _run.alerts = [_alert.id]
+    _run.commit()
+    
+    sender(_alert._local_staging_file.parents[1], 1, 10, ["folders", "runs", "alerts"])
+    time.sleep(1) 
+    
+    _alert.set_status(_run.id, "critical")
+    _alert.commit()
+    import pdb; pdb.set_trace()
+    time.sleep(1)
+    _run.delete()
+    _folder.delete(recursive=True, runs_only=False, delete_runs=True)
+    _alert.delete()