Ticket8359 create p4p wrapper

README.md

@@ -104,6 +104,14 @@ you can use:
 
 >  `python -u run_tests.py  --test_and_emulator C:\Instrument\Apps\EPICS\support\CCD100\master\system_tests`
 
+### Run tests using PV Access
+
+To run tests using PV Access (via P4P) instead of Channel Access, you can use:
+
+>  `python -u run_tests.py -pva`
+or
+>  `python -u run_tests.py --pv-access`
+
 ## Troubleshooting
 
 If all tests are failing then it is likely that the PV prefix is incorrect.

global_settings.py

@@ -0,0 +1 @@
+DEFAULT_USE_PVA = False

requirements.txt

@@ -0,0 +1 @@
+genie_python>=15.1.0-rc1

run_tests.py

@@ -14,6 +14,7 @@
 import xmlrunner
 from genie_python.utilities import cleanup_subprocs_on_process_exit
 
+import global_settings
 from run_utils import ModuleTests, modified_environment, package_contents
 from utils.build_architectures import BuildArchitectures
 from utils.device_launcher import device_collection_launcher, device_launcher
@@ -49,7 +50,10 @@ def check_and_do_pre_ioc_launch_hook(ioc):
 
     :param ioc: A dictionary representing an ioc.
     """
-    do_nothing = lambda *args: None
+
+    def do_nothing(*args):
+        return None
+
     pre_ioc_launch_hook = ioc.get("pre_ioc_launch_hook", do_nothing)
     if callable(pre_ioc_launch_hook):
         pre_ioc_launch_hook()
@@ -176,8 +180,10 @@ def load_and_run_tests(
     test_results = []
 
     arch = get_build_architecture()
-    print("Running tests for arch {}".format(arch.name))
-
+    print(
+        f"Running tests for arch {arch.name}, "
+        f"defaulting to {'PV' if global_settings.DEFAULT_USE_PVA else 'Channel'} Access."
+    )
     for mode in modes:
         if tests_mode is not None and mode != tests_mode:
             continue
@@ -190,7 +196,9 @@ def load_and_run_tests(
             # Skip tests that cannot be run with a 32-bit architecture
             if arch not in module.architectures:
                 print(
-                    f"Skipped module tests.{module.name} in {TestModes.name(mode)}: suite not available with a {BuildArchitectures.archname(arch)} build architecture"
+                    f"Skipped module tests.{module.name} in {TestModes.name(mode)}: "
+                    f"suite not available with a {BuildArchitectures.archname(arch)} "
+                    f"build architecture"
                 )
                 continue
 
@@ -346,7 +354,6 @@ def run_tests(
     settings = {"EPICS_CA_ADDR_LIST": "127.255.255.255"}
 
     test_names = [f"tests.{test}" for test in tests_to_run]
-
     runner = xmlrunner.XMLTestRunner(
         output="test-reports", stream=sys.stdout, failfast=failfast_switch, verbosity=3
     )
@@ -452,6 +459,13 @@ def run_tests(
         help="""Specify a folder that holds both the tests (in a folder called tests) and a lewis 
                         emulator (in a folder called lewis_emulators).""",
     )
+    parser.add_argument(
+        "-pva",
+        "--pv-access",
+        action="store_true",
+        help="""Run tests using PV Access instead of Channel Access. (Note: tests can locally
+        override this).""",
+    )
 
     arguments = parser.parse_args()
 
@@ -499,7 +513,7 @@ def run_tests(
     failfast = arguments.failfast
     report_coverage = arguments.report_coverage
     ask_before_running_tests = arguments.ask_before_running
-
+    global_settings.DEFAULT_USE_PVA = arguments.pv_access
     tests_mode = None
     if arguments.tests_mode == "RECSIM":
         tests_mode = TestModes.RECSIM

run_utils.py

@@ -2,11 +2,14 @@
 import importlib
 import os
 from contextlib import contextmanager
+from types import ModuleType
+from typing import Generator, List, Optional, Set
 
 from utils.build_architectures import BuildArchitectures
+from utils.test_modes import TestModes
 
 
-def package_contents(package_path, filter_files):
+def package_contents(package_path: str, filter_files: str) -> Set[str]:
     """
     Finds all the files in a package.
 
@@ -24,7 +27,7 @@ def package_contents(package_path, filter_files):
 
 
 @contextmanager
-def modified_environment(**kwargs):
+def modified_environment(**kwargs: str) -> Generator[None, None, None]:
     """
     Modifies the environment variables as required then returns them to their original state.
 
@@ -53,49 +56,49 @@ class ModuleTests(object):
         modes: Modes to run the tests in.
     """
 
-    def __init__(self, name):
+    def __init__(self, name: str) -> None:
         self.__name = name
-        self.tests = None
+        self.tests: Optional[List[str]] = None
         self.__file = self.__get_file_reference()
         self.__modes = self.__get_modes()
         self.__architectures = self.__get_architectures()
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Returns the name of the module."""
         return self.__name
 
     @property
-    def modes(self):
+    def modes(self) -> Set[TestModes]:
         """Returns the modes to run the tests in."""
         return self.__modes
 
     @property
-    def file(self):
+    def file(self) -> ModuleType:
         """Returns a reference to the module file."""
         return self.__file
 
     @property
-    def architectures(self):
+    def architectures(self) -> Set[BuildArchitectures]:
         """Returns the architectures the test can be run in."""
         return self.__architectures
 
-    def __get_file_reference(self):
+    def __get_file_reference(self) -> ModuleType:
         module = load_module("tests.{}".format(self.__name))
         return module
 
-    def __get_modes(self):
+    def __get_modes(self) -> Set[TestModes]:
         if not self.__file:
             self.__get_file_reference()
         return check_test_modes(self.__file)
 
-    def __get_architectures(self):
+    def __get_architectures(self) -> Set[BuildArchitectures]:
         if not self.__file:
             self.__get_file_reference()
         return check_build_architectures(self.__file)
 
 
-def load_module(name):
+def load_module(name: str) -> ModuleType:
     """
     Loads a module based on its name.
 
@@ -107,7 +110,7 @@ def load_module(name):
     )
 
 
-def check_test_modes(module):
+def check_test_modes(module: ModuleType) -> Set[TestModes]:
     """
     Checks for RECSIM and DEVSIM test modes.
 
@@ -124,7 +127,7 @@ def check_test_modes(module):
     return modes
 
 
-def check_build_architectures(module):
+def check_build_architectures(module: ModuleType) -> Set[BuildArchitectures]:
     """
     Checks for which build architectures the test can run in.
     If not specified, default to both 64 and 32 bit allowed.

tests/galil.py

@@ -1,5 +1,6 @@
 import os
 import unittest
+from typing import Literal
 
 from utils.channel_access import ChannelAccess
 from utils.ioc_launcher import IOCRegister, get_default_ioc_dir
@@ -72,8 +73,6 @@
     "MOT:DMC{}:SEND_CMD_STR": "CN-1,-1",
     "MOT:DMC{}:LIMITTYPE_CMD": "NO",
     "MOT:DMC{}:HOMETYPE_CMD": "NO",
-    "MOT:DMC{}:LIMITTYPE_CMD": "NO",
-    "MOT:DMC{}:HOMETYPE_CMD": "NO",
 }
 
 # TEST_MODES = [TestModes.DEVSIM,TestModes.NOSIM]
@@ -90,21 +89,21 @@ class GalilTests(unittest.TestCase):
 
     def zero_motors(self):
         for motor in ["{:02d}".format(mtr) for mtr in range(1, self.num_motors + 1)]:
-            self.ca.set_pv_value("MOT:MTR{}{}".format(self.controller, motor), 0)
-            self.ca.assert_that_pv_is("MOT:MTR{}{}".format(self.controller, motor), 0)
-            self.ca.assert_that_pv_is("MOT:MTR{}{}.RBV".format(self.controller, motor), 0)
+            self.pv.set_pv_value("MOT:MTR{}{}".format(self.controller, motor), 0)
+            self.pv.assert_that_pv_is("MOT:MTR{}{}".format(self.controller, motor), 0)
+            self.pv.assert_that_pv_is("MOT:MTR{}{}.RBV".format(self.controller, motor), 0)
 
     def stop_motors(self):
         for motor in ["{:02d}".format(mtr) for mtr in range(1, self.num_motors + 1)]:
-            self.ca.set_pv_value("MOT:MTR{}{}.STOP".format(self.controller, motor), 1)
-            self.ca.assert_that_pv_is("MOT:MTR{}{}.DMOV".format(self.controller, motor), 1)
-            self.ca.assert_that_pv_is("MOT:MTR{}{}.MOVN".format(self.controller, motor), 0)
+            self.pv.set_pv_value("MOT:MTR{}{}.STOP".format(self.controller, motor), 1)
+            self.pv.assert_that_pv_is("MOT:MTR{}{}.DMOV".format(self.controller, motor), 1)
+            self.pv.assert_that_pv_is("MOT:MTR{}{}.MOVN".format(self.controller, motor), 0)
 
     def setUp(self):
         self._ioc = IOCRegister.get_running(DEVICE_PREFIX)
         self.assertIsNotNone(self._ioc)
 
-        self.ca = ChannelAccess(device_prefix=None, default_timeout=20, default_wait_time=0.0)
+        self.pv = ChannelAccess(device_prefix=None, default_timeout=20, default_wait_time=0.0)
         # test galil hardware does not currently have an encoder, software simulated motors do
         if IOCRegister.test_mode == TestModes.NOSIM:
             ueip = "No"
@@ -120,67 +119,69 @@ def test_GIVEN_ioc_started_THEN_pvs_for_all_motors_exist(self):
         check for real motors
         """
         for motor in ["{:02d}".format(mtr) for mtr in range(1, self.num_motors + 1)]:
-            self.ca.assert_that_pv_exists("MOT:MTR{}{}".format(self.controller, motor))
+            self.pv.assert_that_pv_exists("MOT:MTR{}{}".format(self.controller, motor))
 
     def test_GIVEN_ioc_started_THEN_axes_for_all_motors_exist(self):
         for motor in range(1, 8 + 1):
-            self.ca.assert_that_pv_exists("GALIL_01:AXIS{}".format(motor))
+            self.pv.assert_that_pv_exists("GALIL_01:AXIS{}".format(motor))
 
     def test_GIVEN_motor_requested_to_move_THEN_motor_moves(self):
         self.zero_motors()
 
         # Move motor 0101
         val = 3.0
-        self.ca.set_pv_value("MOT:MTR0101", val)
-        self.ca.assert_that_pv_is("MOT:MTR0101", val)
-        self.ca.assert_that_pv_is("MOT:MTR0101.RBV", val)
+        self.pv.set_pv_value("MOT:MTR0101", val)
+        self.pv.assert_that_pv_is("MOT:MTR0101", val)
+        self.pv.assert_that_pv_is("MOT:MTR0101.RBV", val)
 
     def test_GIVEN_axis_requested_to_move_THEN_axis_moves(self):
         self.zero_motors()
 
         # Move axis 1
         val = 4.0
-        self.ca.set_pv_value("GALIL_01:AXIS1:SP", val)
-        self.ca.assert_that_pv_is("GALIL_01:AXIS1:SP:RBV", val)
-        self.ca.assert_that_pv_is("GALIL_01:AXIS1", val)
-        self.ca.assert_that_pv_is("MOT:MTR0101", val)
-        self.ca.assert_that_pv_is("MOT:MTR0101.RBV", val)
+        self.pv.set_pv_value("GALIL_01:AXIS1:SP", val)
+        self.pv.assert_that_pv_is("GALIL_01:AXIS1:SP:RBV", val)
+        self.pv.assert_that_pv_is("GALIL_01:AXIS1", val)
+        self.pv.assert_that_pv_is("MOT:MTR0101", val)
+        self.pv.assert_that_pv_is("MOT:MTR0101.RBV", val)
 
     # @skip_always("Not working")
     @skip_if_nosim("No encoder on test real motor")
     def test_GIVEN_motors_THEN_check_motor_encoder_diff_works(self):
         val = 2.0
         # setup motor using encoder
-        self.ca.set_pv_value("MOT:MTR0101.UEIP", "Yes")
-        self.ca.assert_that_pv_is("MOT:MTR0101.UEIP", "Yes")
-        mres = self.ca.get_pv_value("MOT:MTR0101.MRES")
-        eres = self.ca.get_pv_value("MOT:MTR0101.ERES")
+        self.pv.set_pv_value("MOT:MTR0101.UEIP", "Yes")
+        self.pv.assert_that_pv_is("MOT:MTR0101.UEIP", "Yes")
+        mres = self.pv.get_pv_value("MOT:MTR0101.MRES")
+        eres = self.pv.get_pv_value("MOT:MTR0101.ERES")
+        assert isinstance(mres, float)
+        assert isinstance(eres, float)
         self.zero_motors()
 
         # move to initial position and check in step
-        self.ca.set_pv_value("MOT:MTR0101", val, wait=True)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101", val)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101.RBV", val, tolerance=eres)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101.RMP", val / mres, tolerance=mres)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101.REP", val / eres, tolerance=eres)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101_MTRENC_DIFF", 0.0, tolerance=eres)
+        self.pv.set_pv_value("MOT:MTR0101", val, wait=True)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101", val)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101.RBV", val, tolerance=eres)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101.RMP", val / mres, tolerance=mres)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101.REP", val / eres, tolerance=eres)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101_MTRENC_DIFF", 0.0, tolerance=eres)
 
         # now double encoder resolution so encoder now thinks it is at 2*val
         # giving difference (val - 2*val)
-        self.ca.set_pv_value("MOT:MTR0101.ERES", eres * 2.0, wait=True)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101.RMP", val / mres, tolerance=mres)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101.REP", val / eres, tolerance=eres)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101.RBV", 2.0 * val, tolerance=eres)
-        self.ca.assert_that_pv_is_number("MOT:MTR0101_MTRENC_DIFF", -val, tolerance=eres)
+        self.pv.set_pv_value("MOT:MTR0101.ERES", eres * 2.0, wait=True)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101.RMP", val / mres, tolerance=mres)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101.REP", val / eres, tolerance=eres)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101.RBV", 2.0 * val, tolerance=eres)
+        self.pv.assert_that_pv_is_number("MOT:MTR0101_MTRENC_DIFF", -val, tolerance=eres)
 
-    def setup_motors(self, ueip):
+    def setup_motors(self, ueip: Literal["Yes", "No"]):
         for key, value in CONTROLLER_SETUP.items():
-            self.ca.set_pv_value(key.format(self.controller), value)
-            self.ca.assert_that_pv_is(key.format(self.controller), value)
+            self.pv.set_pv_value(key.format(self.controller), value)
+            self.pv.assert_that_pv_is(key.format(self.controller), value)
 
         for motor in ["{:02d}".format(mtr) for mtr in range(1, self.num_motors + 1)]:
             for key, value in MOTOR_SETUP.items():
                 if isinstance(value, str):
                     value = value.format(ueip=ueip)
-                self.ca.set_pv_value(key.format(self.controller, motor), value)
-                self.ca.assert_that_pv_is(key.format(self.controller, motor), value)
+                self.pv.set_pv_value(key.format(self.controller, motor), value)
+                self.pv.assert_that_pv_is(key.format(self.controller, motor), value)

utils/__init__.py

@@ -0,0 +1 @@
+import global_settings as global_settings

utils/build_architectures.py

@@ -14,7 +14,7 @@ class BuildArchitectures(Enum):
     _32BIT = 2
 
     @staticmethod
-    def archname(arch: str) -> str:
+    def archname(arch: Enum) -> str:
         """
         Returns: nice name of architecture
         """