sdk/python/tests/test_events.py

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