mirror of
https://github.com/slsdetectorgroup/slsDetectorPackage.git
synced 2026-08-06 05:02:28 +02:00
fixed python simulator test fixture (#1350)
* 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:
+10
-42
@@ -101,7 +101,7 @@ To run only tests requiring virtual detectors use the following command:
|
||||
#in build
|
||||
python -m pytest -m detectorintegration ../python/tests/
|
||||
|
||||
There is a helper test fixture in ``slsDetectorSoftware/python/tests/conftest.py`` called ``test_with_simulators`` that sets up virtual detectors and yields the test for all detectors. The set up is done for every test automatically.
|
||||
There is a helper test fixture in ``slsDetectorSoftware/python/tests/conftest.py`` called ``session_simulator`` that sets up virtual detectors and yields the test for all detectors. The set up is done for every test automatically. Note that the fixture persist over the entire session e.g. the fixture is setup one detector at a time and runs all tests using this fixture before cleaning up and moving on to the next detector. It saves time if the setup and cleanup is expensive.
|
||||
|
||||
Example usage:
|
||||
|
||||
@@ -110,10 +110,13 @@ Example usage:
|
||||
import pytest
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_example_with_simulator(test_with_simulators):
|
||||
def test_example_with_simulator(session_simulator):
|
||||
# your test code here
|
||||
|
||||
If you want to run the test only for a specific test use the parametrized test fixture:
|
||||
.. Note::
|
||||
As the detector is set up only once makes sure to not change the state of the detector in a way that affects other tests. If you want to change the state of the detector make sure to reset it at the end of your test.
|
||||
|
||||
If you want to run the test only for a specific detector use the parametrized test fixture:
|
||||
|
||||
Example usage:
|
||||
|
||||
@@ -122,45 +125,10 @@ Example usage:
|
||||
import pytest
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
@pytest.mark.parametrize("setup_parameters", [(["<my_detector>"], <num_modules>)], indirect=True)
|
||||
def test_example_with_specific_simulators(test_with_simulators, setup_parameters):
|
||||
@pytest.mark.parametrize("session_simulator", [("<my_detector>", <num_interfaces>, <num_modules>), ("<another_detector>", <num_interfaces>, <num_modules>)], indirect=True)
|
||||
def test_example_with_specific_simulators(session_simulator):
|
||||
# your test code here
|
||||
|
||||
.. Note::
|
||||
The parametrized test fixture is setup per file and not for the entire session.
|
||||
|
||||
There is another helper test fixture in ``slsDetectorSoftware/python/tests/conftest.py`` called ``session_simulator`` that sets up virtual detectors and yields the test for all detectors. The difference with the previous fixture ``test_with_simulators`` is that this fixture will set up one detector at a time and run all the tests using this fixture before cleaning up and moving on to the next detector. It saves time if the setup and cleanup is expensive.
|
||||
|
||||
Example usage:
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
import pytest
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_define_reg(session_simulator, request):
|
||||
""" Test setting define_reg for ctb and xilinx_ctb."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
from slsdet import RegisterAddress
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
# your test code here
|
||||
|
||||
For more specific parameters, you can parametrize the fixture like below:
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
import pytest
|
||||
|
||||
@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
|
||||
# your test code here
|
||||
|
||||
+37
-78
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -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)]
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+469
-15
@@ -12,6 +12,9 @@ from utils_for_test import (
|
||||
)
|
||||
from slsdet import Detector
|
||||
|
||||
from slsdet._slsdet import slsDetectorDefs
|
||||
|
||||
detectorType = slsDetectorDefs.detectorType
|
||||
|
||||
@pytest.fixture(
|
||||
scope="session",
|
||||
@@ -351,7 +354,6 @@ def test_definelist_bit(session_simulator, request):
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
<<<<<<< HEAD:python/tests/test_det_api.py
|
||||
@pytest.mark.detectorintegration
|
||||
def test_parameters_file(session_simulator, request):
|
||||
""" Test using test_parameters_file."""
|
||||
@@ -397,25 +399,477 @@ def test_patternstart(session_simulator, request):
|
||||
assert d is not None
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb', 'mythen3']:
|
||||
=======
|
||||
@pytest.mark.withdetectorsimulators
|
||||
def test_patternstart(simulator, request):
|
||||
""" Test using patternstart for ctb, xilinx_ctb and mythen3."""
|
||||
det_name = simulator
|
||||
# setup
|
||||
d = Detector()
|
||||
d.hostname = f"localhost:{SERVER_START_PORTNO}"
|
||||
|
||||
if det_name in ['ctb', 'xilinx_ctb', 'mythen3']:
|
||||
>>>>>>> e519633e1 (added patternstart to python (#1368)):python/tests/test_CtbAPI.py
|
||||
d.patternstart()
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.patternstart()
|
||||
assert "not implemented" in str(exc_info.value)
|
||||
|
||||
<<<<<<< HEAD:python/tests/test_det_api.py
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
=======
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_adcclk(session_simulator, request):
|
||||
""" Test using adcclk for ctb and xilinx_ctb."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
from slsdet import Hz, MHz, kHz
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
prev_adcclk = d.getADCClock()
|
||||
|
||||
d.adcclk
|
||||
|
||||
# invalid value type
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.adcclk = 5e6
|
||||
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.adcclk = 5 * 1000 * 1000
|
||||
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.adcclk = Hz(5e6)
|
||||
|
||||
d.adcclk = MHz(15)
|
||||
assert d.adcclk.value == 15_000_000
|
||||
|
||||
d.adcclk = MHz(14.5)
|
||||
assert d.adcclk.value == 14_500_000
|
||||
|
||||
d.adcclk = kHz(15000.5)
|
||||
assert d.adcclk.value == 15_000_500
|
||||
|
||||
# invalid values from server
|
||||
# max is 300MHz for xilinx and 54 MHz for ctb
|
||||
if det_type == 'ctb':
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.adcclk = MHz(66)
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.adcclk = MHz(301)
|
||||
|
||||
# min is 2MHz for ctb and 10MHz for xilinx_ctb
|
||||
if det_type == 'ctb':
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.adcclk = MHz(1)
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.adcclk = MHz(9)
|
||||
|
||||
c = MHz(2)
|
||||
for rc in [5, 10, 15, 20]:
|
||||
d.adcclk = rc * c
|
||||
assert d.adcclk.value == 40_000_000
|
||||
|
||||
for i in range(len(d)):
|
||||
d.setADCClock(prev_adcclk[i], [i])
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
>>>>>>> e519633e1 (added patternstart to python (#1368)):python/tests/test_CtbAPI.py
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_dbitclk(session_simulator, request):
|
||||
""" Test using dbitclk for ctb and xilinx_ctb."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
from slsdet import Hz, MHz, kHz
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
prev_dbitclk = d.getDBITClock()
|
||||
|
||||
d.dbitclk
|
||||
|
||||
# invalid value type
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.dbitclk = 5e6
|
||||
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.dbitclk = 5 * 1000 * 1000
|
||||
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.dbitclk = Hz(5e6)
|
||||
|
||||
d.dbitclk = MHz(15)
|
||||
assert d.dbitclk.value == 15_000_000
|
||||
|
||||
d.dbitclk = MHz(14.5)
|
||||
assert d.dbitclk.value == 14_500_000
|
||||
|
||||
d.dbitclk = kHz(15000.5)
|
||||
assert d.dbitclk.value == 15_000_500
|
||||
|
||||
# invalid values from server
|
||||
# max is 300MHz
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.dbitclk = MHz(301)
|
||||
|
||||
# min is 2MHz for ctb and 10MHz for xilinx_ctb
|
||||
if det_type == 'ctb':
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.dbitclk = MHz(1)
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.dbitclk = MHz(9)
|
||||
|
||||
c = MHz(2)
|
||||
for rc in [5, 10, 15, 20]:
|
||||
d.dbitclk = rc * c
|
||||
assert d.dbitclk.value == 40_000_000
|
||||
|
||||
for i in range(len(d)):
|
||||
d.setDBITClock(prev_dbitclk[i], [i])
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_syncclk(session_simulator, request):
|
||||
""" Test using syncclk for ctb."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
if det_type in ['ctb']:
|
||||
d.syncclk
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_v_limit(session_simulator, request):
|
||||
"""Test v_limit."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
|
||||
# save previous value
|
||||
prev_val = d.getVoltageLimit()
|
||||
from slsdet import dacIndex, powerIndex
|
||||
prev_dac_val = d.getDAC(dacIndex.DAC_0, False)
|
||||
prev_power_dac_val = d.getPowerDAC(powerIndex.V_POWER_A)
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.v_limit = (1200, 'mV') #mV unit not supported, should be 'no unit'
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.v_limit = -100 # previously worked but not allowing now
|
||||
|
||||
# setting dac and power dac with no vlimit should work
|
||||
d.v_limit = 0
|
||||
assert d.v_limit == 0
|
||||
d.setDAC(dacIndex.DAC_0, 1200, True, [0])
|
||||
d.setPowerDAC(powerIndex.V_POWER_A, 1200)
|
||||
|
||||
# setting vlimit should throw setting values above vlimit
|
||||
d.v_limit = 1500
|
||||
assert d.v_limit == 1500
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.setDAC(dacIndex.DAC_0, 1501, True, [0])
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.setPowerDAC(powerIndex.V_POWER_A, 1501)
|
||||
|
||||
# setting dac and power dac below vlimit should still work
|
||||
d.setDAC(dacIndex.DAC_0, 1210, True, [0])
|
||||
d.setPowerDAC(powerIndex.V_POWER_A, 1210)
|
||||
|
||||
# restore previous value
|
||||
d.setVoltageLimit(prev_val)
|
||||
d.setPowerDAC(powerIndex.V_POWER_A, prev_power_dac_val)
|
||||
for i in range(len(d)):
|
||||
d.setDAC(dacIndex.DAC_0, prev_dac_val[i], False, [i])
|
||||
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
d.v_limit
|
||||
assert "not implemented" in str(exc_info.value)
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_v_abcd(session_simulator, request):
|
||||
"""Test v_a, v_b, v_c, v_d, v_io are deprecated comands."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.v_a
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.v_b
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.v_c
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.v_d
|
||||
|
||||
with pytest.raises(Exception):
|
||||
d.v_io
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_powers(session_simulator, request):
|
||||
"""Test powers and powerlist."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
from slsdet import Ctb
|
||||
c = Ctb()
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
|
||||
c.powerlist
|
||||
|
||||
# save previous value
|
||||
from slsdet import powerIndex
|
||||
prev_val_dac = {power: c.getPowerDAC(power) for power in c.getPowerList()}
|
||||
prev_val = {power: c.isPowerEnabled(power) for power in c.getPowerList()}
|
||||
|
||||
# invalid
|
||||
invalid_assignments = [
|
||||
(c.powers, "random", True), # set random power
|
||||
(c.powers, "random", True), # set random attribute of power
|
||||
(c.powers.VA, "dac", "1200"),
|
||||
(c.powers.VA, "enabled", "True"),
|
||||
(c.powers, "VA", "-100"),
|
||||
(c.powers, "VA", "-1"),
|
||||
(c.powers, "VA", "4096")
|
||||
]
|
||||
for obj, attr, value in invalid_assignments:
|
||||
with pytest.raises(Exception):
|
||||
setattr(obj, attr, value)
|
||||
# vchip power can only be accessed via pybindings because it cannot be enabled/disabled
|
||||
with pytest.raises(Exception):
|
||||
c.powers.VCHIP
|
||||
|
||||
# valid
|
||||
c.powers
|
||||
c.powers.VA = 1200
|
||||
assert c.powers.VA == 1200
|
||||
assert c.powers.VA.dac == 1200
|
||||
|
||||
c.powers.VA.enable()
|
||||
assert c.powers.VA.enabled == True
|
||||
|
||||
c.setPowerEnabled([powerIndex.V_POWER_B, powerIndex.V_POWER_C], True)
|
||||
assert c.powers.VB.enabled == True
|
||||
assert c.powers.VC.enabled == True
|
||||
|
||||
c.powers.VA = 1500
|
||||
assert c.powers.VA == 1500
|
||||
assert c.powers.VA.dac == 1500
|
||||
|
||||
# change power name and test same value
|
||||
temp = c.powers.VB
|
||||
c.powerlist = ["VA", "m_VB", "VC", "VD", "VIO"]
|
||||
assert c.powers.m_VB.enabled == True
|
||||
assert c.powers.m_VB == temp
|
||||
|
||||
# restore previous value
|
||||
for power in c.getPowerList():
|
||||
c.setPowerDAC(power, prev_val_dac[power])
|
||||
c.setPowerEnabled([power], prev_val[power])
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
c.powerlist
|
||||
assert "only for CTB" in str(exc_info.value)
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_adclist(session_simulator, request):
|
||||
"""Test ADC list."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
from slsdet import Ctb
|
||||
c = Ctb()
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
c.adclist
|
||||
c.adclist = ["1", "2", "3", "test", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "30", "31", "32"]
|
||||
c.adclist
|
||||
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
c.adclist
|
||||
assert "only for CTB" in str(exc_info.value)
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_signallist(session_simulator, request):
|
||||
"""Test signal list."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
from slsdet import Ctb
|
||||
c = Ctb()
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
c.signallist
|
||||
c.signallist = ["1", "2", "3", "test", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "30", "31", "32", "33", "34", "35", "36", "37", "38", "39", "40", "41", "42", "43", "44", "45", "46", "47", "48", "49", "50", "51", "52", "53", "54", "55", "56", "57", "58", "59", "60", "61", "62", "63", "64"]
|
||||
c.signallist
|
||||
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
c.signallist
|
||||
assert "only for CTB" in str(exc_info.value)
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_slowadc(session_simulator, request):
|
||||
"""Test slow ADC and slow adc list."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
|
||||
from slsdet import Ctb
|
||||
c = Ctb()
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
c.slowadc
|
||||
c.slowadc.SLOWADC5
|
||||
c.slowadclist = ["1", "2", "3", "test", "5", "6", "7", "8"]
|
||||
c.slowadc.test
|
||||
|
||||
else:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
c.signallist
|
||||
assert "only for CTB" in str(exc_info.value)
|
||||
|
||||
Log(LogLevel.INFOGREEN, f"✅ {request.node.name} passed")
|
||||
|
||||
|
||||
|
||||
|
||||
@pytest.mark.detectorintegration
|
||||
def test_dac(session_simulator, request):
|
||||
"""Test dac."""
|
||||
det_type, num_interfaces, num_mods, d = session_simulator
|
||||
assert d is not None
|
||||
from slsdet import dacIndex
|
||||
|
||||
if det_type in ['ctb', 'xilinx_ctb']:
|
||||
|
||||
from slsdet import Ctb
|
||||
c = Ctb()
|
||||
|
||||
# valid
|
||||
c.daclist
|
||||
c.dacvalues
|
||||
|
||||
# save previous value
|
||||
prev_val = {dac: c.getDAC(dac, False) for dac in c.getDacList()}
|
||||
prev_dac_list = c.daclist
|
||||
|
||||
# invalid
|
||||
invalid_assignments = [
|
||||
(c.dacs, "vb_comp", "1200"), # set random dac
|
||||
(c.dacs, "DAC18", "1200"), # set dac 18
|
||||
(c.dacs, "DAC0", "-1"),
|
||||
(c.dacs, "DAC0", "4096")
|
||||
]
|
||||
for obj, attr, value in invalid_assignments:
|
||||
with pytest.raises(Exception):
|
||||
setattr(obj, attr, value)
|
||||
|
||||
# valid
|
||||
c.dacs.DAC0 = 1200
|
||||
assert c.getDAC(dacIndex.DAC_0, False)[0] == 1200
|
||||
|
||||
c.dacs.DAC0 = 0
|
||||
assert c.dacs.DAC0[0] == 0
|
||||
|
||||
# restore previous value
|
||||
for dac in c.getDacList():
|
||||
c.setDAC(dac, prev_val[dac][0], False)
|
||||
c.daclist = prev_dac_list
|
||||
|
||||
else:
|
||||
with pytest.raises(Exception):
|
||||
d.dacs.DAC0
|
||||
|
||||
# valid
|
||||
d.daclist
|
||||
d.dacvalues
|
||||
|
||||
# remember first dac name and index to test later
|
||||
dacname = d.daclist[0]
|
||||
assert dacname
|
||||
dacIndex = d.getDacList()[0]
|
||||
|
||||
# save previous value
|
||||
prev_val = d.getDAC(dacIndex, False)
|
||||
|
||||
if det_type == 'eiger':
|
||||
from slsdet import Eiger
|
||||
c = Eiger()
|
||||
elif det_type == 'jungfrau':
|
||||
from slsdet import Jungfrau
|
||||
c = Jungfrau()
|
||||
elif det_type == 'gotthard2':
|
||||
from slsdet import Gotthard2
|
||||
c = Gotthard2()
|
||||
elif det_type == 'mythen3':
|
||||
from slsdet import Mythen3
|
||||
c = Mythen3()
|
||||
elif det_type == 'moench':
|
||||
from slsdet import Moench
|
||||
c = Moench()
|
||||
else:
|
||||
raise RuntimeError("Unknown detector type to test dac: " + det_type)
|
||||
# invalid checks
|
||||
invalid_assignments = [
|
||||
(c.dacs, "random", "1200"), # set random dac
|
||||
(c.dacs, "DAC0", "1200"), # set random dac
|
||||
(c.dacs, dacname, "-1"),
|
||||
(c.dacs, dacname, "4096")
|
||||
]
|
||||
for obj, attr, value in invalid_assignments:
|
||||
with pytest.raises(Exception):
|
||||
setattr(obj, attr, value)
|
||||
|
||||
# valid, have to use setattr because c is different for each detector
|
||||
# and we cannot hardcode the dac name
|
||||
setattr(c.dacs, dacname, 1200)
|
||||
assert c.getDAC(dacIndex, False)[0] == 1200
|
||||
setattr(c.dacs, dacname, 0)
|
||||
assert getattr(c.dacs, dacname)[0] == 0
|
||||
|
||||
# restore previous value
|
||||
for i in range(len(d)):
|
||||
d.setDAC(dacIndex, prev_val[i], False, [i])
|
||||
|
||||
|
||||
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
@@ -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)
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -257,7 +257,7 @@ def connectToVirtualServers(name, num_mods, ctb_object=False):
|
||||
counts_sec = 5
|
||||
while (counts_sec != 0):
|
||||
try:
|
||||
d.virtual = [num_mods, SERVER_START_PORTNO]
|
||||
d.virtual = [num_mods, SERVER_START_PORTNO] # sets the hostnames
|
||||
break
|
||||
except Exception as e:
|
||||
# stop server still not up, wait a bit longer
|
||||
@@ -288,13 +288,16 @@ def startReceiver(num_mods, fp, no_log_file = False, quiet_mode=False):
|
||||
def loadConfig(name, rx_hostname = 'localhost', settingsdir = None, log_file_fp = None, num_mods = 1, num_frames = 1, num_interfaces = 1):
|
||||
Log(LogLevel.INFO, 'Loading config', log_file_fp, True)
|
||||
try:
|
||||
d = connectToVirtualServers(name, num_mods)
|
||||
if name == 'ctb' or name == 'xilinx_ctb':
|
||||
d = connectToVirtualServers(name, num_mods, ctb_object=True)
|
||||
else:
|
||||
d = connectToVirtualServers(name, num_mods)
|
||||
|
||||
if name == 'jungfrau' or name == 'moench':
|
||||
d.numinterfaces = num_interfaces
|
||||
|
||||
d.udp_dstport = DEFAULT_UDP_DST_PORTNO
|
||||
if name == 'eiger' or num_interfaces == 2:
|
||||
if d.numinterfaces == 2:
|
||||
d.udp_dstport2 = DEFAULT_UDP_DST_PORTNO + 1
|
||||
|
||||
d.rx_hostname = rx_hostname
|
||||
|
||||
Reference in New Issue
Block a user