diff --git a/tests/federation/transport/test_knocking.py b/tests/federation/transport/test_knocking.py
index ff589c0b6c..70209ab090 100644
--- a/tests/federation/transport/test_knocking.py
+++ b/tests/federation/transport/test_knocking.py
@@ -12,15 +12,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from collections import OrderedDict
-from typing import Dict, List
+from typing import Any, Dict, List, Optional
+
+from twisted.test.proto_helpers import MemoryReactor
from synapse.api.constants import EventTypes, JoinRules, Membership
-from synapse.api.room_versions import RoomVersions
-from synapse.events import builder
+from synapse.api.room_versions import RoomVersion, RoomVersions
+from synapse.events import EventBase, builder
+from synapse.events.snapshot import EventContext
from synapse.rest import admin
from synapse.rest.client import login, room
from synapse.server import HomeServer
from synapse.types import RoomAlias
+from synapse.util import Clock
from tests.test_utils import event_injection
from tests.unittest import FederatingHomeserverTestCase, HomeserverTestCase
@@ -197,7 +201,9 @@ class FederationKnockingTestCase(
login.register_servlets,
]
- def prepare(self, reactor, clock, homeserver):
+ def prepare(
+ self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
+ ) -> None:
self.store = homeserver.get_datastores().main
# We're not going to be properly signing events as our remote homeserver is fake,
@@ -205,23 +211,29 @@ class FederationKnockingTestCase(
# Note that these checks are not relevant to this test case.
# Have this homeserver auto-approve all event signature checking.
- async def approve_all_signature_checking(_, pdu):
+ async def approve_all_signature_checking(
+ room_version: RoomVersion,
+ pdu: EventBase,
+ record_failure_callback: Any = None,
+ ) -> EventBase:
return pdu
- homeserver.get_federation_server()._check_sigs_and_hash = (
+ homeserver.get_federation_server()._check_sigs_and_hash = ( # type: ignore[assignment]
approve_all_signature_checking
)
# Have this homeserver skip event auth checks. This is necessary due to
# event auth checks ensuring that events were signed by the sender's homeserver.
- async def _check_event_auth(origin, event, context, *args, **kwargs):
- return context
+ async def _check_event_auth(
+ origin: Optional[str], event: EventBase, context: EventContext
+ ) -> None:
+ pass
- homeserver.get_federation_event_handler()._check_event_auth = _check_event_auth
+ homeserver.get_federation_event_handler()._check_event_auth = _check_event_auth # type: ignore[assignment]
return super().prepare(reactor, clock, homeserver)
- def test_room_state_returned_when_knocking(self):
+ def test_room_state_returned_when_knocking(self) -> None:
"""
Tests that specific, stripped state events from a room are returned after
a remote homeserver successfully knocks on a local room.
|
