fixed python simulator test fixture (#1350)
Build and Deploy on local RHEL9 / build (push) Successful in 2m10s
Build on RHEL9 docker image / build (push) Successful in 3m37s
Build on RHEL8 docker image / build (push) Successful in 4m50s
Build and Deploy on local RHEL8 / build (push) Successful in 5m7s
Run Simulator Tests on local RHEL9 / build (push) Successful in 18m16s
Run Simulator Tests on local RHEL8 / build (push) Successful in 21m52s

* fixed python simulator test fixture

* clear_roi after every test to bring it back to default state, test passing multiple parameters

* Exposing the ctb api tests now to CI

* Revert "Exposing the ctb api tests now to CI"

This reverts commit 411fad1b27.

* fixed tests removed uneccessary stuff

* did not save properly

* updated documentation, renamed file

---------

Co-authored-by: Dhanya Thattil <dhanya.thattil@psi.ch>
This commit is contained in:
2026-05-08 13:09:02 +02:00
committed by GitHub
parent 111d10cfa7
commit 74166a1ea1
7 changed files with 141 additions and 210 deletions
+37 -78
View File
@@ -58,39 +58,56 @@ DEFAULT_SIMULATOR_CONFIGS = [
SIMULATOR_IDS = [f"{det_type}_{num_interface}if_{num_mod}mod" for det_type, num_interface, num_mod in DEFAULT_SIMULATOR_CONFIGS]
'''
for more specific parameters
@pytest.mark.detectorintegration
@pytest.mark.parametrize(
"session_simulator",
[
("ctb", 1, 1),
("xilinx_ctb", 1, 1),
],
indirect=True,
)
def test_define_reg(session_simulator):
det_type, num_interfaces, num_mods, d = session_simulator
'''
@pytest.fixture(scope="session")
def session_simulator(request):
"""
Fixture to start the detector server once and clean up at the end.
Expects request.param = (det_type, num_interfaces, num_mods)
"""
det_type, num_interfaces, num_mods = request.param
fp = sys.stdout
try:
det_type, num_interfaces, num_mods = request.param
fp = sys.stdout
# set up: once per server
Log(LogLevel.INFOBLUE,
f'---- {det_type} | interfaces={num_interfaces} | modules={num_mods} ----', fp)
# set up: once per server
Log(LogLevel.INFOBLUE,
f'---- {det_type} | interfaces={num_interfaces} | modules={num_mods} ----', fp)
cleanup(fp)
startDetectorVirtualServer(det_type, num_mods, fp, True)
startReceiver(num_mods, fp, True)
cleanup(fp)
startDetectorVirtualServer(det_type, num_mods, fp, True, True)
startReceiver(num_mods, fp, True)
Log(LogLevel.INFOBLUE, f'Waiting for server to start up and connect', fp)
d = loadConfig(
name=det_type,
log_file_fp=fp,
num_mods=num_mods,
num_frames=1,
num_interfaces=num_interfaces,
)
Log(LogLevel.INFOBLUE, f'Waiting for server to start up and connect', fp)
d = loadConfig(
name=det_type,
log_file_fp=fp,
num_mods=num_mods,
num_frames=1,
num_interfaces=num_interfaces,
)
loadBasicSettings(name=det_type, d=d, fp=fp)
loadBasicSettings(name=det_type, d=d, fp=fp)
yield det_type, num_interfaces, num_mods, d
cleanup(fp)
yield det_type, num_interfaces, num_mods, d
cleanup(fp)
except Exception as e:
Log(LogLevel.ERROR, f'Tests Failed.', fp)
cleanup(fp)
def pytest_generate_tests(metafunc):
if "session_simulator" not in metafunc.fixturenames:
@@ -110,62 +127,4 @@ def pytest_generate_tests(metafunc):
indirect=True
)
'''
for more specific parameters
@pytest.mark.detectorintegration
@pytest.mark.parametrize(
"session_simulator",
[
("ctb", 1, 1),
("xilinx_ctb", 1, 1),
],
indirect=True,
)
def test_define_reg(session_simulator):
det_type, num_interfaces, num_mods, d = session_simulator
'''
#helper fixture for servers
@pytest.fixture(scope='module')
def setup_parameters(request): # only setup once per module if same parameters used for the scopes
try:
servers, nmods = request.param # comes from @pytest.mark.parametrize(..., indirect=True)
return servers, nmods
except AttributeError:
# fallback default if the test did not parametrize
return (['eiger', 'jungfrau', 'mythen3', 'gotthard2', 'ctb', 'moench', 'xilinx_ctb'], 2)
@pytest.fixture(scope='module')
def test_with_simulators(setup_parameters):
""" Fixture to automatically setup virtual detector servers for testing. """
fp = sys.stdout
servers, nmods = setup_parameters
print("servers:", servers)
print("nmods:", nmods)
try:
for server in servers:
for ninterfaces in range(1,2):
if ninterfaces == 2 and server != 'jungfrau' and server != 'moench':
continue
msg = f'Starting Python API Tests for {server}'
if server == 'jungfrau' or server == 'moench':
msg += f' with {ninterfaces} interfaces'
Log(LogLevel.INFOBLUE, msg, fp)
cleanup(fp)
startDetectorVirtualServer(server, nmods, fp)
startReceiver(nmods, fp)
d = loadConfig(name=server, log_file_fp=fp, num_mods=nmods, num_frames=1, num_interfaces=ninterfaces)
#loadBasicSettings(name=server, d=d, fp=fp)
yield # run test
cleanup(fp) # teardown
except Exception as e:
traceback.print_exc(file=fp)
Log(LogLevel.ERROR, f'Tests Failed.', fp)
cleanup(fp)
+59
View File
@@ -0,0 +1,59 @@
import pytest
import sys
from conftest import session_simulator
from slsdet import Detector
from slsdet._slsdet import slsDetectorDefs
detectorType = slsDetectorDefs.detectorType
@pytest.mark.detectorintegration
def test_rx_ROI(session_simulator):
""" Test rx_ROI property of Detector class. """
det_type, num_interfaces, num_mods, d = session_simulator
assert d is not None
if d.type == detectorType.CHIPTESTBOARD or d.type == detectorType.XILINX_CHIPTESTBOARD:
pytest.skip("Skipping ROI test for ctb/xilinx_ctb detector types.")
if(d.type == detectorType.MYTHEN3 or d.type == detectorType.GOTTHARD2):
d.rx_roi = (0, 10)
roi = d.rx_roi
assert roi == [(0, 10, -1, -1)]
#d.rx_roi = [[5,15, 0, 1]] # not allowed for mythen3
d.rx_roi = [0,10, -1, -1]
assert d.rx_roi == [(0,10,-1,-1)]
d.rx_clearroi()
else:
d.rx_roi = (0, 10, 10, 20)
roi = d.rx_roi
assert roi == [(0, 10, 10, 20)]
d.rx_roi = [5,15,15,25]
assert d.rx_roi == [(5,15,15,25)]
if d.nmod > 1 and (d.type != detectorType.JUNGFRAU) or (d.numinterfaces == 2 and d.type != detectorType.EIGER):
d.rx_roi = [[0,10,0,20], [5,20,410,420]]
roi = d.rx_roi
assert roi == [(0,10,0,20), (5,20,410,420)] #in same file for jungfrau
d.rx_clearroi()
roi = d.rx_roi
assert roi == [(-1,-1,-1,-1)]
+17
View File
@@ -12,6 +12,9 @@ from utils_for_test import (
)
from slsdet import Detector
from slsdet._slsdet import slsDetectorDefs
detectorType = slsDetectorDefs.detectorType
@pytest.mark.detectorintegration
@@ -903,3 +906,17 @@ def test_dac(session_simulator, request):
Log(LogLevel.INFOGREEN, f"{request.node.name} passed")
@pytest.mark.detectorintegration
@pytest.mark.parametrize("session_simulator",[("moench", 1, 2)],indirect=True)
def test_type(session_simulator):
d = Detector()
assert d.type == detectorType.MOENCH
@pytest.mark.detectorintegration
@pytest.mark.parametrize("session_simulator",[("moench", 1, 2), ("jungfrau", 1, 2)],indirect=True)
def test_numinterfaces(session_simulator):
d = Detector()
assert d.numinterfaces == 1
+12 -36
View File
@@ -17,46 +17,18 @@ sys.path.append(str(scripts_dir))
from slsdet import Detector, Ctb, freeSharedMemory
from utils_for_test import (
Log,
LogLevel,
cleanup,
startDetectorVirtualServer,
connectToVirtualServers,
SERVER_START_PORTNO
)
'''
scope = module =>Once per test file/module
to share expensive setup like startDetectorVirtualServer
'''
@pytest.fixture(scope="module")
def det_config():
return {
"name": "ctb",
"num_mods": 1
}
@pytest.fixture(scope="module", autouse=True)
def setup_simulator(det_config):
"""Fixture to start the detector server once and clean up at the end."""
fp = sys.stdout
cleanup(fp)
startDetectorVirtualServer(det_config["name"], det_config["num_mods"], fp)
Log(LogLevel.INFOBLUE, f'Waiting for server to start up and connect')
connectToVirtualServers(det_config["name"], det_config["num_mods"])
Log(LogLevel.INFOBLUE, f'Freeing shm before tests')
freeSharedMemory()
yield # tests run here
cleanup(fp)
from conftest import session_simulator
@pytest.mark.detectorintegration
def test_exptime_after_free_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_free_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_free_should_raise')
@@ -78,7 +50,8 @@ def free_and_create_shm():
k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct
@pytest.mark.detectorintegration
def test_exptime_after_not_passing_var_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_not_passing_var_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_not_passing_var_should_raise')
@@ -102,7 +75,8 @@ def free_and_create_shm_passing_ctb_var(k):
k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct
@pytest.mark.detectorintegration
def test_exptime_after_passing_ctb_var_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_passing_ctb_var_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_passing_ctb_var_should_raise')
d = Ctb() # creates multi shm (assuming no shm exists)
@@ -125,7 +99,8 @@ def free_and_create_shm_returning_ctb():
return k
@pytest.mark.detectorintegration
def test_exptime_after_returning_ctb_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_returning_ctb_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_returning_ctb_should_raise')
d = Ctb() # creates multi shm (assuming no shm exists)
@@ -148,7 +123,8 @@ def test_exptime_after_returning_ctb_should_raise(setup_simulator):
assert str(exc_info.value) == "Shared memory is invalid or freed. Close resources before access."
@pytest.mark.detectorintegration
def test_hostname_twice_acess_old_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_hostname_twice_acess_old_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_hostname_twice_acess_old_should_raise')
d = Ctb() # creates multi shm (assuming no shm exists)
-51
View File
@@ -1,51 +0,0 @@
import pytest
import sys
from conftest import test_with_simulators
from slsdet import Detector
from utils_for_test import (
Log,
LogLevel,
)
@pytest.mark.detectorintegration
@pytest.mark.parametrize("setup_parameters", [(["moench"], 2)], indirect=True)
def test_rx_ROI_moench(test_with_simulators, setup_parameters):
""" Test setting and getting rx_ROI property of Detector class for moench. """
d = Detector()
d.rx_roi = (0, 10, 10, 20)
roi = d.rx_roi
assert roi == [(0, 10, 10, 20)]
d.rx_roi = [5,15,15,25]
assert d.rx_roi == [(5,15,15,25)]
d.rx_roi = [[0,10,0,20], [5,20,410,420]]
roi = d.rx_roi
assert roi == [(0,10,0,20), (5,20,410,420)]
d.rx_clearroi()
roi = d.rx_roi
assert roi == [(-1,-1,-1,-1)]
@pytest.mark.detectorintegration
@pytest.mark.parametrize("setup_parameters", [(["mythen3"], 1)], indirect=True)
def test_rx_ROI_mythen(test_with_simulators, setup_parameters):
""" Test setting and getting rx_ROI property of Detector class for mythen. """
d = Detector()
d.rx_roi = (0, 10)
roi = d.rx_roi
assert roi == [(0, 10, -1, -1)]
#d.rx_roi = [[5,15, 0, 1]] # not allowed for mythen3
d.rx_roi = [0,10, -1, -1]
assert d.rx_roi == [(0,10,-1,-1)]