Added demo test

desy-fsec · Aug 22, 2024 · 7c7372b · 7c7372b
1 parent 7a663f0
commit 7c7372b
Show file tree

Hide file tree

Showing 3 changed files with 53 additions and 7 deletions.
diff --git a/src/ophyd_async/tango/base_devices/base_device.py b/src/ophyd_async/tango/base_devices/base_device.py
@@ -101,12 +101,13 @@ def create_children_from_annotations(self):
                     setattr(self, attr_name, SignalW())
                 elif obj_type is SignalX:
                     setattr(self, attr_name, SignalX())
-                elif obj_type is Signal or None:
+                elif obj_type is Signal or obj_type is None:
                     tango_name = attr_name.lstrip("_")
                     setattr(
                         self,
                         attr_name,
                         infer_signal_frontend(trl=f"{self.trl}/" f"{tango_name}"),
                     )
                 else:
+                    print(obj_type, type(obj_type))
                     raise ValueError(f"Invalid signal type {obj_type}")
diff --git a/src/ophyd_async/tango/signal/signal.py b/src/ophyd_async/tango/signal/signal.py
@@ -19,9 +19,6 @@
     "tango_signal_w",
     "tango_signal_x",
     "tango_signal_auto",
-    "infer_signal_frontend",
-    "infer_python_type",
-    "make_backend",
 )
 
 

diff --git a/tests/tango/test_base_device.py b/tests/tango/test_base_device.py
@@ -2,20 +2,23 @@
 from enum import Enum, IntEnum
 from typing import Type
 
+import bluesky.plan_stubs as bps
+import bluesky.plans as bp
 import numpy as np
 import pytest
 from bluesky import RunEngine
-from bluesky.plans import count
 
 from ophyd_async.core import DeviceCollector, T
 from ophyd_async.tango import TangoReadableDevice, tango_signal_auto
 from ophyd_async.tango._backend._tango_transport import get_python_type
+from ophyd_async.tango.demo._tango.servers import DemoCounter, DemoMover
+from ophyd_async.tango.demo.counter import TangoCounter
+from ophyd_async.tango.demo.mover import TangoMover
 from tango import (
     AttrDataFormat,
     AttrQuality,
     AttrWriteType,
     CmdArgType,
-    # DeviceProxy,
     DevState,
 )
 from tango.asyncio import DeviceProxy
@@ -271,6 +274,22 @@ def tango_test_device():
         yield context.get_device_access("test/device/1")
 
 
+# --------------------------------------------------------------------
+@pytest.fixture(scope="module")
+def demo_test_context():
+    content = (
+        {
+            "class": DemoMover,
+            "devices": [{"name": "demo/motor/1"}],
+        },
+        {
+            "class": DemoCounter,
+            "devices": [{"name": "demo/counter/1"}, {"name": "demo/counter/2"}],
+        },
+    )
+    yield MultiDeviceTestContext(content)
+
+
 # --------------------------------------------------------------------
 @pytest.fixture(autouse=True)
 def reset_tango_asyncio():
@@ -313,4 +332,33 @@ async def connect():
     for readable in ophyd_dev._readables:
         readable._backend.allow_events(False)
         readable._backend.set_polling(True, 0.1, 0.1)
-    RE(count([ophyd_dev], 1))
+    RE(bp.count([ophyd_dev], 1))
+
+
+# --------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_tango_demo(demo_test_context):
+    with demo_test_context:
+        motor1 = TangoMover(
+            trl=demo_test_context.get_device_access("demo/motor/1"), name="motor1"
+        )
+        counter1 = TangoCounter(
+            trl=demo_test_context.get_device_access("demo/counter/1"), name="counter1"
+        )
+        counter2 = TangoCounter(
+            trl=demo_test_context.get_device_access("demo/counter/2"), name="counter2"
+        )
+        await motor1.connect()
+        await counter1.connect()
+        await counter2.connect()
+
+        # Events are not supported by the test context so we disable them
+        motor1.position._backend.allow_events(False)
+        motor1.state._backend.allow_events(False)
+        # Enable polling for the position and state attributes
+        motor1.position._backend.set_polling(True, 0.1, 0.1)
+        motor1.state._backend.set_polling(True, 0.1)
+
+        RE = RunEngine()
+        RE(bps.read(motor1.position))
+        RE(bp.count([counter1, counter2]))