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