Merge branch '10881-build-python-futures-package' refs #10881
[arvados.git] / sdk / python / tests / test_events.py
1 import arvados
2 import io
3 import logging
4 import mock
5 import Queue
6 import run_test_server
7 import threading
8 import time
9 import unittest
10
11 import arvados_testutil
12
13 class WebsocketTest(run_test_server.TestCaseWithServers):
14     MAIN_SERVER = {}
15
16     TIME_PAST = time.time()-3600
17     TIME_FUTURE = time.time()+3600
18     MOCK_WS_URL = 'wss://[{}]/'.format(arvados_testutil.TEST_HOST)
19
20     TEST_TIMEOUT = 10.0
21
22     def setUp(self):
23         self.ws = None
24
25     def tearDown(self):
26         try:
27             if self.ws:
28                 self.ws.close()
29         except Exception as e:
30             print("Error in teardown: ", e)
31         super(WebsocketTest, self).tearDown()
32         run_test_server.reset()
33
34     def _test_subscribe(self, poll_fallback, expect_type, start_time=None, expected=1):
35         run_test_server.authorize_with('active')
36         events = Queue.Queue(100)
37
38         # Create ancestor before subscribing.
39         # When listening with start_time in the past, this should also be retrieved.
40         # However, when start_time is omitted in subscribe, this should not be fetched.
41         ancestor = arvados.api('v1').humans().create(body={}).execute()
42
43         filters = [['object_uuid', 'is_a', 'arvados#human']]
44         if start_time:
45             filters.append(['created_at', '>=', start_time])
46
47         self.ws = arvados.events.subscribe(
48             arvados.api('v1'), filters,
49             events.put_nowait,
50             poll_fallback=poll_fallback,
51             last_log_id=(1 if start_time else None))
52         self.assertIsInstance(self.ws, expect_type)
53         self.assertEqual(200, events.get(True, 5)['status'])
54         human = arvados.api('v1').humans().create(body={}).execute()
55
56         want_uuids = []
57         if expected > 0:
58             want_uuids.append(human['uuid'])
59         if expected > 1:
60             want_uuids.append(ancestor['uuid'])
61         log_object_uuids = []
62         while set(want_uuids) - set(log_object_uuids):
63             log_object_uuids.append(events.get(True, 5)['object_uuid'])
64
65         if expected < 2:
66             with self.assertRaises(Queue.Empty):
67                 # assertEqual just serves to show us what unexpected
68                 # thing comes out of the queue when the assertRaises
69                 # fails; when the test passes, this assertEqual
70                 # doesn't get called.
71                 self.assertEqual(events.get(True, 2), None)
72
73     def test_subscribe_websocket(self):
74         self._test_subscribe(
75             poll_fallback=False, expect_type=arvados.events.EventClient, expected=1)
76
77     @mock.patch('arvados.events.EventClient.__init__')
78     def test_subscribe_poll(self, event_client_constr):
79         event_client_constr.side_effect = Exception('All is well')
80         self._test_subscribe(
81             poll_fallback=0.25, expect_type=arvados.events.PollClient, expected=1)
82
83     def test_subscribe_poll_retry(self):
84         api_mock = mock.MagicMock()
85         n = []
86         def on_ev(ev):
87             n.append(ev)
88
89         error_mock = mock.MagicMock()
90         error_mock.resp.status = 0
91         error_mock._get_reason.return_value = "testing"
92         api_mock.logs().list().execute.side_effect = (arvados.errors.ApiError(error_mock, ""),
93                                                       {"items": [{"id": 1}], "items_available": 1},
94                                                       arvados.errors.ApiError(error_mock, ""),
95                                                       {"items": [{"id": 1}], "items_available": 1})
96         pc = arvados.events.PollClient(api_mock, [], on_ev, 15, None)
97         pc.start()
98         while len(n) < 2:
99             time.sleep(.1)
100         pc.close()
101
102     def test_subscribe_websocket_with_start_time_past(self):
103         self._test_subscribe(
104             poll_fallback=False, expect_type=arvados.events.EventClient,
105             start_time=self.localiso(self.TIME_PAST),
106             expected=2)
107
108     @mock.patch('arvados.events.EventClient.__init__')
109     def test_subscribe_poll_with_start_time_past(self, event_client_constr):
110         event_client_constr.side_effect = Exception('All is well')
111         self._test_subscribe(
112             poll_fallback=0.25, expect_type=arvados.events.PollClient,
113             start_time=self.localiso(self.TIME_PAST),
114             expected=2)
115
116     def test_subscribe_websocket_with_start_time_future(self):
117         self._test_subscribe(
118             poll_fallback=False, expect_type=arvados.events.EventClient,
119             start_time=self.localiso(self.TIME_FUTURE),
120             expected=0)
121
122     @mock.patch('arvados.events.EventClient.__init__')
123     def test_subscribe_poll_with_start_time_future(self, event_client_constr):
124         event_client_constr.side_effect = Exception('All is well')
125         self._test_subscribe(
126             poll_fallback=0.25, expect_type=arvados.events.PollClient,
127             start_time=self.localiso(self.TIME_FUTURE),
128             expected=0)
129
130     def test_subscribe_websocket_with_start_time_past_utc(self):
131         self._test_subscribe(
132             poll_fallback=False, expect_type=arvados.events.EventClient,
133             start_time=self.utciso(self.TIME_PAST),
134             expected=2)
135
136     def test_subscribe_websocket_with_start_time_future_utc(self):
137         self._test_subscribe(
138             poll_fallback=False, expect_type=arvados.events.EventClient,
139             start_time=self.utciso(self.TIME_FUTURE),
140             expected=0)
141
142     def utciso(self, t):
143         return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(t))
144
145     def localiso(self, t):
146         return time.strftime('%Y-%m-%dT%H:%M:%S', time.localtime(t)) + self.isotz(-time.timezone/60)
147
148     def isotz(self, offset):
149         """Convert minutes-east-of-UTC to RFC3339- and ISO-compatible time zone designator"""
150         return '{:+03d}:{:02d}'.format(offset/60, offset%60)
151
152     # Test websocket reconnection on (un)execpted close
153     def _test_websocket_reconnect(self, close_unexpected):
154         run_test_server.authorize_with('active')
155         events = Queue.Queue(100)
156
157         logstream = io.BytesIO()
158         rootLogger = logging.getLogger()
159         streamHandler = logging.StreamHandler(logstream)
160         rootLogger.addHandler(streamHandler)
161
162         filters = [['object_uuid', 'is_a', 'arvados#human']]
163         filters.append(['created_at', '>=', self.localiso(self.TIME_PAST)])
164         self.ws = arvados.events.subscribe(
165             arvados.api('v1'), filters,
166             events.put_nowait,
167             poll_fallback=False,
168             last_log_id=None)
169         self.assertIsInstance(self.ws, arvados.events.EventClient)
170         self.assertEqual(200, events.get(True, 5)['status'])
171
172         # create obj
173         human = arvados.api('v1').humans().create(body={}).execute()
174
175         # expect an event
176         self.assertIn(human['uuid'], events.get(True, 5)['object_uuid'])
177         with self.assertRaises(Queue.Empty):
178             self.assertEqual(events.get(True, 2), None)
179
180         # close (im)properly
181         if close_unexpected:
182             self.ws.ec.close_connection()
183         else:
184             self.ws.close()
185
186         # create one more obj
187         human2 = arvados.api('v1').humans().create(body={}).execute()
188
189         # (un)expect the object creation event
190         if close_unexpected:
191             log_object_uuids = []
192             for i in range(0, 2):
193                 event = events.get(True, 5)
194                 if event.get('object_uuid') != None:
195                     log_object_uuids.append(event['object_uuid'])
196             with self.assertRaises(Queue.Empty):
197                 self.assertEqual(events.get(True, 2), None)
198             self.assertNotIn(human['uuid'], log_object_uuids)
199             self.assertIn(human2['uuid'], log_object_uuids)
200         else:
201             with self.assertRaises(Queue.Empty):
202                 self.assertEqual(events.get(True, 2), None)
203
204         # verify log message to ensure that an (un)expected close
205         log_messages = logstream.getvalue()
206         closeLogFound = log_messages.find("Unexpected close. Reconnecting.")
207         retryLogFound = log_messages.find("Error during websocket reconnect. Will retry")
208         if close_unexpected:
209             self.assertNotEqual(closeLogFound, -1)
210         else:
211             self.assertEqual(closeLogFound, -1)
212         rootLogger.removeHandler(streamHandler)
213
214     def test_websocket_reconnect_on_unexpected_close(self):
215         self._test_websocket_reconnect(True)
216
217     def test_websocket_no_reconnect_on_close_by_user(self):
218         self._test_websocket_reconnect(False)
219
220     # Test websocket reconnection retry
221     @mock.patch('arvados.events._EventClient.connect')
222     def test_websocket_reconnect_retry(self, event_client_connect):
223         event_client_connect.side_effect = [None, Exception('EventClient.connect error'), None]
224
225         logstream = io.BytesIO()
226         rootLogger = logging.getLogger()
227         streamHandler = logging.StreamHandler(logstream)
228         rootLogger.addHandler(streamHandler)
229
230         run_test_server.authorize_with('active')
231         events = Queue.Queue(100)
232
233         filters = [['object_uuid', 'is_a', 'arvados#human']]
234         self.ws = arvados.events.subscribe(
235             arvados.api('v1'), filters,
236             events.put_nowait,
237             poll_fallback=False,
238             last_log_id=None)
239         self.assertIsInstance(self.ws, arvados.events.EventClient)
240
241         # simulate improper close
242         self.ws.on_closed()
243
244         # verify log messages to ensure retry happened
245         log_messages = logstream.getvalue()
246         found = log_messages.find("Error 'EventClient.connect error' during websocket reconnect.")
247         self.assertNotEqual(found, -1)
248         rootLogger.removeHandler(streamHandler)
249
250     @mock.patch('arvados.events._EventClient')
251     def test_subscribe_method(self, websocket_client):
252         filters = [['object_uuid', 'is_a', 'arvados#human']]
253         client = arvados.events.EventClient(
254             self.MOCK_WS_URL, [], lambda event: None, None)
255         client.subscribe(filters[:], 99)
256         websocket_client().subscribe.assert_called_with(filters, 99)
257
258     @mock.patch('arvados.events._EventClient')
259     def test_unsubscribe(self, websocket_client):
260         filters = [['object_uuid', 'is_a', 'arvados#human']]
261         client = arvados.events.EventClient(
262             self.MOCK_WS_URL, filters[:], lambda event: None, None)
263         client.unsubscribe(filters[:])
264         websocket_client().unsubscribe.assert_called_with(filters)
265
266     @mock.patch('arvados.events._EventClient')
267     def test_run_forever_survives_reconnects(self, websocket_client):
268         connected = threading.Event()
269         websocket_client().connect.side_effect = connected.set
270         client = arvados.events.EventClient(
271             self.MOCK_WS_URL, [], lambda event: None, None)
272         forever_thread = threading.Thread(target=client.run_forever)
273         forever_thread.start()
274         # Simulate an unexpected disconnect, and wait for reconnect.
275         close_thread = threading.Thread(target=client.on_closed)
276         close_thread.start()
277         self.assertTrue(connected.wait(timeout=self.TEST_TIMEOUT))
278         close_thread.join()
279         run_forever_alive = forever_thread.is_alive()
280         client.close()
281         forever_thread.join()
282         self.assertTrue(run_forever_alive)
283         self.assertEqual(2, websocket_client().connect.call_count)
284
285
286 class PollClientTestCase(unittest.TestCase):
287     TEST_TIMEOUT = 10.0
288
289     class MockLogs(object):
290
291         def __init__(self):
292             self.logs = []
293             self.lock = threading.Lock()
294             self.api_called = threading.Event()
295
296         def add(self, log):
297             with self.lock:
298                 self.logs.append(log)
299
300         def return_list(self, num_retries=None):
301             self.api_called.set()
302             args, kwargs = self.list_func.call_args_list[-1]
303             filters = kwargs.get('filters', [])
304             if not any(True for f in filters if f[0] == 'id' and f[1] == '>'):
305                 # No 'id' filter was given -- this must be the probe
306                 # to determine the most recent id.
307                 return {'items': [{'id': 1}], 'items_available': 1}
308             with self.lock:
309                 retval = self.logs
310                 self.logs = []
311             return {'items': retval, 'items_available': len(retval)}
312
313     def setUp(self):
314         self.logs = self.MockLogs()
315         self.arv = mock.MagicMock(name='arvados.api()')
316         self.arv.logs().list().execute.side_effect = self.logs.return_list
317         # our MockLogs object's "execute" stub will need to inspect
318         # the call history to determine X in
319         # ....logs().list(filters=X).execute():
320         self.logs.list_func = self.arv.logs().list
321         self.status_ok = threading.Event()
322         self.event_received = threading.Event()
323         self.recv_events = []
324
325     def tearDown(self):
326         if hasattr(self, 'client'):
327             self.client.close(timeout=None)
328
329     def callback(self, event):
330         if event.get('status') == 200:
331             self.status_ok.set()
332         else:
333             self.recv_events.append(event)
334             self.event_received.set()
335
336     def build_client(self, filters=None, callback=None, last_log_id=None, poll_time=99):
337         if filters is None:
338             filters = []
339         if callback is None:
340             callback = self.callback
341         self.client = arvados.events.PollClient(
342             self.arv, filters, callback, poll_time, last_log_id)
343
344     def was_filter_used(self, target):
345         return any(target in call[-1].get('filters', [])
346                    for call in self.arv.logs().list.call_args_list)
347
348     def test_callback(self):
349         test_log = {'id': 12345, 'testkey': 'testtext'}
350         self.logs.add({'id': 123})
351         self.build_client(poll_time=.01)
352         self.client.start()
353         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
354         self.assertTrue(self.event_received.wait(self.TEST_TIMEOUT))
355         self.event_received.clear()
356         self.logs.add(test_log.copy())
357         self.assertTrue(self.event_received.wait(self.TEST_TIMEOUT))
358         self.assertIn(test_log, self.recv_events)
359
360     def test_subscribe(self):
361         client_filter = ['kind', '=', 'arvados#test']
362         self.build_client()
363         self.client.unsubscribe([])
364         self.client.subscribe([client_filter[:]])
365         self.client.start()
366         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
367         self.assertTrue(self.logs.api_called.wait(self.TEST_TIMEOUT))
368         self.assertTrue(self.was_filter_used(client_filter))
369
370     def test_unsubscribe(self):
371         should_filter = ['foo', '=', 'foo']
372         should_not_filter = ['foo', '=', 'bar']
373         self.build_client(poll_time=0.01)
374         self.client.unsubscribe([])
375         self.client.subscribe([should_not_filter[:]])
376         self.client.subscribe([should_filter[:]])
377         self.client.unsubscribe([should_not_filter[:]])
378         self.client.start()
379         self.logs.add({'id': 123})
380         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
381         self.assertTrue(self.event_received.wait(self.TEST_TIMEOUT))
382         self.assertTrue(self.was_filter_used(should_filter))
383         self.assertFalse(self.was_filter_used(should_not_filter))
384
385     def test_run_forever(self):
386         self.build_client()
387         self.client.start()
388         forever_thread = threading.Thread(target=self.client.run_forever)
389         forever_thread.start()
390         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
391         self.assertTrue(forever_thread.is_alive())
392         self.client.close()
393         forever_thread.join()
394         del self.client