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
TIME_FUTURE = time.time()+3600
MOCK_WS_URL = 'wss://[{}]/'.format(arvados_testutil.TEST_HOST)
+ TEST_TIMEOUT = 10.0
+
def setUp(self):
self.ws = None
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)
# close (im)properly
if close_unexpected:
- self.ws.close_connection()
+ self.ws.ec.close_connection()
else:
self.ws.close()
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)
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()
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):
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:
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)
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))
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()