sdk/python/tests/test_events.py

   1 from __future__ import print_function
   2 from __future__ import absolute_import
   3 import arvados
   4 import io
   5 import logging
   6 import mock
   7 import Queue
   8 from . import run_test_server
   9 import threading
  10 import time
  11 import unittest
  12
  13 from . import arvados_testutil
  14
  15 class WebsocketTest(run_test_server.TestCaseWithServers):
  16     MAIN_SERVER = {}
  17
  18     TIME_PAST = time.time()-3600
  19     TIME_FUTURE = time.time()+3600
  20     MOCK_WS_URL = 'wss://[{}]/'.format(arvados_testutil.TEST_HOST)
  21
  22     TEST_TIMEOUT = 10.0
  23
  24     def setUp(self):
  25         self.ws = None
  26
  27     def tearDown(self):
  28         try:
  29             if self.ws:
  30                 self.ws.close()
  31         except Exception as e:
  32             print(("Error in teardown: ", e))
  33         super(WebsocketTest, self).tearDown()
  34         run_test_server.reset()
  35
  36     def _test_subscribe(self, poll_fallback, expect_type, start_time=None, expected=1):
  37         run_test_server.authorize_with('active')
  38         events = Queue.Queue(100)
  39
  40         # Create ancestor before subscribing.
  41         # When listening with start_time in the past, this should also be retrieved.
  42         # However, when start_time is omitted in subscribe, this should not be fetched.
  43         ancestor = arvados.api('v1').humans().create(body={}).execute()
  44
  45         filters = [['object_uuid', 'is_a', 'arvados#human']]
  46         if start_time:
  47             filters.append(['created_at', '>=', start_time])
  48
  49         self.ws = arvados.events.subscribe(
  50             arvados.api('v1'), filters,
  51             events.put_nowait,
  52             poll_fallback=poll_fallback,
  53             last_log_id=(1 if start_time else None))
  54         self.assertIsInstance(self.ws, expect_type)
  55         self.assertEqual(200, events.get(True, 5)['status'])
  56         human = arvados.api('v1').humans().create(body={}).execute()
  57
  58         want_uuids = []
  59         if expected > 0:
  60             want_uuids.append(human['uuid'])
  61         if expected > 1:
  62             want_uuids.append(ancestor['uuid'])
  63         log_object_uuids = []
  64         while set(want_uuids) - set(log_object_uuids):
  65             log_object_uuids.append(events.get(True, 5)['object_uuid'])
  66
  67         if expected < 2:
  68             with self.assertRaises(Queue.Empty):
  69                 # assertEqual just serves to show us what unexpected
  70                 # thing comes out of the queue when the assertRaises
  71                 # fails; when the test passes, this assertEqual
  72                 # doesn't get called.
  73                 self.assertEqual(events.get(True, 2), None)
  74
  75     def test_subscribe_websocket(self):
  76         self._test_subscribe(
  77             poll_fallback=False, expect_type=arvados.events.EventClient, expected=1)
  78
  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')
  82         self._test_subscribe(
  83             poll_fallback=0.25, expect_type=arvados.events.PollClient, expected=1)
  84
  85     def test_subscribe_poll_retry(self):
  86         api_mock = mock.MagicMock()
  87         n = []
  88         def on_ev(ev):
  89             n.append(ev)
  90
  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)
  99         pc.start()
 100         while len(n) < 2:
 101             time.sleep(.1)
 102         pc.close()
 103
 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),
 108             expected=2)
 109
 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),
 116             expected=2)
 117
 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),
 122             expected=0)
 123
 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),
 130             expected=0)
 131
 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),
 136             expected=2)
 137
 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),
 142             expected=0)
 143
 144     def utciso(self, t):
 145         return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(t))
 146
 147     def localiso(self, t):
 148         return time.strftime('%Y-%m-%dT%H:%M:%S', time.localtime(t)) + self.isotz(-time.timezone/60)
 149
 150     def isotz(self, offset):
 151         """Convert minutes-east-of-UTC to RFC3339- and ISO-compatible time zone designator"""
 152         return '{:+03d}:{:02d}'.format(offset/60, offset%60)
 153
 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)
 158
 159         logstream = io.BytesIO()
 160         rootLogger = logging.getLogger()
 161         streamHandler = logging.StreamHandler(logstream)
 162         rootLogger.addHandler(streamHandler)
 163
 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,
 168             events.put_nowait,
 169             poll_fallback=False,
 170             last_log_id=None)
 171         self.assertIsInstance(self.ws, arvados.events.EventClient)
 172         self.assertEqual(200, events.get(True, 5)['status'])
 173
 174         # create obj
 175         human = arvados.api('v1').humans().create(body={}).execute()
 176
 177         # expect an event
 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)
 181
 182         # close (im)properly
 183         if close_unexpected:
 184             self.ws.ec.close_connection()
 185         else:
 186             self.ws.close()
 187
 188         # create one more obj
 189         human2 = arvados.api('v1').humans().create(body={}).execute()
 190
 191         # (un)expect the object creation event
 192         if close_unexpected:
 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)
 202         else:
 203             with self.assertRaises(Queue.Empty):
 204                 self.assertEqual(events.get(True, 2), None)
 205
 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")
 210         if close_unexpected:
 211             self.assertNotEqual(closeLogFound, -1)
 212         else:
 213             self.assertEqual(closeLogFound, -1)
 214         rootLogger.removeHandler(streamHandler)
 215
 216     def test_websocket_reconnect_on_unexpected_close(self):
 217         self._test_websocket_reconnect(True)
 218
 219     def test_websocket_no_reconnect_on_close_by_user(self):
 220         self._test_websocket_reconnect(False)
 221
 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]
 226
 227         logstream = io.BytesIO()
 228         rootLogger = logging.getLogger()
 229         streamHandler = logging.StreamHandler(logstream)
 230         rootLogger.addHandler(streamHandler)
 231
 232         run_test_server.authorize_with('active')
 233         events = Queue.Queue(100)
 234
 235         filters = [['object_uuid', 'is_a', 'arvados#human']]
 236         self.ws = arvados.events.subscribe(
 237             arvados.api('v1'), filters,
 238             events.put_nowait,
 239             poll_fallback=False,
 240             last_log_id=None)
 241         self.assertIsInstance(self.ws, arvados.events.EventClient)
 242
 243         # simulate improper close
 244         self.ws.on_closed()
 245
 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)
 251
 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)
 259
 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)
 267
 268     @mock.patch('arvados.events._EventClient')
 269     def test_run_forever_survives_reconnects(self, websocket_client):
 270         connected = threading.Event()
 271         websocket_client().connect.side_effect = connected.set
 272         client = arvados.events.EventClient(
 273             self.MOCK_WS_URL, [], lambda event: None, None)
 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         self.assertTrue(connected.wait(timeout=self.TEST_TIMEOUT))
 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     TEST_TIMEOUT = 10.0
 290
 291     class MockLogs(object):
 292
 293         def __init__(self):
 294             self.logs = []
 295             self.lock = threading.Lock()
 296             self.api_called = threading.Event()
 297
 298         def add(self, log):
 299             with self.lock:
 300                 self.logs.append(log)
 301
 302         def return_list(self, num_retries=None):
 303             self.api_called.set()
 304             args, kwargs = self.list_func.call_args_list[-1]
 305             filters = kwargs.get('filters', [])
 306             if not any(True for f in filters if f[0] == 'id' and f[1] == '>'):
 307                 # No 'id' filter was given -- this must be the probe
 308                 # to determine the most recent id.
 309                 return {'items': [{'id': 1}], 'items_available': 1}
 310             with self.lock:
 311                 retval = self.logs
 312                 self.logs = []
 313             return {'items': retval, 'items_available': len(retval)}
 314
 315     def setUp(self):
 316         self.logs = self.MockLogs()
 317         self.arv = mock.MagicMock(name='arvados.api()')
 318         self.arv.logs().list().execute.side_effect = self.logs.return_list
 319         # our MockLogs object's "execute" stub will need to inspect
 320         # the call history to determine X in
 321         # ....logs().list(filters=X).execute():
 322         self.logs.list_func = self.arv.logs().list
 323         self.status_ok = threading.Event()
 324         self.event_received = threading.Event()
 325         self.recv_events = []
 326
 327     def tearDown(self):
 328         if hasattr(self, 'client'):
 329             self.client.close(timeout=None)
 330
 331     def callback(self, event):
 332         if event.get('status') == 200:
 333             self.status_ok.set()
 334         else:
 335             self.recv_events.append(event)
 336             self.event_received.set()
 337
 338     def build_client(self, filters=None, callback=None, last_log_id=None, poll_time=99):
 339         if filters is None:
 340             filters = []
 341         if callback is None:
 342             callback = self.callback
 343         self.client = arvados.events.PollClient(
 344             self.arv, filters, callback, poll_time, last_log_id)
 345
 346     def was_filter_used(self, target):
 347         return any(target in call[-1].get('filters', [])
 348                    for call in self.arv.logs().list.call_args_list)
 349
 350     def test_callback(self):
 351         test_log = {'id': 12345, 'testkey': 'testtext'}
 352         self.logs.add({'id': 123})
 353         self.build_client(poll_time=.01)
 354         self.client.start()
 355         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
 356         self.assertTrue(self.event_received.wait(self.TEST_TIMEOUT))
 357         self.event_received.clear()
 358         self.logs.add(test_log.copy())
 359         self.assertTrue(self.event_received.wait(self.TEST_TIMEOUT))
 360         self.assertIn(test_log, self.recv_events)
 361
 362     def test_subscribe(self):
 363         client_filter = ['kind', '=', 'arvados#test']
 364         self.build_client()
 365         self.client.unsubscribe([])
 366         self.client.subscribe([client_filter[:]])
 367         self.client.start()
 368         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
 369         self.assertTrue(self.logs.api_called.wait(self.TEST_TIMEOUT))
 370         self.assertTrue(self.was_filter_used(client_filter))
 371
 372     def test_unsubscribe(self):
 373         should_filter = ['foo', '=', 'foo']
 374         should_not_filter = ['foo', '=', 'bar']
 375         self.build_client(poll_time=0.01)
 376         self.client.unsubscribe([])
 377         self.client.subscribe([should_not_filter[:]])
 378         self.client.subscribe([should_filter[:]])
 379         self.client.unsubscribe([should_not_filter[:]])
 380         self.client.start()
 381         self.logs.add({'id': 123})
 382         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
 383         self.assertTrue(self.event_received.wait(self.TEST_TIMEOUT))
 384         self.assertTrue(self.was_filter_used(should_filter))
 385         self.assertFalse(self.was_filter_used(should_not_filter))
 386
 387     def test_run_forever(self):
 388         self.build_client()
 389         self.client.start()
 390         forever_thread = threading.Thread(target=self.client.run_forever)
 391         forever_thread.start()
 392         self.assertTrue(self.status_ok.wait(self.TEST_TIMEOUT))
 393         self.assertTrue(forever_thread.is_alive())
 394         self.client.close()
 395         forever_thread.join()
 396         del self.client