test_scheduler.py 9.7 KB

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