diff --git a/tests/app/test_frontend_proxy.py b/tests/app/test_frontend_proxy.py
index 641093d349..4a301b84e1 100644
--- a/tests/app/test_frontend_proxy.py
+++ b/tests/app/test_frontend_proxy.py
@@ -22,7 +22,7 @@ class FrontendProxyTests(HomeserverTestCase):
def make_homeserver(self, reactor, clock):
hs = self.setup_test_homeserver(
- http_client=None, homeserverToUse=GenericWorkerServer
+ http_client=None, homeserver_to_use=GenericWorkerServer
)
return hs
diff --git a/tests/app/test_openid_listener.py b/tests/app/test_openid_listener.py
index 0f016c32eb..c2b10d2c70 100644
--- a/tests/app/test_openid_listener.py
+++ b/tests/app/test_openid_listener.py
@@ -26,7 +26,7 @@ from tests.unittest import HomeserverTestCase
class FederationReaderOpenIDListenerTests(HomeserverTestCase):
def make_homeserver(self, reactor, clock):
hs = self.setup_test_homeserver(
- http_client=None, homeserverToUse=GenericWorkerServer
+ http_client=None, homeserver_to_use=GenericWorkerServer
)
return hs
@@ -84,7 +84,7 @@ class FederationReaderOpenIDListenerTests(HomeserverTestCase):
class SynapseHomeserverOpenIDListenerTests(HomeserverTestCase):
def make_homeserver(self, reactor, clock):
hs = self.setup_test_homeserver(
- http_client=None, homeserverToUse=SynapseHomeServer
+ http_client=None, homeserver_to_use=SynapseHomeServer
)
return hs
diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py
index 68a4caabbf..2acb8b7603 100644
--- a/tests/appservice/test_scheduler.py
+++ b/tests/appservice/test_scheduler.py
@@ -60,7 +60,7 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase):
self.successResultOf(defer.ensureDeferred(self.txnctrl.send(service, events)))
self.store.create_appservice_txn.assert_called_once_with(
- service=service, events=events # txn made and saved
+ service=service, events=events, ephemeral=[] # txn made and saved
)
self.assertEquals(0, len(self.txnctrl.recoverers)) # no recoverer made
txn.complete.assert_called_once_with(self.store) # txn completed
@@ -81,7 +81,7 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase):
self.successResultOf(defer.ensureDeferred(self.txnctrl.send(service, events)))
self.store.create_appservice_txn.assert_called_once_with(
- service=service, events=events # txn made and saved
+ service=service, events=events, ephemeral=[] # txn made and saved
)
self.assertEquals(0, txn.send.call_count) # txn not sent though
self.assertEquals(0, txn.complete.call_count) # or completed
@@ -106,7 +106,7 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase):
self.successResultOf(defer.ensureDeferred(self.txnctrl.send(service, events)))
self.store.create_appservice_txn.assert_called_once_with(
- service=service, events=events
+ service=service, events=events, ephemeral=[]
)
self.assertEquals(1, self.recoverer_fn.call_count) # recoverer made
self.assertEquals(1, self.recoverer.recover.call_count) # and invoked
@@ -202,26 +202,28 @@ class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase):
# Expect the event to be sent immediately.
service = Mock(id=4)
event = Mock()
- self.queuer.enqueue(service, event)
- self.txn_ctrl.send.assert_called_once_with(service, [event])
+ self.queuer.enqueue_event(service, event)
+ self.txn_ctrl.send.assert_called_once_with(service, [event], [])
def test_send_single_event_with_queue(self):
d = defer.Deferred()
- self.txn_ctrl.send = Mock(side_effect=lambda x, y: make_deferred_yieldable(d))
+ self.txn_ctrl.send = Mock(
+ side_effect=lambda x, y, z: make_deferred_yieldable(d)
+ )
service = Mock(id=4)
event = Mock(event_id="first")
event2 = Mock(event_id="second")
event3 = Mock(event_id="third")
# Send an event and don't resolve it just yet.
- self.queuer.enqueue(service, event)
+ self.queuer.enqueue_event(service, event)
# Send more events: expect send() to NOT be called multiple times.
- self.queuer.enqueue(service, event2)
- self.queuer.enqueue(service, event3)
- self.txn_ctrl.send.assert_called_with(service, [event])
+ self.queuer.enqueue_event(service, event2)
+ self.queuer.enqueue_event(service, event3)
+ self.txn_ctrl.send.assert_called_with(service, [event], [])
self.assertEquals(1, self.txn_ctrl.send.call_count)
# Resolve the send event: expect the queued events to be sent
d.callback(service)
- self.txn_ctrl.send.assert_called_with(service, [event2, event3])
+ self.txn_ctrl.send.assert_called_with(service, [event2, event3], [])
self.assertEquals(2, self.txn_ctrl.send.call_count)
def test_multiple_service_queues(self):
@@ -239,21 +241,58 @@ class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase):
send_return_list = [srv_1_defer, srv_2_defer]
- def do_send(x, y):
+ def do_send(x, y, z):
return make_deferred_yieldable(send_return_list.pop(0))
self.txn_ctrl.send = Mock(side_effect=do_send)
# send events for different ASes and make sure they are sent
- self.queuer.enqueue(srv1, srv_1_event)
- self.queuer.enqueue(srv1, srv_1_event2)
- self.txn_ctrl.send.assert_called_with(srv1, [srv_1_event])
- self.queuer.enqueue(srv2, srv_2_event)
- self.queuer.enqueue(srv2, srv_2_event2)
- self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event])
+ self.queuer.enqueue_event(srv1, srv_1_event)
+ self.queuer.enqueue_event(srv1, srv_1_event2)
+ self.txn_ctrl.send.assert_called_with(srv1, [srv_1_event], [])
+ self.queuer.enqueue_event(srv2, srv_2_event)
+ self.queuer.enqueue_event(srv2, srv_2_event2)
+ self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event], [])
# make sure callbacks for a service only send queued events for THAT
# service
srv_2_defer.callback(srv2)
- self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event2])
+ self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event2], [])
self.assertEquals(3, self.txn_ctrl.send.call_count)
+
+ def test_send_single_ephemeral_no_queue(self):
+ # Expect the event to be sent immediately.
+ service = Mock(id=4, name="service")
+ event_list = [Mock(name="event")]
+ self.queuer.enqueue_ephemeral(service, event_list)
+ self.txn_ctrl.send.assert_called_once_with(service, [], event_list)
+
+ def test_send_multiple_ephemeral_no_queue(self):
+ # Expect the event to be sent immediately.
+ service = Mock(id=4, name="service")
+ event_list = [Mock(name="event1"), Mock(name="event2"), Mock(name="event3")]
+ self.queuer.enqueue_ephemeral(service, event_list)
+ self.txn_ctrl.send.assert_called_once_with(service, [], event_list)
+
+ def test_send_single_ephemeral_with_queue(self):
+ d = defer.Deferred()
+ self.txn_ctrl.send = Mock(
+ side_effect=lambda x, y, z: make_deferred_yieldable(d)
+ )
+ service = Mock(id=4)
+ event_list_1 = [Mock(event_id="event1"), Mock(event_id="event2")]
+ event_list_2 = [Mock(event_id="event3"), Mock(event_id="event4")]
+ event_list_3 = [Mock(event_id="event5"), Mock(event_id="event6")]
+
+ # Send an event and don't resolve it just yet.
+ self.queuer.enqueue_ephemeral(service, event_list_1)
+ # Send more events: expect send() to NOT be called multiple times.
+ self.queuer.enqueue_ephemeral(service, event_list_2)
+ self.queuer.enqueue_ephemeral(service, event_list_3)
+ self.txn_ctrl.send.assert_called_with(service, [], event_list_1)
+ self.assertEquals(1, self.txn_ctrl.send.call_count)
+ # Resolve txn_ctrl.send
+ d.callback(service)
+ # Expect the queued events to be sent
+ self.txn_ctrl.send.assert_called_with(service, [], event_list_2 + event_list_3)
+ self.assertEquals(2, self.txn_ctrl.send.call_count)
diff --git a/tests/handlers/test_message.py b/tests/handlers/test_message.py
index 64e28bc639..9f6f21a6e2 100644
--- a/tests/handlers/test_message.py
+++ b/tests/handlers/test_message.py
@@ -66,7 +66,6 @@ class EventCreationTestCase(unittest.HomeserverTestCase):
"sender": self.requester.user.to_string(),
"content": {"msgtype": "m.text", "body": random_string(5)},
},
- token_id=self.token_id,
txn_id=txn_id,
)
)
diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py
index 914c82e7a8..8ed67640f8 100644
--- a/tests/handlers/test_presence.py
+++ b/tests/handlers/test_presence.py
@@ -615,7 +615,7 @@ class PresenceJoinTestCase(unittest.HomeserverTestCase):
self.store.get_latest_event_ids_in_room(room_id)
)
- event = self.get_success(builder.build(prev_event_ids))
+ event = self.get_success(builder.build(prev_event_ids, None))
self.get_success(self.federation_handler.on_receive_pdu(hostname, event))
diff --git a/tests/replication/_base.py b/tests/replication/_base.py
index 81ea985b9f..093e2faac7 100644
--- a/tests/replication/_base.py
+++ b/tests/replication/_base.py
@@ -59,7 +59,7 @@ class BaseStreamTestCase(unittest.HomeserverTestCase):
self.reactor.lookups["testserv"] = "1.2.3.4"
self.worker_hs = self.setup_test_homeserver(
http_client=None,
- homeserverToUse=GenericWorkerServer,
+ homeserver_to_use=GenericWorkerServer,
config=self._get_worker_hs_config(),
reactor=self.reactor,
)
@@ -266,7 +266,7 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
config.update(extra_config)
worker_hs = self.setup_test_homeserver(
- homeserverToUse=GenericWorkerServer,
+ homeserver_to_use=GenericWorkerServer,
config=config,
reactor=self.reactor,
**kwargs
diff --git a/tests/replication/test_federation_ack.py b/tests/replication/test_federation_ack.py
index 23be1167a3..1853667558 100644
--- a/tests/replication/test_federation_ack.py
+++ b/tests/replication/test_federation_ack.py
@@ -31,7 +31,7 @@ class FederationAckTestCase(HomeserverTestCase):
return config
def make_homeserver(self, reactor, clock):
- hs = self.setup_test_homeserver(homeserverToUse=GenericWorkerServer)
+ hs = self.setup_test_homeserver(homeserver_to_use=GenericWorkerServer)
return hs
diff --git a/tests/replication/test_federation_sender_shard.py b/tests/replication/test_federation_sender_shard.py
index 9c4a9c3563..779745ae9d 100644
--- a/tests/replication/test_federation_sender_shard.py
+++ b/tests/replication/test_federation_sender_shard.py
@@ -226,7 +226,7 @@ class FederationSenderTestCase(BaseMultiWorkerStreamTestCase):
}
builder = factory.for_room_version(room_version, event_dict)
- join_event = self.get_success(builder.build(prev_event_ids))
+ join_event = self.get_success(builder.build(prev_event_ids, None))
self.get_success(federation.on_send_join_request(remote_server, join_event))
self.replicate()
diff --git a/tests/replication/test_sharded_event_persister.py b/tests/replication/test_sharded_event_persister.py
index 6068d14905..82cf033d4e 100644
--- a/tests/replication/test_sharded_event_persister.py
+++ b/tests/replication/test_sharded_event_persister.py
@@ -14,8 +14,12 @@
# limitations under the License.
import logging
+from mock import patch
+
+from synapse.api.room_versions import RoomVersion
from synapse.rest import admin
from synapse.rest.client.v1 import login, room
+from synapse.rest.client.v2_alpha import sync
from tests.replication._base import BaseMultiWorkerStreamTestCase
from tests.utils import USE_POSTGRES_FOR_TESTS
@@ -36,6 +40,7 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
admin.register_servlets_for_client_rest_resource,
room.register_servlets,
login.register_servlets,
+ sync.register_servlets,
]
def prepare(self, reactor, clock, hs):
@@ -43,6 +48,9 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
self.other_user_id = self.register_user("otheruser", "pass")
self.other_access_token = self.login("otheruser", "pass")
+ self.room_creator = self.hs.get_room_creation_handler()
+ self.store = hs.get_datastore()
+
def default_config(self):
conf = super().default_config()
conf["redis"] = {"enabled": "true"}
@@ -53,6 +61,29 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
}
return conf
+ def _create_room(self, room_id: str, user_id: str, tok: str):
+ """Create a room with given room_id
+ """
+
+ # We control the room ID generation by patching out the
+ # `_generate_room_id` method
+ async def generate_room(
+ creator_id: str, is_public: bool, room_version: RoomVersion
+ ):
+ await self.store.store_room(
+ room_id=room_id,
+ room_creator_user_id=creator_id,
+ is_public=is_public,
+ room_version=room_version,
+ )
+ return room_id
+
+ with patch(
+ "synapse.handlers.room.RoomCreationHandler._generate_room_id"
+ ) as mock:
+ mock.side_effect = generate_room
+ self.helper.create_room_as(user_id, tok=tok)
+
def test_basic(self):
"""Simple test to ensure that multiple rooms can be created and joined,
and that different rooms get handled by different instances.
@@ -100,3 +131,189 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
self.assertTrue(persisted_on_1)
self.assertTrue(persisted_on_2)
+
+ def test_vector_clock_token(self):
+ """Tests that using a stream token with a vector clock component works
+ correctly with basic /sync and /messages usage.
+ """
+
+ self.make_worker_hs(
+ "synapse.app.generic_worker", {"worker_name": "worker1"},
+ )
+
+ worker_hs2 = self.make_worker_hs(
+ "synapse.app.generic_worker", {"worker_name": "worker2"},
+ )
+
+ sync_hs = self.make_worker_hs(
+ "synapse.app.generic_worker", {"worker_name": "sync"},
+ )
+
+ # Specially selected room IDs that get persisted on different workers.
+ room_id1 = "!foo:test"
+ room_id2 = "!baz:test"
+
+ self.assertEqual(
+ self.hs.config.worker.events_shard_config.get_instance(room_id1), "worker1"
+ )
+ self.assertEqual(
+ self.hs.config.worker.events_shard_config.get_instance(room_id2), "worker2"
+ )
+
+ user_id = self.register_user("user", "pass")
+ access_token = self.login("user", "pass")
+
+ store = self.hs.get_datastore()
+
+ # Create two room on the different workers.
+ self._create_room(room_id1, user_id, access_token)
+ self._create_room(room_id2, user_id, access_token)
+
+ # The other user joins
+ self.helper.join(
+ room=room_id1, user=self.other_user_id, tok=self.other_access_token
+ )
+ self.helper.join(
+ room=room_id2, user=self.other_user_id, tok=self.other_access_token
+ )
+
+ # Do an initial sync so that we're up to date.
+ request, channel = self.make_request("GET", "/sync", access_token=access_token)
+ self.render_on_worker(sync_hs, request)
+ next_batch = channel.json_body["next_batch"]
+
+ # We now gut wrench into the events stream MultiWriterIdGenerator on
+ # worker2 to mimic it getting stuck persisting an event. This ensures
+ # that when we send an event on worker1 we end up in a state where
+ # worker2 events stream position lags that on worker1, resulting in a
+ # RoomStreamToken with a non-empty instance map component.
+ #
+ # Worker2's event stream position will not advance until we call
+ # __aexit__ again.
+ actx = worker_hs2.get_datastore()._stream_id_gen.get_next()
+ self.get_success(actx.__aenter__())
+
+ response = self.helper.send(room_id1, body="Hi!", tok=self.other_access_token)
+ first_event_in_room1 = response["event_id"]
+
+ # Assert that the current stream token has an instance map component, as
+ # we are trying to test vector clock tokens.
+ room_stream_token = store.get_room_max_token()
+ self.assertNotEqual(len(room_stream_token.instance_map), 0)
+
+ # Check that syncing still gets the new event, despite the gap in the
+ # stream IDs.
+ request, channel = self.make_request(
+ "GET", "/sync?since={}".format(next_batch), access_token=access_token
+ )
+ self.render_on_worker(sync_hs, request)
+
+ # We should only see the new event and nothing else
+ self.assertIn(room_id1, channel.json_body["rooms"]["join"])
+ self.assertNotIn(room_id2, channel.json_body["rooms"]["join"])
+
+ events = channel.json_body["rooms"]["join"][room_id1]["timeline"]["events"]
+ self.assertListEqual(
+ [first_event_in_room1], [event["event_id"] for event in events]
+ )
+
+ # Get the next batch and makes sure its a vector clock style token.
+ vector_clock_token = channel.json_body["next_batch"]
+ self.assertTrue(vector_clock_token.startswith("m"))
+
+ # Now that we've got a vector clock token we finish the fake persisting
+ # an event we started above.
+ self.get_success(actx.__aexit__(None, None, None))
+
+ # Now try and send an event to the other rooom so that we can test that
+ # the vector clock style token works as a `since` token.
+ response = self.helper.send(room_id2, body="Hi!", tok=self.other_access_token)
+ first_event_in_room2 = response["event_id"]
+
+ request, channel = self.make_request(
+ "GET",
+ "/sync?since={}".format(vector_clock_token),
+ access_token=access_token,
+ )
+ self.render_on_worker(sync_hs, request)
+
+ self.assertNotIn(room_id1, channel.json_body["rooms"]["join"])
+ self.assertIn(room_id2, channel.json_body["rooms"]["join"])
+
+ events = channel.json_body["rooms"]["join"][room_id2]["timeline"]["events"]
+ self.assertListEqual(
+ [first_event_in_room2], [event["event_id"] for event in events]
+ )
+
+ next_batch = channel.json_body["next_batch"]
+
+ # We also want to test that the vector clock style token works with
+ # pagination. We do this by sending a couple of new events into the room
+ # and syncing again to get a prev_batch token for each room, then
+ # paginating from there back to the vector clock token.
+ self.helper.send(room_id1, body="Hi again!", tok=self.other_access_token)
+ self.helper.send(room_id2, body="Hi again!", tok=self.other_access_token)
+
+ request, channel = self.make_request(
+ "GET", "/sync?since={}".format(next_batch), access_token=access_token
+ )
+ self.render_on_worker(sync_hs, request)
+
+ prev_batch1 = channel.json_body["rooms"]["join"][room_id1]["timeline"][
+ "prev_batch"
+ ]
+ prev_batch2 = channel.json_body["rooms"]["join"][room_id2]["timeline"][
+ "prev_batch"
+ ]
+
+ # Paginating back in the first room should not produce any results, as
+ # no events have happened in it. This tests that we are correctly
+ # filtering results based on the vector clock portion.
+ request, channel = self.make_request(
+ "GET",
+ "/rooms/{}/messages?from={}&to={}&dir=b".format(
+ room_id1, prev_batch1, vector_clock_token
+ ),
+ access_token=access_token,
+ )
+ self.render_on_worker(sync_hs, request)
+ self.assertListEqual([], channel.json_body["chunk"])
+
+ # Paginating back on the second room should produce the first event
+ # again. This tests that pagination isn't completely broken.
+ request, channel = self.make_request(
+ "GET",
+ "/rooms/{}/messages?from={}&to={}&dir=b".format(
+ room_id2, prev_batch2, vector_clock_token
+ ),
+ access_token=access_token,
+ )
+ self.render_on_worker(sync_hs, request)
+ self.assertEqual(len(channel.json_body["chunk"]), 1)
+ self.assertEqual(
+ channel.json_body["chunk"][0]["event_id"], first_event_in_room2
+ )
+
+ # Paginating forwards should give the same results
+ request, channel = self.make_request(
+ "GET",
+ "/rooms/{}/messages?from={}&to={}&dir=f".format(
+ room_id1, vector_clock_token, prev_batch1
+ ),
+ access_token=access_token,
+ )
+ self.render_on_worker(sync_hs, request)
+ self.assertListEqual([], channel.json_body["chunk"])
+
+ request, channel = self.make_request(
+ "GET",
+ "/rooms/{}/messages?from={}&to={}&dir=f".format(
+ room_id2, vector_clock_token, prev_batch2,
+ ),
+ access_token=access_token,
+ )
+ self.render_on_worker(sync_hs, request)
+ self.assertEqual(len(channel.json_body["chunk"]), 1)
+ self.assertEqual(
+ channel.json_body["chunk"][0]["event_id"], first_event_in_room2
+ )
diff --git a/tests/rest/client/test_third_party_rules.py b/tests/rest/client/test_third_party_rules.py
index b737625e33..0048bea54a 100644
--- a/tests/rest/client/test_third_party_rules.py
+++ b/tests/rest/client/test_third_party_rules.py
@@ -114,10 +114,10 @@ class ThirdPartyRulesTestCase(unittest.HomeserverTestCase):
self.render(request)
self.assertEquals(channel.result["code"], b"403", channel.result)
- def test_modify_event(self):
- """Tests that the module can successfully tweak an event before it is persisted.
- """
- # first patch the event checker so that it will modify the event
+ def test_cannot_modify_event(self):
+ """cannot accidentally modify an event before it is persisted"""
+
+ # first patch the event checker so that it will try to modify the event
async def check(ev: EventBase, state):
ev.content = {"x": "y"}
return True
@@ -132,6 +132,26 @@ class ThirdPartyRulesTestCase(unittest.HomeserverTestCase):
access_token=self.tok,
)
self.render(request)
+ self.assertEqual(channel.result["code"], b"500", channel.result)
+
+ def test_modify_event(self):
+ """The module can return a modified version of the event"""
+ # first patch the event checker so that it will modify the event
+ async def check(ev: EventBase, state):
+ d = ev.get_dict()
+ d["content"] = {"x": "y"}
+ return d
+
+ current_rules_module().check_event_allowed = check
+
+ # now send the event
+ request, channel = self.make_request(
+ "PUT",
+ "/_matrix/client/r0/rooms/%s/send/modifyme/1" % self.room_id,
+ {"x": "x"},
+ access_token=self.tok,
+ )
+ self.render(request)
self.assertEqual(channel.result["code"], b"200", channel.result)
event_id = channel.json_body["event_id"]
diff --git a/tests/storage/test__base.py b/tests/storage/test__base.py
index f5afed017c..8e69b1e9cc 100644
--- a/tests/storage/test__base.py
+++ b/tests/storage/test__base.py
@@ -20,82 +20,11 @@ from mock import Mock
from twisted.internet import defer
from synapse.util.async_helpers import ObservableDeferred
-from synapse.util.caches.descriptors import Cache, cached
+from synapse.util.caches.descriptors import cached
from tests import unittest
-class CacheTestCase(unittest.HomeserverTestCase):
- def prepare(self, reactor, clock, homeserver):
- self.cache = Cache("test")
-
- def test_empty(self):
- failed = False
- try:
- self.cache.get("foo")
- except KeyError:
- failed = True
-
- self.assertTrue(failed)
-
- def test_hit(self):
- self.cache.prefill("foo", 123)
-
- self.assertEquals(self.cache.get("foo"), 123)
-
- def test_invalidate(self):
- self.cache.prefill(("foo",), 123)
- self.cache.invalidate(("foo",))
-
- failed = False
- try:
- self.cache.get(("foo",))
- except KeyError:
- failed = True
-
- self.assertTrue(failed)
-
- def test_eviction(self):
- cache = Cache("test", max_entries=2)
-
- cache.prefill(1, "one")
- cache.prefill(2, "two")
- cache.prefill(3, "three") # 1 will be evicted
-
- failed = False
- try:
- cache.get(1)
- except KeyError:
- failed = True
-
- self.assertTrue(failed)
-
- cache.get(2)
- cache.get(3)
-
- def test_eviction_lru(self):
- cache = Cache("test", max_entries=2)
-
- cache.prefill(1, "one")
- cache.prefill(2, "two")
-
- # Now access 1 again, thus causing 2 to be least-recently used
- cache.get(1)
-
- cache.prefill(3, "three")
-
- failed = False
- try:
- cache.get(2)
- except KeyError:
- failed = True
-
- self.assertTrue(failed)
-
- cache.get(1)
- cache.get(3)
-
-
class CacheDecoratorTestCase(unittest.HomeserverTestCase):
@defer.inlineCallbacks
def test_passthrough(self):
diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py
index c905a38930..c5c7987349 100644
--- a/tests/storage/test_appservice.py
+++ b/tests/storage/test_appservice.py
@@ -244,7 +244,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase):
service = Mock(id=self.as_list[0]["id"])
events = [Mock(event_id="e1"), Mock(event_id="e2")]
txn = yield defer.ensureDeferred(
- self.store.create_appservice_txn(service, events)
+ self.store.create_appservice_txn(service, events, [])
)
self.assertEquals(txn.id, 1)
self.assertEquals(txn.events, events)
@@ -258,7 +258,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase):
yield self._insert_txn(service.id, 9644, events)
yield self._insert_txn(service.id, 9645, events)
txn = yield defer.ensureDeferred(
- self.store.create_appservice_txn(service, events)
+ self.store.create_appservice_txn(service, events, [])
)
self.assertEquals(txn.id, 9646)
self.assertEquals(txn.events, events)
@@ -270,7 +270,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase):
events = [Mock(event_id="e1"), Mock(event_id="e2")]
yield self._set_last_txn(service.id, 9643)
txn = yield defer.ensureDeferred(
- self.store.create_appservice_txn(service, events)
+ self.store.create_appservice_txn(service, events, [])
)
self.assertEquals(txn.id, 9644)
self.assertEquals(txn.events, events)
@@ -293,7 +293,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase):
yield self._insert_txn(self.as_list[3]["id"], 9643, events)
txn = yield defer.ensureDeferred(
- self.store.create_appservice_txn(service, events)
+ self.store.create_appservice_txn(service, events, [])
)
self.assertEquals(txn.id, 9644)
self.assertEquals(txn.events, events)
diff --git a/tests/storage/test_redaction.py b/tests/storage/test_redaction.py
index 1ea35d60c1..d4f9e809db 100644
--- a/tests/storage/test_redaction.py
+++ b/tests/storage/test_redaction.py
@@ -236,9 +236,9 @@ class RedactionTestCase(unittest.HomeserverTestCase):
self._event_id = event_id
@defer.inlineCallbacks
- def build(self, prev_event_ids):
+ def build(self, prev_event_ids, auth_event_ids):
built_event = yield defer.ensureDeferred(
- self._base_builder.build(prev_event_ids)
+ self._base_builder.build(prev_event_ids, auth_event_ids)
)
built_event._event_id = self._event_id
diff --git a/tests/test_metrics.py b/tests/test_metrics.py
index f5f63d8ed6..759e4cd048 100644
--- a/tests/test_metrics.py
+++ b/tests/test_metrics.py
@@ -15,7 +15,7 @@
# limitations under the License.
from synapse.metrics import REGISTRY, InFlightGauge, generate_latest
-from synapse.util.caches.descriptors import Cache
+from synapse.util.caches.deferred_cache import DeferredCache
from tests import unittest
@@ -138,7 +138,7 @@ class CacheMetricsTests(unittest.HomeserverTestCase):
Caches produce metrics reflecting their state when scraped.
"""
CACHE_NAME = "cache_metrics_test_fgjkbdfg"
- cache = Cache(CACHE_NAME, max_entries=777)
+ cache = DeferredCache(CACHE_NAME, max_entries=777)
items = {
x.split(b"{")[0].decode("ascii"): x.split(b" ")[1].decode("ascii")
diff --git a/tests/unittest.py b/tests/unittest.py
index 6c1661c92c..040b126a27 100644
--- a/tests/unittest.py
+++ b/tests/unittest.py
@@ -20,7 +20,7 @@ import hmac
import inspect
import logging
import time
-from typing import Optional, Tuple, Type, TypeVar, Union
+from typing import Optional, Tuple, Type, TypeVar, Union, overload
from mock import Mock, patch
@@ -364,6 +364,36 @@ class HomeserverTestCase(TestCase):
Function to optionally be overridden in subclasses.
"""
+ # Annoyingly mypy doesn't seem to pick up the fact that T is SynapseRequest
+ # when the `request` arg isn't given, so we define an explicit override to
+ # cover that case.
+ @overload
+ def make_request(
+ self,
+ method: Union[bytes, str],
+ path: Union[bytes, str],
+ content: Union[bytes, dict] = b"",
+ access_token: Optional[str] = None,
+ shorthand: bool = True,
+ federation_auth_origin: str = None,
+ content_is_form: bool = False,
+ ) -> Tuple[SynapseRequest, FakeChannel]:
+ ...
+
+ @overload
+ def make_request(
+ self,
+ method: Union[bytes, str],
+ path: Union[bytes, str],
+ content: Union[bytes, dict] = b"",
+ access_token: Optional[str] = None,
+ request: Type[T] = SynapseRequest,
+ shorthand: bool = True,
+ federation_auth_origin: str = None,
+ content_is_form: bool = False,
+ ) -> Tuple[T, FakeChannel]:
+ ...
+
def make_request(
self,
method: Union[bytes, str],
diff --git a/tests/util/caches/test_deferred_cache.py b/tests/util/caches/test_deferred_cache.py
new file mode 100644
index 0000000000..9717be56b6
--- /dev/null
+++ b/tests/util/caches/test_deferred_cache.py
@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 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.
+
+import unittest
+from functools import partial
+
+from twisted.internet import defer
+
+from synapse.util.caches.deferred_cache import DeferredCache
+
+
+class DeferredCacheTestCase(unittest.TestCase):
+ def test_empty(self):
+ cache = DeferredCache("test")
+ failed = False
+ try:
+ cache.get("foo")
+ except KeyError:
+ failed = True
+
+ self.assertTrue(failed)
+
+ def test_hit(self):
+ cache = DeferredCache("test")
+ cache.prefill("foo", 123)
+
+ self.assertEquals(cache.get("foo"), 123)
+
+ def test_invalidate(self):
+ cache = DeferredCache("test")
+ cache.prefill(("foo",), 123)
+ cache.invalidate(("foo",))
+
+ failed = False
+ try:
+ cache.get(("foo",))
+ except KeyError:
+ failed = True
+
+ self.assertTrue(failed)
+
+ def test_invalidate_all(self):
+ cache = DeferredCache("testcache")
+
+ callback_record = [False, False]
+
+ def record_callback(idx):
+ callback_record[idx] = True
+
+ # add a couple of pending entries
+ d1 = defer.Deferred()
+ cache.set("key1", d1, partial(record_callback, 0))
+
+ d2 = defer.Deferred()
+ cache.set("key2", d2, partial(record_callback, 1))
+
+ # lookup should return observable deferreds
+ self.assertFalse(cache.get("key1").has_called())
+ self.assertFalse(cache.get("key2").has_called())
+
+ # let one of the lookups complete
+ d2.callback("result2")
+
+ # for now at least, the cache will return real results rather than an
+ # observabledeferred
+ self.assertEqual(cache.get("key2"), "result2")
+
+ # now do the invalidation
+ cache.invalidate_all()
+
+ # lookup should return none
+ self.assertIsNone(cache.get("key1", None))
+ self.assertIsNone(cache.get("key2", None))
+
+ # both callbacks should have been callbacked
+ self.assertTrue(callback_record[0], "Invalidation callback for key1 not called")
+ self.assertTrue(callback_record[1], "Invalidation callback for key2 not called")
+
+ # letting the other lookup complete should do nothing
+ d1.callback("result1")
+ self.assertIsNone(cache.get("key1", None))
+
+ def test_eviction(self):
+ cache = DeferredCache(
+ "test", max_entries=2, apply_cache_factor_from_config=False
+ )
+
+ cache.prefill(1, "one")
+ cache.prefill(2, "two")
+ cache.prefill(3, "three") # 1 will be evicted
+
+ failed = False
+ try:
+ cache.get(1)
+ except KeyError:
+ failed = True
+
+ self.assertTrue(failed)
+
+ cache.get(2)
+ cache.get(3)
+
+ def test_eviction_lru(self):
+ cache = DeferredCache(
+ "test", max_entries=2, apply_cache_factor_from_config=False
+ )
+
+ cache.prefill(1, "one")
+ cache.prefill(2, "two")
+
+ # Now access 1 again, thus causing 2 to be least-recently used
+ cache.get(1)
+
+ cache.prefill(3, "three")
+
+ failed = False
+ try:
+ cache.get(2)
+ except KeyError:
+ failed = True
+
+ self.assertTrue(failed)
+
+ cache.get(1)
+ cache.get(3)
diff --git a/tests/util/caches/test_descriptors.py b/tests/util/caches/test_descriptors.py
index 677e925477..3d1f960869 100644
--- a/tests/util/caches/test_descriptors.py
+++ b/tests/util/caches/test_descriptors.py
@@ -14,7 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
-from functools import partial
import mock
@@ -42,49 +41,6 @@ def run_on_reactor():
return make_deferred_yieldable(d)
-class CacheTestCase(unittest.TestCase):
- def test_invalidate_all(self):
- cache = descriptors.Cache("testcache")
-
- callback_record = [False, False]
-
- def record_callback(idx):
- callback_record[idx] = True
-
- # add a couple of pending entries
- d1 = defer.Deferred()
- cache.set("key1", d1, partial(record_callback, 0))
-
- d2 = defer.Deferred()
- cache.set("key2", d2, partial(record_callback, 1))
-
- # lookup should return observable deferreds
- self.assertFalse(cache.get("key1").has_called())
- self.assertFalse(cache.get("key2").has_called())
-
- # let one of the lookups complete
- d2.callback("result2")
-
- # for now at least, the cache will return real results rather than an
- # observabledeferred
- self.assertEqual(cache.get("key2"), "result2")
-
- # now do the invalidation
- cache.invalidate_all()
-
- # lookup should return none
- self.assertIsNone(cache.get("key1", None))
- self.assertIsNone(cache.get("key2", None))
-
- # both callbacks should have been callbacked
- self.assertTrue(callback_record[0], "Invalidation callback for key1 not called")
- self.assertTrue(callback_record[1], "Invalidation callback for key2 not called")
-
- # letting the other lookup complete should do nothing
- d1.callback("result1")
- self.assertIsNone(cache.get("key1", None))
-
-
class DescriptorTestCase(unittest.TestCase):
@defer.inlineCallbacks
def test_cache(self):
diff --git a/tests/utils.py b/tests/utils.py
index 0c09f5457f..acec74e9e9 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -21,6 +21,7 @@ import time
import uuid
import warnings
from inspect import getcallargs
+from typing import Type
from urllib import parse as urlparse
from mock import Mock, patch
@@ -194,8 +195,8 @@ def setup_test_homeserver(
name="test",
config=None,
reactor=None,
- homeserverToUse=TestHomeServer,
- **kargs
+ homeserver_to_use: Type[HomeServer] = TestHomeServer,
+ **kwargs
):
"""
Setup a homeserver suitable for running tests against. Keyword arguments
@@ -218,8 +219,8 @@ def setup_test_homeserver(
config.ldap_enabled = False
- if "clock" not in kargs:
- kargs["clock"] = MockClock()
+ if "clock" not in kwargs:
+ kwargs["clock"] = MockClock()
if USE_POSTGRES_FOR_TESTS:
test_db = "synapse_test_%s" % uuid.uuid4().hex
@@ -264,18 +265,20 @@ def setup_test_homeserver(
cur.close()
db_conn.close()
- hs = homeserverToUse(
- name,
- config=config,
- version_string="Synapse/tests",
- tls_server_context_factory=Mock(),
- tls_client_options_factory=Mock(),
- reactor=reactor,
- **kargs
+ hs = homeserver_to_use(
+ name, config=config, version_string="Synapse/tests", reactor=reactor,
)
+ # Install @cache_in_self attributes
+ for key, val in kwargs.items():
+ setattr(hs, key, val)
+
+ # Mock TLS
+ hs.tls_server_context_factory = Mock()
+ hs.tls_client_options_factory = Mock()
+
hs.setup()
- if homeserverToUse.__name__ == "TestHomeServer":
+ if homeserver_to_use == TestHomeServer:
hs.setup_background_tasks()
if isinstance(db_engine, PostgresEngine):
@@ -339,7 +342,7 @@ def setup_test_homeserver(
hs.get_auth_handler().validate_hash = validate_hash
- fed = kargs.get("resource_for_federation", None)
+ fed = kwargs.get("resource_for_federation", None)
if fed:
register_federation_servlets(hs, fed)
|
