Add test for fly_and_collect stub

pyproject.toml

@@ -15,8 +15,8 @@ description = "Common Diamond specific Bluesky plans and functions"
 dependencies = [
  "blueapi",
  "ophyd",
- "ophyd_async @ git+https://github.com/bluesky/ophyd-async.git@add-detector-logic",
- "scanspec"
+ "ophyd_async @ git+https://github.com/bluesky/ophyd-async.git",
+ "scanspec",
 ] # Add project dependencies here, e.g. ["click", "numpy"]
 dynamic = ["version"]
 license.file = "LICENSE"
@@ -85,6 +85,7 @@ addopts = """
 filterwarnings = "error"
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests"
+asyncio_mode = "auto"
 
 [tool.coverage.run]
 data_file = "/tmp/dls_bluesky_core.coverage"

tests/conftest.py

@@ -0,0 +1,24 @@
+import asyncio
+
+import pytest
+from bluesky.run_engine import RunEngine, TransitionError
+
+
+@pytest.fixture(scope="function")
+def RE(request):
+ loop = asyncio.new_event_loop()
+ loop.set_debug(True)
+ RE = RunEngine({}, call_returns_result=True, loop=loop)
+
+ def clean_event_loop():
+ if RE.state not in ("idle", "panicked"):
+ try:
+ RE.halt()
+ except TransitionError:
+ pass
+ loop.call_soon_threadsafe(loop.stop)
+ RE._th.join()
+ loop.close()
+
+ request.addfinalizer(clean_event_loop)
+ return RE

tests/stubs/test_flyables.py

@@ -0,0 +1,52 @@
+import asyncio
+from typing import Dict
+
+import bluesky.plan_stubs as bps
+import pytest
+from bluesky.protocols import Collectable, Descriptor, Flyable
+from ophyd_async.core import AsyncStatus
+
+from dls_bluesky_core.stubs.flyables import fly_and_collect
+
+
+class DummyFlyer(Flyable, Collectable):
+ def __init__(self, name: str) -> None:
+ self._name = name
+ self.has_flown = False
+
+ @property
+ def name(self) -> str:
+ return self._name
+
+ @AsyncStatus.wrap
+ async def kickoff(self) -> None:
+ self._fly_status = AsyncStatus(self._fly())
+
+ async def _fly(self) -> None:
+ self.has_flown = True
+ await asyncio.sleep(0.1)
+
+ def complete(self) -> AsyncStatus:
+ return self._fly_status
+
+ def describe_collect(self) -> Dict[str, Descriptor]:
+ return {
+ self.name: Descriptor(
+ source="some:source", shape=[], dtype="array", external="STREAM:"
+ )
+ }
+
+
+@pytest.fixture
+def flyer() -> Flyable:
+ return DummyFlyer("test")
+
+
+async def test_fly_and_collect(RE, flyer: DummyFlyer):
+ def open_and_close_run_for_fly_and_collect():
+ yield from bps.open_run()
+ yield from fly_and_collect(flyer, flush_period=0.01, checkpoint_every_collect=True)
+ yield from bps.close_run()
+
+ RE(open_and_close_run_for_fly_and_collect())
+ assert flyer.has_flown is True