Windows: fix unit tests & enable CI (#633)

* Fix race condition in mock_signal_backend

* Fix numpy datatypes in test_mock_signal_backend

* Fix adaravis URI paths

* Fix path in test_kinetix

* Fix paths in simdetector tests

* Loosen timings against race condition in test_sim_detector

* Correct paths in test_advimba

* Fix paths in test_eiger

* Fix broken tests in test_signals

* Fix tange test_base_device tests on Windows

Missing process=True argument was causing all subsequent tests to fail on windows

* Fix race conditions in tango tests

* Use tango FQTRL to placate windows unit tests

* Correct paths in fastcs tests

* Correct path logic for windows compatibility

* Slacken timings for race condition in test_sim_motor

* Slacken race condition timings in test_motor

* Enable windows CI builds

* Try allowing asyncio time to clean up tasks

* Object IDs on windows are uppercase

* Try wait_for_wakeups

.github/workflows/ci.yml

@@ -20,7 +20,7 @@ jobs:
     if: needs.check.outputs.branch-pr == ''
     strategy:
       matrix:
-        runs-on: ["ubuntu-latest"] # can add windows-latest, macos-latest
+        runs-on: ["ubuntu-latest", "windows-latest"] # can add macos-latest
         python-version: ["3.10","3.11"] # 3.12 should be added when p4p is updated
         include:
           # Include one that runs in the dev environment

tests/core/test_flyer.py

@@ -1,3 +1,4 @@
+import asyncio
 import time
 from collections.abc import AsyncGenerator, AsyncIterator, Sequence
 from typing import Any
@@ -356,6 +357,9 @@ def flying_plan():
     with pytest.raises(Exception, match=match_msg):
         RE(flying_plan())
 
+    # Try explicitly letting event loop clean up tasks...?
+    await asyncio.sleep(1.0)
+
 
 @pytest.mark.parametrize(
     ["kwargs", "error_msg"],

tests/core/test_mock_signal_backend.py

@@ -135,7 +135,7 @@ async def test_mock_utils_throw_error_if_backend_isnt_mock_signal_backend():
 
     for msg in exc_msgs:
         assert re.match(
-            r"Signal <ophyd_async.core._signal.SignalRW object at [0-9a-z]+> "
+            r"Signal <ophyd_async.core._signal.SignalRW object at [0-9a-zA-Z]+> "
             r"not connected in mock mode",
             msg,
         )
@@ -190,7 +190,7 @@ async def test_blocks_during_put(mock_signals):
         assert not status1.done
         assert not status2.done
 
-    await asyncio.sleep(1e-4)
+    await asyncio.sleep(0.1)
 
     assert status1.done
     assert status2.done

tests/core/test_soft_signal_backend.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 import time
 from collections.abc import Callable, Sequence
 from typing import Any
@@ -67,10 +68,16 @@ def close(self):
         self.backend.set_callback(None)
 
 
+# Can be removed once numpy >=2 is pinned.
+default_int_type = (
+    "<i4" if os.name == "nt" and np.version.version.startswith("1.") else "<i8"
+)
+
+
 @pytest.mark.parametrize(
     "datatype, initial_value, put_value, descriptor, dtype_numpy",
     [
-        (int, 0, 43, integer_d, "<i8"),
+        (int, 0, 43, integer_d, default_int_type),
         (float, 0.0, 43.5, number_d, "<f8"),
         (str, "", "goodbye", string_d, "|S40"),
         (MyEnum, MyEnum.a, MyEnum.c, enum_d, "|S40"),

tests/epics/adaravis/test_aravis.py

@@ -112,7 +112,9 @@ async def test_can_collect(
     stream_resource = docs[0][1]
     sr_uid = stream_resource["uid"]
     assert stream_resource["data_key"] == "test_adaravis1"
-    assert stream_resource["uri"] == "file://localhost" + str(full_file_name)
+    assert stream_resource["uri"] == "file://localhost/" + str(full_file_name).lstrip(
+        "/"
+    )
     assert stream_resource["parameters"] == {
         "dataset": "/entry/data/data",
         "swmr": False,

tests/epics/adkinetix/test_kinetix.py

@@ -93,7 +93,9 @@ async def test_can_collect(
     stream_resource = docs[0][1]
     sr_uid = stream_resource["uid"]
     assert stream_resource["data_key"] == "test_adkinetix1"
-    assert stream_resource["uri"] == "file://localhost" + str(full_file_name)
+    assert stream_resource["uri"] == "file://localhost/" + str(full_file_name).lstrip(
+        "/"
+    )
     assert stream_resource["parameters"] == {
         "dataset": "/entry/data/data",
         "swmr": False,

tests/epics/adsimdetector/test_sim.py

@@ -52,7 +52,7 @@ def count_sim(dets: list[adsimdetector.SimDetector], times: int = 1):
         for det in dets:
             yield from bps.trigger(det, wait=False, group="wait_for_trigger")
 
-        yield from bps.sleep(0.2)
+        yield from bps.sleep(1.0)
         [
             set_mock_value(
                 cast(adcore.ADHDFWriter, det.writer).hdf.num_captured,
@@ -207,11 +207,15 @@ def plan():
     assert sdb["stream_resource"] == srb["uid"]
     assert (
         srb["uri"]
-        == "file://localhost" + str(info_b.directory_path / info_b.filename) + ".h5"
+        == "file://localhost/"
+        + str(info_b.directory_path / info_b.filename).lstrip("/")
+        + ".h5"
     )
     assert (
         sra["uri"]
-        == "file://localhost" + str(info_a.directory_path / info_a.filename) + ".h5"
+        == "file://localhost/"
+        + str(info_a.directory_path / info_a.filename).lstrip("/")
+        + ".h5"
     )
 
     assert event["data"] == {}

tests/epics/advimba/test_vimba.py

@@ -108,7 +108,9 @@ async def test_can_collect(
     stream_resource = docs[0][1]
     sr_uid = stream_resource["uid"]
     assert stream_resource["data_key"] == "test_advimba1"
-    assert stream_resource["uri"] == "file://localhost" + str(full_file_name)
+    assert stream_resource["uri"] == "file://localhost/" + str(full_file_name).lstrip(
+        "/"
+    )
     assert stream_resource["parameters"] == {
         "dataset": "/entry/data/data",
         "swmr": False,

tests/epics/eiger/test_odin_io.py

@@ -18,18 +18,17 @@ def odin_driver_and_writer(RE) -> OdinDriverAndWriter:
 
 
 async def test_when_open_called_then_file_correctly_set(
-    odin_driver_and_writer: OdinDriverAndWriter,
+    odin_driver_and_writer: OdinDriverAndWriter, tmp_path: Path
 ):
     driver, writer = odin_driver_and_writer
     path_info = writer._path_provider.return_value
-    expected_path = "/tmp"
     expected_filename = "filename.h5"
-    path_info.directory_path = Path(expected_path)
+    path_info.directory_path = tmp_path
     path_info.filename = expected_filename
 
     await writer.open()
 
-    get_mock_put(driver.file_path).assert_called_once_with(expected_path, wait=ANY)
+    get_mock_put(driver.file_path).assert_called_once_with(str(tmp_path), wait=ANY)
     get_mock_put(driver.file_name).assert_called_once_with(expected_filename, wait=ANY)
 
 

tests/epics/signal/test_signals.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 import random
 import string
 import subprocess
@@ -99,7 +100,7 @@ def ioc(request: pytest.FixtureRequest):
     # close backend caches before the event loop
     purge_channel_caches()
     try:
-        print(process.communicate("exit")[0])
+        print(process.communicate("exit()")[0])
     except ValueError:
         # Someone else already called communicate
         pass
@@ -260,7 +261,11 @@ def get_dtype_numpy(suffix: str) -> str:  # type: ignore
             elif "64" in suffix:
                 int_str += "8"
             else:
-                int_str += "8"
+                int_str += (
+                    "4"
+                    if os.name == "nt" and np.version.version.startswith("1.")
+                    else "8"
+                )
             return int_str
         if "str" in suffix or "enum" in suffix:
             return "|S40"
@@ -908,6 +913,7 @@ async def test_bool_works_for_mismatching_enums(ioc: IOC):
     await sig.connect()
 
 
+@pytest.mark.skipif(os.name == "nt", reason="Hangs on windows for unknown reasons")
 async def test_can_read_using_ophyd_async_then_ophyd(ioc: IOC):
     oa_read = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:float_prec_1"
     ophyd_read = f"{PV_PREFIX}:{ioc.protocol}:float_prec_0"

tests/epics/test_motor.py

@@ -18,9 +18,16 @@
 )
 from ophyd_async.epics import motor
 
-# Long enough for multiple asyncio event loop cycles to run so
-# all the tasks have a chance to run
-A_BIT = 0.001
+
+async def wait_for_wakeups(max_yields=10):
+    loop = asyncio.get_event_loop()
+    # If anything has called loop.call_soon or is scheduled a wakeup
+    # then let it run
+    for _ in range(max_yields):
+        await asyncio.sleep(0)
+        if not loop._ready:
+            return
+    raise RuntimeError(f"Tasks still scheduling wakeups after {max_yields} yields")
 
 
 @pytest.fixture
@@ -37,7 +44,7 @@ async def sim_motor():
 async def wait_for_eq(item, attribute, comparison, timeout):
     timeout_time = time.monotonic() + timeout
     while getattr(item, attribute) != comparison:
-        await asyncio.sleep(A_BIT)
+        await wait_for_wakeups()
         if time.monotonic() > timeout_time:
             raise TimeoutError
 
@@ -49,6 +56,7 @@ async def test_motor_moving_well(sim_motor: motor.Motor) -> None:
     s.watch(watcher)
     done = Mock()
     s.add_callback(done)
+    await wait_for_wakeups()
     await wait_for_eq(watcher, "call_count", 1, 1)
     assert watcher.call_args == call(
         name="sim_motor",
@@ -78,7 +86,7 @@ async def test_motor_moving_well(sim_motor: motor.Motor) -> None:
     set_mock_value(sim_motor.motor_done_move, True)
     set_mock_value(sim_motor.user_readback, 0.55)
     set_mock_put_proceeds(sim_motor.user_setpoint, True)
-    await asyncio.sleep(A_BIT)
+    await wait_for_wakeups()
     await wait_for_eq(s, "done", True, 1)
     done.assert_called_once_with(s)
 
@@ -90,7 +98,7 @@ async def test_motor_moving_well_2(sim_motor: motor.Motor) -> None:
     s.watch(watcher)
     done = Mock()
     s.add_callback(done)
-    await asyncio.sleep(A_BIT)
+    await wait_for_wakeups()
     assert watcher.call_count == 1
     assert watcher.call_args == call(
         name="sim_motor",
@@ -99,7 +107,7 @@ async def test_motor_moving_well_2(sim_motor: motor.Motor) -> None:
         target=0.55,
         unit="mm",
         precision=3,
-        time_elapsed=pytest.approx(0.0, abs=0.05),
+        time_elapsed=pytest.approx(0.0, abs=0.2),
     )
     watcher.reset_mock()
     assert 0.55 == await sim_motor.user_setpoint.get_value()
@@ -115,10 +123,10 @@ async def test_motor_moving_well_2(sim_motor: motor.Motor) -> None:
         target=0.55,
         unit="mm",
         precision=3,
-        time_elapsed=pytest.approx(0.1, abs=0.05),
+        time_elapsed=pytest.approx(0.1, abs=0.2),
     )
     set_mock_put_proceeds(sim_motor.user_setpoint, True)
-    await asyncio.sleep(A_BIT)
+    await wait_for_wakeups()
     assert s.done
     done.assert_called_once_with(s)
 
@@ -157,7 +165,7 @@ async def test_motor_moving_stopped(sim_motor: motor.Motor):
     assert not s.done
     await sim_motor.stop()
     set_mock_put_proceeds(sim_motor.user_setpoint, True)
-    await asyncio.sleep(A_BIT)
+    await wait_for_wakeups()
     assert s.done
     assert s.success is False
 

tests/fastcs/panda/test_hdf_panda.py

@@ -149,7 +149,7 @@ def flying_plan():
             "uid": ANY,
             "data_key": data_key_name,
             "mimetype": "application/x-hdf5",
-            "uri": "file://localhost" + str(tmp_path / "test-panda.h5"),
+            "uri": "file://localhost/" + str(tmp_path / "test-panda.h5").lstrip("/"),
             "parameters": {
                 "dataset": f"/{dataset_name}",
                 "swmr": False,

tests/fastcs/panda/test_writer.py

@@ -187,15 +187,16 @@ def assert_resource_document(name, resource_doc):
             "uid": ANY,
             "data_key": name,
             "mimetype": "application/x-hdf5",
-            "uri": "file://localhost" + str(tmp_path / "mock_panda" / "data.h5"),
+            "uri": "file://localhost/"
+            + str(tmp_path / "mock_panda" / "data.h5").lstrip("/"),
             "parameters": {
                 "dataset": f"/{name}",
                 "swmr": False,
                 "multiplier": 1,
                 "chunk_shape": (1024,),
             },
         }
-        assert "mock_panda/data.h5" in resource_doc["uri"]
+        assert os.path.join("mock_panda", "data.h5") in resource_doc["uri"]
 
     [item async for item in mock_writer.collect_stream_docs(1)]
     assert type(mock_writer._file) is HDFFile

tests/sim/demo/test_sim_motor.py

@@ -42,9 +42,9 @@ async def test_stop():
     async with DeviceCollector():
         m1 = SimMotor("M1", instant=False)
 
-    # this move should take 10 seconds but we will stop it after 0.2
+    # this move should take 10 seconds but we will stop it after 0.5
     move_status = m1.set(10)
-    await asyncio.sleep(0.2)
+    await asyncio.sleep(0.5)
     await m1.stop(success=False)
     new_pos = await m1.user_readback.get_value()
     assert new_pos < 10

tests/sim/test_sim_detector.py

@@ -1,3 +1,4 @@
+import os
 from collections import defaultdict
 
 import bluesky.plans as bp
@@ -44,6 +45,8 @@ def plan():
         docs, start=1, descriptor=1, stream_resource=2, stream_datum=2, event=1, stop=1
     )
     path = docs["stream_resource"][0]["uri"].split("://localhost")[-1]
+    if os.name == "nt":
+        path = path.lstrip("/")
     h5file = h5py.File(path)
     assert list(h5file["/entry"]) == ["data", "sum"]
     assert list(h5file["/entry/sum"]) == [44540.0]

tests/tango/test_base_device.py

@@ -290,7 +290,7 @@ def demo_test_context():
             "devices": [{"name": "demo/counter/1"}, {"name": "demo/counter/2"}],
         },
     )
-    yield MultiDeviceTestContext(content)
+    yield MultiDeviceTestContext(content, process=True)
 
 
 # --------------------------------------------------------------------
@@ -384,7 +384,7 @@ async def test_tango_demo(demo_test_context):
         RE(bp.count(list(detector.counters.values())))
 
         set_status = detector.set(1.0)
-        await asyncio.sleep(0.1)
+        await asyncio.sleep(1.0)
         stop_status = detector.stop()
         await set_status
         await stop_status

tests/tango/test_tango_transport.py

@@ -244,6 +244,7 @@ async def test_attribute_proxy_put(tango_test_device, attr, wait):
     else:
         if not wait:
             raise AssertionError("If wait is False, put should return a status object")
+    await asyncio.sleep(1.0)
     updated_value = await attr_proxy.get()
     if isinstance(new_value, np.ndarray):
         assert np.all(updated_value == new_value)
@@ -284,6 +285,7 @@ async def test_attribute_proxy_get_w_value(tango_test_device, attr, new_value):
     device_proxy = await DeviceProxy(tango_test_device)
     attr_proxy = AttributeProxy(device_proxy, attr)
     await attr_proxy.put(new_value)
+    await asyncio.sleep(1.0)
     attr_proxy_value = await attr_proxy.get()
     if isinstance(new_value, np.ndarray):
         assert np.all(attr_proxy_value == new_value)
@@ -794,9 +796,9 @@ async def test_tango_transport_allow_events(echo_device, allow):
 @pytest.mark.asyncio
 async def test_tango_transport_read_and_write_trl(tango_test_device):
     device_proxy = await DeviceProxy(tango_test_device)
-    trl = device_proxy.dev_name()
-    read_trl = trl + "/" + "readback"
-    write_trl = trl + "/" + "setpoint"
+    # Must use a FQTRL, at least on windows.
+    read_trl = tango_test_device + "/" + "readback"
+    write_trl = tango_test_device + "/" + "setpoint"
 
     # Test with existing proxy
     transport = TangoSignalBackend(float, read_trl, write_trl, device_proxy)