80 lines
3.4 KiB
Python
80 lines
3.4 KiB
Python
"""The display shows the newest acquisition: after polls, Stop drains and settings changes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import numpy as np
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
from PyQt6.QtWidgets import QApplication # noqa: E402
|
|
|
|
from python_app.gui.controllers.renderers.trace import TraceRenderer # noqa: E402
|
|
from python_app.models.dataset_model import ComboKey, ResultBlock, ResultCollection, ResultPayload # noqa: E402
|
|
from python_app.models.gui_profile_schema import GuiPassThroughStateModel # noqa: E402
|
|
from python_app.tests.test_reader_recovery import pipeline_fixture # noqa: E402
|
|
|
|
|
|
def _result(collection_id: int, capture_start_ns: int, revision: int = 0) -> ResultCollection:
|
|
return ResultCollection(collection_id, capture_start_ns + 1, settings_revision=revision,
|
|
capture_start_ns=capture_start_ns, capture_end_ns=capture_start_ns + 1)
|
|
|
|
|
|
class NewestResultDisplayTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.pipeline, _supervisor = pipeline_fixture(self)
|
|
self.pipeline.pump_events = mock.Mock()
|
|
draw = mock.patch.object(self.pipeline.plots, "draw")
|
|
self.draw = draw.start()
|
|
self.addCleanup(draw.stop)
|
|
|
|
def read(self, *results: ResultCollection | None) -> None:
|
|
reader = mock.Mock()
|
|
reader.pop_result_collection.side_effect = [*results, *[None] * 10]
|
|
self.pipeline._result_reader = reader
|
|
|
|
def test_poll_shows_the_newest_acquisition_rather_than_the_last_one_read(self) -> None:
|
|
newest = _result(2, 200)
|
|
self.read(newest, _result(1, 100)) # a re-fed earlier sweep arrives after the newest
|
|
self.pipeline.poll()
|
|
self.draw.assert_called_once_with(newest)
|
|
|
|
def test_results_drained_by_stop_are_shown_once_stopped(self) -> None:
|
|
last = _result(2, 200)
|
|
self.read(_result(1, 100), last)
|
|
self.pipeline.stop()
|
|
self.draw.assert_not_called()
|
|
self.pipeline.poll()
|
|
self.pipeline.poll()
|
|
self.draw.assert_called_once_with(last)
|
|
|
|
def test_revision_drain_waits_past_quiet_polls_for_the_published_revision(self) -> None:
|
|
stale, fresh = _result(1, 100, revision=1), _result(1, 100, revision=2)
|
|
for revision, expected in ((None, stale), (2, fresh)):
|
|
with self.subTest(revision=revision):
|
|
self.pipeline.history.results.clear()
|
|
self.read(stale, None, None, None, fresh)
|
|
self.pipeline.drain_results(timeout_s=60, poll_s=0, revision=revision)
|
|
self.assertIs(self.pipeline.history.results[-1], expected)
|
|
|
|
|
|
class TraceViewTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def test_draws_only_pass_through_payloads(self) -> None:
|
|
renderer = TraceRenderer(mock.Mock())
|
|
self.addCleanup(renderer.widget.close)
|
|
payloads = [ResultPayload(name, 1, frequency_hz=np.array([1e9, 2e9]), trace=np.ones(2, np.complex64))
|
|
for name in ("bscan", "pass_through")]
|
|
collection = ResultCollection(1, 1, blocks=[ResultBlock(ComboKey(0, 0), payloads)])
|
|
self.assertTrue(renderer.draw(collection, GuiPassThroughStateModel()))
|
|
self.assertEqual([name for _combo, _index, name in renderer.magnitude.curves], ["pass_through"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|