Add a test for karabo-bridge-serve-run

extra_data/read_machinery.py

@@ -4,6 +4,7 @@
 """
 import logging
 import math
+import os
 import os.path as osp
 import re
 import time
@@ -17,7 +18,7 @@
 DETECTOR_NAMES = {'AGIPD', 'DSSC', 'LPD'}
 DETECTOR_SOURCE_RE = re.compile(r'(.+)/DET/(\d+)CH')
 
-DATA_ROOT_DIR = '/gpfs/exfel/exp'
+DATA_ROOT_DIR = os.environ.get('EXTRA_DATA_DATA_ROOT', '/gpfs/exfel/exp')
 
 
 class _SliceConstructor(type):

extra_data/tests/test_streamer.py

@@ -2,6 +2,7 @@
 
 import os
 
+import numpy as np
 import pytest
 from subprocess import PIPE, Popen
 
@@ -73,6 +74,48 @@ def test_serve_files(mock_fxe_raw_run, tmp_path):
             assert rc == -9  # process terminated by kill signal
 
 
+@pytest.mark.skipif(os.name != 'posix', reason="Test uses Unix socket")
+def test_serve_run(mock_spb_raw_and_proc_run, tmp_path):
+    mock_data_root, _, _ = mock_spb_raw_and_proc_run
+    xgm_src = 'SPB_XTD9_XGM/DOOCS/MAIN'
+    agipd_m0_src = 'SPB_DET_AGIPD1M-1/DET/0CH0:xtdf'
+    args = ['karabo-bridge-serve-run', '2012', '238',
+            '--port', f'ipc://{tmp_path}/socket',
+            '--include', f'{xgm_src}[beamPosition.i*Pos]',
+            '--include', '*AGIPD1M-1/DET/0CH0:xtdf'
+           ]
+    interface = None
+
+    p = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE,
+               env=dict(os.environ,
+                        PYTHONUNBUFFERED='1',
+                        EXTRA_DATA_DATA_ROOT=mock_data_root))
+    try:
+        for line in p.stdout:
+            line = line.decode('utf-8')
+            if line.startswith('Streamer started on:'):
+                interface = line.partition(':')[2].strip()
+                break
+
+        print('interface:', interface)
+        assert interface is not None, p.stderr.read().decode()
+
+        with Client(interface, timeout=30) as c:
+            data, meta = c.next()
+
+        tid = next(m['timestamp.tid'] for m in meta.values())
+        assert tid == 10000
+        assert set(data) == {xgm_src, agipd_m0_src}
+        assert set(data[xgm_src]) == \
+               {f'beamPosition.i{xy}Pos.value' for xy in 'xy'} | {'metadata'}
+        assert data[agipd_m0_src]['image.data'].dtype == np.float32
+    finally:
+        if p.poll() is None:
+            p.kill()
+            rc = p.wait(timeout=2)
+            assert rc == -9  # process terminated by kill signal
+
+
 def test_deprecated_server():
     with pytest.deprecated_call():
         with ZMQStreamer(2222):