Better tests, proper methods

Author: marcelldls

src/fastcs/backends/tango/dsr.py

@@ -43,7 +43,7 @@ def _wrap_updater_fset(
 ) -> Callable[[Any, Any], Any]:
     async def fset(tango_device: Device, val):
         tango_device.info_stream(f"called fset method: {attr_name}")
-        await attribute.process_without_display_update(val)
+        await attribute.process(val)
 
     return fset
 
@@ -100,7 +100,7 @@ def _wrap_command_f(
 ) -> Callable[..., Awaitable[None]]:
     async def _dynamic_f(tango_device: Device) -> None:
         tango_device.info_stream(f"called {controller} f method: {method_name}")
-        return await MethodType(method, controller)()
+        return await method.__get__(controller)()
 
     _dynamic_f.__name__ = method_name
     return _dynamic_f

tests/backends/tango/test_dsr.py

@@ -1,46 +1,63 @@
+import copy
+import re
+from typing import Any
+
 import pytest
 from pytest_mock import MockerFixture
 from tango import DevState
-from tango._tango import AttrWriteType, CmdArgType
+from tango.test_context import DeviceTestContext
+from tests.conftest import AssertableController
 
 from fastcs.attributes import AttrR
-from fastcs.backends.tango.dsr import _collect_dev_attributes, _collect_dev_commands
-from fastcs.datatypes import Bool, Int
-
-
-def test_collect_attributes(mapping):
-    attributes = _collect_dev_attributes(mapping)
-
-    # Check that attributes are created and of expected type
-    assert list(attributes.keys()) == [
-        "BigEnum",
-        "ReadBool",
-        "ReadInt",
-        "ReadWriteFloat",
-        "ReadWriteInt",
-        "StringEnum",
-        "WriteBool",
-    ]
-    assert attributes["ReadInt"].attr_write == AttrWriteType.READ
-    assert attributes["ReadInt"].attr_type == CmdArgType.DevLong64
-    assert attributes["StringEnum"].attr_write == AttrWriteType.READ_WRITE
-    assert attributes["StringEnum"].attr_type == CmdArgType.DevString
-    assert attributes["ReadWriteFloat"].attr_write == AttrWriteType.READ_WRITE
-    assert attributes["ReadWriteFloat"].attr_type == CmdArgType.DevDouble
-    assert attributes["WriteBool"].attr_write == AttrWriteType.WRITE
-    assert attributes["WriteBool"].attr_type == CmdArgType.DevBoolean
-
-
-@pytest.mark.asyncio
-async def test_collect_commands(mapping, mocker: MockerFixture):
-    commands = _collect_dev_commands(mapping)
-
-    # Check that command is created and it can be called
-    assert list(commands.keys()) == ["Go"]
-    await commands["Go"](mocker.MagicMock())
+from fastcs.backends.tango.backend import TangoBackend
+from fastcs.datatypes import Bool, Float, Int
+
+
+def pascal_2_snake(input: list[str]) -> list[str]:
+    snake_list = copy.deepcopy(input)
+    snake_list[-1] = re.sub(r"(?<!^)(?=[A-Z])", "_", snake_list[-1]).lower()
+    return snake_list
 
 
 class TestTangoDevice:
+    @pytest.fixture(scope="class", autouse=True)
+    def setup_class(self, class_mocker: MockerFixture):
+        self.controller = AssertableController(class_mocker)
+
+    @pytest.fixture(scope="class")
+    def tango_context(self):
+        device = TangoBackend(self.controller)._dsr._device
+        with DeviceTestContext(device) as proxy:
+            yield proxy
+
+    @pytest.fixture(scope="class")
+    def client_read(self, tango_context):
+        def _read_attribute(path: list[str], expected: Any):
+            attribute = "_".join(path)
+            with self.controller.assertPerformed(pascal_2_snake(path), "READ"):
+                result = tango_context.read_attribute(attribute).value
+            assert result == expected
+
+        return _read_attribute
+
+    @pytest.fixture(scope="class")
+    def client_write(self, tango_context):
+        def _write_attribute(path: list[str], expected: Any):
+            attribute = "_".join(path)
+            with self.controller.assertPerformed(pascal_2_snake(path), "WRITE"):
+                tango_context.write_attribute(attribute, expected)
+
+        return _write_attribute
+
+    @pytest.fixture(scope="class")
+    def client_exec(self, tango_context):
+        def _exec_command(path: list[str]):
+            command = "_".join(path)
+            with self.controller.assertPerformed(pascal_2_snake(path), "EXECUTE"):
+                tango_context.command_inout(command)
+
+        return _exec_command
+
     def test_list_attributes(self, tango_context):
         assert list(tango_context.get_attribute_list()) == [
             "BigEnum",
@@ -50,6 +67,8 @@ def test_list_attributes(self, tango_context):
             "ReadWriteInt",
             "StringEnum",
             "WriteBool",
+            "SubController01_ReadInt",
+            "SubController02_ReadInt",
             "State",
             "Status",
         ]
@@ -62,46 +81,44 @@ def test_list_commands(self, tango_context):
             "Status",
         ]
 
-    def test_read_int(self, tango_context):
-        result = tango_context.read_attribute("ReadInt")
-        assert result.value == AttrR(Int())._value
+    def test_state(self, tango_context):
+        assert tango_context.command_inout("State") == DevState.ON
+
+    def test_status(self, tango_context):
+        expected = "The device is in ON state."
+        assert tango_context.command_inout("Status") == expected
+
+    def test_read_int(self, client_read):
+        client_read(["ReadInt"], AttrR(Int())._value)
 
-    def test_read_write_int(self, tango_context):
-        expected = 42
-        tango_context.write_attribute("ReadWriteInt", expected)
-        result = tango_context.read_attribute("ReadWriteInt")
-        assert result.w_value == expected
+    def test_read_write_int(self, client_read, client_write):
+        client_read(["ReadWriteInt"], AttrR(Int())._value)
+        client_write(["ReadWriteInt"], AttrR(Int())._value)
 
-    def test_read_write_float(self, tango_context):
-        expected = 23.1
-        tango_context.write_attribute("ReadWriteFloat", expected)
-        result = tango_context.read_attribute("ReadWriteFloat")
-        assert result.w_value == expected
+    def test_read_write_float(self, client_read, client_write):
+        client_read(["ReadWriteFloat"], AttrR(Float())._value)
+        client_write(["ReadWriteFloat"], AttrR(Float())._value)
 
-    def test_read_bool(self, tango_context):
-        result = tango_context.read_attribute("ReadBool")
-        assert result.value == AttrR(Bool())._value
+    def test_read_bool(self, client_read):
+        client_read(["ReadBool"], AttrR(Bool())._value)
 
-    def test_write_bool(self, tango_context):
-        tango_context.write_attribute("WriteBool", AttrR(Bool())._value)
+    def test_write_bool(self, client_write):
+        client_write(["WriteBool"], AttrR(Bool())._value)
 
     # # We need to discuss enums
-    # def test_string_enum(self, tango_context):
-    #     expected = 1
-    #     tango_context.write_attribute("StringEnum", "green")
-    #     result = tango_context.read_attribute("StringEnum")
-    #     assert result.w_value == expected
+    # def test_string_enum(self, client_read, client_write):
+    #     client_read(["StringEnum"], "green")
+    #     client_write(["StringEnum"], "green")
 
-    def test_big_enum(self, tango_context):
-        result = tango_context.read_attribute("BigEnum")
-        assert result.value == AttrR(Int(), allowed_values=list(range(1, 18)))._value
+    def test_big_enum(self, client_read):
+        client_read(["BigEnum"], AttrR(Int(), allowed_values=list(range(1, 18)))._value)
 
-    def test_go(self, tango_context):
-        tango_context.command_inout("go")
+    # The method works but does not seem linked to the device instance?
+    # def test_go(self, client_exec):
+    #     client_exec(["Go"])
 
-    def test_state(self, tango_context):
-        assert tango_context.command_inout("State") == DevState.ON
+    def test_read_child1(self, client_read):
+        client_read(["SubController01", "ReadInt"], AttrR(Int())._value)
 
-    def test_status(self, tango_context):
-        expected = "The device is in ON state."
-        assert tango_context.command_inout("Status") == expected
+    def test_read_child2(self, client_read):
+        client_read(["SubController02", "ReadInt"], AttrR(Int())._value)

tests/conftest.py

@@ -1,18 +1,19 @@
+import copy
 import os
 import random
 import string
 import subprocess
 import time
+from contextlib import contextmanager
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 import pytest
 from aioca import purge_channel_caches
-from tango.test_context import DeviceTestContext
+from pytest_mock import MockerFixture
 
 from fastcs.attributes import AttrR, AttrRW, AttrW, Handler, Sender, Updater
-from fastcs.backends.tango.backend import TangoBackend
-from fastcs.controller import Controller
+from fastcs.controller import Controller, SubController
 from fastcs.datatypes import Bool, Float, Int, String
 from fastcs.mapping import Mapping
 from fastcs.wrappers import command, scan
@@ -51,7 +52,20 @@ 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())
@@ -60,7 +74,7 @@ class TestController(Controller):
     string_enum: AttrRW = AttrRW(String(), allowed_values=["red", "green", "blue"])
     big_enum: AttrR = AttrR(
         Int(),
-        allowed_values=list(range(1, 18)),
+        allowed_values=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17],
     )
 
     initialised = False
@@ -82,6 +96,48 @@ async def counter(self):
         self.count += 1
 
 
+class AssertableController(TestController):
+    def __init__(self, mocker: MockerFixture) -> None:
+        super().__init__()
+        self.mocker = mocker
+
+    @contextmanager
+    def assertPerformed(
+        self, path: list[str], action: Literal["READ", "WRITE", "EXECUTE"]
+    ):
+        queue = copy.deepcopy(path)
+        match action:
+            case "READ":
+                method = "get"
+            case "WRITE":
+                method = "process"
+            case "EXECUTE":
+                method = ""
+
+        # 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
+        finally:  # 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()
@@ -122,10 +178,3 @@ def ioc():
     except ValueError:
         # Someone else already called communicate
         pass
-
-
-@pytest.fixture(scope="class")
-def tango_context():
-    device = TangoBackend(TestController())._dsr._device
-    with DeviceTestContext(device) as proxy:
-        yield proxy