Deprecation message

2026-08-08 22:43:01 +01:00
parent 2735e6e96b
commit 26ca2d3256
33 changed files with 889 additions and 637 deletions
@@ -0,0 +1,81 @@
+import unittest
+from unittest.mock import patch
+from uuid import uuid4
+
+from dynalab.connectors.events import (
+    SignalConnectedEvent,
+    SignalDisconnectedEvent,
+    SignalValueEvent,
+)
+from dynalab.state import AppState
+from dynalab.ui.live_data.controller import handle_connector_event
+from dynalab.ui.live_data.tags import signal_gauge_tag
+from dynalab_protocol.models import SignalDescriptor
+
+
+class ApplicationStateTests(unittest.TestCase):
+    def test_instances_do_not_share_threading_state(self) -> None:
+        first = AppState()
+        second = AppState()
+
+        first.stop_event.set()
+        first.connector_events.put(SignalValueEvent(uuid4(), 1.0))
+
+        self.assertFalse(second.stop_event.is_set())
+        self.assertTrue(second.connector_events.empty())
+
+
+class LiveDataControllerTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.state = AppState()
+        self.connector_id = uuid4()
+        self.signal = SignalDescriptor(
+            id=uuid4(),
+            name="Engine speed",
+            type="number",
+            min_value=0,
+            max_value=9000,
+            unit="rpm",
+        )
+
+    @patch("dynalab.ui.live_data.controller.dpgg")
+    def test_signal_lifecycle_updates_state_and_gauge(self, gauges) -> None:
+        handle_connector_event(
+            self.state, SignalConnectedEvent(self.connector_id, self.signal)
+        )
+        handle_connector_event(
+            self.state, SignalValueEvent(self.signal.id, 3250.0)
+        )
+        handle_connector_event(
+            self.state, SignalDisconnectedEvent(self.connector_id, self.signal.id)
+        )
+
+        gauges.digital_gauge.assert_called_once()
+        gauges.set_value.assert_called_once_with(
+            tag=signal_gauge_tag(self.signal.id), value=3250.0
+        )
+        gauges.delete_gauge.assert_called_once_with(
+            tag=signal_gauge_tag(self.signal.id)
+        )
+        self.assertNotIn(self.signal.id, self.state.signals.descriptors)
+        self.assertNotIn(self.signal.id, self.state.signals.values)
+
+    @patch("dynalab.ui.live_data.controller.dpgg")
+    def test_shared_signal_remains_until_last_connector_disconnects(self, gauges) -> None:
+        other_connector_id = uuid4()
+        handle_connector_event(
+            self.state, SignalConnectedEvent(self.connector_id, self.signal)
+        )
+        handle_connector_event(
+            self.state, SignalConnectedEvent(other_connector_id, self.signal)
+        )
+        handle_connector_event(
+            self.state, SignalDisconnectedEvent(self.connector_id, self.signal.id)
+        )
+
+        gauges.delete_gauge.assert_not_called()
+        self.assertIn(self.signal.id, self.state.signals.descriptors)
+
+
+if __name__ == "__main__":
+    unittest.main()