Update HomeserverTestCase.get_success(...) and friends to drive async Rust (Tokio runtime/thread pool)

changelog.d/19871.misc

@@ -0,0 +1 @@
+Update `HomeserverTestCase.get_success(...)` and friends to drive async Rust (Tokio runtime/thread pool).

tests/app/test_homeserver_shutdown.py

@@ -76,6 +76,13 @@ async def shutdown() -> None:
 
         self.get_success(shutdown())
 
+        # XXX: There can be a few already dispatched database queries (from normal
+        # background tasks in Synapse) and the threadless `ThreadPool` that we use in
+        # tests uses *untracked* clock calls to pass database results back so `shutdown`
+        # doesn't cancel those calls. This is a quirk of our test infrastructure
+        # (threadless `ThreadPool`) so this kind of "hack" is fine.
+        self.reactor.advance(0)
+
         # Cleanup the internal reference in our test case
         del self.hs
 
@@ -106,7 +113,7 @@ def test_clean_homeserver_shutdown_mid_background_updates(self) -> None:
         # Pump the background updates by a single iteration, just to ensure any extra
         # resources it uses have been started.
         store = weakref.proxy(self.hs.get_datastores().main)
-        self.get_success(store.db_pool.updates.do_next_background_update(False), by=0.1)
+        self.get_success(store.db_pool.updates.do_next_background_update(False))
 
         hs_ref = weakref.ref(self.hs)
 
@@ -127,6 +134,13 @@ async def shutdown() -> None:
 
         self.get_success(shutdown())
 
+        # XXX: There can be a few already dispatched database queries (from normal
+        # background tasks in Synapse) and the threadless `ThreadPool` that we use in
+        # tests uses *untracked* clock calls to pass database results back so `shutdown`
+        # doesn't cancel those calls. This is a quirk of our test infrastructure
+        # (threadless `ThreadPool`) so this kind of "hack" is fine.
+        self.reactor.advance(0)
+
         # Cleanup the internal reference in our test case
         del self.hs
 

tests/handlers/test_federation.py

@@ -357,7 +357,6 @@ def create_invite() -> EventBase:
                 event.room_version,
             ),
             exc=LimitExceededError,
-            by=0.5,
         )
 
     def _build_and_send_join_event(

tests/handlers/test_oauth_delegation.py

@@ -21,11 +21,10 @@
 
 import json
 import threading
-import time
 from http import HTTPStatus
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from io import BytesIO
-from typing import Any, ClassVar, Coroutine, Generator, TypeVar, Union
+from typing import Any, ClassVar, TypeVar
 from unittest.mock import ANY, AsyncMock, Mock
 from urllib.parse import parse_qs
 
@@ -37,7 +36,6 @@
 )
 from signedjson.sign import sign_json
 
-from twisted.internet.defer import Deferred, ensureDeferred
 from twisted.internet.testing import MemoryReactor
 
 from synapse.api.auth.mas import MasDelegatedAuth
@@ -809,31 +807,6 @@ class MasAuthDelegation(HomeserverTestCase):
     def device_scope(self) -> str:
         return self.device_scope_prefix + DEVICE
 
-    def till_deferred_has_result(
-        self,
-        awaitable: Union[
-            "Coroutine[Deferred[Any], Any, T]",
-            "Generator[Deferred[Any], Any, T]",
-            "Deferred[T]",
-        ],
-    ) -> "Deferred[T]":
-        """Wait until a deferred has a result.
-
-        This is useful because the Rust HTTP client will resolve the deferred
-        using reactor.callFromThread, which are only run when we call
-        reactor.advance.
-        """
-        deferred = ensureDeferred(awaitable)
-        tries = 0
-        while not deferred.called:
-            time.sleep(0.1)
-            self.reactor.advance(0)
-            tries += 1
-            if tries > 100:
-                raise Exception("Timed out waiting for deferred to resolve")
-
-        return deferred
-
     def default_config(self) -> dict[str, Any]:
         config = super().default_config()
         config["public_baseurl"] = BASE_URL
@@ -883,11 +856,7 @@ def test_simple_introspection(self) -> None:
             "expires_in": 60,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.user.to_string(), USER_ID)
         self.assertEqual(requester.device_id, DEVICE)
@@ -906,11 +875,7 @@ def test_unexpiring_token(self) -> None:
             "username": USERNAME,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.user.to_string(), USER_ID)
         self.assertEqual(requester.device_id, DEVICE)
@@ -931,9 +896,7 @@ def test_inexistent_device(self) -> None:
         }
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             InvalidClientTokenError,
         )
         self.assertEqual(failure.value.code, 401)
@@ -948,9 +911,7 @@ def test_inexistent_user(self) -> None:
         }
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             AuthError,
         )
         # This is a 500, it should never happen really
@@ -966,9 +927,7 @@ def test_missing_scope(self) -> None:
         }
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             InvalidClientTokenError,
         )
         self.assertEqual(failure.value.code, 401)
@@ -977,9 +936,7 @@ def test_invalid_response(self) -> None:
         self.server.introspection_response = {}
 
         failure = self.get_failure(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            ),
+            self._auth.get_user_by_access_token("some_token"),
             SynapseError,
         )
         self.assertEqual(failure.value.code, 503)
@@ -994,11 +951,7 @@ def test_device_id_in_body(self) -> None:
             "device_id": DEVICE,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.device_id, DEVICE)
 
@@ -1011,11 +964,7 @@ def test_admin_scope(self) -> None:
             "expires_in": 60,
         }
 
-        requester = self.get_success(
-            self.till_deferred_has_result(
-                self._auth.get_user_by_access_token("some_token")
-            )
-        )
+        requester = self.get_success(self._auth.get_user_by_access_token("some_token"))
 
         self.assertEqual(requester.user.to_string(), USER_ID)
         self.assertTrue(self.get_success(self._auth.is_server_admin(requester)))
@@ -1040,17 +989,15 @@ def test_cached_expired_introspection(self) -> None:
         request.requestHeaders.getRawHeaders = mock_getRawHeaders()
 
         # The first CS-API request causes a successful introspection
-        self.get_success(
-            self.till_deferred_has_result(self._auth.get_user_by_req(request))
-        )
+        self.get_success(self._auth.get_user_by_req(request))
         self.assertEqual(self.server.calls, 1)
 
         # Sleep for 60 seconds so the token expires.
         self.reactor.advance(60.0)
 
         # Now the CS-API request fails because the token expired
-        self.assertFailure(
-            self.till_deferred_has_result(self._auth.get_user_by_req(request)),
+        self.get_failure(
+            self._auth.get_user_by_req(request),
             InvalidClientTokenError,
         )
         # Ensure another introspection request was not sent

tests/handlers/test_presence.py

@@ -29,6 +29,7 @@
     get_verify_key,
 )
 
+from twisted.internet.defer import ensureDeferred
 from twisted.internet.testing import MemoryReactor
 
 from synapse.api.constants import EventTypes, Membership, PresenceState
@@ -58,6 +59,7 @@
 from synapse.storage.keys import FetchKeyResult
 from synapse.types import JsonDict, UserID, get_domain_from_id
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 
 from tests import unittest
 from tests.replication._base import BaseMultiWorkerStreamTestCase
@@ -948,12 +950,17 @@ def test_external_process_timeout(self) -> None:
         )
         worker_presence_handler = worker_to_sync_against.get_presence_handler()
 
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 self.user_id, self.device_id, True, PresenceState.ONLINE
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # Check that if we wait a while without telling the handler the user has
         # stopped syncing that their presence state doesn't get timed out.
@@ -1264,30 +1271,40 @@ def test_set_presence_from_syncing_multi_device(
             worker_presence_handler = worker_to_sync_against.get_presence_handler()
 
         # 1. Sync with the first device.
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-1",
                 affect_presence=dev_1_state != PresenceState.OFFLINE,
                 presence_state=dev_1_state,
-            ),
-            by=0.01,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # 2. Wait half the idle timer.
         self.reactor.advance(IDLE_TIMER / 1000 / 2)
         self.reactor.pump([0.1])
 
         # 3. Sync with the second device.
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-2",
                 affect_presence=dev_2_state != PresenceState.OFFLINE,
                 presence_state=dev_2_state,
-            ),
-            by=0.01,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # 4. Assert the expected presence state.
         state = self.get_success(
@@ -1305,15 +1322,21 @@ def test_set_presence_from_syncing_multi_device(
         #
         # This is due to EXTERNAL_PROCESS_EXPIRY being equivalent to IDLE_TIMER.
         if test_with_workers:
-            with self.get_success(
+            sync_d = ensureDeferred(
                 worker_presence_handler.user_syncing(
                     f"@other-user:{self.hs.config.server.server_name}",
                     "dev-3",
                     affect_presence=True,
                     presence_state=PresenceState.ONLINE,
-                ),
-                by=0.01,
-            ):
+                )
+            )
+            # `user_syncing` proxies the presence write to the main process over an HTTP
+            # replication request. The request body is streamed by a `Cooperator` that uses
+            # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+            # we need to actually advance the clock for it to fire.
+            self.reactor.advance(Duration(microseconds=1).as_secs())
+
+            with self.get_success(sync_d):
                 pass
 
         # 5. Advance such that the first device should be discarded (the idle timer),
@@ -1501,26 +1524,36 @@ def test_set_presence_from_non_syncing_multi_device(
             worker_presence_handler = worker_to_sync_against.get_presence_handler()
 
         # 1. Sync with the first device.
-        sync_1 = self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-1",
                 affect_presence=dev_1_state != PresenceState.OFFLINE,
                 presence_state=dev_1_state,
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        sync_1 = self.get_success(sync_d)
 
         # 2. Sync with the second device.
-        sync_2 = self.get_success(
+        sync_d = ensureDeferred(
             worker_presence_handler.user_syncing(
                 user_id,
                 "dev-2",
                 affect_presence=dev_2_state != PresenceState.OFFLINE,
                 presence_state=dev_2_state,
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        sync_2 = self.get_success(sync_d)
 
         # 3. Assert the expected presence state.
         state = self.get_success(
@@ -1622,12 +1655,17 @@ def test_set_presence_from_syncing_keeps_busy(
         # Perform a sync with a presence state other than busy. This should NOT change
         # our presence status; we only change from busy if we explicitly set it via
         # /presence/*.
-        self.get_success(
+        sync_d = ensureDeferred(
             worker_to_sync_against.get_presence_handler().user_syncing(
                 self.user_id, self.device_id, True, PresenceState.ONLINE
-            ),
-            by=0.1,
+            )
         )
+        # `user_syncing` proxies the presence write to the main process over an HTTP
+        # replication request. The request body is streamed by a `Cooperator` that uses
+        # the clock to schedule each chunk at a tiny *non-zero* delay (`_EPSILON`), so
+        # we need to actually advance the clock for it to fire.
+        self.reactor.advance(Duration(microseconds=1).as_secs())
+        self.get_success(sync_d)
 
         # Check against the main process that the user's presence did not change.
         state = self.get_success(self.presence_handler.get_state(self.user_id_obj))