4 from datetime import datetime, timedelta, tzinfo
6 import logging.handlers
16 import arvados_testutil
18 class WebsocketTest(run_test_server.TestCaseWithServers):
21 TIME_PAST = time.time()-3600
22 TIME_FUTURE = time.time()+3600
23 MOCK_WS_URL = 'wss://[{}]/'.format(arvados_testutil.TEST_HOST)
32 except Exception as e:
33 print("Error in teardown: ", e)
34 super(WebsocketTest, self).tearDown()
35 run_test_server.reset()
37 def _test_subscribe(self, poll_fallback, expect_type, start_time=None, expected=1):
38 run_test_server.authorize_with('active')
39 events = Queue.Queue(100)
41 # Create ancestor before subscribing.
42 # When listening with start_time in the past, this should also be retrieved.
43 # However, when start_time is omitted in subscribe, this should not be fetched.
44 ancestor = arvados.api('v1').humans().create(body={}).execute()
46 filters = [['object_uuid', 'is_a', 'arvados#human']]
48 filters.append(['created_at', '>=', start_time])
50 self.ws = arvados.events.subscribe(
51 arvados.api('v1'), filters,
53 poll_fallback=poll_fallback,
54 last_log_id=(1 if start_time else None))
55 self.assertIsInstance(self.ws, expect_type)
56 self.assertEqual(200, events.get(True, 5)['status'])
57 human = arvados.api('v1').humans().create(body={}).execute()
60 for i in range(0, expected):
61 log_object_uuids.append(events.get(True, 5)['object_uuid'])
64 self.assertIn(human['uuid'], log_object_uuids)
67 self.assertIn(ancestor['uuid'], log_object_uuids)
69 with self.assertRaises(Queue.Empty):
70 # assertEqual just serves to show us what unexpected thing
71 # comes out of the queue when the assertRaises fails; when
72 # the test passes, this assertEqual doesn't get called.
73 self.assertEqual(events.get(True, 2), None)
75 def test_subscribe_websocket(self):
77 poll_fallback=False, expect_type=arvados.events.EventClient, expected=1)
79 @mock.patch('arvados.events.EventClient.__init__')
80 def test_subscribe_poll(self, event_client_constr):
81 event_client_constr.side_effect = Exception('All is well')
83 poll_fallback=0.25, expect_type=arvados.events.PollClient, expected=1)
85 def test_subscribe_poll_retry(self):
86 api_mock = mock.MagicMock()
91 error_mock = mock.MagicMock()
92 error_mock.resp.status = 0
93 error_mock._get_reason.return_value = "testing"
94 api_mock.logs().list().execute.side_effect = (arvados.errors.ApiError(error_mock, ""),
95 {"items": [{"id": 1}], "items_available": 1},
96 arvados.errors.ApiError(error_mock, ""),
97 {"items": [{"id": 1}], "items_available": 1})
98 pc = arvados.events.PollClient(api_mock, [], on_ev, 15, None)
104 def test_subscribe_websocket_with_start_time_past(self):
105 self._test_subscribe(
106 poll_fallback=False, expect_type=arvados.events.EventClient,
107 start_time=self.localiso(self.TIME_PAST),
110 @mock.patch('arvados.events.EventClient.__init__')
111 def test_subscribe_poll_with_start_time_past(self, event_client_constr):
112 event_client_constr.side_effect = Exception('All is well')
113 self._test_subscribe(
114 poll_fallback=0.25, expect_type=arvados.events.PollClient,
115 start_time=self.localiso(self.TIME_PAST),
118 def test_subscribe_websocket_with_start_time_future(self):
119 self._test_subscribe(
120 poll_fallback=False, expect_type=arvados.events.EventClient,
121 start_time=self.localiso(self.TIME_FUTURE),
124 @mock.patch('arvados.events.EventClient.__init__')
125 def test_subscribe_poll_with_start_time_future(self, event_client_constr):
126 event_client_constr.side_effect = Exception('All is well')
127 self._test_subscribe(
128 poll_fallback=0.25, expect_type=arvados.events.PollClient,
129 start_time=self.localiso(self.TIME_FUTURE),
132 def test_subscribe_websocket_with_start_time_past_utc(self):
133 self._test_subscribe(
134 poll_fallback=False, expect_type=arvados.events.EventClient,
135 start_time=self.utciso(self.TIME_PAST),
138 def test_subscribe_websocket_with_start_time_future_utc(self):
139 self._test_subscribe(
140 poll_fallback=False, expect_type=arvados.events.EventClient,
141 start_time=self.utciso(self.TIME_FUTURE),
145 return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(t))
147 def localiso(self, t):
148 return time.strftime('%Y-%m-%dT%H:%M:%S', time.localtime(t)) + self.isotz(-time.timezone/60)
150 def isotz(self, offset):
151 """Convert minutes-east-of-UTC to ISO8601 time zone designator"""
152 return '{:+03d}{:02d}'.format(offset/60, offset%60)
154 # Test websocket reconnection on (un)execpted close
155 def _test_websocket_reconnect(self, close_unexpected):
156 run_test_server.authorize_with('active')
157 events = Queue.Queue(100)
159 logstream = StringIO.StringIO()
160 rootLogger = logging.getLogger()
161 streamHandler = logging.StreamHandler(logstream)
162 rootLogger.addHandler(streamHandler)
164 filters = [['object_uuid', 'is_a', 'arvados#human']]
165 filters.append(['created_at', '>=', self.localiso(self.TIME_PAST)])
166 self.ws = arvados.events.subscribe(
167 arvados.api('v1'), filters,
171 self.assertIsInstance(self.ws, arvados.events.EventClient)
172 self.assertEqual(200, events.get(True, 5)['status'])
175 human = arvados.api('v1').humans().create(body={}).execute()
178 self.assertIn(human['uuid'], events.get(True, 5)['object_uuid'])
179 with self.assertRaises(Queue.Empty):
180 self.assertEqual(events.get(True, 2), None)
184 self.ws.close_connection()
188 # create one more obj
189 human2 = arvados.api('v1').humans().create(body={}).execute()
191 # (un)expect the object creation event
193 log_object_uuids = []
194 for i in range(0, 2):
195 event = events.get(True, 5)
196 if event.get('object_uuid') != None:
197 log_object_uuids.append(event['object_uuid'])
198 with self.assertRaises(Queue.Empty):
199 self.assertEqual(events.get(True, 2), None)
200 self.assertNotIn(human['uuid'], log_object_uuids)
201 self.assertIn(human2['uuid'], log_object_uuids)
203 with self.assertRaises(Queue.Empty):
204 self.assertEqual(events.get(True, 2), None)
206 # verify log message to ensure that an (un)expected close
207 log_messages = logstream.getvalue()
208 closeLogFound = log_messages.find("Unexpected close. Reconnecting.")
209 retryLogFound = log_messages.find("Error during websocket reconnect. Will retry")
211 self.assertNotEqual(closeLogFound, -1)
213 self.assertEqual(closeLogFound, -1)
214 rootLogger.removeHandler(streamHandler)
216 def test_websocket_reconnect_on_unexpected_close(self):
217 self._test_websocket_reconnect(True)
219 def test_websocket_no_reconnect_on_close_by_user(self):
220 self._test_websocket_reconnect(False)
222 # Test websocket reconnection retry
223 @mock.patch('arvados.events._EventClient.connect')
224 def test_websocket_reconnect_retry(self, event_client_connect):
225 event_client_connect.side_effect = [None, Exception('EventClient.connect error'), None]
227 logstream = StringIO.StringIO()
228 rootLogger = logging.getLogger()
229 streamHandler = logging.StreamHandler(logstream)
230 rootLogger.addHandler(streamHandler)
232 run_test_server.authorize_with('active')
233 events = Queue.Queue(100)
235 filters = [['object_uuid', 'is_a', 'arvados#human']]
236 self.ws = arvados.events.subscribe(
237 arvados.api('v1'), filters,
241 self.assertIsInstance(self.ws, arvados.events.EventClient)
243 # simulate improper close
246 # verify log messages to ensure retry happened
247 log_messages = logstream.getvalue()
248 found = log_messages.find("Error 'EventClient.connect error' during websocket reconnect.")
249 self.assertNotEqual(found, -1)
250 rootLogger.removeHandler(streamHandler)
252 @mock.patch('arvados.events._EventClient')
253 def test_subscribe_method(self, websocket_client):
254 filters = [['object_uuid', 'is_a', 'arvados#human']]
255 client = arvados.events.EventClient(
256 self.MOCK_WS_URL, [], lambda event: None, None)
257 client.subscribe(filters[:], 99)
258 websocket_client().subscribe.assert_called_with(filters, 99)
260 @mock.patch('arvados.events._EventClient')
261 def test_unsubscribe(self, websocket_client):
262 filters = [['object_uuid', 'is_a', 'arvados#human']]
263 client = arvados.events.EventClient(
264 self.MOCK_WS_URL, filters[:], lambda event: None, None)
265 client.unsubscribe(filters[:])
266 websocket_client().unsubscribe.assert_called_with(filters)
268 @unittest.expectedFailure
269 @mock.patch('arvados.events._EventClient')
270 def test_run_forever(self, websocket_client):
271 client = arvados.events.EventClient(
272 self.MOCK_WS_URL, [], lambda event: None, None)
274 websocket_client().run_forever.assert_called_with()
277 class PollClientTestCase(unittest.TestCase):
278 class MockLogs(object):
281 self.lock = threading.Lock()
285 self.logs.append(log)
287 def return_list(self, num_retries=None):
291 return {'items': retval, 'items_available': len(retval)}
295 self.logs = self.MockLogs()
296 self.arv = mock.MagicMock(name='arvados.api()')
297 self.arv.logs().list().execute.side_effect = self.logs.return_list
298 self.callback_cond = threading.Condition()
299 self.recv_events = []
302 if hasattr(self, 'client'):
303 self.client.close(timeout=None)
305 def callback(self, event):
306 with self.callback_cond:
307 self.recv_events.append(event)
308 self.callback_cond.notify_all()
310 def build_client(self, filters=None, callback=None, last_log_id=None, poll_time=99):
314 callback = self.callback
315 self.client = arvados.events.PollClient(
316 self.arv, filters, callback, poll_time, last_log_id)
318 def was_filter_used(self, target):
319 return any(target in call[-1].get('filters', [])
320 for call in self.arv.logs().list.call_args_list)
322 def test_callback(self):
323 test_log = {'id': 12345, 'testkey': 'testtext'}
324 self.logs.add({'id': 123})
325 self.build_client(poll_time=.01)
326 with self.callback_cond:
328 self.callback_cond.wait()
329 self.logs.add(test_log.copy())
330 self.callback_cond.wait()
331 self.client.close(timeout=None)
332 self.assertIn(test_log, self.recv_events)
334 def test_subscribe(self):
335 client_filter = ['kind', '=', 'arvados#test']
337 self.client.subscribe([client_filter[:]])
338 with self.callback_cond:
340 self.callback_cond.wait()
341 self.client.close(timeout=None)
342 self.assertTrue(self.was_filter_used(client_filter))
344 def test_unsubscribe(self):
345 client_filter = ['kind', '=', 'arvados#test']
347 self.client.subscribe([client_filter[:]])
348 self.client.unsubscribe([client_filter[:]])
350 self.client.close(timeout=None)
351 self.assertFalse(self.was_filter_used(client_filter))
353 def test_run_forever(self):
355 with self.callback_cond:
357 forever_thread = threading.Thread(target=self.client.run_forever)
358 forever_thread.start()
359 self.callback_cond.wait()
360 self.assertTrue(forever_thread.is_alive())
362 forever_thread.join()