fix(ble): fix race condition in test_ble_driver_connect_stream (#825)

Author: mangelajo

python/packages/jumpstarter-driver-ble/jumpstarter_driver_ble/driver_test.py

@@ -2,6 +2,7 @@
 
 import anyio
 import pytest
+from jumpstarter_testing.eventually import eventually
 
 from .driver import BleWriteNotifyStream, _ble_notify_handler
 from jumpstarter.common.utils import serve
@@ -76,7 +77,11 @@ def test_ble_driver_connect_stream():
             with client.stream() as stream:
                 # Send data through the stream
                 stream.send(b"hello")
-                mock_client.write_gatt_char.assert_called()
+
+                # stream.send() only guarantees data was written to the
+                # gRPC transport, not that the server has called
+                # write_gatt_char yet — poll until it has.
+                eventually(mock_client.write_gatt_char.assert_called)
 
                 # Verify start_notify was called for the notify characteristic
                 mock_client.start_notify.assert_called_once()

python/packages/jumpstarter-driver-ble/pyproject.toml

@@ -36,7 +36,7 @@ requires = ["hatchling", "hatch-vcs", "hatch-pin-jumpstarter"]
 build-backend = "hatchling.build"
 
 [dependency-groups]
-dev = ["pytest-anyio>=0.0.0", "pytest-cov>=6.0.0", "pytest>=8.3.3"]
+dev = ["jumpstarter-testing", "pytest-anyio>=0.0.0", "pytest-cov>=6.0.0", "pytest>=8.3.3"]
 
 [tool.hatch.build.hooks.pin_jumpstarter]
 name = "pin_jumpstarter"

python/packages/jumpstarter-testing/jumpstarter_testing/__init__.py

@@ -1,3 +1,4 @@
+from .eventually import eventually
 from .pytest import JumpstarterTest
 
-__all__ = ["JumpstarterTest"]
+__all__ = ["JumpstarterTest", "eventually"]

python/packages/jumpstarter-testing/jumpstarter_testing/eventually.py

@@ -0,0 +1,33 @@
+import time
+
+
+def eventually(fn, *, timeout=5, interval=0.05):
+    """Poll ``fn`` until it succeeds or *timeout* seconds elapse.
+
+    Similar to Ginkgo's ``Eventually``: calls *fn* repeatedly, catching
+    ``AssertionError`` and ``Exception``, sleeping *interval* seconds between
+    attempts.  If *fn* does not succeed before *timeout*, the last captured
+    exception is re-raised.
+
+    Args:
+        fn: A callable (typically a lambda wrapping an assertion).
+        timeout: Maximum time in seconds to keep retrying (default 5).
+        interval: Time in seconds between attempts (default 0.05).
+
+    Example::
+
+        from jumpstarter_testing import eventually
+
+        mock.write_gatt_char = AsyncMock()
+        stream.send(b"hello")
+        eventually(mock.write_gatt_char.assert_called)
+    """
+    deadline = time.monotonic() + timeout
+    while True:
+        try:
+            fn()
+            return
+        except (AssertionError, Exception):
+            if time.monotonic() >= deadline:
+                raise
+            time.sleep(interval)

python/packages/jumpstarter-testing/jumpstarter_testing/eventually_test.py

@@ -0,0 +1,31 @@
+import pytest
+
+from jumpstarter_testing.eventually import eventually
+
+
+def test_eventually_succeeds_immediately():
+    """Callable that passes on the first try returns immediately."""
+    called = []
+    def fn():
+        called.append(1)
+    eventually(fn, timeout=1)
+    assert len(called) == 1
+
+
+def test_eventually_retries_until_success():
+    """Callable that fails initially but succeeds after a few attempts."""
+    counter = {"n": 0}
+    def fn():
+        counter["n"] += 1
+        if counter["n"] < 3:
+            raise AssertionError("not yet")
+    eventually(fn, timeout=5, interval=0.01)
+    assert counter["n"] == 3
+
+
+def test_eventually_raises_on_timeout():
+    """Callable that never succeeds raises the last error after timeout."""
+    def fn():
+        raise AssertionError("always fails")
+    with pytest.raises(AssertionError, match="always fails"):
+        eventually(fn, timeout=0.1, interval=0.02)