Work on tango tests

marcelldls · marcelldls · commit 1a079b9c4ae4 · 2024-11-11T14:38:05.000Z
diff --git a/src/fastcs/backends/tango/dsr.py b/src/fastcs/backends/tango/dsr.py
@@ -16,7 +16,6 @@
 @dataclass
 class TangoDSROptions:
     dev_name: str = "MY/DEVICE/NAME"
-    dev_class: str = "FAST_CS_DEVICE"
     dsr_instance: str = "MY_SERVER_INSTANCE"
     debug: bool = False
 
@@ -44,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
 
@@ -101,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
@@ -155,11 +154,10 @@ def _collect_dsr_args(options: TangoDSROptions) -> list[str]:
 class TangoDSR:
     def __init__(self, mapping: Mapping):
         self._mapping = mapping
+        self.dev_class = self._mapping.controller.__class__.__name__
+        self._device = self._create_device()
 
-    def run(self, options: TangoDSROptions | None = None) -> None:
-        if options is None:
-            options = TangoDSROptions()
-
+    def _create_device(self):
         class_dict: dict = {
             **_collect_dev_attributes(self._mapping),
             **_collect_dev_commands(self._mapping),
@@ -169,13 +167,18 @@ def run(self, options: TangoDSROptions | None = None) -> None:
         }
 
         class_bases = (server.Device,)
-        pytango_class = type(options.dev_class, class_bases, class_dict)
+        pytango_class = type(self.dev_class, class_bases, class_dict)
+        return pytango_class
+
+    def run(self, options: TangoDSROptions | None = None) -> None:
+        if options is None:
+            options = TangoDSROptions()
 
         dsr_args = _collect_dsr_args(options)
 
         server.run(
-            (pytango_class,),
-            [options.dev_class, options.dsr_instance, *dsr_args],
+            (self._device,),
+            [self.dev_class, options.dsr_instance, *dsr_args],
             green_mode=server.GreenMode.Asyncio,
         )
 
diff --git a/tests/backends/tango/compose/compose.yaml b/tests/backends/tango/compose/compose.yaml
diff --git a/tests/backends/tango/test_dsr.py b/tests/backends/tango/test_dsr.py
@@ -1,37 +1,124 @@
+import copy
+import re
+from typing import Any
+
 import pytest
 from pytest_mock import MockerFixture
-from tango._tango import AttrWriteType, CmdArgType
-
-from fastcs.backends.tango.dsr import _collect_dev_attributes, _collect_dev_commands
-
-
-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 tango import DevState
+from tango.test_context import DeviceTestContext
+from tests.conftest import AssertableController
+
+from fastcs.attributes import AttrR
+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",
+            "ReadBool",
+            "ReadInt",
+            "ReadWriteFloat",
+            "ReadWriteInt",
+            "StringEnum",
+            "WriteBool",
+            "SubController01_ReadInt",
+            "SubController02_ReadInt",
+            "State",
+            "Status",
+        ]
+
+    def test_list_commands(self, tango_context):
+        assert list(tango_context.get_command_list()) == [
+            "Go",
+            "Init",
+            "State",
+            "Status",
+        ]
+
+    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, client_read, client_write):
+        client_read(["ReadWriteInt"], AttrR(Int())._value)
+        client_write(["ReadWriteInt"], AttrR(Int())._value)
+
+    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, client_read):
+        client_read(["ReadBool"], 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, client_read, client_write):
+    #     client_read(["StringEnum"], "green")
+    #     client_write(["StringEnum"], "green")
+
+    def test_big_enum(self, client_read):
+        client_read(["BigEnum"], AttrR(Int(), allowed_values=list(range(1, 18)))._value)
+
+    # The method works but does not seem linked to the device instance?
+    # def test_go(self, client_exec):
+    #     client_exec(["Go"])
+
+    def test_read_child1(self, client_read):
+        client_read(["SubController01", "ReadInt"], AttrR(Int())._value)
+
+    def test_read_child2(self, client_read):
+        client_read(["SubController02", "ReadInt"], AttrR(Int())._value)
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -1,16 +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 pytest_mock import MockerFixture
 
 from fastcs.attributes import AttrR, AttrRW, AttrW, Handler, Sender, Updater
-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
@@ -49,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())
@@ -80,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()
