Ticket6777

common_tests/eurotherm.py

@@ -12,6 +12,7 @@
 SENSOR_DISCONNECTED_VALUE = 1529
 NONE_TXT_CALIBRATION_MAX_TEMPERATURE = 10000.0
 NONE_TXT_CALIBRATION_MIN_TEMPERATURE = 0.0
+SCALING = "1.0"
 
 TEST_VALUES = [-50, 0.1, 50, 3000]
 
@@ -23,9 +24,29 @@ class EurothermBaseTests(metaclass=abc.ABCMeta):
     """
     Tests for the Eurotherm temperature controller.
     """
+    @abc.abstractmethod
+    def get_device(self):
+        pass
+
+    @abc.abstractmethod
+    def get_emulator_device(self):
+        pass
+
+    @abc.abstractmethod
+    def _get_temperature_setter_wrapper(self):
+        pass
+
+    @abc.abstractmethod
+    def get_scaling(self):
+        pass
+
+    def get_prefix(self):
+        return "{}:A01".format(self.get_device())
+
     def setUp(self):
         self._setup_lewis_and_channel_access()
         self._reset_device_state()
+        self.ca_no_prefix = ChannelAccess()
 
     def _setup_lewis_and_channel_access(self):
         self._lewis, self._ioc = get_running_lewis_and_ioc("eurotherm", "EUROTHRM_01")
@@ -51,6 +72,7 @@ def _reset_device_state(self, sensor="A01"):
         self.ca.assert_that_pv_value_is_unchanged(f"{sensor}:TEMP", 5)
 
     def _set_setpoint_and_current_temperature(self, temperature, sensor="A01"):
+
         if IOCRegister.uses_rec_sim:
             self.ca.set_pv_value(f"{sensor}:SIM:TEMP:SP", temperature)
             self.ca.assert_that_pv_is(f"{sensor}:SIM:TEMP", temperature)
@@ -61,28 +83,30 @@ def _set_setpoint_and_current_temperature(self, temperature, sensor="A01"):
             self.ca.assert_that_pv_is_number(f"{sensor}:TEMP", temperature, 0.1, timeout=30)
             self._lewis.backdoor_set_on_device("ramp_setpoint_temperature", temperature)
             self.ca.assert_that_pv_is_number(f"{sensor}:TEMP:SP:RBV", temperature, 0.1, timeout=30)
-
+    
     def test_WHEN_read_rbv_temperature_THEN_rbv_value_is_same_as_backdoor(self):
-        expected_temperature = 10.0
+        expected_temperature = 10.0 
         self._set_setpoint_and_current_temperature(expected_temperature)
         self.ca.assert_that_pv_is("A01:RBV", expected_temperature)
 
     def test_GIVEN_a_sp_WHEN_sp_read_rbv_temperature_THEN_rbv_value_is_same_as_sp(self):
         expected_temperature = 10.0
         self.ca.assert_setting_setpoint_sets_readback(expected_temperature, "A01:SP:RBV", "A01:SP")
 
+
     def test_WHEN_set_ramp_rate_in_K_per_min_THEN_current_temperature_reaches_set_point_in_expected_time(self):
-        start_temperature = 5.0
+        start_temperature = 5.0 
         ramp_on = 1
         ramp_rate = 60.0
-        setpoint_temperature = 25.0
+        setpoint_temperature = 25.0 
 
         self._set_setpoint_and_current_temperature(start_temperature)
 
         self.ca.set_pv_value("A01:RATE:SP", ramp_rate)
         self.ca.assert_that_pv_is_number("A01:RATE", ramp_rate, 0.1)
         self.ca.set_pv_value("A01:RAMPON:SP", ramp_on)
-        self.ca.set_pv_value("A01:TEMP:SP", setpoint_temperature)
+        with self._get_temperature_setter_wrapper():
+            self.ca.set_pv_value("A01:TEMP:SP", setpoint_temperature)
 
         start = time.time()
         self.ca.assert_that_pv_is_number("A01:TEMP:SP:RBV", setpoint_temperature, tolerance=0.1, timeout=60)
@@ -134,13 +158,15 @@ def test_GIVEN_temperature_setpoint_followed_by_calibration_change_WHEN_same_set
         tolerance = 0.2
         self.ca.set_pv_value("A01:RAMPON:SP", 0)
         reset_calibration_file(self.ca, prefix="A01:")
-        self.ca.set_pv_value("A01:TEMP:SP", temperature)
+        with self._get_temperature_setter_wrapper():
+            self.ca.set_pv_value("A01:TEMP:SP", temperature)
         self.ca.assert_that_pv_is_number("A01:TEMP:SP:RBV", temperature, tolerance=tolerance, timeout=rbv_change_timeout)
         with use_calibration_file(self.ca, "C006.txt", prefix="A01:"):
             self.ca.assert_that_pv_is_not_number("A01:TEMP:SP:RBV", temperature, tolerance=tolerance, timeout=rbv_change_timeout)
 
             # Act
-            self.ca.set_pv_value("A01:TEMP:SP", temperature)
+            with self._get_temperature_setter_wrapper():
+                self.ca.set_pv_value("A01:TEMP:SP", temperature)
 
             # Assert
             self.ca.assert_that_pv_is_number("A01:TEMP:SP:RBV", temperature, tolerance=tolerance, timeout=rbv_change_timeout)
@@ -227,7 +253,8 @@ def test_WHEN_disconnected_THEN_in_alarm(self, record):
         with self._lewis.backdoor_simulate_disconnected_device():
             self.ca.assert_that_pv_alarm_is(record, ChannelAccess.Alarms.INVALID, timeout=30)
         # Assert alarms clear on reconnection
-        self.ca.assert_that_pv_alarm_is(record, ChannelAccess.Alarms.NONE, timeout=30)
+        with self._get_temperature_setter_wrapper():
+            self.ca.assert_that_pv_alarm_is(record, ChannelAccess.Alarms.NONE, timeout=30)
 
     @parameterized.expand(parameterized_list(PID_TEST_VALUES))
     @skip_if_recsim("Backdoor not available in recsim")

tests/eurotherm_eibisynch.py

@@ -1,4 +1,5 @@
 import unittest
+import contextlib
 
 from parameterized import parameterized
 
@@ -11,6 +12,7 @@
 
 DEVICE = "EUROTHRM_01"
 EMULATOR = "eurotherm"
+SCALING = "1.0"
 
 IOCS = [
     {
@@ -39,6 +41,23 @@
 
 
 class EurothermTests(EurothermBaseTests, unittest.TestCase):
+    def setUp(self):
+        super(EurothermTests, self).setUp()
+        self._lewis.backdoor_set_on_device("scaling", 1.0 / float(SCALING))
+
+    def get_device(self):
+        return DEVICE
+
+    def get_emulator_device(self):
+        return EMULATOR
+
+    def get_scaling(self):
+        return SCALING
+
+    def _get_temperature_setter_wrapper(self):
+        return contextlib.nullcontext()
+
+
     @parameterized.expand([
         ("over_range_calc_pv_is_over_range", NONE_TXT_CALIBRATION_MAX_TEMPERATURE + 5.0, 1.0),
         ("over_range_calc_pv_is_within_range", NONE_TXT_CALIBRATION_MAX_TEMPERATURE - 200, 0.0),

tests/eurotherm_modbus.py

@@ -1,15 +1,16 @@
 import unittest
+import contextlib
 
 from parameterized import parameterized
 
 from utils.test_modes import TestModes
 from utils.ioc_launcher import get_default_ioc_dir, ProcServLauncher
 from common_tests.eurotherm import EurothermBaseTests, PID_TEST_VALUES
-
 from utils.testing import parameterized_list
 
 DEVICE = "EUROTHRM_01"
 EMULATOR = "eurotherm"
+SCALING = "0.1"
 
 IOCS = [
     {
@@ -18,6 +19,7 @@
         "ioc_launcher_class": ProcServLauncher,
         "macros": {
             "COMMS_MODE": "modbus",
+            "NEEDLE_VALVE": "no",
             "ADDR_1": "01",
             "ADDR_2": "",
             "ADDR_3": "",
@@ -28,11 +30,12 @@
             "ADDR_8": "",
             "ADDR_9": "",
             "ADDR_10": "",
-            "TEMP_SCALING_1": "0.1",
+            "TEMP_SCALING_1": SCALING,
             "P_SCALING_1": "1",
             "I_SCALING_1": "1",
             "D_SCALING_1": "1",
-            "OUTPUT_SCALING_1": "0.1",
+            "OUTPUT_SCALING_1": SCALING,
+            "NEEDLE_VALVE": "no"
         },
         "emulator": EMULATOR,
         "lewis_protocol": "eurotherm_modbus",
@@ -42,13 +45,28 @@
 
 TEST_MODES = [TestModes.DEVSIM]
 
-
 class EurothermModbusTests(EurothermBaseTests, unittest.TestCase):
+    def setUp(self):
+        super(EurothermModbusTests, self).setUp()
+        self._lewis.backdoor_set_on_device("scaling", 1.0 / float(SCALING))
+
+    def get_device(self):
+        return DEVICE
+
+    def get_emulator_device(self):
+        return EMULATOR
+
+    def get_scaling(self):
+        return SCALING
+
+    def _get_temperature_setter_wrapper(self):
+        return contextlib.nullcontext()
+
     def test_WHEN_autotune_set_THEN_readback_updates(self):
         for state in [0, 1]:
             self.ca.set_pv_value("A01:AUTOTUNE:SP", state)
             self.ca.assert_that_pv_is("A01:AUTOTUNE", state)
-
+            
     @parameterized.expand(parameterized_list(PID_TEST_VALUES))
     def test_WHEN_p_set_THEN_p_updates(self, _, val):
         self.ca.assert_setting_setpoint_sets_readback(value=val, readback_pv="A01:P", timeout=15)
@@ -64,3 +82,4 @@ def test_WHEN_d_set_THEN_d_updates(self, _, val):
     @parameterized.expand(parameterized_list([0, 0.5, 100]))
     def test_WHEN_max_output_set_THEN_max_output_updates(self, _, val):
         self.ca.assert_setting_setpoint_sets_readback(value=val, readback_pv="A01:MAX_OUTPUT", timeout=15)
+