1
0

test_scheduler.py 9.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2015, 2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. from mock import Mock
  16. from twisted.internet import defer
  17. from synapse.appservice import ApplicationServiceState
  18. from synapse.appservice.scheduler import (
  19. _Recoverer,
  20. _ServiceQueuer,
  21. _TransactionController,
  22. )
  23. from synapse.logging.context import make_deferred_yieldable
  24. from tests import unittest
  25. from ..utils import MockClock
  26. class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase):
  27. def setUp(self):
  28. self.clock = MockClock()
  29. self.store = Mock()
  30. self.as_api = Mock()
  31. self.recoverer = Mock()
  32. self.recoverer_fn = Mock(return_value=self.recoverer)
  33. self.txnctrl = _TransactionController(
  34. clock=self.clock, store=self.store, as_api=self.as_api
  35. )
  36. self.txnctrl.RECOVERER_CLASS = self.recoverer_fn
  37. def test_single_service_up_txn_sent(self):
  38. # Test: The AS is up and the txn is successfully sent.
  39. service = Mock()
  40. events = [Mock(), Mock()]
  41. txn_id = "foobar"
  42. txn = Mock(id=txn_id, service=service, events=events)
  43. # mock methods
  44. self.store.get_appservice_state = Mock(
  45. return_value=defer.succeed(ApplicationServiceState.UP)
  46. )
  47. txn.send = Mock(return_value=defer.succeed(True))
  48. self.store.create_appservice_txn = Mock(return_value=defer.succeed(txn))
  49. # actual call
  50. self.txnctrl.send(service, events)
  51. self.store.create_appservice_txn.assert_called_once_with(
  52. service=service, events=events # txn made and saved
  53. )
  54. self.assertEquals(0, len(self.txnctrl.recoverers)) # no recoverer made
  55. txn.complete.assert_called_once_with(self.store) # txn completed
  56. def test_single_service_down(self):
  57. # Test: The AS is down so it shouldn't push; Recoverers will do it.
  58. # It should still make a transaction though.
  59. service = Mock()
  60. events = [Mock(), Mock()]
  61. txn = Mock(id="idhere", service=service, events=events)
  62. self.store.get_appservice_state = Mock(
  63. return_value=defer.succeed(ApplicationServiceState.DOWN)
  64. )
  65. self.store.create_appservice_txn = Mock(return_value=defer.succeed(txn))
  66. # actual call
  67. self.txnctrl.send(service, events)
  68. self.store.create_appservice_txn.assert_called_once_with(
  69. service=service, events=events # txn made and saved
  70. )
  71. self.assertEquals(0, txn.send.call_count) # txn not sent though
  72. self.assertEquals(0, txn.complete.call_count) # or completed
  73. def test_single_service_up_txn_not_sent(self):
  74. # Test: The AS is up and the txn is not sent. A Recoverer is made and
  75. # started.
  76. service = Mock()
  77. events = [Mock(), Mock()]
  78. txn_id = "foobar"
  79. txn = Mock(id=txn_id, service=service, events=events)
  80. # mock methods
  81. self.store.get_appservice_state = Mock(
  82. return_value=defer.succeed(ApplicationServiceState.UP)
  83. )
  84. self.store.set_appservice_state = Mock(return_value=defer.succeed(True))
  85. txn.send = Mock(return_value=defer.succeed(False)) # fails to send
  86. self.store.create_appservice_txn = Mock(return_value=defer.succeed(txn))
  87. # actual call
  88. self.txnctrl.send(service, events)
  89. self.store.create_appservice_txn.assert_called_once_with(
  90. service=service, events=events
  91. )
  92. self.assertEquals(1, self.recoverer_fn.call_count) # recoverer made
  93. self.assertEquals(1, self.recoverer.recover.call_count) # and invoked
  94. self.assertEquals(1, len(self.txnctrl.recoverers)) # and stored
  95. self.assertEquals(0, txn.complete.call_count) # txn not completed
  96. self.store.set_appservice_state.assert_called_once_with(
  97. service, ApplicationServiceState.DOWN # service marked as down
  98. )
  99. class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase):
  100. def setUp(self):
  101. self.clock = MockClock()
  102. self.as_api = Mock()
  103. self.store = Mock()
  104. self.service = Mock()
  105. self.callback = Mock()
  106. self.recoverer = _Recoverer(
  107. clock=self.clock,
  108. as_api=self.as_api,
  109. store=self.store,
  110. service=self.service,
  111. callback=self.callback,
  112. )
  113. def test_recover_single_txn(self):
  114. txn = Mock()
  115. # return one txn to send, then no more old txns
  116. txns = [txn, None]
  117. def take_txn(*args, **kwargs):
  118. return defer.succeed(txns.pop(0))
  119. self.store.get_oldest_unsent_txn = Mock(side_effect=take_txn)
  120. self.recoverer.recover()
  121. # shouldn't have called anything prior to waiting for exp backoff
  122. self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count)
  123. txn.send = Mock(return_value=True)
  124. # wait for exp backoff
  125. self.clock.advance_time(2)
  126. self.assertEquals(1, txn.send.call_count)
  127. self.assertEquals(1, txn.complete.call_count)
  128. # 2 because it needs to get None to know there are no more txns
  129. self.assertEquals(2, self.store.get_oldest_unsent_txn.call_count)
  130. self.callback.assert_called_once_with(self.recoverer)
  131. self.assertEquals(self.recoverer.service, self.service)
  132. def test_recover_retry_txn(self):
  133. txn = Mock()
  134. txns = [txn, None]
  135. pop_txn = False
  136. def take_txn(*args, **kwargs):
  137. if pop_txn:
  138. return defer.succeed(txns.pop(0))
  139. else:
  140. return defer.succeed(txn)
  141. self.store.get_oldest_unsent_txn = Mock(side_effect=take_txn)
  142. self.recoverer.recover()
  143. self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count)
  144. txn.send = Mock(return_value=False)
  145. self.clock.advance_time(2)
  146. self.assertEquals(1, txn.send.call_count)
  147. self.assertEquals(0, txn.complete.call_count)
  148. self.assertEquals(0, self.callback.call_count)
  149. self.clock.advance_time(4)
  150. self.assertEquals(2, txn.send.call_count)
  151. self.assertEquals(0, txn.complete.call_count)
  152. self.assertEquals(0, self.callback.call_count)
  153. self.clock.advance_time(8)
  154. self.assertEquals(3, txn.send.call_count)
  155. self.assertEquals(0, txn.complete.call_count)
  156. self.assertEquals(0, self.callback.call_count)
  157. txn.send = Mock(return_value=True) # successfully send the txn
  158. pop_txn = True # returns the txn the first time, then no more.
  159. self.clock.advance_time(16)
  160. self.assertEquals(1, txn.send.call_count) # new mock reset call count
  161. self.assertEquals(1, txn.complete.call_count)
  162. self.callback.assert_called_once_with(self.recoverer)
  163. class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase):
  164. def setUp(self):
  165. self.txn_ctrl = Mock()
  166. self.queuer = _ServiceQueuer(self.txn_ctrl, MockClock())
  167. def test_send_single_event_no_queue(self):
  168. # Expect the event to be sent immediately.
  169. service = Mock(id=4)
  170. event = Mock()
  171. self.queuer.enqueue(service, event)
  172. self.txn_ctrl.send.assert_called_once_with(service, [event])
  173. def test_send_single_event_with_queue(self):
  174. d = defer.Deferred()
  175. self.txn_ctrl.send = Mock(side_effect=lambda x, y: make_deferred_yieldable(d))
  176. service = Mock(id=4)
  177. event = Mock(event_id="first")
  178. event2 = Mock(event_id="second")
  179. event3 = Mock(event_id="third")
  180. # Send an event and don't resolve it just yet.
  181. self.queuer.enqueue(service, event)
  182. # Send more events: expect send() to NOT be called multiple times.
  183. self.queuer.enqueue(service, event2)
  184. self.queuer.enqueue(service, event3)
  185. self.txn_ctrl.send.assert_called_with(service, [event])
  186. self.assertEquals(1, self.txn_ctrl.send.call_count)
  187. # Resolve the send event: expect the queued events to be sent
  188. d.callback(service)
  189. self.txn_ctrl.send.assert_called_with(service, [event2, event3])
  190. self.assertEquals(2, self.txn_ctrl.send.call_count)
  191. def test_multiple_service_queues(self):
  192. # Tests that each service has its own queue, and that they don't block
  193. # on each other.
  194. srv1 = Mock(id=4)
  195. srv_1_defer = defer.Deferred()
  196. srv_1_event = Mock(event_id="srv1a")
  197. srv_1_event2 = Mock(event_id="srv1b")
  198. srv2 = Mock(id=6)
  199. srv_2_defer = defer.Deferred()
  200. srv_2_event = Mock(event_id="srv2a")
  201. srv_2_event2 = Mock(event_id="srv2b")
  202. send_return_list = [srv_1_defer, srv_2_defer]
  203. def do_send(x, y):
  204. return make_deferred_yieldable(send_return_list.pop(0))
  205. self.txn_ctrl.send = Mock(side_effect=do_send)
  206. # send events for different ASes and make sure they are sent
  207. self.queuer.enqueue(srv1, srv_1_event)
  208. self.queuer.enqueue(srv1, srv_1_event2)
  209. self.txn_ctrl.send.assert_called_with(srv1, [srv_1_event])
  210. self.queuer.enqueue(srv2, srv_2_event)
  211. self.queuer.enqueue(srv2, srv_2_event2)
  212. self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event])
  213. # make sure callbacks for a service only send queued events for THAT
  214. # service
  215. srv_2_defer.callback(srv2)
  216. self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event2])
  217. self.assertEquals(3, self.txn_ctrl.send.call_count)