X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/c60995ff0471ce31f0858d1f19a6788014112438..b9e5c8b32858338850da3e12ce27570b828898b3:/sdk/python/tests/test_events.py diff --git a/sdk/python/tests/test_events.py b/sdk/python/tests/test_events.py index c637efe342..0199724339 100644 --- a/sdk/python/tests/test_events.py +++ b/sdk/python/tests/test_events.py @@ -1,14 +1,9 @@ import arvados -import arvados.events -import arvados.errors -from datetime import datetime, timedelta, tzinfo +import io import logging -import logging.handlers import mock import Queue import run_test_server -import StringIO -import tempfile import threading import time import unittest @@ -22,6 +17,8 @@ class WebsocketTest(run_test_server.TestCaseWithServers): TIME_FUTURE = time.time()+3600 MOCK_WS_URL = 'wss://[{}]/'.format(arvados_testutil.TEST_HOST) + TEST_TIMEOUT = 10.0 + def setUp(self): self.ws = None @@ -156,7 +153,7 @@ class WebsocketTest(run_test_server.TestCaseWithServers): run_test_server.authorize_with('active') events = Queue.Queue(100) - logstream = StringIO.StringIO() + logstream = io.BytesIO() rootLogger = logging.getLogger() streamHandler = logging.StreamHandler(logstream) rootLogger.addHandler(streamHandler) @@ -181,7 +178,7 @@ class WebsocketTest(run_test_server.TestCaseWithServers): # close (im)properly if close_unexpected: - self.ws.close_connection() + self.ws.ec.close_connection() else: self.ws.close() @@ -224,7 +221,7 @@ class WebsocketTest(run_test_server.TestCaseWithServers): def test_websocket_reconnect_retry(self, event_client_connect): event_client_connect.side_effect = [None, Exception('EventClient.connect error'), None] - logstream = StringIO.StringIO() + logstream = io.BytesIO() rootLogger = logging.getLogger() streamHandler = logging.StreamHandler(logstream) rootLogger.addHandler(streamHandler) @@ -265,17 +262,31 @@ class WebsocketTest(run_test_server.TestCaseWithServers): client.unsubscribe(filters[:]) websocket_client().unsubscribe.assert_called_with(filters) - @unittest.expectedFailure @mock.patch('arvados.events._EventClient') - def test_run_forever(self, websocket_client): + def test_run_forever_survives_reconnects(self, websocket_client): + connected = threading.Event() + websocket_client().connect.side_effect = connected.set client = arvados.events.EventClient( self.MOCK_WS_URL, [], lambda event: None, None) - client.run_forever() - websocket_client().run_forever.assert_called_with() + forever_thread = threading.Thread(target=client.run_forever) + forever_thread.start() + # Simulate an unexpected disconnect, and wait for reconnect. + close_thread = threading.Thread(target=client.on_closed) + close_thread.start() + self.assertTrue(connected.wait(timeout=self.TEST_TIMEOUT)) + close_thread.join() + run_forever_alive = forever_thread.is_alive() + client.close() + forever_thread.join() + self.assertTrue(run_forever_alive) + self.assertEqual(2, websocket_client().connect.call_count) class PollClientTestCase(unittest.TestCase): + TEST_TIMEOUT = 10.0 + class MockLogs(object): + def __init__(self): self.logs = [] self.lock = threading.Lock() @@ -290,12 +301,11 @@ class PollClientTestCase(unittest.TestCase): self.logs = [] return {'items': retval, 'items_available': len(retval)} - def setUp(self): self.logs = self.MockLogs() self.arv = mock.MagicMock(name='arvados.api()') self.arv.logs().list().execute.side_effect = self.logs.return_list - self.callback_cond = threading.Condition() + self.callback_called = threading.Event() self.recv_events = [] def tearDown(self): @@ -303,9 +313,8 @@ class PollClientTestCase(unittest.TestCase): self.client.close(timeout=None) def callback(self, event): - with self.callback_cond: - self.recv_events.append(event) - self.callback_cond.notify_all() + self.recv_events.append(event) + self.callback_called.set() def build_client(self, filters=None, callback=None, last_log_id=None, poll_time=99): if filters is None: @@ -323,11 +332,11 @@ class PollClientTestCase(unittest.TestCase): test_log = {'id': 12345, 'testkey': 'testtext'} self.logs.add({'id': 123}) self.build_client(poll_time=.01) - with self.callback_cond: - self.client.start() - self.callback_cond.wait() - self.logs.add(test_log.copy()) - self.callback_cond.wait() + self.client.start() + self.assertTrue(self.callback_called.wait(self.TEST_TIMEOUT)) + self.callback_called.clear() + self.logs.add(test_log.copy()) + self.assertTrue(self.callback_called.wait(self.TEST_TIMEOUT)) self.client.close(timeout=None) self.assertIn(test_log, self.recv_events) @@ -335,9 +344,8 @@ class PollClientTestCase(unittest.TestCase): client_filter = ['kind', '=', 'arvados#test'] self.build_client() self.client.subscribe([client_filter[:]]) - with self.callback_cond: - self.client.start() - self.callback_cond.wait() + self.client.start() + self.assertTrue(self.callback_called.wait(self.TEST_TIMEOUT)) self.client.close(timeout=None) self.assertTrue(self.was_filter_used(client_filter)) @@ -352,11 +360,10 @@ class PollClientTestCase(unittest.TestCase): def test_run_forever(self): self.build_client() - with self.callback_cond: - self.client.start() - forever_thread = threading.Thread(target=self.client.run_forever) - forever_thread.start() - self.callback_cond.wait() + self.client.start() + forever_thread = threading.Thread(target=self.client.run_forever) + forever_thread.start() + self.assertTrue(self.callback_called.wait(self.TEST_TIMEOUT)) self.assertTrue(forever_thread.is_alive()) self.client.close() forever_thread.join()