test: add wait_until utility function for polling predicates in tests

tests/integration/conftest.py

@@ -17,7 +17,12 @@
 
 @pytest.fixture()
 def setup_delete_syftboxes():
-    """Clean up syftboxes from drive before running integration tests."""
+    """Clear both Drive accounts around every integration test.
+
+    Cleaning up afterwards as well as before is what keeps one test's leftovers
+    out of the next run: the accounts are shared, so state that outlives a run
+    is state the next run inherits.
+    """
     if os.environ.get("INTEGRATION_TEST_MOCK_MODE", "").lower() == "true":
         yield
         return
@@ -29,4 +34,7 @@ def setup_delete_syftboxes():
             as token_do.json and token_ds.json. Also set the environment variables AI_AUDIT_EMAIL_DO and AI_AUDIT_EMAIL_DS to the email addresses of the DO and DS."""
         )
     remove_syftboxes_from_drive()
-    yield
+    try:
+        yield
+    finally:
+        remove_syftboxes_from_drive()

tests/integration/utils.py

@@ -4,7 +4,7 @@
 import uuid
 import time
 from pathlib import Path
-from typing import List
+from typing import Callable, List
 from syft.sync.utils.syftbox_utils import get_event_hash_from_content
 from syft.sync.syftbox_manager import SyftboxManager
 
@@ -21,16 +21,45 @@
 token_path_ds = CREDENTIALS_DIR / FILE_DS
 
 
+def wait_until(
+    predicate: Callable[[], bool], timeout: float = 30.0, interval: float = 0.5
+) -> bool:
+    """Poll ``predicate`` until it holds, and say whether it did.
+
+    Drive gives no upper bound on how long a write takes to reach the other
+    side, so a fixed sleep is a guess: too short and the test fails on a slow
+    day, too long and every run pays for it. Returns False on timeout, leaving
+    the assertion to the caller so the failure names what never arrived.
+    """
+    deadline = time.monotonic() + timeout
+    while True:
+        if predicate():
+            return True
+        if time.monotonic() >= deadline:
+            return False
+        time.sleep(interval)
+
+
 def remove_syftboxes_from_drive():
+    """Clear both accounts on Drive, including the state outside the folder tree.
+
+    ``delete_syftbox`` walks the /SyftBox tree, and Drive's eventual consistency
+    can leave a recent file out of that listing. SYFT_peers.json is the one that
+    matters: a surviving entry marks the peer accepted or rejected, and the next
+    run's peer request is then filtered out of the folder scan and never seen.
+    ``delete_unversioned_state`` removes it by name, and runs first because
+    ``get_syftbox_folder_id`` recreates the folder it needs.
+    """
     manager_ds, manager_do = SyftboxManager._pair_with_google_drive_testing_connection(
         do_email=EMAIL_DO,
         ds_email=EMAIL_DS,
         do_token_path=token_path_do,
         ds_token_path=token_path_ds,
         add_peers=False,
     )
-    manager_ds.delete_syftbox(broadcast_delete_events=False)
-    manager_do.delete_syftbox(broadcast_delete_events=False)
+    for manager in (manager_ds, manager_do):
+        manager._connection_router.connection_for_own_syftbox().delete_unversioned_state()
+        manager.delete_syftbox(broadcast_delete_events=False)
 
 
 def get_mock_event(path: str = "email@email.com/test.job") -> FileChangeEvent:

tests/integration/without_unit_coverage/test_sync_manager.py

@@ -11,6 +11,8 @@
 from time import sleep
 import pytest
 
+from tests.integration.utils import wait_until
+
 
 SYFT_DIR = Path(__file__).parent.parent.parent.parent
 # These are in gitignore, create yourself
@@ -59,12 +61,28 @@ def test_peer_request_blocks_sync_until_approved():
     # Step 1: DS makes peer request by adding DO
     ds_manager.add_peer(do_manager.email)
 
-    # Wait for sync
-    sleep(1)
-
-    # Verify: DO sees this as a pending request
-    do_manager.load_peers()
-    assert len(do_manager.peer_manager.requested_by_peer_peers) == 1
+    # Verify: DO sees this as a pending request, once it reaches them.
+    # force_download because the DS is an external writer of SYFT_peers.json;
+    # the cached copy cannot show a request made after it was read.
+    def _peer_request_arrived() -> bool:
+        do_manager.load_peers(force_download=True)
+        return len(do_manager.peer_manager.requested_by_peer_peers) == 1
+
+    def _peer_states() -> str:
+        """What the DO thinks of every peer, for when the request never shows."""
+        router = do_manager.peer_manager.connection_router
+        known = ", ".join(
+            f"{p.email}={p.state.value}"
+            for p in router.get_all_peers_from_json(force_download=True)
+        )
+        requests = ", ".join(p.email for p in router.get_peer_requests())
+        return f"peers.json: [{known or 'empty'}], folder scan: [{requests or 'empty'}]"
+
+    assert wait_until(_peer_request_arrived), (
+        f"peer request from {ds_manager.email} never reached {do_manager.email}. "
+        f"{_peer_states()}. A stale 'accepted' or 'rejected' entry for the DS "
+        f"filters the request out of the folder scan."
+    )
     assert len(do_manager.peer_manager.approved_peers) == 0
     assert do_manager.peer_manager.requested_by_peer_peers[0].email == ds_manager.email