Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-05-23 13:11:07 -04:00
Code Issues Projects Releases Packages Wiki Activity

Fix-up type hints for tests.push module. (#14816)

Browse source
This commit is contained in:
Patrick Cloke 2023-01-11 07:35:40 -05:00 • committed by GitHub
parent d6bda5addd
commit 7f2cabf271
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 67 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

46
tests/push/test_email.py
View file

@ -13,25 +13,28 @@
# limitations under the License.
import email.message
import os
from typing import Dict, List, Sequence, Tuple
from typing import Any, Dict, List, Sequence, Tuple
import attr
import pkg_resources
from twisted.internet.defer import Deferred
from twisted.test.proto_helpers import MemoryReactor
import synapse.rest.admin
from synapse.api.errors import Codes, SynapseError
from synapse.rest.client import login, room
from synapse.server import HomeServer
from synapse.util import Clock
from tests.unittest import HomeserverTestCase
@attr.s
@attr.s(auto_attribs=True)
class _User:
"Helper wrapper for user ID and access token"
id = attr.ib()
token = attr.ib()
id: str
token: str
class EmailPusherTests(HomeserverTestCase):
@ -41,10 +44,9 @@ class EmailPusherTests(HomeserverTestCase):
room.register_servlets,
login.register_servlets,
]
user_id = True
hijack_auth = False
def make_homeserver(self, reactor, clock):
def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
config = self.default_config()
config["email"] = {
@ -72,17 +74,17 @@ class EmailPusherTests(HomeserverTestCase):
# List[Tuple[Deferred, args, kwargs]]
self.email_attempts: List[Tuple[Deferred, Sequence, Dict]] = []
def sendmail(*args, **kwargs):
def sendmail(*args: Any, **kwargs: Any) -> Deferred:
# This mocks out synapse.reactor.send_email._sendmail.
d = Deferred()
d: Deferred = Deferred()
self.email_attempts.append((d, args, kwargs))
return d
hs.get_send_email_handler()._sendmail = sendmail
hs.get_send_email_handler()._sendmail = sendmail # type: ignore[assignment]
return hs
def prepare(self, reactor, clock, hs):
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
# Register the user who gets notified
self.user_id = self.register_user("user", "pass")
self.access_token = self.login("user", "pass")
@ -129,7 +131,7 @@ class EmailPusherTests(HomeserverTestCase):
self.auth_handler = hs.get_auth_handler()
self.store = hs.get_datastores().main
def test_need_validated_email(self):
def test_need_validated_email(self) -> None:
"""Test that we can only add an email pusher if the user has validated
their email.
"""
@ -151,7 +153,7 @@ class EmailPusherTests(HomeserverTestCase):
self.assertEqual(400, cm.exception.code)
self.assertEqual(Codes.THREEPID_NOT_FOUND, cm.exception.errcode)
def test_simple_sends_email(self):
def test_simple_sends_email(self) -> None:
# Create a simple room with two users
room = self.helper.create_room_as(self.user_id, tok=self.access_token)
self.helper.invite(
@ -171,7 +173,7 @@ class EmailPusherTests(HomeserverTestCase):
self._check_for_mail()
def test_invite_sends_email(self):
def test_invite_sends_email(self) -> None:
# Create a room and invite the user to it
room = self.helper.create_room_as(self.others[0].id, tok=self.others[0].token)
self.helper.invite(
@ -184,7 +186,7 @@ class EmailPusherTests(HomeserverTestCase):
# We should get emailed about the invite
self._check_for_mail()
def test_invite_to_empty_room_sends_email(self):
def test_invite_to_empty_room_sends_email(self) -> None:
# Create a room and invite the user to it
room = self.helper.create_room_as(self.others[0].id, tok=self.others[0].token)
self.helper.invite(
@ -200,7 +202,7 @@ class EmailPusherTests(HomeserverTestCase):
# We should get emailed about the invite
self._check_for_mail()
def test_multiple_members_email(self):
def test_multiple_members_email(self) -> None:
# We want to test multiple notifications, so we pause processing of push
# while we send messages.
self.pusher._pause_processing()
@ -227,7 +229,7 @@ class EmailPusherTests(HomeserverTestCase):
# We should get emailed about those messages
self._check_for_mail()
def test_multiple_rooms(self):
def test_multiple_rooms(self) -> None:
# We want to test multiple notifications from multiple rooms, so we pause
# processing of push while we send messages.
self.pusher._pause_processing()
@ -257,7 +259,7 @@ class EmailPusherTests(HomeserverTestCase):
# We should get emailed about those messages
self._check_for_mail()
def test_room_notifications_include_avatar(self):
def test_room_notifications_include_avatar(self) -> None:
# Create a room and set its avatar.
room = self.helper.create_room_as(self.user_id, tok=self.access_token)
self.helper.send_state(
@ -290,7 +292,7 @@ class EmailPusherTests(HomeserverTestCase):
)
self.assertIn("_matrix/media/v1/thumbnail/DUMMY_MEDIA_ID", html)
def test_empty_room(self):
def test_empty_room(self) -> None:
"""All users leaving a room shouldn't cause the pusher to break."""
# Create a simple room with two users
room = self.helper.create_room_as(self.user_id, tok=self.access_token)
@ -309,7 +311,7 @@ class EmailPusherTests(HomeserverTestCase):
# We should get emailed about that message
self._check_for_mail()
def test_empty_room_multiple_messages(self):
def test_empty_room_multiple_messages(self) -> None:
"""All users leaving a room shouldn't cause the pusher to break."""
# Create a simple room with two users
room = self.helper.create_room_as(self.user_id, tok=self.access_token)
@ -329,7 +331,7 @@ class EmailPusherTests(HomeserverTestCase):
# We should get emailed about that message
self._check_for_mail()
def test_encrypted_message(self):
def test_encrypted_message(self) -> None:
room = self.helper.create_room_as(self.user_id, tok=self.access_token)
self.helper.invite(
room=room, src=self.user_id, tok=self.access_token, targ=self.others[0].id
@ -342,7 +344,7 @@ class EmailPusherTests(HomeserverTestCase):
# We should get emailed about that message
self._check_for_mail()
def test_no_email_sent_after_removed(self):
def test_no_email_sent_after_removed(self) -> None:
# Create a simple room with two users
room = self.helper.create_room_as(self.user_id, tok=self.access_token)
self.helper.invite(
@ -379,7 +381,7 @@ class EmailPusherTests(HomeserverTestCase):
pushers = list(pushers)
self.assertEqual(len(pushers), 0)
def test_remove_unlinked_pushers_background_job(self):
def test_remove_unlinked_pushers_background_job(self) -> None:
"""Checks that all existing pushers associated with unlinked email addresses are removed
upon running the remove_deleted_email_pushers background update.
"""