Create common fixtures which can be reused by other repos (#1657)

* Create common fixtures which can be reused by other repos

* Fixed tests, made run_engine only module

* Move event_loop_fuzzing to dodal.testing.fixtures.utils

* omit fixtures from code coverage report

Author: oliwenmandiamond

conftest.py

@@ -1,14 +1,10 @@
-import asyncio
 import logging
 import sys
-import time
-from collections.abc import Mapping
 from os import environ
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
-from bluesky.run_engine import RunEngine
 from ophyd.status import Status
 from ophyd_async.core import (
     PathInfo,
@@ -49,6 +45,10 @@
 environ["DODAL_TEST_MODE"] = "true"
 
 
+# Add run_engine fixtures to be used in tests
+pytest_plugins = ["dodal.testing.fixtures.run_engine"]
+
+
 @pytest.fixture(autouse=True)
 def patch_open_to_prevent_dls_reads_in_tests():
     unpatched_open = open
@@ -105,37 +105,7 @@ async def static_path_provider(
     return svpp
 
 
-@pytest.fixture
-def run_engine_documents(run_engine: RunEngine) -> Mapping[str, list[dict]]:
-    docs: dict[str, list[dict]] = {}
-
-    def append_and_print(name, doc):
-        if name not in docs:
-            docs[name] = []
-        docs[name] += [doc]
-
-    run_engine.subscribe(append_and_print)
-    return docs
-
-
 def failed_status(failure: Exception) -> Status:
     status = Status()
     status.set_exception(failure)
     return status
-
-
-@pytest.fixture(scope="session", autouse=True)
-async def _ensure_running_bluesky_event_loop():
-    run_engine = RunEngine()
-    # make sure the event loop is thoroughly up and running before we try to create
-    # any ophyd_async devices which might need it
-    timeout = time.monotonic() + 1
-    while not run_engine.loop.is_running():
-        await asyncio.sleep(0)
-        if time.monotonic() > timeout:
-            raise TimeoutError("This really shouldn't happen but just in case...")
-
-
-@pytest.fixture()
-async def run_engine():
-    yield RunEngine()

pyproject.toml

@@ -136,6 +136,8 @@ testpaths = "docs src tests system_tests"
 exclude_also = [
     '^"""', # Ignore the start/end of a file-level triple quoted docstring
 ]
+# Exclude pytest.fixtures from code coverage report
+omit = ["*/dodal/testing/fixtures/*"]
 
 [tool.coverage.run]
 data_file = "/tmp/dodal.coverage"

src/dodal/testing/fixtures/__init__.py

@@ -0,0 +1,46 @@
+"""
+Allow external repos to reuse these fixtures so defined in single place.
+"""
+
+import asyncio
+import time
+from collections.abc import Mapping
+
+import pytest
+from bluesky.run_engine import RunEngine
+from bluesky.simulators import RunEngineSimulator
+
+
+@pytest.fixture(scope="session", autouse=True)
+async def _ensure_running_bluesky_event_loop():
+    run_engine = RunEngine()
+    # make sure the event loop is thoroughly up and running before we try to create
+    # any ophyd_async devices which might need it
+    timeout = time.monotonic() + 1
+    while not run_engine.loop.is_running():
+        await asyncio.sleep(0)
+        if time.monotonic() > timeout:
+            raise TimeoutError("This really shouldn't happen but just in case...")
+
+
+@pytest.fixture()
+async def run_engine():
+    yield RunEngine()
+
+
+@pytest.fixture
+def sim_run_engine() -> RunEngineSimulator:
+    return RunEngineSimulator()
+
+
+@pytest.fixture
+def run_engine_documents(run_engine: RunEngine) -> Mapping[str, list[dict]]:
+    docs: dict[str, list[dict]] = {}
+
+    def append_and_print(name, doc):
+        if name not in docs:
+            docs[name] = []
+        docs[name] += [doc]
+
+    run_engine.subscribe(append_and_print)
+    return docs

src/dodal/testing/fixtures/run_engine.py

@@ -0,0 +1,57 @@
+import asyncio
+import threading
+import time
+from random import random
+from threading import Thread
+
+import pytest
+
+
+@pytest.fixture
+async def event_loop_fuzzing():
+    """
+    This fixture can be used to try and detect / reproduce intermittent test failures
+    caused by race conditions and timing issues, which are often difficult to replicate
+    due to caching etc. causing timing to be different on a development machine compared
+    to when the test runs in CI.
+
+    It works by attaching a fuzzer to the current event loop which randomly schedules
+    a fixed delay into the event loop thread every few milliseconds. The idea is that
+    over a number of iterations, there should be sufficient timing variation introduced
+    that the failure can be reproduced.
+
+    Examples:
+        Example usage:
+    >>> import pytest
+    >>> # repeat the test a number of times
+    >>> @pytest.mark.parametrize("i", range(0, 100))
+    ... async def my_unreliable_test(i, event_loop_fuzzing):
+    ...     # Do some stuff in here
+    ...     ...
+    """
+    fuzz_probability = 0.05
+    fuzz_delay_s = 0.05
+    fuzz_period_s = 0.001
+    stop_running = threading.Event()
+    event_loop = asyncio.get_running_loop()
+
+    def delay(finished_event: threading.Event):
+        time.sleep(fuzz_delay_s)  # noqa: TID251
+        finished_event.set()
+
+    def fuzz():
+        while not stop_running.is_set():
+            if random() < fuzz_probability:
+                delay_is_finished = threading.Event()
+                event_loop.call_soon_threadsafe(delay, delay_is_finished)
+                delay_is_finished.wait()
+
+            time.sleep(fuzz_period_s)  # noqa: TID251
+
+    fuzzer_thread = Thread(group=None, target=fuzz, name="Event loop fuzzer")
+    fuzzer_thread.start()
+    try:
+        yield None
+    finally:
+        stop_running.set()
+        fuzzer_thread.join()

src/dodal/testing/fixtures/utils.py

@@ -1,12 +1,7 @@
-import asyncio
 import importlib
 import os
-import threading
-import time
 from collections.abc import AsyncGenerator
 from pathlib import Path
-from random import random
-from threading import Thread
 from types import ModuleType
 from unittest.mock import patch
 
@@ -28,6 +23,9 @@
 from tests.devices.test_data import TEST_LUT_TXT
 from tests.test_data import I04_BEAMLINE_PARAMETERS
 
+# Add utils fixtures to be used in tests
+pytest_plugins = ["dodal.testing.fixtures.utils"]
+
 
 @pytest.fixture(scope="function")
 def module_and_devices_for_beamline(request: pytest.FixtureRequest):
@@ -81,53 +79,3 @@ async def baton_in_commissioning_mode() -> AsyncGenerator[Baton]:
     set_mock_value(baton.commissioning, True)
     yield baton
     set_commissioning_signal(None)
-
-
-@pytest.fixture
-async def event_loop_fuzzing():
-    """
-    This fixture can be used to try and detect / reproduce intermittent test failures
-    caused by race conditions and timing issues, which are often difficult to replicate
-    due to caching etc. causing timing to be different on a development machine compared
-    to when the test runs in CI.
-
-    It works by attaching a fuzzer to the current event loop which randomly schedules
-    a fixed delay into the event loop thread every few milliseconds. The idea is that
-    over a number of iterations, there should be sufficient timing variation introduced
-    that the failure can be reproduced.
-
-    Examples:
-        Example usage:
-    >>> import pytest
-    >>> # repeat the test a number of times
-    >>> @pytest.mark.parametrize("i", range(0, 100))
-    ... async def my_unreliable_test(i, event_loop_fuzzing):
-    ...     # Do some stuff in here
-    ...     ...
-    """
-    fuzz_probability = 0.05
-    fuzz_delay_s = 0.05
-    fuzz_period_s = 0.001
-    stop_running = threading.Event()
-    event_loop = asyncio.get_running_loop()
-
-    def delay(finished_event: threading.Event):
-        time.sleep(fuzz_delay_s)
-        finished_event.set()
-
-    def fuzz():
-        while not stop_running.is_set():
-            if random() < fuzz_probability:
-                delay_is_finished = threading.Event()
-                event_loop.call_soon_threadsafe(delay, delay_is_finished)
-                delay_is_finished.wait()
-
-            time.sleep(fuzz_period_s)
-
-    fuzzer_thread = Thread(group=None, target=fuzz, name="Event loop fuzzer")
-    fuzzer_thread.start()
-    try:
-        yield None
-    finally:
-        stop_running.set()
-        fuzzer_thread.join()