diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py
index b6941f6da9..84d705c6db 100644
--- a/synapse/appservice/scheduler.py
+++ b/synapse/appservice/scheduler.py
@@ -185,10 +185,18 @@ class _ServiceQueuer:
if not events and not ephemeral and not to_device_messages_to_send:
return
+ # Don't pass kwargs unless necessary. This makes unit testing calls of
+ # txn_ctrl.send much more elegant.
+ additional_send_kwargs = {}
+ if ephemeral:
+ additional_send_kwargs["ephemeral"] = ephemeral
+ if to_device_messages_to_send:
+ additional_send_kwargs[
+ "to_device_messages"
+ ] = to_device_messages_to_send
+
try:
- await self.txn_ctrl.send(
- service, events, ephemeral, to_device_messages_to_send
- )
+ await self.txn_ctrl.send(service, events, **additional_send_kwargs)
except Exception:
logger.exception("AS request failed")
finally:
diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py
index 55f0899bae..13c34efc80 100644
--- a/tests/appservice/test_scheduler.py
+++ b/tests/appservice/test_scheduler.py
@@ -14,14 +14,17 @@
from unittest.mock import Mock
from twisted.internet import defer
+from twisted.internet.testing import MemoryReactor
from synapse.appservice import ApplicationServiceState
from synapse.appservice.scheduler import (
+ ApplicationServiceScheduler,
_Recoverer,
- _ServiceQueuer,
_TransactionController,
)
from synapse.logging.context import make_deferred_yieldable
+from synapse.server import HomeServer
+from synapse.util import Clock
from tests import unittest
from tests.test_utils import simple_async_mock
@@ -58,7 +61,10 @@ 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, ephemeral=[] # txn made and saved
+ service=service,
+ events=events,
+ ephemeral=[],
+ to_device_messages=[], # txn made and saved
)
self.assertEquals(0, len(self.txnctrl.recoverers)) # no recoverer made
txn.complete.assert_called_once_with(self.store) # txn completed
@@ -79,7 +85,10 @@ 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, ephemeral=[] # txn made and saved
+ service=service,
+ events=events,
+ ephemeral=[],
+ to_device_messages=[], # txn made and saved
)
self.assertEquals(0, txn.send.call_count) # txn not sent though
self.assertEquals(0, txn.complete.call_count) # or completed
@@ -102,7 +111,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, ephemeral=[]
+ service=service, events=events, ephemeral=[], to_device_messages=[]
)
self.assertEquals(1, self.recoverer_fn.call_count) # recoverer made
self.assertEquals(1, self.recoverer.recover.call_count) # and invoked
@@ -189,38 +198,41 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase):
self.callback.assert_called_once_with(self.recoverer)
-class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase):
- def setUp(self):
+class ApplicationServiceSchedulerQueuerTestCase(unittest.HomeserverTestCase):
+ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer):
+ self.scheduler = ApplicationServiceScheduler(hs)
self.txn_ctrl = Mock()
self.txn_ctrl.send = simple_async_mock()
- self.queuer = _ServiceQueuer(self.txn_ctrl, MockClock())
+
+ # Replace instantiated _TransactionController instances with our Mock
+ self.scheduler.txn_ctrl = self.txn_ctrl
+ self.scheduler.queuer.txn_ctrl = self.txn_ctrl
def test_send_single_event_no_queue(self):
# Expect the event to be sent immediately.
service = Mock(id=4)
event = Mock()
- self.queuer.enqueue_event(service, event)
- self.txn_ctrl.send.assert_called_once_with(service, [event], [])
+ self.scheduler.enqueue_for_appservice(service, events=[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, z: make_deferred_yieldable(d)
- )
+ self.txn_ctrl.send = Mock(return_value=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_event(service, event)
+ self.scheduler.enqueue_for_appservice(service, events=[event])
# Send more events: expect send() to NOT be called multiple times.
- self.queuer.enqueue_event(service, event2)
- self.queuer.enqueue_event(service, event3)
- self.txn_ctrl.send.assert_called_with(service, [event], [])
+ # (call enqueue_for_appservice multiple times deliberately)
+ self.scheduler.enqueue_for_appservice(service, events=[event2])
+ self.scheduler.enqueue_for_appservice(service, events=[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):
@@ -238,23 +250,23 @@ class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase):
send_return_list = [srv_1_defer, srv_2_defer]
- def do_send(x, y, z):
+ def do_send(*args, **kwargs):
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_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], [])
+ self.scheduler.enqueue_for_appservice(srv1, events=[srv_1_event])
+ self.scheduler.enqueue_for_appservice(srv1, events=[srv_1_event2])
+ self.txn_ctrl.send.assert_called_with(srv1, [srv_1_event])
+ self.scheduler.enqueue_for_appservice(srv2, events=[srv_2_event])
+ self.scheduler.enqueue_for_appservice(srv2, events=[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_large_txns(self):
@@ -262,7 +274,7 @@ class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase):
srv_2_defer = defer.Deferred()
send_return_list = [srv_1_defer, srv_2_defer]
- def do_send(x, y, z):
+ def do_send(*args, **kwargs):
return make_deferred_yieldable(send_return_list.pop(0))
self.txn_ctrl.send = Mock(side_effect=do_send)
@@ -270,67 +282,65 @@ class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase):
service = Mock(id=4, name="service")
event_list = [Mock(name="event%i" % (i + 1)) for i in range(200)]
for event in event_list:
- self.queuer.enqueue_event(service, event)
+ self.scheduler.enqueue_for_appservice(service, events=[event])
# Expect the first event to be sent immediately.
- self.txn_ctrl.send.assert_called_with(service, [event_list[0]], [])
+ self.txn_ctrl.send.assert_called_with(service, [event_list[0]])
srv_1_defer.callback(service)
# Then send the next 100 events
- self.txn_ctrl.send.assert_called_with(service, event_list[1:101], [])
+ self.txn_ctrl.send.assert_called_with(service, event_list[1:101])
srv_2_defer.callback(service)
# Then the final 99 events
- self.txn_ctrl.send.assert_called_with(service, event_list[101:], [])
+ self.txn_ctrl.send.assert_called_with(service, event_list[101:])
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)
+ self.scheduler.enqueue_for_appservice(service, ephemeral=event_list)
+ self.txn_ctrl.send.assert_called_once_with(service, [], ephemeral=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)
+ self.scheduler.enqueue_for_appservice(service, ephemeral=event_list)
+ self.txn_ctrl.send.assert_called_once_with(service, [], ephemeral=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)
- )
+ self.txn_ctrl.send = Mock(return_value=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)
+ self.scheduler.enqueue_for_appservice(service, ephemeral=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.scheduler.enqueue_for_appservice(service, ephemeral=event_list_2)
+ self.scheduler.enqueue_for_appservice(service, ephemeral=event_list_3)
+ self.txn_ctrl.send.assert_called_with(service, [], ephemeral=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.txn_ctrl.send.assert_called_with(
+ service, [], ephemeral=event_list_2 + event_list_3
+ )
self.assertEquals(2, self.txn_ctrl.send.call_count)
def test_send_large_txns_ephemeral(self):
d = defer.Deferred()
- self.txn_ctrl.send = Mock(
- side_effect=lambda x, y, z: make_deferred_yieldable(d)
- )
+ self.txn_ctrl.send = Mock(return_value=make_deferred_yieldable(d))
# Expect the event to be sent immediately.
service = Mock(id=4, name="service")
first_chunk = [Mock(name="event%i" % (i + 1)) for i in range(100)]
second_chunk = [Mock(name="event%i" % (i + 101)) for i in range(50)]
event_list = first_chunk + second_chunk
- self.queuer.enqueue_ephemeral(service, event_list)
- self.txn_ctrl.send.assert_called_once_with(service, [], first_chunk)
+ self.scheduler.enqueue_for_appservice(service, ephemeral=event_list)
+ self.txn_ctrl.send.assert_called_once_with(service, [], ephemeral=first_chunk)
d.callback(service)
- self.txn_ctrl.send.assert_called_with(service, [], second_chunk)
+ self.txn_ctrl.send.assert_called_with(service, [], ephemeral=second_chunk)
self.assertEquals(2, self.txn_ctrl.send.call_count)
diff --git a/tests/handlers/test_appservice.py b/tests/handlers/test_appservice.py
index 6ca9c7f9f1..0c82eadfef 100644
--- a/tests/handlers/test_appservice.py
+++ b/tests/handlers/test_appservice.py
@@ -26,7 +26,7 @@ from synapse.types import RoomStreamToken
from synapse.util.stringutils import random_string
from tests import unittest
-from tests.test_utils import make_awaitable
+from tests.test_utils import make_awaitable, simple_async_mock
from tests.utils import MockClock
@@ -68,8 +68,8 @@ class AppServiceHandlerTestCase(unittest.TestCase):
]
self.handler.notify_interested_services(RoomStreamToken(None, 1))
- self.mock_scheduler.submit_event_for_as.assert_called_once_with(
- interested_service, event
+ self.mock_scheduler.enqueue_for_appservice.assert_called_once_with(
+ interested_service, events=[event]
)
def test_query_user_exists_unknown_user(self):
@@ -279,8 +279,8 @@ class AppServiceHandlerTestCase(unittest.TestCase):
self.handler.notify_interested_services_ephemeral(
"receipt_key", 580, ["@fakerecipient:example.com"]
)
- self.mock_scheduler.submit_ephemeral_events_for_as.assert_called_once_with(
- interested_service, [event]
+ self.mock_scheduler.enqueue_for_appservice.assert_called_once_with(
+ interested_service, ephemeral=[event]
)
self.mock_store.set_appservice_stream_type_pos.assert_called_once_with(
interested_service,
@@ -310,8 +310,8 @@ class AppServiceHandlerTestCase(unittest.TestCase):
"receipt_key", 580, ["@fakerecipient:example.com"]
)
# This method will be called, but with an empty list of events
- self.mock_scheduler.submit_ephemeral_events_for_as.assert_called_once_with(
- interested_service, []
+ self.mock_scheduler.enqueue_for_appservice.assert_called_once_with(
+ interested_service, ephemeral=[]
)
def _mkservice(self, is_interested, protocols=None):
@@ -345,11 +345,10 @@ class ApplicationServicesHandlerSendEventsTestCase(unittest.HomeserverTestCase):
]
def prepare(self, reactor, clock, hs):
- # Mock the ApplicationServiceScheduler queuer so that we can track any
- # outgoing ephemeral events
- self.mock_service_queuer = Mock()
- self.mock_service_queuer.enqueue_ephemeral = Mock()
- hs.get_application_service_handler().scheduler.queuer = self.mock_service_queuer
+ # Mock the ApplicationServiceScheduler's _TransactionController's send method so that
+ # we can track any outgoing ephemeral events
+ self.send_mock = simple_async_mock()
+ hs.get_application_service_handler().scheduler.txn_ctrl.send = self.send_mock
# Mock out application services, and allow defining our own in tests
self._services: List[ApplicationService] = []
@@ -423,19 +422,22 @@ class ApplicationServicesHandlerSendEventsTestCase(unittest.HomeserverTestCase):
# Only the local_user -> exclusive_as_user to-device message should have been forwarded to the AS.
#
# The uninterested application service should not have been notified at all.
- self.mock_service_queuer.enqueue_ephemeral.assert_called_once()
- service, events = self.mock_service_queuer.enqueue_ephemeral.call_args[0]
+ self.send_mock.assert_called_once()
+ service, _events = self.send_mock.call_args[0]
+ to_device_messages = self.send_mock.call_args[1]["to_device_messages"]
# Assert that this was the same to-device message that local_user sent
self.assertEqual(service, interested_appservice)
- self.assertEqual(events[0]["type"], "m.room_key_request")
- self.assertEqual(events[0]["sender"], self.local_user)
+ self.assertEqual(to_device_messages[0]["type"], "m.room_key_request")
+ self.assertEqual(to_device_messages[0]["sender"], self.local_user)
# Additional fields 'to_user_id' and 'to_device_id' specifically for
# to-device messages via the AS API
- self.assertEqual(events[0]["to_user_id"], self.exclusive_as_user)
- self.assertEqual(events[0]["to_device_id"], self.exclusive_as_user_device_id)
- self.assertEqual(events[0]["content"], message_content)
+ self.assertEqual(to_device_messages[0]["to_user_id"], self.exclusive_as_user)
+ self.assertEqual(
+ to_device_messages[0]["to_device_id"], self.exclusive_as_user_device_id
+ )
+ self.assertEqual(to_device_messages[0]["content"], message_content)
@unittest.override_config(
{"experimental_features": {"msc2409_to_device_messages_enabled": True}}
@@ -525,22 +527,23 @@ class ApplicationServicesHandlerSendEventsTestCase(unittest.HomeserverTestCase):
)
self.assertEqual(chan.code, 200, chan.result)
- self.mock_service_queuer.enqueue_ephemeral.assert_called()
+ self.send_mock.assert_called()
# Count the total number of to-device messages that were sent out per-service.
# Ensure that we only sent to-device messages to interested services, and that
# each interested service received the full count of to-device messages.
service_id_to_message_count: Dict[str, int] = {}
- for call in self.mock_service_queuer.enqueue_ephemeral.call_args_list:
- service, events = call[0]
+ for call in self.send_mock.call_args_list:
+ service, _events = call[0]
+ to_device_messages = call[1]["to_device_messages"]
# Check that this was made to an interested service
self.assertIn(service, interested_appservices)
# Add to the count of messages for this application service
service_id_to_message_count.setdefault(service.id, 0)
- service_id_to_message_count[service.id] += len(events)
+ service_id_to_message_count[service.id] += len(to_device_messages)
# Assert that each interested service received the full count of messages
for count in service_id_to_message_count.values():
|
