Unit test for abs. timing parameters

tests/conftest.py

@@ -1,6 +1,7 @@
 import asyncio
 import os
 import threading
+
 from collections import deque
 from io import BufferedReader
 from pathlib import Path
@@ -42,6 +43,25 @@ def fast_dump():
         yield chunked_read(f, 500)
 
 
+@pytest_asyncio.fixture
+def fast_dump_with_extra_header_params():
+    """
+    Add header parameters to `fast_dump.bin` binary stream. The fixture
+    is used for testing absolute timing parameters passed in the header.
+    """
+    def fast_dump(extra_header_params):
+        param_to_replace = "sample_bytes=\"52\""
+        params = " ".join(f"{k}=\"{v}\"" for k, v in extra_header_params.items())
+        params = " ".join([param_to_replace, params]) if params else param_to_replace
+        with open(Path(__file__).parent / "data_dumps/fast_dump.bin", "rb") as f:
+            # Simulate larger chunked read
+            data = chunked_read(f, 500)
+            for buffer in data:
+                yield buffer.replace(param_to_replace.encode(), params.encode())
+
+    return fast_dump
+
+
 @pytest_asyncio.fixture
 def raw_dump():
     with open(Path(__file__).parent / "data_dumps/raw_dump.bin", "rb") as f:

tests/test_asyncio.py

@@ -1,4 +1,5 @@
 import asyncio
+import copy
 
 import pytest
 
@@ -50,6 +51,48 @@ async def test_asyncio_data(
     assert fast_dump_expected == events
 
 
+@pytest.mark.parametrize("timing_params",[
+    {},
+    {
+        "arm_time": "2024-03-05T20:27:12.607841574Z",
+        "start_time": "2024-03-05T20:27:12.608875498Z",
+    },
+    {
+        "arm_time": "2024-03-05T20:27:12.607841574Z",
+        "start_time": "2024-03-05T20:27:12.608875498Z",
+        "hw_time_offset_ns": 100555,
+    },
+])
+async def test_asyncio_data_with_abs_timing(
+    dummy_server_async,
+    fast_dump_with_extra_header_params,
+    fast_dump_expected,
+    timing_params,
+):
+    """
+    The test for handling of `arm_time`, `start_time` and `hw_time_offset_ns`
+    parameters passed in the header. The test is reusing the existing `fast_dump`
+    and `fast_dump_expected` by adding timing parameters to the header in
+    the binary stream and replacing the expected `StartData` attributes with
+    the expected values.
+    """
+    dummy_server_async.data = fast_dump_with_extra_header_params(timing_params)
+    events = []
+    async with AsyncioClient("localhost") as client:
+        async for data in client.data(frame_timeout=1):
+            events.append(data)
+            if len(events) == len(fast_dump_expected):
+                break
+    fast_dump_expected = list(fast_dump_expected)
+
+    # Replace attributes in `StartData` with the expected values
+    fast_dump_expected[1] = copy.deepcopy(fast_dump_expected[1])
+    for attr_name in timing_params:
+        setattr(fast_dump_expected[1], attr_name, timing_params[attr_name])
+
+    assert fast_dump_expected == events
+
+
 async def test_asyncio_data_timeout(dummy_server_async, fast_dump):
     dummy_server_async.data = fast_dump
     async with AsyncioClient("localhost") as client:

tests/test_cli.py

@@ -7,13 +7,10 @@
 from click.testing import CliRunner
 
 from pandablocks import cli
-from pandablocks.hdf import GATE_DURATION_FIELD, SAMPLES_FIELD, HDFDataOverrunException
-from tests.conftest import (
-    STATE_RESPONSES,
-    STATE_SAVEFILE,
-    DummyServer,
-    assert_all_data_in_hdf_file,
-)
+from pandablocks.hdf import (GATE_DURATION_FIELD, SAMPLES_FIELD,
+                             HDFDataOverrunException)
+from tests.conftest import (STATE_RESPONSES, STATE_SAVEFILE, DummyServer,
+                            assert_all_data_in_hdf_file)
 
 
 @pytest.mark.parametrize("samples_name", [GATE_DURATION_FIELD, SAMPLES_FIELD])