Gave each set of backend tests it's own AssertableController

This is needed since not all datatypes are supported on every backend.

Author: evvaaaa

src/fastcs/transport/epics/gui.py

@@ -1,4 +1,3 @@
-
 from pvi._format.dls import DLSFormatter
 from pvi.device import (
     LED,

src/fastcs/transport/epics/util.py

@@ -39,7 +39,6 @@
     "max_alarm": "HOPR",
     "znam": "ZNAM",
     "onam": "ONAM",
-    "shape": "length",
 }
 
 
@@ -50,12 +49,23 @@ def get_record_metadata_from_attribute(
 
 
 def get_record_metadata_from_datatype(datatype: DataType[T]) -> dict[str, str]:
-    return {
+    arguments = {
         DATATYPE_FIELD_TO_RECORD_FIELD[field]: value
         for field, value in asdict(datatype).items()
         if field in DATATYPE_FIELD_TO_RECORD_FIELD
     }
 
+    match datatype:
+        case WaveForm():
+            if len(datatype.shape) != 1:
+                raise TypeError(
+                    f"Unsupported shape {datatype.shape}, the EPICS backend only "
+                    "supports to 1D arrays"
+                )
+            arguments["length"] = datatype.shape[0]
+
+    return arguments
+
 
 def get_cast_method_to_epics_type(datatype: DataType[T]) -> Callable[[T], object]:
     match datatype:

tests/__init__.py

@@ -0,0 +1,111 @@
+import copy
+from contextlib import contextmanager
+from typing import Literal
+
+from pytest_mock import MockerFixture
+
+from fastcs.attributes import AttrR, Handler, Sender, Updater
+from fastcs.controller import Controller, SubController
+from fastcs.datatypes import Int
+from fastcs.wrappers import command, scan
+
+
+class TestUpdater(Updater):
+    update_period = 1
+
+    async def update(self, controller, attr):
+        print(f"{controller} update {attr}")
+
+
+class TestSender(Sender):
+    async def put(self, controller, attr, value):
+        print(f"{controller}: {attr} = {value}")
+
+
+class TestHandler(Handler, TestUpdater, TestSender):
+    pass
+
+
+class TestSubController(SubController):
+    read_int: AttrR = AttrR(Int(), handler=TestUpdater())
+
+
+class TestController(Controller):
+    def __init__(self) -> None:
+        super().__init__()
+
+        self._sub_controllers: list[TestSubController] = []
+        for index in range(1, 3):
+            controller = TestSubController()
+            self._sub_controllers.append(controller)
+            self.register_sub_controller(f"SubController{index:02d}", controller)
+
+    initialised = False
+    connected = False
+    count = 0
+
+    async def initialise(self) -> None:
+        self.initialised = True
+
+    async def connect(self) -> None:
+        self.connected = True
+
+    @command()
+    async def go(self):
+        pass
+
+    @scan(0.01)
+    async def counter(self):
+        self.count += 1
+
+
+class AssertableController(TestController):
+    def __init__(self, mocker: MockerFixture) -> None:
+        self.mocker = mocker
+        super().__init__()
+
+    @contextmanager
+    def assert_read_here(self, path: list[str]):
+        yield from self._assert_method(path, "get")
+
+    @contextmanager
+    def assert_write_here(self, path: list[str]):
+        yield from self._assert_method(path, "process")
+
+    @contextmanager
+    def assert_execute_here(self, path: list[str]):
+        yield from self._assert_method(path, "")
+
+    def _assert_method(self, path: list[str], method: Literal["get", "process", ""]):
+        """
+        This context manager can be used to confirm that a fastcs
+        controller's respective attribute or command methods are called
+        a single time within a context block
+        """
+        queue = copy.deepcopy(path)
+
+        # Navigate to subcontroller
+        controller = self
+        item_name = queue.pop(-1)
+        for item in queue:
+            controllers = controller.get_sub_controllers()
+            controller = controllers[item]
+
+        # create probe
+        if method:
+            attr = getattr(controller, item_name)
+            spy = self.mocker.spy(attr, method)
+        else:
+            spy = self.mocker.spy(controller, item_name)
+        initial = spy.call_count
+
+        try:
+            yield  # Enter context
+        except Exception as e:
+            raise e
+        else:  # Exit context
+            final = spy.call_count
+            assert final == initial + 1, (
+                f"Expected {'.'.join(path + [method] if method else path)} "
+                f"to be called once, but it was called {final - initial} times."
+            )

tests/assertable_controller.py

@@ -1,27 +1,45 @@
-import copy
-import enum
 import os
 import random
 import string
 import subprocess
 import time
-from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Literal
+from typing import Any
 
-import numpy as np
 import pytest
 from aioca import purge_channel_caches
-from pytest_mock import MockerFixture
 
-from fastcs.attributes import AttrR, AttrRW, AttrW, Handler, Sender, Updater
-from fastcs.controller import Controller, SubController
-from fastcs.datatypes import Bool, Enum, Float, Int, String, WaveForm
-from fastcs.wrappers import command, scan
+from fastcs.attributes import AttrR, AttrRW, AttrW
+from fastcs.datatypes import Bool, Float, Int, String
+from tests.assertable_controller import (
+    TestController,
+    TestHandler,
+    TestSender,
+    TestUpdater,
+)
 
 DATA_PATH = Path(__file__).parent / "data"
 
 
+class BackendTestController(TestController):
+    read_int: AttrR = AttrR(Int(), handler=TestUpdater())
+    read_write_int: AttrRW = AttrRW(Int(), handler=TestHandler())
+    read_write_float: AttrRW = AttrRW(Float())
+    read_bool: AttrR = AttrR(Bool())
+    write_bool: AttrW = AttrW(Bool(), handler=TestSender())
+    read_string: AttrRW = AttrRW(String())
+    big_enum: AttrR = AttrR(
+        Int(
+            allowed_values=list(range(17)),
+        ),
+    )
+
+
+@pytest.fixture
+def controller():
+    return BackendTestController()
+
+
 @pytest.fixture
 def data() -> Path:
     return DATA_PATH
@@ -45,131 +63,6 @@ def pytest_internalerror(excinfo: pytest.ExceptionInfo[Any]):
         raise excinfo.value
 
 
-class TestUpdater(Updater):
-    update_period = 1
-
-    async def update(self, controller, attr):
-        print(f"{controller} update {attr}")
-
-
-class TestSender(Sender):
-    async def put(self, controller, attr, value):
-        print(f"{controller}: {attr} = {value}")
-
-
-class TestHandler(Handler, TestUpdater, TestSender):
-    pass
-
-
-class TestSubController(SubController):
-    read_int: AttrR = AttrR(Int(), handler=TestUpdater())
-
-
-class TestController(Controller):
-    def __init__(self) -> None:
-        super().__init__()
-
-        self._sub_controllers: list[TestSubController] = []
-        for index in range(1, 3):
-            controller = TestSubController()
-            self._sub_controllers.append(controller)
-            self.register_sub_controller(f"SubController{index:02d}", controller)
-
-    read_int: AttrR = AttrR(Int(), handler=TestUpdater())
-    read_write_int: AttrRW = AttrRW(Int(), handler=TestHandler())
-    read_write_float: AttrRW = AttrRW(Float())
-    read_bool: AttrR = AttrR(Bool())
-    write_bool: AttrW = AttrW(Bool(), handler=TestSender())
-    read_string: AttrRW = AttrRW(String())
-    enum: AttrRW = AttrRW(Enum(enum.IntEnum("Enum", {"RED": 0, "GREEN": 1, "BLUE": 2})))
-    one_d_waveform: AttrRW = AttrRW(WaveForm(np.int32, (10,)))
-    two_d_waveform: AttrRW = AttrRW(WaveForm(np.int32, (10, 10)))
-    big_enum: AttrR = AttrR(
-        Int(
-            allowed_values=list(range(17)),
-        ),
-    )
-
-    initialised = False
-    connected = False
-    count = 0
-
-    async def initialise(self) -> None:
-        self.initialised = True
-
-    async def connect(self) -> None:
-        self.connected = True
-
-    @command()
-    async def go(self):
-        pass
-
-    @scan(0.01)
-    async def counter(self):
-        self.count += 1
-
-
-class AssertableController(TestController):
-    def __init__(self, mocker: MockerFixture) -> None:
-        super().__init__()
-        self.mocker = mocker
-
-    @contextmanager
-    def assert_read_here(self, path: list[str]):
-        yield from self._assert_method(path, "get")
-
-    @contextmanager
-    def assert_write_here(self, path: list[str]):
-        yield from self._assert_method(path, "process")
-
-    @contextmanager
-    def assert_execute_here(self, path: list[str]):
-        yield from self._assert_method(path, "")
-
-    def _assert_method(self, path: list[str], method: Literal["get", "process", ""]):
-        """
-        This context manager can be used to confirm that a fastcs
-        controller's respective attribute or command methods are called
-        a single time within a context block
-        """
-        queue = copy.deepcopy(path)
-
-        # Navigate to subcontroller
-        controller = self
-        item_name = queue.pop(-1)
-        for item in queue:
-            controllers = controller.get_sub_controllers()
-            controller = controllers[item]
-
-        # create probe
-        if method:
-            attr = getattr(controller, item_name)
-            spy = self.mocker.spy(attr, method)
-        else:
-            spy = self.mocker.spy(controller, item_name)
-        initial = spy.call_count
-        try:
-            yield  # Enter context
-        except Exception as e:
-            raise e
-        else:  # Exit context
-            final = spy.call_count
-            assert final == initial + 1, (
-                f"Expected {'.'.join(path + [method] if method else path)} "
-                f"to be called once, but it was called {final - initial} times."
-            )
-
-
-@pytest.fixture
-def controller():
-    return TestController()
-
-
-@pytest.fixture(scope="class")
-def assertable_controller(class_mocker: MockerFixture):
-    return AssertableController(class_mocker)
-
-
 PV_PREFIX = "".join(random.choice(string.ascii_lowercase) for _ in range(12))
 HERE = Path(os.path.dirname(os.path.abspath(__file__)))
 

tests/conftest.py

@@ -92,7 +92,7 @@ def test_over_defined_schema():
     error = (
         ""
         "Expected no more than 2 arguments for 'ManyArgs.__init__' "
-        "but received 3 as `(self, arg: test_launch.SomeConfig, too_many)`"
+        "but received 3 as `(self, arg: tests.test_launch.SomeConfig, too_many)`"
     )
 
     with pytest.raises(LaunchError) as exc_info:

tests/test_launch.py

@@ -1,14 +1,12 @@
 from pvi.device import (
     LED,
     ButtonPanel,
-    ComboBox,
     Group,
     SignalR,
     SignalRW,
     SignalW,
     SignalX,
     SubScreen,
-    TextFormat,
     TextRead,
     TextWrite,
     ToggleButton,
@@ -47,19 +45,12 @@ def test_get_components(controller):
             children=[
                 SignalR(
                     name="ReadInt",
-                    read_pv="DEVICE:SubController01:ReadInt",
+                    read_pv="DEVICE:SubController02:ReadInt",
                     read_widget=TextRead(),
                 )
             ],
         ),
         SignalR(name="BigEnum", read_pv="DEVICE:BigEnum", read_widget=TextRead()),
-        SignalRW(
-            name="Enum",
-            read_pv="DEVICE:Enum_RBV",
-            read_widget=TextRead(format=TextFormat.string),
-            write_pv="DEVICE:Enum",
-            write_widget=ComboBox(choices=["RED", "GREEN", "BLUE"]),
-        ),
         SignalR(name="ReadBool", read_pv="DEVICE:ReadBool", read_widget=LED()),
         SignalR(
             name="ReadInt",