+ try:
+ client.on_closed()
+ self.assertTrue(forever_thread.is_alive())
+ self.assertEqual(2, websocket_client.call_count)
+ finally:
+ client.close()
+ forever_thread.join()
+
+ @staticmethod
+ def fake_client(conn_patch, filters=None, url=MOCK_WS_URL):
+ """Set up EventClient test infrastructure
+
+ Given a patch of `arvados.events.ws_client.connect`,
+ this returns a 3-tuple:
+
+ * `msg_event` is a `threading.Event` that is set as the test client
+ event callback. You can wait for this event to confirm that a
+ sent message has been acknowledged and processed.
+
+ * `mock_client` is a `mock.Mock` wrapper around `FakeWebsocketClient`.
+ Use this to assert `EventClient` calls the right methods. It tests
+ that `EventClient` acquires a lock before calling `send`.
+
+ * `client` is the `EventClient` that uses `mock_client` under the hood
+ that you exercise methods of.
+
+ Other arguments are passed to initialize `EventClient`.
+ """
+ msg_event = threading.Event()
+ fake_client = FakeWebsocketClient()
+ mock_client = fake_client.mock_wrapper()
+ conn_patch.return_value = mock_client
+ client = arvados.events.EventClient(url, filters, lambda _: msg_event.set())
+ fake_client._check_lock = client._subscribe_lock
+ return msg_event, mock_client, client
+
+ @mock.patch('arvados.events.ws_client.connect')
+ def test_subscribe_locking(self, ws_conn):
+ f = [['created_at', '>=', '2023-12-01T00:00:00.000Z']]
+ msg_event, wss_client, self.ws = self.fake_client(ws_conn)
+ self.assertTrue(msg_event.wait(timeout=1), "timed out waiting for setup callback")
+ msg_event.clear()
+ wss_client.send.reset_mock()
+ self.ws.subscribe(f)
+ self.assertTrue(msg_event.wait(timeout=1), "timed out waiting for subscribe callback")
+ wss_client.send.assert_called()
+ (msg,), _ = wss_client.send.call_args
+ self.assertEqual(
+ json.loads(msg),
+ {'method': 'subscribe', 'filters': f},
+ )
+
+ @mock.patch('arvados.events.ws_client.connect')
+ def test_unsubscribe_locking(self, ws_conn):
+ f = [['created_at', '>=', '2023-12-01T01:00:00.000Z']]
+ msg_event, wss_client, self.ws = self.fake_client(ws_conn, f)
+ self.assertTrue(msg_event.wait(timeout=1), "timed out waiting for setup callback")
+ msg_event.clear()
+ wss_client.send.reset_mock()
+ self.ws.unsubscribe(f)
+ self.assertTrue(msg_event.wait(timeout=1), "timed out waiting for unsubscribe callback")
+ wss_client.send.assert_called()
+ (msg,), _ = wss_client.send.call_args
+ self.assertEqual(
+ json.loads(msg),
+ {'method': 'unsubscribe', 'filters': f},
+ )
+
+ @mock.patch('arvados.events.ws_client.connect')
+ def test_resubscribe_locking(self, ws_conn):
+ f = [['created_at', '>=', '2023-12-01T02:00:00.000Z']]
+ msg_event, wss_client, self.ws = self.fake_client(ws_conn, f)
+ self.assertTrue(msg_event.wait(timeout=1), "timed out waiting for setup callback")
+ msg_event.clear()
+ wss_client.send.reset_mock()
+ wss_client.force_disconnect()
+ self.assertTrue(msg_event.wait(timeout=1), "timed out waiting for resubscribe callback")
+ wss_client.send.assert_called()
+ (msg,), _ = wss_client.send.call_args
+ self.assertEqual(
+ json.loads(msg),
+ {'method': 'subscribe', 'filters': f},
+ )