Amint2 l convert

.gitignore

@@ -12,3 +12,7 @@ dllPath.bat
 runIOC.bat
 runIOC.sh
 relPaths.sh
+__pycache__/
+*.py[cod]
+*$py.class
+/test-reports/

Makefile

@@ -27,3 +27,6 @@ iocBoot_DEPEND_DIRS += $(filter %App,$(DIRS))
 # Add any additional dependency rules here:
 
 include $(TOP)/configure/RULES_TOP
+
+ioctests:
+	.\system_tests\run_tests.bat

system_tests/lewis_emulators/__init__.py

@@ -0,0 +1,2 @@
+# DO NOT DELETE THIS FILE - LEWIS FRAMEWORK REQUIRES THE DIRECTORY TO BE IMPORTABLE
+from __future__ import absolute_import

system_tests/lewis_emulators/amint2l/__init__.py

@@ -0,0 +1,5 @@
+from .device import SimulatedAmint2l
+from ..lewis_versions import LEWIS_LATEST
+
+framework_version = LEWIS_LATEST
+__all__ = ['SimulatedAmint2l']

system_tests/lewis_emulators/amint2l/device.py

@@ -0,0 +1,36 @@
+from collections import OrderedDict
+
+from lewis.devices import StateMachineDevice
+from .states import DefaultState
+
+
+class SimulatedAmint2l(StateMachineDevice):
+    """
+    Simulated AM Int2-L pressure transducer.
+    """
+
+    def _initialize_data(self):
+        """
+        Sets the initial state of the device.
+        """
+        self.connected = True
+        self.pressure = 2.0
+        self.address = "AB"
+
+    def _get_state_handlers(self):
+        """
+        Returns: states and their names
+        """
+        return {DefaultState.NAME: DefaultState()}
+
+    def _get_initial_state(self):
+        """
+        Returns: the name of the initial state
+        """
+        return DefaultState.NAME
+
+    def _get_transition_handlers(self):
+        """
+        Returns: the state transitions
+        """
+        return OrderedDict()

system_tests/lewis_emulators/amint2l/interfaces/__init__.py

@@ -0,0 +1,3 @@
+from .stream_interface import Amint2lStreamInterface
+
+__all__ = ['Amint2lStreamInterface']

system_tests/lewis_emulators/amint2l/interfaces/stream_interface.py

@@ -0,0 +1,63 @@
+"""
+Stream device for amint2l
+"""
+
+from lewis.adapters.stream import StreamInterface
+from lewis.core.logging import has_log
+
+from lewis.utils.command_builder import CmdBuilder
+from lewis.utils.replies import conditional_reply
+
+if_connected = conditional_reply("connected")
+
+
+@has_log
+class Amint2lStreamInterface(StreamInterface):
+    """
+    Stream interface for the serial port
+    """
+
+    in_terminator = chr(3)
+    out_terminator = chr(3)
+
+    def __init__(self):
+
+        super(Amint2lStreamInterface, self).__init__()
+        self.commands = {
+            CmdBuilder(self.get_pressure).stx().arg("[A-Fa-f0-9]+").escape("r").build()
+        }
+
+    @if_connected
+    def handle_error(self, request, error):
+        """
+        If command is not recognised print and error
+
+        Args:
+            request: requested string
+            error: problem
+
+        """
+        self.log.error("An error occurred at request " + repr(request) + ": " + repr(error))
+
+    @if_connected
+    def get_pressure(self, address):
+
+        """
+        Gets the current pressure
+
+        :param address: address of request
+        Returns: pressure in correct format if pressure has a value; if None returns None as if it is disconnected
+
+        """
+        if address.upper() != self._device.address.upper():
+            self.log.error("unknown address {0}".format(address))
+            return None
+        self.log.info("Pressure: {0}".format(self._device.pressure))
+        if self._device.pressure is None:
+            return None
+        else:
+            try:
+                return "{stx}{pressure:+8.3f}".format(stx=chr(2), pressure=self._device.pressure)
+            except ValueError:
+                # pressure contains string probably OR (over range) or UR (under range)
+                return "{stx}{pressure:8s}".format(stx=chr(2), pressure=self._device.pressure)

system_tests/lewis_emulators/amint2l/states.py

@@ -0,0 +1,8 @@
+from lewis.core.statemachine import State
+
+
+class DefaultState(State):
+    """
+    Device is in default state.
+    """
+    NAME = 'Default'

system_tests/lewis_emulators/lewis_versions.py

@@ -0,0 +1,2 @@
+LEWIS_1_3_0 = "1.3.0"
+LEWIS_LATEST = LEWIS_1_3_0

system_tests/run_tests.bat

@@ -0,0 +1,13 @@
+@echo off
+REM Run this directory's tests using the IOC Testing Framework
+
+SET CurrentDir=%~dp0
+
+call "%~dp0..\..\..\..\config_env.bat"
+
+set "PYTHONUNBUFFERED=1"
+
+REM Command line arguments always passed to the test script
+SET ARGS=--test_and_emulator %~dp0
+call %PYTHON3% "%EPICS_KIT_ROOT%\support\IocTestFramework\master\run_tests.py" %ARGS% %*
+IF %ERRORLEVEL% NEQ 0 EXIT /b %errorlevel%

system_tests/tests/amint2l.py

@@ -0,0 +1,94 @@
+import unittest
+
+from utils.test_modes import TestModes
+from utils.channel_access import ChannelAccess
+from utils.ioc_launcher import get_default_ioc_dir
+from utils.testing import skip_if_recsim, get_running_lewis_and_ioc
+
+# Internal Address of device (must be 2 characters)
+ADDRESS = "01"
+
+# Device prefix
+DEVICE_PREFIX = "AMINT2L_01"
+
+IOCS = [
+    {
+        "name": DEVICE_PREFIX,
+        "directory": get_default_ioc_dir("AMINT2L"),
+        "macros": {
+            "ADDR": ADDRESS,
+        },
+        "emulator": "amint2l",
+    },
+]
+
+
+TEST_MODES = [TestModes.RECSIM, TestModes.DEVSIM]
+
+
+class Amint2lTests(unittest.TestCase):
+    """
+    Tests for the AM Int2-L.
+    """
+
+    def setUp(self):
+        self._lewis, self._ioc = get_running_lewis_and_ioc("amint2l", DEVICE_PREFIX)
+        self.assertIsNotNone(self._lewis)
+        self.assertIsNotNone(self._ioc)
+
+        self.ca = ChannelAccess(device_prefix=DEVICE_PREFIX)
+        self._lewis.backdoor_set_on_device('connected', True)
+        self._lewis.backdoor_set_on_device("address", ADDRESS)
+
+    def _set_pressure(self, expected_pressure):
+        self._lewis.backdoor_set_on_device("pressure", expected_pressure)
+        self._ioc.set_simulated_value("SIM:PRESSURE", expected_pressure)
+
+    def test_GIVEN_pressure_set_WHEN_read_THEN_pressure_is_as_expected(self):
+        expected_pressure = 1.23
+        self._set_pressure(expected_pressure)
+
+        self.ca.assert_that_pv_is("PRESSURE", expected_pressure)
+        self.ca.assert_that_pv_alarm_is("PRESSURE", self.ca.Alarms.NONE)
+        self.ca.assert_that_pv_is("RANGE:ERROR", "No Error")
+
+    def test_GIVEN_negative_pressure_set_WHEN_read_THEN_pressure_is_as_expected(self):
+        expected_pressure = -123.34
+        self._set_pressure(expected_pressure)
+
+        self.ca.assert_that_pv_is("PRESSURE", expected_pressure)
+
+    def test_GIVEN_pressure_with_no_decimal_places_set_WHEN_read_THEN_pressure_is_as_expected(self):
+        expected_pressure = 7
+        self._set_pressure(expected_pressure)
+
+        self.ca.assert_that_pv_is("PRESSURE", expected_pressure)
+
+    @skip_if_recsim("In rec sim this test fails")
+    def test_GIVEN_pressure_over_range_set_WHEN_read_THEN_error(self):
+        expected_pressure = "OR"
+        self._set_pressure(expected_pressure)
+
+        self.ca.assert_that_pv_alarm_is("PRESSURE", self.ca.Alarms.INVALID)
+        self.ca.assert_that_pv_is("RANGE:ERROR", "Over Range")
+
+    @skip_if_recsim("In rec sim this test fails")
+    def test_GIVEN_pressure_under_range_set_WHEN_read_THEN_error(self):
+        expected_pressure = "UR"
+        self._set_pressure(expected_pressure)
+
+        self.ca.assert_that_pv_alarm_is("PRESSURE", self.ca.Alarms.INVALID)
+        self.ca.assert_that_pv_is("RANGE:ERROR", "Under Range")
+
+    @skip_if_recsim("In rec sim this test fails")
+    def test_GIVEN_device_disconnected_WHEN_read_THEN_pv_shows_disconnect(self):
+        self._lewis.backdoor_set_on_device("pressure", None)
+        # Setting none simulates no response from device which is like pulling the serial cable. Disconnecting the
+        # emulator using the backdoor makes the record go udf not timeout which is what the actual device does.
+
+        self.ca.assert_that_pv_alarm_is("PRESSURE", self.ca.Alarms.INVALID)
+
+    @skip_if_recsim("Can not test disconnection in recsim")
+    def test_GIVEN_device_not_connected_WHEN_get_pressure_THEN_alarm(self):
+        self._lewis.backdoor_set_on_device('connected', False)
+        self.ca.assert_that_pv_alarm_is('PRESSURE', ChannelAccess.Alarms.INVALID)