diff --git a/tests/push/test_email.py b/tests/push/test_email.py
index 57b2f0536e..ab8bb417e7 100644
--- a/tests/push/test_email.py
+++ b/tests/push/test_email.py
@@ -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.
"""
|
