Archived
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
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()
|