diff options
Diffstat (limited to '')
| -rw-r--r-- | tests/unittest.py | 85 |
1 files changed, 57 insertions, 28 deletions
diff --git a/tests/unittest.py b/tests/unittest.py index 5b19065c71..9afa68c164 100644 --- a/tests/unittest.py +++ b/tests/unittest.py @@ -22,10 +22,11 @@ import secrets import time from typing import ( Any, - AnyStr, + Awaitable, Callable, ClassVar, Dict, + Generic, Iterable, List, Optional, @@ -39,6 +40,7 @@ from unittest.mock import Mock, patch import canonicaljson import signedjson.key import unpaddedbase64 +from typing_extensions import Protocol from twisted.internet.defer import Deferred, ensureDeferred from twisted.python.failure import Failure @@ -49,7 +51,7 @@ from twisted.web.resource import Resource from twisted.web.server import Request from synapse import events -from synapse.api.constants import EventTypes, Membership +from synapse.api.constants import EventTypes from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion from synapse.config.homeserver import HomeServerConfig from synapse.config.server import DEFAULT_ROOM_VERSION @@ -70,7 +72,13 @@ from synapse.types import JsonDict, UserID, create_requester from synapse.util import Clock from synapse.util.httpresourcetree import create_resource_tree -from tests.server import FakeChannel, get_clock, make_request, setup_test_homeserver +from tests.server import ( + CustomHeaderType, + FakeChannel, + get_clock, + make_request, + setup_test_homeserver, +) from tests.test_utils import event_injection, setup_awaitable_errors from tests.test_utils.logging_setup import setup_logging from tests.utils import default_config, setupdb @@ -78,6 +86,17 @@ from tests.utils import default_config, setupdb setupdb() setup_logging() +TV = TypeVar("TV") +_ExcType = TypeVar("_ExcType", bound=BaseException, covariant=True) + + +class _TypedFailure(Generic[_ExcType], Protocol): + """Extension to twisted.Failure, where the 'value' has a certain type.""" + + @property + def value(self) -> _ExcType: + ... + def around(target): """A CLOS-style 'around' modifier, which wraps the original method of the @@ -276,6 +295,7 @@ class HomeserverTestCase(TestCase): if hasattr(self, "user_id"): if self.hijack_auth: + assert self.helper.auth_user_id is not None # We need a valid token ID to satisfy foreign key constraints. token_id = self.get_success( @@ -288,6 +308,7 @@ class HomeserverTestCase(TestCase): ) async def get_user_by_access_token(token=None, allow_guest=False): + assert self.helper.auth_user_id is not None return { "user": UserID.from_string(self.helper.auth_user_id), "token_id": token_id, @@ -295,6 +316,7 @@ class HomeserverTestCase(TestCase): } async def get_user_by_req(request, allow_guest=False, rights="access"): + assert self.helper.auth_user_id is not None return create_requester( UserID.from_string(self.helper.auth_user_id), token_id, @@ -311,7 +333,7 @@ class HomeserverTestCase(TestCase): ) if self.needs_threadpool: - self.reactor.threadpool = ThreadPool() + self.reactor.threadpool = ThreadPool() # type: ignore[assignment] self.addCleanup(self.reactor.threadpool.stop) self.reactor.threadpool.start() @@ -426,7 +448,7 @@ class HomeserverTestCase(TestCase): federation_auth_origin: Optional[bytes] = None, content_is_form: bool = False, await_result: bool = True, - custom_headers: Optional[Iterable[Tuple[AnyStr, AnyStr]]] = None, + custom_headers: Optional[Iterable[CustomHeaderType]] = None, client_ip: str = "127.0.0.1", ) -> FakeChannel: """ @@ -511,30 +533,36 @@ class HomeserverTestCase(TestCase): return hs - def pump(self, by=0.0): + def pump(self, by: float = 0.0) -> None: """ Pump the reactor enough that Deferreds will fire. """ self.reactor.pump([by] * 100) - def get_success(self, d, by=0.0): - deferred: Deferred[TV] = ensureDeferred(d) + def get_success( + self, + d: Awaitable[TV], + by: float = 0.0, + ) -> TV: + deferred: Deferred[TV] = ensureDeferred(d) # type: ignore[arg-type] self.pump(by=by) return self.successResultOf(deferred) - def get_failure(self, d, exc): + def get_failure( + self, d: Awaitable[Any], exc: Type[_ExcType] + ) -> _TypedFailure[_ExcType]: """ Run a Deferred and get a Failure from it. The failure must be of the type `exc`. """ - deferred: Deferred[Any] = ensureDeferred(d) + deferred: Deferred[Any] = ensureDeferred(d) # type: ignore[arg-type] self.pump() return self.failureResultOf(deferred, exc) - def get_success_or_raise(self, d, by=0.0): + def get_success_or_raise(self, d: Awaitable[TV], by: float = 0.0) -> TV: """Drive deferred to completion and return result or raise exception on failure. """ - deferred: Deferred[TV] = ensureDeferred(d) + deferred: Deferred[TV] = ensureDeferred(d) # type: ignore[arg-type] results: list = [] deferred.addBoth(results.append) @@ -642,11 +670,11 @@ class HomeserverTestCase(TestCase): def login( self, - username, - password, - device_id=None, - custom_headers: Optional[Iterable[Tuple[AnyStr, AnyStr]]] = None, - ): + username: str, + password: str, + device_id: Optional[str] = None, + custom_headers: Optional[Iterable[CustomHeaderType]] = None, + ) -> str: """ Log in a user, and get an access token. Requires the Login API be registered. @@ -668,18 +696,22 @@ class HomeserverTestCase(TestCase): return access_token def create_and_send_event( - self, room_id, user, soft_failed=False, prev_event_ids=None - ): + self, + room_id: str, + user: UserID, + soft_failed: bool = False, + prev_event_ids: Optional[List[str]] = None, + ) -> str: """ Create and send an event. Args: - soft_failed (bool): Whether to create a soft failed event or not - prev_event_ids (list[str]|None): Explicitly set the prev events, + soft_failed: Whether to create a soft failed event or not + prev_event_ids: Explicitly set the prev events, or if None just use the default Returns: - str: The new event's ID. + The new event's ID. """ event_creator = self.hs.get_event_creation_handler() requester = create_requester(user) @@ -706,7 +738,7 @@ class HomeserverTestCase(TestCase): return event.event_id - def inject_room_member(self, room: str, user: str, membership: Membership) -> None: + def inject_room_member(self, room: str, user: str, membership: str) -> None: """ Inject a membership event into a room. @@ -766,7 +798,7 @@ class FederatingHomeserverTestCase(HomeserverTestCase): path: str, content: Optional[JsonDict] = None, await_result: bool = True, - custom_headers: Optional[Iterable[Tuple[AnyStr, AnyStr]]] = None, + custom_headers: Optional[Iterable[CustomHeaderType]] = None, client_ip: str = "127.0.0.1", ) -> FakeChannel: """Make an inbound signed federation request to this server @@ -799,7 +831,7 @@ class FederatingHomeserverTestCase(HomeserverTestCase): self.site, method=method, path=path, - content=content, + content=content or "", shorthand=False, await_result=await_result, custom_headers=custom_headers, @@ -878,9 +910,6 @@ def override_config(extra_config): return decorator -TV = TypeVar("TV") - - def skip_unless(condition: bool, reason: str) -> Callable[[TV], TV]: """A test decorator which will skip the decorated test unless a condition is set |