Merge remote-tracking branch 'origin/develop' into hs/hacked-together-event-cache

14 files changed, 282 insertions, 236 deletions

diff --git a/tests/app/test_frontend_proxy.py b/tests/app/test_frontend_proxy.py
deleted file mode 100644
index 3d45da38ab..0000000000
--- a/tests/app/test_frontend_proxy.py
+++ /dev/null
@@ -1,83 +0,0 @@
-# Copyright 2018 New Vector Ltd
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-from synapse.app.generic_worker import GenericWorkerServer
-
-from tests.server import make_request
-from tests.unittest import HomeserverTestCase
-
-
-class FrontendProxyTests(HomeserverTestCase):
-    def make_homeserver(self, reactor, clock):
-
-        hs = self.setup_test_homeserver(
-            federation_http_client=None, homeserver_to_use=GenericWorkerServer
-        )
-
-        return hs
-
-    def default_config(self):
-        c = super().default_config()
-        c["worker_app"] = "synapse.app.frontend_proxy"
-
-        c["worker_listeners"] = [
-            {
-                "type": "http",
-                "port": 8080,
-                "bind_addresses": ["0.0.0.0"],
-                "resources": [{"names": ["client"]}],
-            }
-        ]
-
-        return c
-
-    def test_listen_http_with_presence_enabled(self):
-        """
-        When presence is on, the stub servlet will not register.
-        """
-        # Presence is on
-        self.hs.config.use_presence = True
-
-        # Listen with the config
-        self.hs._listen_http(self.hs.config.worker.worker_listeners[0])
-
-        # Grab the resource from the site that was told to listen
-        self.assertEqual(len(self.reactor.tcpServers), 1)
-        site = self.reactor.tcpServers[0][1]
-
-        channel = make_request(self.reactor, site, "PUT", "presence/a/status")
-
-        # 400 + unrecognised, because nothing is registered
-        self.assertEqual(channel.code, 400)
-        self.assertEqual(channel.json_body["errcode"], "M_UNRECOGNIZED")
-
-    def test_listen_http_with_presence_disabled(self):
-        """
-        When presence is off, the stub servlet will register.
-        """
-        # Presence is off
-        self.hs.config.use_presence = False
-
-        # Listen with the config
-        self.hs._listen_http(self.hs.config.worker.worker_listeners[0])
-
-        # Grab the resource from the site that was told to listen
-        self.assertEqual(len(self.reactor.tcpServers), 1)
-        site = self.reactor.tcpServers[0][1]
-
-        channel = make_request(self.reactor, site, "PUT", "presence/a/status")
-
-        # 401, because the stub servlet still checks authentication
-        self.assertEqual(channel.code, 401)
-        self.assertEqual(channel.json_body["errcode"], "M_MISSING_TOKEN")
diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py
index 61271cd084..ce330e79cc 100644
--- a/tests/handlers/test_presence.py
+++ b/tests/handlers/test_presence.py
@@ -509,6 +509,14 @@ class PresenceFederationQueueTestCase(unittest.HomeserverTestCase):
 
         self.assertCountEqual(rows, expected_rows)
 
+        now_token = self.queue.get_current_token(self.instance_name)
+        rows, upto_token, limited = self.get_success(
+            self.queue.get_replication_rows("master", upto_token, now_token, 10)
+        )
+        self.assertEqual(upto_token, now_token)
+        self.assertFalse(limited)
+        self.assertCountEqual(rows, [])
+
     def test_send_and_get_split(self):
         state1 = UserPresenceState.default("@user1:test")
         state2 = UserPresenceState.default("@user2:test")
@@ -538,6 +546,20 @@ class PresenceFederationQueueTestCase(unittest.HomeserverTestCase):
 
         self.assertCountEqual(rows, expected_rows)
 
+        now_token = self.queue.get_current_token(self.instance_name)
+        rows, upto_token, limited = self.get_success(
+            self.queue.get_replication_rows("master", upto_token, now_token, 10)
+        )
+
+        self.assertEqual(upto_token, now_token)
+        self.assertFalse(limited)
+
+        expected_rows = [
+            (2, ("dest3", "@user3:test")),
+        ]
+
+        self.assertCountEqual(rows, expected_rows)
+
     def test_clear_queue_all(self):
         state1 = UserPresenceState.default("@user1:test")
         state2 = UserPresenceState.default("@user2:test")
diff --git a/tests/http/test_fedclient.py b/tests/http/test_fedclient.py
index 9e97185507..ed9a884d76 100644
--- a/tests/http/test_fedclient.py
+++ b/tests/http/test_fedclient.py
@@ -26,6 +26,7 @@ from twisted.web.http import HTTPChannel
 
 from synapse.api.errors import RequestSendFailed
 from synapse.http.matrixfederationclient import (
+    MAX_RESPONSE_SIZE,
     MatrixFederationHttpClient,
     MatrixFederationRequest,
 )
@@ -560,3 +561,61 @@ class FederationClientTests(HomeserverTestCase):
 
         f = self.failureResultOf(test_d)
         self.assertIsInstance(f.value, RequestSendFailed)
+
+    def test_too_big(self):
+        """
+        Test what happens if a huge response is returned from the remote endpoint.
+        """
+
+        test_d = defer.ensureDeferred(self.cl.get_json("testserv:8008", "foo/bar"))
+
+        self.pump()
+
+        # Nothing happened yet
+        self.assertNoResult(test_d)
+
+        # Make sure treq is trying to connect
+        clients = self.reactor.tcpClients
+        self.assertEqual(len(clients), 1)
+        (host, port, factory, _timeout, _bindAddress) = clients[0]
+        self.assertEqual(host, "1.2.3.4")
+        self.assertEqual(port, 8008)
+
+        # complete the connection and wire it up to a fake transport
+        protocol = factory.buildProtocol(None)
+        transport = StringTransport()
+        protocol.makeConnection(transport)
+
+        # that should have made it send the request to the transport
+        self.assertRegex(transport.value(), b"^GET /foo/bar")
+        self.assertRegex(transport.value(), b"Host: testserv:8008")
+
+        # Deferred is still without a result
+        self.assertNoResult(test_d)
+
+        # Send it a huge HTTP response
+        protocol.dataReceived(
+            b"HTTP/1.1 200 OK\r\n"
+            b"Server: Fake\r\n"
+            b"Content-Type: application/json\r\n"
+            b"\r\n"
+        )
+
+        self.pump()
+
+        # should still be waiting
+        self.assertNoResult(test_d)
+
+        sent = 0
+        chunk_size = 1024 * 512
+        while not test_d.called:
+            protocol.dataReceived(b"a" * chunk_size)
+            sent += chunk_size
+            self.assertLessEqual(sent, MAX_RESPONSE_SIZE)
+
+        self.assertEqual(sent, MAX_RESPONSE_SIZE)
+
+        f = self.failureResultOf(test_d)
+        self.assertIsInstance(f.value, RequestSendFailed)
+
+        self.assertTrue(transport.disconnecting)
diff --git a/tests/http/test_site.py b/tests/http/test_site.py
new file mode 100644
index 0000000000..8c13b4f693
--- /dev/null
+++ b/tests/http/test_site.py
@@ -0,0 +1,83 @@
+# Copyright 2021 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from twisted.internet.address import IPv6Address
+from twisted.test.proto_helpers import StringTransport
+
+from synapse.app.homeserver import SynapseHomeServer
+
+from tests.unittest import HomeserverTestCase
+
+
+class SynapseRequestTestCase(HomeserverTestCase):
+    def make_homeserver(self, reactor, clock):
+        return self.setup_test_homeserver(homeserver_to_use=SynapseHomeServer)
+
+    def test_large_request(self):
+        """overlarge HTTP requests should be rejected"""
+        self.hs.start_listening()
+
+        # find the HTTP server which is configured to listen on port 0
+        (port, factory, _backlog, interface) = self.reactor.tcpServers[0]
+        self.assertEqual(interface, "::")
+        self.assertEqual(port, 0)
+
+        # as a control case, first send a regular request.
+
+        # complete the connection and wire it up to a fake transport
+        client_address = IPv6Address("TCP", "::1", "2345")
+        protocol = factory.buildProtocol(client_address)
+        transport = StringTransport()
+        protocol.makeConnection(transport)
+
+        protocol.dataReceived(
+            b"POST / HTTP/1.1\r\n"
+            b"Connection: close\r\n"
+            b"Transfer-Encoding: chunked\r\n"
+            b"\r\n"
+            b"0\r\n"
+            b"\r\n"
+        )
+
+        while not transport.disconnecting:
+            self.reactor.advance(1)
+
+        # we should get a 404
+        self.assertRegex(transport.value().decode(), r"^HTTP/1\.1 404 ")
+
+        # now send an oversized request
+        protocol = factory.buildProtocol(client_address)
+        transport = StringTransport()
+        protocol.makeConnection(transport)
+
+        protocol.dataReceived(
+            b"POST / HTTP/1.1\r\n"
+            b"Connection: close\r\n"
+            b"Transfer-Encoding: chunked\r\n"
+            b"\r\n"
+        )
+
+        # we deliberately send all the data in one big chunk, to ensure that
+        # twisted isn't buffering the data in the chunked transfer decoder.
+        # we start with the chunk size, in hex. (We won't actually send this much)
+        protocol.dataReceived(b"10000000\r\n")
+        sent = 0
+        while not transport.disconnected:
+            self.assertLess(sent, 0x10000000, "connection did not drop")
+            protocol.dataReceived(b"\0" * 1024)
+            sent += 1024
+
+        # default max upload size is 50M, so it should drop on the next buffer after
+        # that.
+        self.assertEqual(sent, 50 * 1024 * 1024 + 1024)
diff --git a/tests/replication/_base.py b/tests/replication/_base.py
index 6ec4e6c867..802c3ec724 100644
--- a/tests/replication/_base.py
+++ b/tests/replication/_base.py
@@ -12,14 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import Any, Callable, Dict, List, Optional, Tuple, Type
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
-from twisted.internet.interfaces import IConsumer, IPullProducer, IReactorTime
 from twisted.internet.protocol import Protocol
-from twisted.internet.task import LoopingCall
-from twisted.web.http import HTTPChannel
 from twisted.web.resource import Resource
-from twisted.web.server import Request, Site
 
 from synapse.app.generic_worker import GenericWorkerServer
 from synapse.http.server import JsonResource
@@ -33,7 +29,6 @@ from synapse.replication.tcp.resource import (
     ServerReplicationStreamProtocol,
 )
 from synapse.server import HomeServer
-from synapse.util import Clock
 
 from tests import unittest
 from tests.server import FakeTransport
@@ -154,7 +149,19 @@ class BaseStreamTestCase(unittest.HomeserverTestCase):
         client_protocol = client_factory.buildProtocol(None)
 
         # Set up the server side protocol
-        channel = _PushHTTPChannel(self.reactor, SynapseRequest, self.site)
+        channel = self.site.buildProtocol(None)
+
+        # hook into the channel's request factory so that we can keep a record
+        # of the requests
+        requests: List[SynapseRequest] = []
+        real_request_factory = channel.requestFactory
+
+        def request_factory(*args, **kwargs):
+            request = real_request_factory(*args, **kwargs)
+            requests.append(request)
+            return request
+
+        channel.requestFactory = request_factory
 
         # Connect client to server and vice versa.
         client_to_server_transport = FakeTransport(
@@ -176,7 +183,10 @@ class BaseStreamTestCase(unittest.HomeserverTestCase):
         server_to_client_transport.loseConnection()
         client_to_server_transport.loseConnection()
 
-        return channel.request
+        # there should have been exactly one request
+        self.assertEqual(len(requests), 1)
+
+        return requests[0]
 
     def assert_request_is_get_repl_stream_updates(
         self, request: SynapseRequest, stream_name: str
@@ -349,6 +359,8 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
             config=worker_hs.config.server.listeners[0],
             resource=resource,
             server_version_string="1",
+            max_request_body_size=4096,
+            reactor=self.reactor,
         )
 
         if worker_hs.config.redis.redis_enabled:
@@ -386,7 +398,7 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
         client_protocol = client_factory.buildProtocol(None)
 
         # Set up the server side protocol
-        channel = _PushHTTPChannel(self.reactor, SynapseRequest, self._hs_to_site[hs])
+        channel = self._hs_to_site[hs].buildProtocol(None)
 
         # Connect client to server and vice versa.
         client_to_server_transport = FakeTransport(
@@ -444,112 +456,6 @@ class TestReplicationDataHandler(ReplicationDataHandler):
             self.received_rdata_rows.append((stream_name, token, r))
 
 
-class _PushHTTPChannel(HTTPChannel):
-    """A HTTPChannel that wraps pull producers to push producers.
-
-    This is a hack to get around the fact that HTTPChannel transparently wraps a
-    pull producer (which is what Synapse uses to reply to requests) with
-    `_PullToPush` to convert it to a push producer. Unfortunately `_PullToPush`
-    uses the standard reactor rather than letting us use our test reactor, which
-    makes it very hard to test.
-    """
-
-    def __init__(
-        self, reactor: IReactorTime, request_factory: Type[Request], site: Site
-    ):
-        super().__init__()
-        self.reactor = reactor
-        self.requestFactory = request_factory
-        self.site = site
-
-        self._pull_to_push_producer = None  # type: Optional[_PullToPushProducer]
-
-    def registerProducer(self, producer, streaming):
-        # Convert pull producers to push producer.
-        if not streaming:
-            self._pull_to_push_producer = _PullToPushProducer(
-                self.reactor, producer, self
-            )
-            producer = self._pull_to_push_producer
-
-        super().registerProducer(producer, True)
-
-    def unregisterProducer(self):
-        if self._pull_to_push_producer:
-            # We need to manually stop the _PullToPushProducer.
-            self._pull_to_push_producer.stop()
-
-    def checkPersistence(self, request, version):
-        """Check whether the connection can be re-used"""
-        # We hijack this to always say no for ease of wiring stuff up in
-        # `handle_http_replication_attempt`.
-        request.responseHeaders.setRawHeaders(b"connection", [b"close"])
-        return False
-
-    def requestDone(self, request):
-        # Store the request for inspection.
-        self.request = request
-        super().requestDone(request)
-
-
-class _PullToPushProducer:
-    """A push producer that wraps a pull producer."""
-
-    def __init__(
-        self, reactor: IReactorTime, producer: IPullProducer, consumer: IConsumer
-    ):
-        self._clock = Clock(reactor)
-        self._producer = producer
-        self._consumer = consumer
-
-        # While running we use a looping call with a zero delay to call
-        # resumeProducing on given producer.
-        self._looping_call = None  # type: Optional[LoopingCall]
-
-        # We start writing next reactor tick.
-        self._start_loop()
-
-    def _start_loop(self):
-        """Start the looping call to"""
-
-        if not self._looping_call:
-            # Start a looping call which runs every tick.
-            self._looping_call = self._clock.looping_call(self._run_once, 0)
-
-    def stop(self):
-        """Stops calling resumeProducing."""
-        if self._looping_call:
-            self._looping_call.stop()
-            self._looping_call = None
-
-    def pauseProducing(self):
-        """Implements IPushProducer"""
-        self.stop()
-
-    def resumeProducing(self):
-        """Implements IPushProducer"""
-        self._start_loop()
-
-    def stopProducing(self):
-        """Implements IPushProducer"""
-        self.stop()
-        self._producer.stopProducing()
-
-    def _run_once(self):
-        """Calls resumeProducing on producer once."""
-
-        try:
-            self._producer.resumeProducing()
-        except Exception:
-            logger.exception("Failed to call resumeProducing")
-            try:
-                self._consumer.unregisterProducer()
-            except Exception:
-                pass
-
-            self.stopProducing()
-
-
 class FakeRedisPubSubServer:
     """A fake Redis server for pub/sub."""
 
diff --git a/tests/rest/admin/test_room.py b/tests/rest/admin/test_room.py
index 6b84188120..ee071c2477 100644
--- a/tests/rest/admin/test_room.py
+++ b/tests/rest/admin/test_room.py
@@ -17,6 +17,8 @@ import urllib.parse
 from typing import List, Optional
 from unittest.mock import Mock
 
+from parameterized import parameterized_class
+
 import synapse.rest.admin
 from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import Codes
@@ -144,6 +146,13 @@ class ShutdownRoomTestCase(unittest.HomeserverTestCase):
         )
 
 
+@parameterized_class(
+    ("method", "url_template"),
+    [
+        ("POST", "/_synapse/admin/v1/rooms/%s/delete"),
+        ("DELETE", "/_synapse/admin/v1/rooms/%s"),
+    ],
+)
 class DeleteRoomTestCase(unittest.HomeserverTestCase):
     servlets = [
         synapse.rest.admin.register_servlets,
@@ -175,7 +184,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         self.room_id = self.helper.create_room_as(
             self.other_user, tok=self.other_user_tok
         )
-        self.url = "/_synapse/admin/v1/rooms/%s/delete" % self.room_id
+        self.url = self.url_template % self.room_id
 
     def test_requester_is_no_admin(self):
         """
@@ -183,7 +192,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         """
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url,
             json.dumps({}),
             access_token=self.other_user_tok,
@@ -196,10 +205,10 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         """
         Check that unknown rooms/server return error 404.
         """
-        url = "/_synapse/admin/v1/rooms/!unknown:test/delete"
+        url = self.url_template % "!unknown:test"
 
         channel = self.make_request(
-            "POST",
+            self.method,
             url,
             json.dumps({}),
             access_token=self.admin_user_tok,
@@ -212,10 +221,10 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         """
         Check that invalid room names, return an error 400.
         """
-        url = "/_synapse/admin/v1/rooms/invalidroom/delete"
+        url = self.url_template % "invalidroom"
 
         channel = self.make_request(
-            "POST",
+            self.method,
             url,
             json.dumps({}),
             access_token=self.admin_user_tok,
@@ -234,7 +243,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         body = json.dumps({"new_room_user_id": "@unknown:test"})
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url,
             content=body.encode(encoding="utf_8"),
             access_token=self.admin_user_tok,
@@ -253,7 +262,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         body = json.dumps({"new_room_user_id": "@not:exist.bla"})
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url,
             content=body.encode(encoding="utf_8"),
             access_token=self.admin_user_tok,
@@ -272,7 +281,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         body = json.dumps({"block": "NotBool"})
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url,
             content=body.encode(encoding="utf_8"),
             access_token=self.admin_user_tok,
@@ -288,7 +297,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         body = json.dumps({"purge": "NotBool"})
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url,
             content=body.encode(encoding="utf_8"),
             access_token=self.admin_user_tok,
@@ -314,7 +323,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         body = json.dumps({"block": True, "purge": True})
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url.encode("ascii"),
             content=body.encode(encoding="utf_8"),
             access_token=self.admin_user_tok,
@@ -347,7 +356,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         body = json.dumps({"block": False, "purge": True})
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url.encode("ascii"),
             content=body.encode(encoding="utf_8"),
             access_token=self.admin_user_tok,
@@ -381,7 +390,7 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         body = json.dumps({"block": False, "purge": False})
 
         channel = self.make_request(
-            "POST",
+            self.method,
             self.url.encode("ascii"),
             content=body.encode(encoding="utf_8"),
             access_token=self.admin_user_tok,
@@ -426,10 +435,9 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         self._is_member(room_id=self.room_id, user_id=self.other_user)
 
         # Test that the admin can still send shutdown
-        url = "/_synapse/admin/v1/rooms/%s/delete" % self.room_id
         channel = self.make_request(
-            "POST",
-            url.encode("ascii"),
+            self.method,
+            self.url,
             json.dumps({"new_room_user_id": self.admin_user}),
             access_token=self.admin_user_tok,
         )
@@ -473,10 +481,9 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         self._is_member(room_id=self.room_id, user_id=self.other_user)
 
         # Test that the admin can still send shutdown
-        url = "/_synapse/admin/v1/rooms/%s/delete" % self.room_id
         channel = self.make_request(
-            "POST",
-            url.encode("ascii"),
+            self.method,
+            self.url,
             json.dumps({"new_room_user_id": self.admin_user}),
             access_token=self.admin_user_tok,
         )
diff --git a/tests/rest/admin/test_user.py b/tests/rest/admin/test_user.py
index b3afd51522..d599a4c984 100644
--- a/tests/rest/admin/test_user.py
+++ b/tests/rest/admin/test_user.py
@@ -18,7 +18,7 @@ import json
 import urllib.parse
 from binascii import unhexlify
 from typing import List, Optional
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import synapse.rest.admin
 from synapse.api.constants import UserTypes
@@ -54,8 +54,6 @@ class UserRegisterTestCase(unittest.HomeserverTestCase):
         self.datastore = Mock(return_value=Mock())
         self.datastore.get_current_state_deltas = Mock(return_value=(0, []))
 
-        self.secrets = Mock()
-
         self.hs = self.setup_test_homeserver()
 
         self.hs.config.registration_shared_secret = "shared"
@@ -84,14 +82,13 @@ class UserRegisterTestCase(unittest.HomeserverTestCase):
         Calling GET on the endpoint will return a randomised nonce, using the
         homeserver's secrets provider.
         """
-        secrets = Mock()
-        secrets.token_hex = Mock(return_value="abcd")
-
-        self.hs.get_secrets = Mock(return_value=secrets)
+        with patch("secrets.token_hex") as token_hex:
+            # Patch secrets.token_hex for the duration of this context
+            token_hex.return_value = "abcd"
 
-        channel = self.make_request("GET", self.url)
+            channel = self.make_request("GET", self.url)
 
-        self.assertEqual(channel.json_body, {"nonce": "abcd"})
+            self.assertEqual(channel.json_body, {"nonce": "abcd"})
 
     def test_expired_nonce(self):
         """
diff --git a/tests/rest/client/v1/test_presence.py b/tests/rest/client/v1/test_presence.py
index 3a050659ca..409f3949dc 100644
--- a/tests/rest/client/v1/test_presence.py
+++ b/tests/rest/client/v1/test_presence.py
@@ -16,6 +16,7 @@ from unittest.mock import Mock
 
 from twisted.internet import defer
 
+from synapse.handlers.presence import PresenceHandler
 from synapse.rest.client.v1 import presence
 from synapse.types import UserID
 
@@ -32,7 +33,7 @@ class PresenceTestCase(unittest.HomeserverTestCase):
 
     def make_homeserver(self, reactor, clock):
 
-        presence_handler = Mock()
+        presence_handler = Mock(spec=PresenceHandler)
         presence_handler.set_state.return_value = defer.succeed(None)
 
         hs = self.setup_test_homeserver(
@@ -59,12 +60,12 @@ class PresenceTestCase(unittest.HomeserverTestCase):
         self.assertEqual(channel.code, 200)
         self.assertEqual(self.hs.get_presence_handler().set_state.call_count, 1)
 
+    @unittest.override_config({"use_presence": False})
     def test_put_presence_disabled(self):
         """
         PUT to the status endpoint with use_presence disabled will NOT call
         set_state on the presence handler.
         """
-        self.hs.config.use_presence = False
 
         body = {"presence": "here", "status_msg": "beep boop"}
         channel = self.make_request(
diff --git a/tests/rest/client/v2_alpha/test_register.py b/tests/rest/client/v2_alpha/test_register.py
index 98695b05d5..1cad5f00eb 100644
--- a/tests/rest/client/v2_alpha/test_register.py
+++ b/tests/rest/client/v2_alpha/test_register.py
@@ -310,6 +310,57 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase):
 
         self.assertIsNotNone(channel.json_body.get("sid"))
 
+    @unittest.override_config(
+        {
+            "public_baseurl": "https://test_server",
+            "email": {
+                "smtp_host": "mail_server",
+                "smtp_port": 2525,
+                "notif_from": "sender@host",
+            },
+        }
+    )
+    def test_reject_invalid_email(self):
+        """Check that bad emails are rejected"""
+
+        # Test for email with multiple @
+        channel = self.make_request(
+            "POST",
+            b"register/email/requestToken",
+            {"client_secret": "foobar", "email": "email@@email", "send_attempt": 1},
+        )
+        self.assertEquals(400, channel.code, channel.result)
+        # Check error to ensure that we're not erroring due to a bug in the test.
+        self.assertEquals(
+            channel.json_body,
+            {"errcode": "M_UNKNOWN", "error": "Unable to parse email address"},
+        )
+
+        # Test for email with no @
+        channel = self.make_request(
+            "POST",
+            b"register/email/requestToken",
+            {"client_secret": "foobar", "email": "email", "send_attempt": 1},
+        )
+        self.assertEquals(400, channel.code, channel.result)
+        self.assertEquals(
+            channel.json_body,
+            {"errcode": "M_UNKNOWN", "error": "Unable to parse email address"},
+        )
+
+        # Test for super long email
+        email = "a@" + "a" * 1000
+        channel = self.make_request(
+            "POST",
+            b"register/email/requestToken",
+            {"client_secret": "foobar", "email": email, "send_attempt": 1},
+        )
+        self.assertEquals(400, channel.code, channel.result)
+        self.assertEquals(
+            channel.json_body,
+            {"errcode": "M_UNKNOWN", "error": "Unable to parse email address"},
+        )
+
 
 class AccountValidityTestCase(unittest.HomeserverTestCase):
 
diff --git a/tests/server.py b/tests/server.py
index b535a5d886..9df8cda24f 100644
--- a/tests/server.py
+++ b/tests/server.py
@@ -603,12 +603,6 @@ class FakeTransport:
         if self.disconnected:
             return
 
-        if not hasattr(self.other, "transport"):
-            # the other has no transport yet; reschedule
-            if self.autoflush:
-                self._reactor.callLater(0.0, self.flush)
-            return
-
         if maxbytes is not None:
             to_write = self.buffer[:maxbytes]
         else:
diff --git a/tests/storage/test__base.py b/tests/storage/test__base.py
index 6339a43f0c..200b9198f9 100644
--- a/tests/storage/test__base.py
+++ b/tests/storage/test__base.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import secrets
 
 from tests import unittest
 
@@ -21,7 +22,7 @@ class UpsertManyTests(unittest.HomeserverTestCase):
     def prepare(self, reactor, clock, hs):
         self.storage = hs.get_datastore()
 
-        self.table_name = "table_" + hs.get_secrets().token_hex(6)
+        self.table_name = "table_" + secrets.token_hex(6)
         self.get_success(
             self.storage.db_pool.runInteraction(
                 "create",
diff --git a/tests/test_server.py b/tests/test_server.py
index 55cde7f62f..407e172e41 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -202,6 +202,8 @@ class OptionsResourceTests(unittest.TestCase):
             parse_listener_def({"type": "http", "port": 0}),
             self.resource,
             "1.0",
+            max_request_body_size=1234,
+            reactor=self.reactor,
         )
 
         # render the request and return the channel
diff --git a/tests/unittest.py b/tests/unittest.py
index ee22a53849..74db7c08f1 100644
--- a/tests/unittest.py
+++ b/tests/unittest.py
@@ -18,6 +18,7 @@ import hashlib
 import hmac
 import inspect
 import logging
+import secrets
 import time
 from typing import Callable, Dict, Iterable, Optional, Tuple, Type, TypeVar, Union
 from unittest.mock import Mock, patch
@@ -247,6 +248,8 @@ class HomeserverTestCase(TestCase):
             config=self.hs.config.server.listeners[0],
             resource=self.resource,
             server_version_string="1",
+            max_request_body_size=1234,
+            reactor=self.reactor,
         )
 
         from tests.rest.client.v1.utils import RestHelper
@@ -624,7 +627,6 @@ class HomeserverTestCase(TestCase):
             str: The new event's ID.
         """
         event_creator = self.hs.get_event_creation_handler()
-        secrets = self.hs.get_secrets()
         requester = create_requester(user)
 
         event, context = self.get_success(
diff --git a/tests/utils.py b/tests/utils.py
index 63d52b9140..6bd008dcfe 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -153,6 +153,10 @@ def default_config(name, parse=False):
             "local": {"per_second": 10000, "burst_count": 10000},
             "remote": {"per_second": 10000, "burst_count": 10000},
         },
+        "rc_invites": {
+            "per_room": {"per_second": 10000, "burst_count": 10000},
+            "per_user": {"per_second": 10000, "burst_count": 10000},
+        },
         "rc_3pid_validation": {"per_second": 10000, "burst_count": 10000},
         "saml2_enabled": False,
         "public_baseurl": None,