Skip to content

Commit

Permalink
formatted lsicorr and mclennan tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@LowriJenkins
LowriJenkins committed Jul 12, 2024
1 parent c2adcdb commit 9eb0875
Show file tree
Hide file tree
Showing 2 changed files with 143 additions and 122 deletions.
208 changes: 116 additions & 92 deletions tests/lsicorr.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,59 +24,62 @@
"macros": {
"FILEPATH": os.path.join(dir_path, "..", "test-reports"),
"ADDR": "127.0.0.1",
"FIRMWARE_REVISION": "0"
}
"FIRMWARE_REVISION": "0",
},
}

]


TEST_MODES = [TestModes.DEVSIM]

PV_NAMES = ["CORRELATIONTYPE",
"NORMALIZATION",
"MEASUREMENTDURATION",
"SWAPCHANNELS",
"SAMPLINGTIMEMULTIT",
"TRANSFERRATE",
"OVERLOADLIMIT",
"OVERLOADINTERVAL",
"ERRORMSG",
"EXPERIMENTNAME",
"OUTPUTFILE",
"START",
"STOP",
"CORRELATION_FUNCTION",
"LAGS",
"REPETITIONS",
"CURRENT_REPETITION",
"RUNNING",
"CONNECTED",
"SCATTERING_ANGLE",
"SAMPLE_TEMP",
"SOLVENT_VISCOSITY",
"SOLVENT_REFRACTIVE_INDEX",
"LASER_WAVELENGTH"]


SETTING_PVS = [("CORRELATIONTYPE", "CROSS"),
("NORMALIZATION", "SYMMETRIC"),
("MEASUREMENTDURATION", 30),
("SWAPCHANNELS", "ChB_ChA"),
("SAMPLINGTIMEMULTIT", "ns200"),
("TRANSFERRATE", "ms700"),
("OVERLOADLIMIT", 15),
("OVERLOADINTERVAL", 450),
("START", "YES"),
("STOP", "YES"),
("REPETITIONS", 5),
("SCATTERING_ANGLE", 110),
("SAMPLE_TEMP", 298),
("SOLVENT_VISCOSITY", 1),
("SOLVENT_REFRACTIVE_INDEX", 1.33),
("LASER_WAVELENGTH", 642),
("WAIT", 1000),
("MIN_TIME_LAG", 50)]
PV_NAMES = [
"CORRELATIONTYPE",
"NORMALIZATION",
"MEASUREMENTDURATION",
"SWAPCHANNELS",
"SAMPLINGTIMEMULTIT",
"TRANSFERRATE",
"OVERLOADLIMIT",
"OVERLOADINTERVAL",
"ERRORMSG",
"EXPERIMENTNAME",
"OUTPUTFILE",
"START",
"STOP",
"CORRELATION_FUNCTION",
"LAGS",
"REPETITIONS",
"CURRENT_REPETITION",
"RUNNING",
"CONNECTED",
"SCATTERING_ANGLE",
"SAMPLE_TEMP",
"SOLVENT_VISCOSITY",
"SOLVENT_REFRACTIVE_INDEX",
"LASER_WAVELENGTH",
]


SETTING_PVS = [
("CORRELATIONTYPE", "CROSS"),
("NORMALIZATION", "SYMMETRIC"),
("MEASUREMENTDURATION", 30),
("SWAPCHANNELS", "ChB_ChA"),
("SAMPLINGTIMEMULTIT", "ns200"),
("TRANSFERRATE", "ms700"),
("OVERLOADLIMIT", 15),
("OVERLOADINTERVAL", 450),
("START", "YES"),
("STOP", "YES"),
("REPETITIONS", 5),
("SCATTERING_ANGLE", 110),
("SAMPLE_TEMP", 298),
("SOLVENT_VISCOSITY", 1),
("SOLVENT_REFRACTIVE_INDEX", 1.33),
("LASER_WAVELENGTH", 642),
("WAIT", 1000),
("MIN_TIME_LAG", 50),
]


class LSITests(unittest.TestCase):
Expand All @@ -86,9 +89,11 @@ class LSITests(unittest.TestCase):

def setUp(self):

Check failure on line 90 in tests/lsicorr.py

View workflow job for this annotation

GitHub Actions / call-workflow / ruff

Ruff (D102) 

tests/lsicorr.py:90:9: D102 Missing docstring in public method
self._ioc = IOCRegister.get_running("LSI")
self.ca = ChannelAccess(default_timeout=30, device_prefix=DEVICE_PREFIX, default_wait_time=0.0)
self.ca.set_pv_value('WAIT', 0)
self.ca.assert_that_pv_is("TAKING_DATA", "NO", timeout=10)
self.ca = ChannelAccess(
default_timeout=30, device_prefix=DEVICE_PREFIX, default_wait_time=0.0
)
self.ca.set_pv_value("WAIT", 0)
self.ca.assert_that_pv_is("TAKING_DATA", "NO", timeout=10)

def test_GIVEN_setting_pv_WHEN_pv_written_to_THEN_new_value_read_back(self):

Check failure on line 98 in tests/lsicorr.py

View workflow job for this annotation

GitHub Actions / call-workflow / ruff

Ruff (N802) 

tests/lsicorr.py:98:9: N802 Function name `test_GIVEN_setting_pv_WHEN_pv_written_to_THEN_new_value_read_back` should be lowercase

Check failure on line 98 in tests/lsicorr.py

View workflow job for this annotation

GitHub Actions / call-workflow / ruff

Ruff (D102) 

tests/lsicorr.py:98:9: D102 Missing docstring in public method
pv_name = "MEASUREMENTDURATION"
Expand All @@ -104,7 +109,9 @@ def test_GIVEN_setting_pv_WHEN_pv_written_to_with_invalid_value_THEN_value_not_u
self.ca.set_pv_value(pv_name, -1)
self.ca.assert_that_pv_is_number(pv_name, original_value)

def test_GIVEN_integer_device_setting_WHEN_pv_written_to_with_a_float_THEN_value_is_rounded_before_setting(self):
def test_GIVEN_integer_device_setting_WHEN_pv_written_to_with_a_float_THEN_value_is_rounded_before_setting(
self,
):
pv_name = "MEASUREMENTDURATION"
new_value = 12.3

Expand All @@ -120,41 +127,58 @@ def test_GIVEN_monitor_on_setting_pv_WHEN_pv_changed_THEN_monitor_gets_updated(s
with self.ca.assert_that_pv_monitor_is(pv_name, expected_value):
self.ca.set_pv_value(pv_name + ":SP", new_value)

def test_GIVEN_invalid_value_for_setting_WHEN_setting_pv_written_THEN_status_pv_updates_with_error(self):
def test_GIVEN_invalid_value_for_setting_WHEN_setting_pv_written_THEN_status_pv_updates_with_error(
self,
):
setting_pv = "MEASUREMENTDURATION"
self.ca.set_pv_value(setting_pv, -1)
error_message = "LSI --- wrong value assigned to MeasurementDuration"

self.ca.assert_that_pv_is("ERRORMSG", error_message)

@parameterized.expand([
("NORMALIZATION", ("SYMMETRIC", "COMPENSATED")),
("SWAPCHANNELS", ("ChA_ChB", "ChB_ChA")),
("CORRELATIONTYPE", ("AUTO", "CROSS")),
("TRANSFERRATE", ("ms100", "ms150", "ms200", "ms250", "ms300", "ms400", "ms500", "ms600", "ms700")),
("SAMPLINGTIMEMULTIT", ("ns12_5", "ns200", "ns400", "ns800", "ns1600", "ns3200"))
])
def test_GIVEN_enum_setting_WHEN_setting_pv_written_to_THEN_new_value_read_back(self, pv, values):
@parameterized.expand(
[
("NORMALIZATION", ("SYMMETRIC", "COMPENSATED")),
("SWAPCHANNELS", ("ChA_ChB", "ChB_ChA")),
("CORRELATIONTYPE", ("AUTO", "CROSS")),
(
"TRANSFERRATE",
("ms100", "ms150", "ms200", "ms250", "ms300", "ms400", "ms500", "ms600", "ms700"),
),
("SAMPLINGTIMEMULTIT", ("ns12_5", "ns200", "ns400", "ns800", "ns1600", "ns3200")),
]
)
def test_GIVEN_enum_setting_WHEN_setting_pv_written_to_THEN_new_value_read_back(
self, pv, values
):
for value in values:
self.ca.set_pv_value(pv, value, sleep_after_set=0.0)
self.ca.assert_that_pv_is(pv, value)

@parameterized.expand([
("OVERLOADLIMIT", "Mcps"),
("SCATTERING_ANGLE", "degree"),
("SAMPLE_TEMP", "K"),
("SOLVENT_VISCOSITY", "mPas"),
("SOLVENT_REFRACTIVE_INDEX", ""),
("LASER_WAVELENGTH", "nm")
])
def test_GIVEN_pv_with_unit_WHEN_EGU_field_read_from_THEN_unit_returned(self, pv, expected_unit):
@parameterized.expand(
[
("OVERLOADLIMIT", "Mcps"),
("SCATTERING_ANGLE", "degree"),
("SAMPLE_TEMP", "K"),
("SOLVENT_VISCOSITY", "mPas"),
("SOLVENT_REFRACTIVE_INDEX", ""),
("LASER_WAVELENGTH", "nm"),
]
)
def test_GIVEN_pv_with_unit_WHEN_EGU_field_read_from_THEN_unit_returned(
self, pv, expected_unit
):
self.ca.assert_that_pv_is("{pv}.EGU".format(pv=pv), expected_unit)

@parameterized.expand([
("CORRELATION_FUNCTION", 400),
("LAGS", 400),
])
def test_GIVEN_array_pv_WHEN_NELM_field_read_THEN_length_of_array_returned(self, pv, expected_length):
@parameterized.expand(
[
("CORRELATION_FUNCTION", 400),
("LAGS", 400),
]
)
def test_GIVEN_array_pv_WHEN_NELM_field_read_THEN_length_of_array_returned(
self, pv, expected_length
):
self.ca.assert_that_pv_is_number("{pv}.NELM".format(pv=pv), expected_length)

@parameterized.expand(parameterized_list(SETTING_PVS))
Expand All @@ -169,25 +193,25 @@ def test_GIVEN_pv_name_THEN_val_field_exists_for_that_pv(self, _, pv):
def test_GIVEN_pv_WHEN_pv_read_THEN_pv_has_no_alarms(self, _, pv):
self.ca.assert_that_pv_alarm_is(pv, self.ca.Alarms.NONE)

@parameterized.expand(parameterized_list([
"CORRELATION_FUNCTION",
"LAGS"
]))
def test_GIVEN_start_pressed_WHEN_measurement_is_possible_THEN_correlation_and_lags_populated(self, _, pv):
@parameterized.expand(parameterized_list(["CORRELATION_FUNCTION", "LAGS"]))
def test_GIVEN_start_pressed_WHEN_measurement_is_possible_THEN_correlation_and_lags_populated(
self, _, pv
):
self.ca.assert_setting_setpoint_sets_readback(0, "MIN_TIME_LAG")
self.ca.assert_that_pv_is("TAKING_DATA", "NO", timeout=10)
self.ca.set_pv_value("START", 1, sleep_after_set=0.0)
array_size = self.ca.get_pv_value("{pv}.NELM".format(pv=pv))
test_data = np.linspace(0, array_size, array_size)
self.ca.assert_that_pv_value_causes_func_to_return_true(pv, lambda pv_value: np.allclose(pv_value, test_data))
self.ca.assert_that_pv_value_causes_func_to_return_true(
pv, lambda pv_value: np.allclose(pv_value, test_data)
)

@parameterized.expand (parameterized_list([
"CORRELATION_FUNCTION",
"LAGS"
]))
def test_GIVEN_start_pressed_WHEN_measurement_is_possible_THEN_lags_data_below_min_time_lag_calculated(self, _, pv):
@parameterized.expand(parameterized_list(["CORRELATION_FUNCTION", "LAGS"]))
def test_GIVEN_start_pressed_WHEN_measurement_is_possible_THEN_lags_data_below_min_time_lag_calculated(
self, _, pv
):
# Arrange
pv_name = 'MIN_TIME_LAG'
pv_name = "MIN_TIME_LAG"
min_time_lag = 50
self.ca.assert_setting_setpoint_sets_readback(min_time_lag, pv_name)
self.ca.assert_that_pv_is("TAKING_DATA", "NO", timeout=10)
Expand All @@ -197,13 +221,14 @@ def test_GIVEN_start_pressed_WHEN_measurement_is_possible_THEN_lags_data_below_m
array_size = self.ca.get_pv_value("{pv}.NELM".format(pv=pv))
test_data = np.linspace(0, array_size, array_size)
# Scale minimum time lag from nanoseconds to seconds before comparison with test data
min_time_lag_seconds = min_time_lag/1e9
min_time_lag_seconds = min_time_lag / 1e9
test_data = np.delete(test_data, np.argwhere(test_data < min_time_lag_seconds))
test_data.resize(array_size)

self.ca.assert_that_pv_value_causes_func_to_return_true(
pv, lambda pv_value: np.allclose(pv_value, test_data),
message=f"PV {pv} data not all close to test data.\n Test data: {test_data}"
pv,
lambda pv_value: np.allclose(pv_value, test_data),
message=f"PV {pv} data not all close to test data.\n Test data: {test_data}",
)

def test_GIVEN_start_pressed_WHEN_measurement_already_on_THEN_error_raised(self):
Expand All @@ -212,7 +237,7 @@ def test_GIVEN_start_pressed_WHEN_measurement_already_on_THEN_error_raised(self)

error_message = "LSI --- Cannot configure: Measurement active"
self.ca.assert_that_pv_is("ERRORMSG", error_message)

def test_GIVEN_wait_at_start_AND_wait_set_WHEN_start_run_THEN_waits_happen(self):
# Arrange
self.ca.set_pv_value("WAIT_AT_START", "YES")
Expand All @@ -227,7 +252,7 @@ def test_GIVEN_wait_at_start_AND_wait_set_WHEN_start_run_THEN_waits_happen(self)
self.ca.assert_that_pv_is("WAITING", "NO")
self.ca.assert_that_pv_is("RUNNING", "YES")
self.ca.assert_that_pv_is("RUNNING", "NO")

def test_GIVEN_wait_set_WHEN_start_run_THEN_waits_happen(self):
# Arrange
self.ca.set_pv_value("WAIT_AT_START", False)
Expand All @@ -244,4 +269,3 @@ def test_GIVEN_wait_set_WHEN_start_run_THEN_waits_happen(self):
self.ca.assert_that_pv_is("WAITING", "NO")
self.ca.assert_that_pv_is("RUNNING", "YES")
self.ca.assert_that_pv_is("RUNNING", "NO")

Loading

0 comments on commit 9eb0875

Please sign in to comment.