[arvados] created: 2.7.0-5502-g187f790d8b
git repository hosting
git at public.arvados.org
Fri Dec 1 15:39:06 UTC 2023
at 187f790d8b71c5c7b54a599bec6e58950d7116dc (commit)
commit 187f790d8b71c5c7b54a599bec6e58950d7116dc
Author: Brett Smith <brett.smith at curii.com>
Date: Fri Dec 1 10:18:13 2023 -0500
21146: Add last_log_id argument to PollClient.subscribe
This helps avoid a crash if `subscribe` returns a `PollClient` and then
the user calls `subscribe` with two arguments. I don't know whether
ignoring `last_log_id` is *correct*, but that's the current behavior,
and anything else is out of scope for the current ticket.
Arvados-DCO-1.1-Signed-off-by: Brett Smith <brett.smith at curii.com>
diff --git a/sdk/python/arvados/events.py b/sdk/python/arvados/events.py
index 56650558e8..917c876706 100644
--- a/sdk/python/arvados/events.py
+++ b/sdk/python/arvados/events.py
@@ -468,12 +468,15 @@ class PollClient(threading.Thread):
# to do so raises the same exception."
pass
- def subscribe(self, f: Filter) -> None:
+ def subscribe(self, f: Filter, last_log_id: Optional[int]=None) -> None:
"""Subscribe to another set of events from the server
Arguments:
* f: arvados.events.Filter | None --- One filter to subscribe to.
+
+ * last_log_id: Optional[int] --- Ignored; this argument exists for
+ API compatibility with `EventClient.subscribe`.
"""
self.on_event({'status': 200})
self.filters.append(f)
commit d768d69e85e12e8f300687ea27b995fc419a0eea
Author: Brett Smith <brett.smith at curii.com>
Date: Fri Dec 1 10:16:56 2023 -0500
19825: Document arvados.events module
Arvados-DCO-1.1-Signed-off-by: Brett Smith <brett.smith at curii.com>
diff --git a/sdk/python/arvados/events.py b/sdk/python/arvados/events.py
index 8b81bf3939..56650558e8 100644
--- a/sdk/python/arvados/events.py
+++ b/sdk/python/arvados/events.py
@@ -1,6 +1,17 @@
# Copyright (C) The Arvados Authors. All rights reserved.
#
# SPDX-License-Identifier: Apache-2.0
+"""Follow events on an Arvados cluster
+
+This module provides different ways to get notified about events that happen
+on an Arvados cluster. You indicate which events you want updates about, and
+provide a function that is called any time one of those events is received
+from the server.
+
+`subscribe` is the main entry point. It helps you construct one of the two
+API-compatible client classes: `EventClient` (which uses WebSockets) or
+`PollClient` (which periodically queries the logs list methods).
+"""
import enum
import json
@@ -269,7 +280,43 @@ class EventClient(threading.Thread):
class PollClient(threading.Thread):
- def __init__(self, api, filters, on_event, poll_time, last_log_id):
+ """Follow Arvados events via polling logs
+
+ PollClient follows events on Arvados cluster by periodically running
+ logs list API calls. Users can select the events they want to follow and
+ run their own callback function on each.
+ """
+ def __init__(
+ self,
+ api: 'arvados.api_resources.ArvadosAPIClient',
+ filters: Optional[Filter],
+ on_event: EventCallback,
+ poll_time: float=15,
+ last_log_id: Optional[int]=None,
+ ) -> None:
+ """Initialize a polling client
+
+ Constructor arguments:
+
+ * api: arvados.api_resources.ArvadosAPIClient --- The Arvados API
+ client used to query logs. It will be used in a separate thread,
+ so if it is not an instance of `arvados.safeapi.ThreadSafeApiCache`
+ it should not be reused after the thread is started.
+
+ * filters: arvados.events.Filter | None --- One event filter to
+ subscribe to after connecting to the WebSocket server. If not
+ specified, the client will subscribe to all events.
+
+ * on_event: arvados.events.EventCallback --- When the client
+ receives an event from the WebSocket server, it calls this
+ function with the event object.
+
+ * poll_time: float --- The number of seconds to wait between querying
+ logs. Default 15.
+
+ * last_log_id: int | None --- If specified, queries will include a
+ filter for logs with an `id` at least this value.
+ """
super(PollClient, self).__init__()
self.api = api
if filters:
@@ -294,6 +341,11 @@ class PollClient(threading.Thread):
self._skip_old_events = False
def run(self):
+ """Run the client loop
+
+ This method runs in a separate thread to poll and process events
+ from the server.
+ """
self.on_event({'status': 200})
while not self._closing.is_set():
@@ -382,23 +434,29 @@ class PollClient(threading.Thread):
self._closing.wait(self.poll_time)
def run_forever(self):
+ """Run the polling client indefinitely
+
+ This method blocks until the `close` method is called (e.g., from
+ another thread) or the client permanently loses its connection.
+ """
# Have to poll here, otherwise KeyboardInterrupt will never get processed.
while not self._closing.is_set():
self._closing.wait(1)
- def close(self, code=None, reason=None, timeout=0):
- """Close poll client and optionally wait for it to finish.
+ def close(self, code: Optional[int]=None, reason: Optional[str]=None, timeout: float=0) -> None:
+ """Stop polling and processing events
- If an :on_event: handler is running in a different thread,
- first wait (indefinitely) for it to return.
+ Arguments:
- After closing, wait up to :timeout: seconds for the thread to
- finish the poll request in progress (if any).
+ * code: Optional[int] --- Ignored; this argument exists for API
+ compatibility with `EventClient.close`.
- :code: and :reason: are ignored. They are present for
- interface compatibility with EventClient.
- """
+ * reason: Optional[str] --- Ignored; this argument exists for API
+ compatibility with `EventClient.close`.
+ * timeout: float --- How long to wait for the client thread to finish
+ processing events. Default 0, which means no timeout.
+ """
with self._closing_lock:
self._closing.set()
try:
@@ -410,11 +468,24 @@ class PollClient(threading.Thread):
# to do so raises the same exception."
pass
- def subscribe(self, f):
+ def subscribe(self, f: Filter) -> None:
+ """Subscribe to another set of events from the server
+
+ Arguments:
+
+ * f: arvados.events.Filter | None --- One filter to subscribe to.
+ """
self.on_event({'status': 200})
self.filters.append(f)
def unsubscribe(self, f):
+ """Unsubscribe from an event stream
+
+ Arguments:
+
+ * f: arvados.events.Filter | None --- One event filter to stop
+ receiving events for.
+ """
del self.filters[self.filters.index(f)]
@@ -432,21 +503,42 @@ def _subscribe_websocket(api, filters, on_event, last_log_id=None):
else:
return client
-
-def subscribe(api, filters, on_event, poll_fallback=15, last_log_id=None):
- """
- :api:
- a client object retrieved from arvados.api(). The caller should not use this client object for anything else after calling subscribe().
- :filters:
- Initial subscription filters.
- :on_event:
- The callback when a message is received.
- :poll_fallback:
- If websockets are not available, fall back to polling every N seconds. If poll_fallback=False, this will return None if websockets are not available.
- :last_log_id:
- Log rows that are newer than the log id
+def subscribe(
+ api: 'arvados.api_resources.ArvadosAPIClient',
+ filters: Optional[Filter],
+ on_event: EventCallback,
+ poll_fallback: float=15,
+ last_log_id: Optional[int]=None,
+) -> Union[EventClient, PollClient]:
+ """Start a thread to monitor events
+
+ This method tries to construct an `EventClient` to process Arvados
+ events via WebSockets. If that fails, or the
+ `ARVADOS_DISABLE_WEBSOCKETS` flag is set in user configuration, it falls
+ back to constructing a `PollClient` to process the events via API
+ polling.
+
+ Arguments:
+
+ * api: arvados.api_resources.ArvadosAPIClient --- The Arvados API
+ client used to query logs. It may be used in a separate thread,
+ so if it is not an instance of `arvados.safeapi.ThreadSafeApiCache`
+ it should not be reused after this method returns.
+
+ * filters: arvados.events.Filter | None --- One event filter to
+ subscribe to after initializing the client. If not specified, the
+ client will subscribe to all events.
+
+ * on_event: arvados.events.EventCallback --- When the client receives an
+ event, it calls this function with the event object.
+
+ * poll_time: float --- The number of seconds to wait between querying
+ logs. If 0, this function will refuse to construct a `PollClient`.
+ Default 15.
+
+ * last_log_id: int | None --- If specified, start processing events with
+ at least this `id` value.
"""
-
if not poll_fallback:
return _subscribe_websocket(api, filters, on_event, last_log_id)
commit eb2aed26a98fddbe6efc6b10464fd188b890f273
Author: Brett Smith <brett.smith at curii.com>
Date: Fri Nov 24 16:38:32 2023 -0500
21146: Migrate PySDK to websockets library
This library is more mature than ws4py. This commit introduces a new
EventClient implementation based on it while remaining API-compatible.
Arvados-DCO-1.1-Signed-off-by: Brett Smith <brett.smith at curii.com>
diff --git a/sdk/python/arvados/events.py b/sdk/python/arvados/events.py
index 2c5e33c2e5..8b81bf3939 100644
--- a/sdk/python/arvados/events.py
+++ b/sdk/python/arvados/events.py
@@ -2,145 +2,270 @@
#
# SPDX-License-Identifier: Apache-2.0
+import enum
import json
import logging
import os
import re
import ssl
+import sys
import _thread
import threading
import time
+import websockets.exceptions as ws_exc
+import websockets.sync.client as ws_client
+
from . import config
from . import errors
from . import util
from .retry import RetryLoop
-
-from ws4py.client.threadedclient import WebSocketClient
+from ._version import __version__
+
+from typing import (
+ Any,
+ Callable,
+ Dict,
+ Iterable,
+ List,
+ Optional,
+ Union,
+)
+
+EventCallback = Callable[[Dict[str, Any]], object]
+"""Type signature for an event handler callback"""
+FilterCondition = List[Union[None, str, 'Filter']]
+"""Type signature for a single filter condition"""
+Filter = List[FilterCondition]
+"""Type signature for an entire filter"""
_logger = logging.getLogger('arvados.events')
-class _EventClient(WebSocketClient):
- def __init__(self, url, filters, on_event, last_log_id, on_closed):
- ssl_options = {'ca_certs': util.ca_certs_path()}
- if config.flag_is_true('ARVADOS_API_HOST_INSECURE'):
- ssl_options['cert_reqs'] = ssl.CERT_NONE
- else:
- ssl_options['cert_reqs'] = ssl.CERT_REQUIRED
+class WSMethod(enum.Enum):
+ """Arvados WebSocket methods
- # Warning: If the host part of url resolves to both IPv6 and
- # IPv4 addresses (common with "localhost"), only one of them
- # will be attempted -- and it might not be the right one. See
- # ws4py's WebSocketBaseClient.__init__.
- super(_EventClient, self).__init__(url, ssl_options=ssl_options)
+ This enum represents valid values for the `method` field in messages
+ sent to an Arvados WebSocket server.
+ """
+ SUBSCRIBE = 'subscribe'
+ SUB = SUBSCRIBE
+ UNSUBSCRIBE = 'unsubscribe'
+ UNSUB = UNSUBSCRIBE
- self.filters = filters
- self.on_event = on_event
+
+class EventClient(threading.Thread):
+ """Follow Arvados events via WebSocket
+
+ EventClient follows events on Arvados cluster published by the WebSocket
+ server. Users can select the events they want to follow and run their own
+ callback function on each.
+ """
+ _USER_AGENT = 'Python/{}.{}.{} arvados.events/{}'.format(
+ *sys.version_info[:3],
+ __version__,
+ )
+
+ def __init__(
+ self,
+ url: str,
+ filters: Optional[Filter],
+ on_event_cb: EventCallback,
+ last_log_id: Optional[int]=None,
+ *,
+ insecure: Optional[bool]=None,
+ ) -> None:
+ """Initialize a WebSocket client
+
+ Constructor arguments:
+
+ * url: str --- The `wss` URL for an Arvados WebSocket server.
+
+ * filters: arvados.events.Filter | None --- One event filter to
+ subscribe to after connecting to the WebSocket server. If not
+ specified, the client will subscribe to all events.
+
+ * on_event_cb: arvados.events.EventCallback --- When the client
+ receives an event from the WebSocket server, it calls this
+ function with the event object.
+
+ * last_log_id: int | None --- If specified, this will be used as the
+ value for the `last_log_id` field in subscribe messages sent by
+ the client.
+
+ Constructor keyword arguments:
+
+ * insecure: bool | None --- If `True`, the client will not check the
+ validity of the server's TLS certificate. If not specified, uses
+ the value from the user's `ARVADOS_API_HOST_INSECURE` setting.
+ """
+ self.url = url
+ self.filters = [filters or []]
+ self.on_event_cb = on_event_cb
self.last_log_id = last_log_id
- self._closing_lock = threading.RLock()
- self._closing = False
- self._closed = threading.Event()
- self.on_closed = on_closed
+ self.is_closed = threading.Event()
+ self._ssl_ctx = ssl.create_default_context(
+ purpose=ssl.Purpose.SERVER_AUTH,
+ cafile=util.ca_certs_path(),
+ )
+ if insecure is None:
+ insecure = config.flag_is_true('ARVADOS_API_HOST_INSECURE')
+ if insecure:
+ self._ssl_ctx.check_hostname = False
+ self._ssl_ctx.verify_mode = ssl.CERT_NONE
+ self._subscribe_lock = threading.Lock()
+ self._connect()
+ super().__init__(daemon=True)
+ self.start()
+
+ def _connect(self) -> None:
+ # There are no locks protecting this method. After the thread starts,
+ # it should only be called from inside.
+ self._client = ws_client.connect(
+ self.url,
+ logger=_logger,
+ ssl_context=self._ssl_ctx,
+ user_agent_header=self._USER_AGENT,
+ )
+ self._client_ok = True
+
+ def _subscribe(self, f: Filter, last_log_id: Optional[int]) -> None:
+ extra = {}
+ if last_log_id is not None:
+ extra['last_log_id'] = last_log_id
+ return self._update_sub(WSMethod.SUBSCRIBE, f, **extra)
- def opened(self):
- for f in self.filters:
- self.subscribe(f, self.last_log_id)
+ def _update_sub(self, method: WSMethod, f: Filter, **extra: Any) -> None:
+ msg = json.dumps({
+ 'method': method.value,
+ 'filters': f,
+ **extra,
+ })
+ self._client.send(msg)
- def closed(self, code, reason=None):
- self._closed.set()
- self.on_closed()
+ def close(self, code: int=1000, reason: str='', timeout: float=0) -> None:
+ """Close the WebSocket connection and stop processing events
- def received_message(self, m):
- with self._closing_lock:
- if not self._closing:
- self.on_event(json.loads(str(m)))
+ Arguments:
+
+ * code: int --- The WebSocket close code sent to the server when
+ disconnecting. Default 1000.
- def close(self, code=1000, reason='', timeout=0):
- """Close event client and optionally wait for it to finish.
+ * reason: str --- The WebSocket close reason sent to the server when
+ disconnecting. Default is the empty string.
- :timeout: is the number of seconds to wait for ws4py to
- indicate that the connection has closed.
+ * timeout: float --- How long to wait for the WebSocket server to
+ acknowledge the disconnection, in seconds. Default 0, which means
+ no timeout.
"""
- super(_EventClient, self).close(code, reason)
- with self._closing_lock:
- # make sure we don't process any more messages.
- self._closing = True
- # wait for ws4py to tell us the connection is closed.
- self._closed.wait(timeout=timeout)
+ self.is_closed.set()
+ self._client.close_timeout = timeout or None
+ self._client.close(code, reason)
- def subscribe(self, f, last_log_id=None):
- m = {"method": "subscribe", "filters": f}
- if last_log_id is not None:
- m["last_log_id"] = last_log_id
- self.send(json.dumps(m))
+ def run_forever(self) -> None:
+ """Run the WebSocket client indefinitely
- def unsubscribe(self, f):
- self.send(json.dumps({"method": "unsubscribe", "filters": f}))
+ This method blocks until the `close` method is called (e.g., from
+ another thread) or the client permanently loses its connection.
+ """
+ # Have to poll here to let KeyboardInterrupt get raised.
+ while not self.is_closed.wait(1):
+ pass
+ def subscribe(self, f: Filter, last_log_id: Optional[int]=None) -> None:
+ """Subscribe to another set of events from the server
-class EventClient(object):
- def __init__(self, url, filters, on_event_cb, last_log_id):
- self.url = url
- if filters:
- self.filters = [filters]
- else:
- self.filters = [[]]
- self.on_event_cb = on_event_cb
- self.last_log_id = last_log_id
- self.is_closed = threading.Event()
- self._setup_event_client()
+ Arguments:
- def _setup_event_client(self):
- self.ec = _EventClient(self.url, self.filters, self.on_event,
- self.last_log_id, self.on_closed)
- self.ec.daemon = True
- try:
- self.ec.connect()
- except Exception:
- self.ec.close_connection()
- raise
+ * f: arvados.events.Filter | None --- One filter to subscribe to
+ events for.
- def subscribe(self, f, last_log_id=None):
- self.filters.append(f)
- self.ec.subscribe(f, last_log_id)
+ * last_log_id: int | None --- If specified, request events starting
+ from this id. If not specified, the server will only send events
+ that occur after processing the subscription.
+ """
+ with self._subscribe_lock:
+ self._subscribe(f, last_log_id)
+ self.filters.append(f)
- def unsubscribe(self, f):
- del self.filters[self.filters.index(f)]
- self.ec.unsubscribe(f)
+ def unsubscribe(self, f: Filter) -> None:
+ """Unsubscribe from an event stream
- def close(self, code=1000, reason='', timeout=0):
- self.is_closed.set()
- self.ec.close(code, reason, timeout)
+ Arguments:
+
+ * f: arvados.events.Filter | None --- One event filter to stop
+ receiving events for.
+ """
+ with self._subscribe_lock:
+ try:
+ index = self.filters.index(f)
+ except ValueError:
+ raise ValueError(f"filter not subscribed: {f!r}") from None
+ self._update_sub(WSMethod.UNSUBSCRIBE, f)
+ del self.filters[index]
+
+ def on_closed(self) -> None:
+ """Handle disconnection from the WebSocket server
+
+ This method is called when the client loses its connection from
+ receiving events. This implementation tries to establish a new
+ connection if it was not closed client-side.
+ """
+ if self.is_closed.is_set():
+ return
+ _logger.warning("Unexpected close. Reconnecting.")
+ for _ in RetryLoop(num_retries=25, backoff_start=.1, max_wait=15):
+ try:
+ self._connect()
+ except Exception as e:
+ _logger.warning("Error '%s' during websocket reconnect.", e)
+ else:
+ _logger.warning("Reconnect successful.")
+ break
+ else:
+ _logger.error("EventClient thread could not contact websocket server.")
+ self.is_closed.set()
+ _thread.interrupt_main()
+
+ def on_event(self, m: Dict[str, Any]) -> None:
+ """Handle an event from the WebSocket server
+
+ This method is called whenever the client receives an event from the
+ server. This implementation records the `id` field internally, then
+ calls the callback function provided at initialization time.
- def on_event(self, m):
- if m.get('id') != None:
- self.last_log_id = m.get('id')
+ Arguments:
+
+ * m: Dict[str, Any] --- The event object, deserialized from JSON.
+ """
+ try:
+ self.last_log_id = m['id']
+ except KeyError:
+ pass
try:
self.on_event_cb(m)
- except Exception as e:
+ except Exception:
_logger.exception("Unexpected exception from event callback.")
_thread.interrupt_main()
- def on_closed(self):
- if not self.is_closed.is_set():
- _logger.warning("Unexpected close. Reconnecting.")
- for tries_left in RetryLoop(num_retries=25, backoff_start=.1, max_wait=15):
- try:
- self._setup_event_client()
- _logger.warning("Reconnect successful.")
- break
- except Exception as e:
- _logger.warning("Error '%s' during websocket reconnect.", e)
- if tries_left == 0:
- _logger.exception("EventClient thread could not contact websocket server.")
- self.is_closed.set()
- _thread.interrupt_main()
- return
+ def run(self) -> None:
+ """Run the client loop
- def run_forever(self):
- # Have to poll here to let KeyboardInterrupt get raised.
- while not self.is_closed.wait(1):
- pass
+ This method runs in a separate thread to receive and process events
+ from the server.
+ """
+ self.setName(f'ArvadosWebsockets-{self.ident}')
+ while self._client_ok and not self.is_closed.is_set():
+ try:
+ with self._subscribe_lock:
+ for f in self.filters:
+ self._subscribe(f, self.last_log_id)
+ for msg_s in self._client:
+ if not self.is_closed.is_set():
+ msg = json.loads(msg_s)
+ self.on_event(msg)
+ except ws_exc.ConnectionClosed:
+ self._client_ok = False
+ self.on_closed()
class PollClient(threading.Thread):
diff --git a/sdk/python/setup.py b/sdk/python/setup.py
index 7f35a590e5..284a460f1a 100644
--- a/sdk/python/setup.py
+++ b/sdk/python/setup.py
@@ -129,7 +129,8 @@ setup(name='arvados-python-client',
'setuptools >=40.3.0',
# As of 4.8.0rc1, typing_extensions does not parse in Python 3.7
'typing_extensions >=3.7.4, <4.8; python_version<"3.8"',
- 'ws4py >=0.4.2',
+ 'websockets >=11.0',
+ 'websockets ~=11.0; python_version<"3.8"',
],
classifiers=[
'Programming Language :: Python :: 3',
diff --git a/sdk/python/tests/test_events.py b/sdk/python/tests/test_events.py
index f31971b1b4..b4e6a0b1cd 100644
--- a/sdk/python/tests/test_events.py
+++ b/sdk/python/tests/test_events.py
@@ -2,6 +2,7 @@
#
# SPDX-License-Identifier: Apache-2.0
+import json
import logging
import mock
import queue
@@ -10,10 +11,63 @@ import threading
import time
import unittest
+import websockets.exceptions as ws_exc
+
import arvados
from . import arvados_testutil as tutil
from . import run_test_server
+class FakeWebsocketClient:
+ """Fake self-contained version of websockets.sync.client.ClientConnection
+
+ This provides enough of the API to test EventClient. It loosely mimics
+ the Arvados WebSocket API by acknowledging subscribe messages. You can use
+ `mock_wrapper` to test calls. You can set `_check_lock` to test that the
+ given lock is acquired before `send` is called.
+ """
+
+ def __init__(self):
+ self._check_lock = None
+ self._closed = threading.Event()
+ self._messages = queue.Queue()
+
+ def mock_wrapper(self):
+ wrapper = mock.Mock(wraps=self)
+ wrapper.__iter__ = lambda _: self.__iter__()
+ return wrapper
+
+ def __iter__(self):
+ while True:
+ msg = self._messages.get()
+ self._messages.task_done()
+ if isinstance(msg, Exception):
+ raise msg
+ else:
+ yield msg
+
+ def close(self, code=1000, reason=''):
+ if not self._closed.is_set():
+ self._closed.set()
+ self.force_disconnect()
+
+ def force_disconnect(self):
+ self._messages.put(ws_exc.ConnectionClosed(None, None))
+
+ def send(self, msg):
+ if self._check_lock is not None and self._check_lock.acquire(blocking=False):
+ self._check_lock.release()
+ raise AssertionError(f"called ws_client.send() without lock")
+ elif self._closed.is_set():
+ raise ws_exc.ConnectionClosed(None, None)
+ try:
+ msg = json.loads(msg)
+ except ValueError:
+ status = 400
+ else:
+ status = 200
+ self._messages.put(json.dumps({'status': status}))
+
+
class WebsocketTest(run_test_server.TestCaseWithServers):
MAIN_SERVER = {}
@@ -193,7 +247,7 @@ class WebsocketTest(run_test_server.TestCaseWithServers):
# close (im)properly
if close_unexpected:
- self.ws.ec.close_connection()
+ self.ws._client.close()
else:
self.ws.close()
@@ -232,69 +286,115 @@ class WebsocketTest(run_test_server.TestCaseWithServers):
self._test_websocket_reconnect(False)
# Test websocket reconnection retry
- @mock.patch('arvados.events._EventClient.connect')
- def test_websocket_reconnect_retry(self, event_client_connect):
- event_client_connect.side_effect = [None, Exception('EventClient.connect error'), None]
-
+ @mock.patch('arvados.events.ws_client.connect')
+ def test_websocket_reconnect_retry(self, ws_conn):
logstream = tutil.StringIO()
rootLogger = logging.getLogger()
streamHandler = logging.StreamHandler(logstream)
rootLogger.addHandler(streamHandler)
-
- run_test_server.authorize_with('active')
- events = queue.Queue(100)
-
- filters = [['object_uuid', 'is_a', 'arvados#human']]
- self.ws = arvados.events.subscribe(
- arvados.api('v1'), filters,
- events.put_nowait,
- poll_fallback=False,
- last_log_id=None)
- self.assertIsInstance(self.ws, arvados.events.EventClient)
-
- # simulate improper close
- self.ws.on_closed()
-
- # verify log messages to ensure retry happened
- log_messages = logstream.getvalue()
- found = log_messages.find("Error 'EventClient.connect error' during websocket reconnect.")
- self.assertNotEqual(found, -1)
- rootLogger.removeHandler(streamHandler)
-
- @mock.patch('arvados.events._EventClient')
- def test_subscribe_method(self, websocket_client):
- filters = [['object_uuid', 'is_a', 'arvados#human']]
- client = arvados.events.EventClient(
- self.MOCK_WS_URL, [], lambda event: None, None)
- client.subscribe(filters[:], 99)
- websocket_client().subscribe.assert_called_with(filters, 99)
-
- @mock.patch('arvados.events._EventClient')
- def test_unsubscribe(self, websocket_client):
- filters = [['object_uuid', 'is_a', 'arvados#human']]
- client = arvados.events.EventClient(
- self.MOCK_WS_URL, filters[:], lambda event: None, None)
- client.unsubscribe(filters[:])
- websocket_client().unsubscribe.assert_called_with(filters)
-
- @mock.patch('arvados.events._EventClient')
+ try:
+ 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()
+ ws_conn.side_effect = [Exception('EventClient.connect error'), wss_client]
+ wss_client.force_disconnect()
+ self.assertTrue(msg_event.wait(timeout=1), "timed out waiting for reconnect callback")
+ # verify log messages to ensure retry happened
+ self.assertIn("Error 'EventClient.connect error' during websocket reconnect.", logstream.getvalue())
+ self.assertEqual(ws_conn.call_count, 3)
+ finally:
+ rootLogger.removeHandler(streamHandler)
+
+ @mock.patch('arvados.events.ws_client.connect')
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)
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)
+ 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},
+ )
class PollClientTestCase(unittest.TestCase):
commit e2a2b9d3ec8f11394df3cc587f9caab5f6132d67
Author: Brett Smith <brett.smith at curii.com>
Date: Fri Nov 24 15:05:18 2023 -0500
21146: Clean test_events imports
Arvados-DCO-1.1-Signed-off-by: Brett Smith <brett.smith at curii.com>
diff --git a/sdk/python/tests/test_events.py b/sdk/python/tests/test_events.py
index f5192160f3..f31971b1b4 100644
--- a/sdk/python/tests/test_events.py
+++ b/sdk/python/tests/test_events.py
@@ -2,13 +2,6 @@
#
# SPDX-License-Identifier: Apache-2.0
-from __future__ import print_function
-from __future__ import absolute_import
-from __future__ import division
-from future import standard_library
-standard_library.install_aliases()
-from builtins import range
-from builtins import object
import logging
import mock
import queue
@@ -21,7 +14,6 @@ import arvados
from . import arvados_testutil as tutil
from . import run_test_server
-
class WebsocketTest(run_test_server.TestCaseWithServers):
MAIN_SERVER = {}
commit 28c06ec49951386a2e1831e058a839346bddb024
Author: Brett Smith <brett.smith at curii.com>
Date: Wed Nov 22 12:13:46 2023 -0500
21146: Clean arvados.events imports
Arvados-DCO-1.1-Signed-off-by: Brett Smith <brett.smith at curii.com>
diff --git a/sdk/python/arvados/events.py b/sdk/python/arvados/events.py
index e53e4980a8..2c5e33c2e5 100644
--- a/sdk/python/arvados/events.py
+++ b/sdk/python/arvados/events.py
@@ -2,32 +2,27 @@
#
# SPDX-License-Identifier: Apache-2.0
-from __future__ import absolute_import
-from future import standard_library
-standard_library.install_aliases()
-from builtins import str
-from builtins import object
-import arvados
-from . import config
-from . import errors
-from .retry import RetryLoop
-
-import logging
import json
-import _thread
-import threading
-import time
+import logging
import os
import re
import ssl
+import _thread
+import threading
+import time
+
+from . import config
+from . import errors
+from . import util
+from .retry import RetryLoop
+
from ws4py.client.threadedclient import WebSocketClient
_logger = logging.getLogger('arvados.events')
-
class _EventClient(WebSocketClient):
def __init__(self, url, filters, on_event, last_log_id, on_closed):
- ssl_options = {'ca_certs': arvados.util.ca_certs_path()}
+ ssl_options = {'ca_certs': util.ca_certs_path()}
if config.flag_is_true('ARVADOS_API_HOST_INSECURE'):
ssl_options['cert_reqs'] = ssl.CERT_NONE
else:
commit b83ec86c35e47c19b01f4f0e6fa74ddac6a5b92b
Author: Brett Smith <brett.smith at curii.com>
Date: Wed Nov 22 11:54:18 2023 -0500
21146: Clean and sort PySDK install_requires
Arvados-DCO-1.1-Signed-off-by: Brett Smith <brett.smith at curii.com>
diff --git a/sdk/python/setup.py b/sdk/python/setup.py
index 9ba9629bca..7f35a590e5 100644
--- a/sdk/python/setup.py
+++ b/sdk/python/setup.py
@@ -116,21 +116,20 @@ setup(name='arvados-python-client',
],
install_requires=[
'ciso8601 >=2.0.0',
+ 'dataclasses; python_version<"3.7"',
'future',
'google-api-core <2.11.0', # 2.11.0rc1 is incompatible with google-auth<2
'google-api-python-client >=2.1.0',
'google-auth <2',
'httplib2 >=0.9.2, <0.20.2',
+ 'protobuf <4.0.0dev',
'pycurl >=7.19.5.1, <7.45.0',
+ 'pyparsing <3',
'ruamel.yaml >=0.15.54, <0.17.22',
'setuptools >=40.3.0',
# As of 4.8.0rc1, typing_extensions does not parse in Python 3.7
'typing_extensions >=3.7.4, <4.8; python_version<"3.8"',
'ws4py >=0.4.2',
- 'protobuf <4.0.0dev',
- 'pyparsing <3',
- 'setuptools >=40.3.0',
- 'dataclasses; python_version<"3.7"',
],
classifiers=[
'Programming Language :: Python :: 3',
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list