Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion common_tests/DAQmx.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
ICPCONFIGNAME = "DAQMX"


class DAQmxTests(object):
class DAQmxTests:
"""
General tests for the DAQmx.
"""
Expand Down
41 changes: 20 additions & 21 deletions common_tests/danfysik.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
HAS_TRIPPED = {True: "Tripped", False: "OK"}


class DanfysikBase(object):
class DanfysikBase:
"""
Tests for danfysik.
"""
Expand Down Expand Up @@ -87,70 +87,69 @@ def _deactivate_interlocks(self):
"""
Most danfysiks have interlocks deactivated on startup anyway
"""
pass

@parameterized.expand(parameterized_list(["VOLT", "CURR"]))
@skip_if_recsim("Can not test disconnection in rec sim")
def test_GIVEN_device_not_connected_WHEN_pv_checked_THEN_pv_in_alarm(self, _, pv):
for id_prefix in self.id_prefixes:
self.ca.assert_that_pv_alarm_is_not(
"{}{}".format(id_prefix, pv), ChannelAccess.Alarms.INVALID, timeout=30
f"{id_prefix}{pv}", ChannelAccess.Alarms.INVALID, timeout=30
)

with self._lewis.backdoor_simulate_disconnected_device():
for id_prefix in self.id_prefixes:
self.ca.assert_that_pv_alarm_is(
"{}{}".format(id_prefix, pv), ChannelAccess.Alarms.INVALID, timeout=30
f"{id_prefix}{pv}", ChannelAccess.Alarms.INVALID, timeout=30
)

for id_prefix in self.id_prefixes:
self.ca.assert_that_pv_alarm_is_not(
"{}{}".format(id_prefix, pv), ChannelAccess.Alarms.INVALID, timeout=30
f"{id_prefix}{pv}", ChannelAccess.Alarms.INVALID, timeout=30
)

def test_WHEN_polarity_setpoint_is_set_THEN_readback_updates_with_set_value(self):
for id_prefix in self.id_prefixes:
for pol in POLARITIES:
self.ca.assert_setting_setpoint_sets_readback(pol, "{}POL".format(id_prefix))
self.ca.assert_setting_setpoint_sets_readback(pol, f"{id_prefix}POL")

def test_WHEN_polarity_setpoint_is_set_with_number_THEN_readback_updates_with_set_value(self):
for id_prefix in self.id_prefixes:
for pol_num, pol in enumerate(POLARITIES):
self.ca.assert_setting_setpoint_sets_readback(
pol_num, "{}POL".format(id_prefix), "{}POL:SP".format(id_prefix), pol
pol_num, f"{id_prefix}POL", f"{id_prefix}POL:SP", pol
)

@skip_if_recsim("Recsim is not set up properly for this test to work")
def test_WHEN_power_setpoint_is_set_THEN_readback_updates_with_set_value(self):
for id_prefix in self.id_prefixes:
for state in POWER_STATES:
self.ca.assert_setting_setpoint_sets_readback(state, "{}POWER".format(id_prefix))
self.ca.assert_setting_setpoint_sets_readback(state, f"{id_prefix}POWER")

@skip_if_recsim("Recsim is not set up properly for this test to work")
def test_WHEN_power_setpoint_is_set_THEN_readback_updates_with_set_value(self):
def test_WHEN_power_setpoint_is_set_THEN_readback_updates_with_set_value_sp(self):
for id_prefix in self.id_prefixes:
for state_num, state in enumerate(POWER_STATES):
self.ca.assert_setting_setpoint_sets_readback(
state_num, "{}POWER".format(id_prefix), "{}POWER:SP".format(id_prefix), state
state_num, f"{id_prefix}POWER", f"{id_prefix}POWER:SP", state
)

def test_WHEN_current_setpoint_is_set_THEN_current_readback_updates_to_set_value(self):
for id_prefix in self.id_prefixes:
for curr in TEST_CURRENTS:
self.ca.set_pv_value("{}CURR:SP".format(id_prefix), curr)
self.ca.set_pv_value(f"{id_prefix}CURR:SP", curr)
expected_value = (
curr * self.current_readback_factor if IOCRegister.uses_rec_sim else curr
)
self.ca.assert_that_pv_is_number(
"{}CURR".format(id_prefix), expected_value, tolerance=0.5
f"{id_prefix}CURR", expected_value, tolerance=0.5
) # Tolerance 0.5 because readback is integer

@skip_if_devsim("In dev sim this test fails as the simulated records are not used")
def test_GIVEN_emulator_not_in_use_WHEN_voltage_is_read_THEN_value_is_as_expected(self):
expected_value = 12
for id_prefix in self.id_prefixes:
self.ca.set_pv_value("{}SIM:VOLT".format(id_prefix), expected_value)
self.ca.assert_that_pv_is("{}VOLT".format(id_prefix), expected_value)
self.ca.set_pv_value(f"{id_prefix}SIM:VOLT", expected_value)
self.ca.assert_that_pv_is(f"{id_prefix}VOLT", expected_value)

@skip_if_recsim("Recsim is unable to simulate comms being uninitialized")
def test_GIVEN_power_supply_comms_become_uninitialized_THEN_ioc_recovers(self):
Expand All @@ -161,7 +160,7 @@ def test_GIVEN_power_supply_comms_become_uninitialized_THEN_ioc_recovers(self):
for id_prefix in self.id_prefixes:
# Should be able to re-initialize comms and read the new voltage
self.ca.assert_that_pv_is_number(
"{}VOLT".format(id_prefix), volt, tolerance=0.5, timeout=30
f"{id_prefix}VOLT", volt, tolerance=0.5, timeout=30
)

finally:
Expand All @@ -171,19 +170,19 @@ def test_GIVEN_power_supply_comms_become_uninitialized_THEN_ioc_recovers(self):
def test_GIVEN_no_interlocks_active_WHEN_getting_overall_interlock_status_THEN_it_is_ok(self):
self._deactivate_interlocks()
for id_prefix in self.id_prefixes:
self.ca.assert_that_pv_is("{}ILK".format(id_prefix), HAS_TRIPPED[False])
self.ca.assert_that_pv_is(f"{id_prefix}ILK", HAS_TRIPPED[False])

@skip_if_recsim(
"In rec sim this test fails as recsim does not set any of the related values "
"which are set by the emulator"
)
def test_WHEN_reset_is_sent_THEN_readbacks_and_power_are_off(self):
for id_prefix in self.id_prefixes:
self.ca.set_pv_value("{}CURR:SP".format(id_prefix), 5)
self.ca.set_pv_value("{}RESET".format(id_prefix), 1)
self.ca.assert_that_pv_is("{}POWER".format(id_prefix), "Off")
self.ca.assert_that_pv_is("{}CURR".format(id_prefix), 0)
self.ca.assert_that_pv_is("{}VOLT".format(id_prefix), 0)
self.ca.set_pv_value(f"{id_prefix}CURR:SP", 5)
self.ca.set_pv_value(f"{id_prefix}RESET", 1)
self.ca.assert_that_pv_is(f"{id_prefix}POWER", "Off")
self.ca.assert_that_pv_is(f"{id_prefix}CURR", 0)
self.ca.assert_that_pv_is(f"{id_prefix}VOLT", 0)

def test_GIVEN_power_on_and_zero_sp_WHEN_enabling_auto_onoff_THEN_device_is_powered_off(self):
self.set_autoonoff(False)
Expand Down
15 changes: 6 additions & 9 deletions common_tests/eurotherm.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
import abc
import os
import contextlib
import time
import typing
import unittest
from typing import ContextManager

from parameterized import parameterized

Expand Down Expand Up @@ -49,15 +48,15 @@ def get_emulator_device(self) -> str:
pass

@abc.abstractmethod
def _get_temperature_setter_wrapper(self) -> ContextManager:
def _get_temperature_setter_wrapper(self) -> contextlib.AbstractContextManager:
pass

@abc.abstractmethod
def get_scaling(self) -> str:
pass

def get_prefix(self) -> str:
return "{}:A01".format(self.get_device())
return f"{self.get_device()}:A01"

def setUp(self):
self._setup_lewis_and_channel_access()
Expand Down Expand Up @@ -274,12 +273,10 @@ def _assert_units(self, units):

def _assert_using_mock_table_location(self):
for pv in ["A01:TEMP", "A01:TEMP:SP:CONV", "A01:TEMP:SP:RBV:CONV"]:
self.ca.assert_that_pv_is(
"{}.TDIR".format(pv), r"eurotherm2k/master/example_temp_sensor"
)
self.ca.assert_that_pv_is(f"{pv}.TDIR", r"eurotherm2k/master/example_temp_sensor")
self.ca.assert_that_pv_is_path(
"{}.BDIR".format(pv),
os.path.join(EPICS_TOP, "support").replace("\\", "/"),
f"{pv}.BDIR",
(EPICS_TOP / "support").as_posix(),
)

def test_WHEN_calibration_file_is_in_units_of_K_THEN_egu_of_temperature_pvs_is_K(
Expand Down
58 changes: 26 additions & 32 deletions common_tests/fermichopper.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
)


class ErrorStrings(object):
class ErrorStrings:
"""
Error messages that we expect to appear in the IOC log in various situations.
"""
Expand All @@ -34,22 +34,11 @@ class ErrorStrings(object):
CONTROLLER_OVERSPEED = "Controller reports speed limit exceeded"


class FermichopperBase(object, metaclass=ABCMeta):
class FermichopperBase(metaclass=ABCMeta):
"""
Tests for the Fermi Chopper IOC.
"""

valid_commands = ["0001", "0002", "0003", "0004", "0005"]

# Values that will be tested in the parametrized tests.
test_chopper_speeds = [100, 350, 600]
test_delay_durations = [0, 2.5, 18]
test_gatewidth_values = [0, 0.5, 5]
test_temperature_values = [20.0, 25.0, 37.5, 47.5]
test_current_values = [0, 1.37, 2.22]
test_voltage_values = [0, 282.9, 333.3]
test_autozero_values = [-5.0, -2.22, 0, 1.23, 5]

@abstractmethod
def _get_device_prefix(self):
pass
Expand All @@ -70,6 +59,17 @@ def setUp(self):
if not IOCRegister.uses_rec_sim:
self._lewis.backdoor_run_function_on_device("reset")

self.valid_commands = ["0001", "0002", "0003", "0004", "0005"]

# Values that will be tested in the parametrized tests.
self.test_chopper_speeds = [100, 350, 600]
self.test_delay_durations = [0, 2.5, 18]
self.test_gatewidth_values = [0, 0.5, 5]
self.test_temperature_values = [20.0, 25.0, 37.5, 47.5]
self.test_current_values = [0, 1.37, 2.22]
self.test_voltage_values = [0, 282.9, 333.3]
self.test_autozero_values = [-5.0, -2.22, 0, 1.23, 5]

def is_device_broken(self):
if IOCRegister.uses_rec_sim:
return False # In recsim, assume device is always ok
Expand Down Expand Up @@ -105,7 +105,7 @@ def test_WHEN_speed_setpoint_is_set_THEN_readback_updates(self):
@skip_if_recsim("Recsim does not handle this")
def test_WHEN_speed_setpoint_is_set_via_gui_pv_THEN_readback_updates(self):
for speed in self.test_chopper_speeds:
self.ca.set_pv_value("SPEED:SP:GUI", "{} Hz".format(speed))
self.ca.set_pv_value("SPEED:SP:GUI", f"{speed} Hz")
self.ca.assert_that_pv_is("SPEED:SP", speed)
self.ca.assert_that_pv_alarm_is("SPEED:SP", self.ca.Alarms.NONE)
self.ca.assert_that_pv_is("SPEED:SP:RBV", speed)
Expand All @@ -132,14 +132,12 @@ def test_WHEN_autozero_voltages_are_set_via_backdoor_THEN_pvs_update(self):
for number, boundary, value in itertools.product(
[1, 2], ["upper", "lower"], self.test_autozero_values
):
self._lewis.backdoor_set_on_device(
"autozero_{n}_{b}".format(n=number, b=boundary), value
)
self._lewis.backdoor_set_on_device(f"autozero_{number}_{boundary}", value)
self.ca.assert_that_pv_is_number(
"AUTOZERO:{n}:{b}".format(n=number, b=boundary.upper()), value, tolerance=0.05
f"AUTOZERO:{number}:{boundary.upper()}", value, tolerance=0.05
)
self.ca.assert_that_pv_alarm_is(
"AUTOZERO:{n}:{b}".format(n=number, b=boundary.upper()), self.ca.Alarms.NONE
f"AUTOZERO:{number}:{boundary.upper()}", self.ca.Alarms.NONE
)

@skip_if_recsim("In rec sim this test fails")
Expand Down Expand Up @@ -256,28 +254,26 @@ def test_GIVEN_autozero_voltages_are_out_of_range_WHEN_chopper_is_moving_THEN_ra
self._ioc, in_time=2, must_contain=ErrorStrings.CONTROLLER_AUTOZERO_OUT_OF_RANGE
):
# Set autozero voltage too high
self._lewis.backdoor_set_on_device(
"autozero_{n}_{p}".format(n=number, p=position), 3.2
)
self._lewis.backdoor_set_on_device(f"autozero_{number}_{position}", 3.2)
# Assert
self.ca.assert_that_pv_is("AUTOZERO:RANGECHECK", 1)

# Reset relevant autozero voltage back to zero
self._lewis.backdoor_set_on_device("autozero_{n}_{p}".format(n=number, p=position), 0)
self._lewis.backdoor_set_on_device(f"autozero_{number}_{position}", 0)
self.ca.assert_that_pv_is_number(
"AUTOZERO:{n}:{p}".format(n=number, p=position.upper()), 0, tolerance=0.1
f"AUTOZERO:{number}:{position.upper()}", 0, tolerance=0.1
)

@contextmanager
def _lie_about(self, lie):
if IOCRegister.uses_rec_sim:
raise IOError("Can't use lewis backdoor in recsim!")
raise OSError("Can't use lewis backdoor in recsim!")

self._lewis.backdoor_set_on_device("is_lying_about_{}".format(lie), True)
self._lewis.backdoor_set_on_device(f"is_lying_about_{lie}", True)
try:
yield
finally:
self._lewis.backdoor_set_on_device("is_lying_about_{}".format(lie), False)
self._lewis.backdoor_set_on_device(f"is_lying_about_{lie}", False)

def _lie_about_delay_setpoint_readback(self):
return self._lie_about("delay_sp_rbv")
Expand Down Expand Up @@ -440,16 +436,14 @@ def test_GIVEN_autozero_voltages_are_out_of_range_WHEN_chopper_is_moving_THEN_sw
self._ioc, in_time=2, must_contain=ErrorStrings.SOFTWARE_AUTOZERO_OUT_OF_RANGE
):
# Set autozero voltage too high and set device moving
self._lewis.backdoor_set_on_device(
"autozero_{n}_{p}".format(n=number, p=position), 3.2
)
self._lewis.backdoor_set_on_device(f"autozero_{number}_{position}", 3.2)
self._lewis.backdoor_set_on_device("speed", 7)

# Assert that "switch drive on and stop" was sent
self.ca.assert_that_pv_is("LASTCOMMAND", "0001")

# Reset relevant autozero voltage back to zero
self._lewis.backdoor_set_on_device("autozero_{n}_{p}".format(n=number, p=position), 0)
self._lewis.backdoor_set_on_device(f"autozero_{number}_{position}", 0)
self.ca.assert_that_pv_is_number(
"AUTOZERO:{n}:{p}".format(n=number, p=position.upper()), 0, tolerance=0.1
f"AUTOZERO:{number}:{position.upper()}", 0, tolerance=0.1
)
Loading
Loading