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