diff --git a/kazoo/client.py b/kazoo/client.py index 4877f3580..dc775ba87 100644 --- a/kazoo/client.py +++ b/kazoo/client.py @@ -99,10 +99,7 @@ _RETRY_COMPAT_DEFAULTS = dict( - max_retries=None, - retry_delay=0.1, - retry_backoff=2, - retry_max_delay=3600, + max_retries=None, retry_delay=0.1, retry_backoff=2, retry_max_delay=3600 ) _RETRY_COMPAT_MAPPING = dict( @@ -177,7 +174,6 @@ def __init__( check_hostname: bool = False, ) -> None: ... - # FIXME This should be deprecated then killed @overload @deprecated( "Passing retry configuration parameters directly to the client" @@ -332,11 +328,7 @@ def __init__( self.logger = logger or log # Record the handler strategy used - # FIXME we need to work on making SequentialGeventHandler being an - # IHandler to make this cleaner but for now this will do. - self.handler = cast( - "IHandler", handler if handler else SequentialThreadingHandler() - ) + self.handler = handler if handler else SequentialThreadingHandler() if inspect.isclass(self.handler): raise ConfigurationError( "Handler must be an instance of a class, " @@ -346,16 +338,8 @@ def __init__( self.auth_data = set(auth_data if auth_data else []) self.default_acl = default_acl self.randomize_hosts = randomize_hosts - # FIXME Note: hosts and chroot are set by set_hosts, which also checks - # for chroot changes at runtime, so we initialize them to None here to - # avoid confusion with the empty string that set_hosts would set them - # to. This is massively hacky as set_hosts is only called from here - # anyway, but I want to make this change minimally invasive. - # we should really do self.hosts, self.chroot = self.set_hosts(hosts) - # and have set_hosts return the hosts and chroot - self.hosts: list[tuple[str, int]] = None # type: ignore[assignment] - self.chroot: str = None # type: ignore[assignment] - self.set_hosts(hosts) + + self.hosts, self.chroot = self._collect_hosts(hosts) self.use_ssl = use_ssl self.verify_certs = verify_certs @@ -395,74 +379,54 @@ def __init__( self._stopped.set() self._writer_stopped.set() - # FIXME This is kind of gross but we need to set these to something so - # that the type checker will understand that they are set by the time - # they are used and that they have the right type. - # We would do better to use a few variables/functions instead of - # overloading self.retry but this is a bit less invasive to the code - # and the type checker can understand it with a few hacks - self.retry: KazooRetry = None # type: ignore[assignment] - self._conn_retry: KazooRetry = None # type: ignore[assignment] - - if type(connection_retry) is dict: - self._conn_retry = KazooRetry(**connection_retry) - elif type(connection_retry) is KazooRetry: - self._conn_retry = connection_retry - - if type(command_retry) is dict: - self.retry = KazooRetry(**command_retry) - elif type(command_retry) is KazooRetry: - self.retry = command_retry - - if type(self._conn_retry) is KazooRetry: + old_retry_keys = dict(_RETRY_COMPAT_DEFAULTS) + for key in old_retry_keys: + try: + old_retry_keys[key] = cast( + "dict[str, float | None]", kwargs + ).pop(key) + warnings.warn( + "Passing retry configuration param %s to the " + "client directly is deprecated, please pass a " + "configured retry object (using param %s)" + % (key, _RETRY_COMPAT_MAPPING[key]), + DeprecationWarning, + stacklevel=2, + ) + except KeyError: + pass + + retry_keys: dict[str, Any] = {} + for oldname, value in old_retry_keys.items(): + retry_keys[_RETRY_COMPAT_MAPPING[oldname]] = value + retry_keys["sleep_func"] = self.handler.sleep_func + + def make_retry( + retry: KazooRetry | KazooRetryParams | None, + ) -> KazooRetry: + if isinstance(retry, dict): + return KazooRetry(**retry) + if isinstance(retry, KazooRetry): + return retry + return KazooRetry(**retry_keys) + + self._conn_retry = make_retry(connection_retry) + self._retry = make_retry(command_retry) + + if self._conn_retry is not None: if self.handler.sleep_func != self._conn_retry.sleep_func: raise ConfigurationError( "Retry handler and event handler " " must use the same sleep func" ) - if type(self.retry) is KazooRetry: - if self.handler.sleep_func != self.retry.sleep_func: + if self._retry is not None: + if self.handler.sleep_func != self._retry.sleep_func: raise ConfigurationError( "Command retry handler and event handler " "must use the same sleep func" ) - if self.retry is None or self._conn_retry is None: - # Note: because of the hacks at line 280, mypy thinks this is - # unreachable - old_retry_keys = dict( # type: ignore[unreachable] - _RETRY_COMPAT_DEFAULTS - ) - for key in old_retry_keys: - try: - old_retry_keys[key] = kwargs.pop(key) - warnings.warn( - "Passing retry configuration param %s to the " - "client directly is deprecated, please pass a " - "configured retry object (using param %s)" - % (key, _RETRY_COMPAT_MAPPING[key]), - DeprecationWarning, - stacklevel=2, - ) - except KeyError: - pass - - retry_keys = {} - for oldname, value in old_retry_keys.items(): - retry_keys[_RETRY_COMPAT_MAPPING[oldname]] = value - - if self._conn_retry is None: - self._conn_retry = KazooRetry( - sleep_func=self.handler.sleep_func, - **retry_keys, - ) - if self.retry is None: - self.retry = KazooRetry( - sleep_func=self.handler.sleep_func, - **retry_keys, - ) - # Managing legacy SASL options for scheme, auth in self.auth_data: if scheme != "sasl": @@ -506,7 +470,6 @@ def __init__( # Every retry call should have its own copy of the retry helper # to avoid shared retry counts - self._retry = self.retry def _retry( func: Callable[GenericArgs, KazooRetry.RETRY_RETURN], @@ -515,14 +478,7 @@ def _retry( ) -> KazooRetry.RETRY_RETURN: return self._retry.copy()(func, *args, **kwargs) - # FIXME - # (expression has type "Callable[[VarArg(Any), KwArg(Any)], Any]", - # variable has type "KazooRetry") so basically self.retry needs to be - # set to that and then the type checker will understand that - # self.retry.copy() is a valid call. This is just a mess and needs the - # code rearranging to be more mypy friendly but this is the least - # invasive way to do it for now - self.retry = _retry # type: ignore[assignment] + self.retry = _retry self.Barrier = partial(Barrier, self) self.Counter = partial(Counter, self) @@ -607,10 +563,14 @@ def connected(self) -> bool: established.""" return self._live.is_set() + def _collect_hosts( + self, hosts: str | list[str] + ) -> tuple[list[tuple[str, int]], str]: + new_hosts, chroot = collect_hosts(hosts) + return new_hosts, normpath(chroot) + def set_hosts( - self, - hosts: str | list[str], - randomize_hosts: bool | None = None, + self, hosts: str | list[str], randomize_hosts: bool | None = None ) -> None: """sets the list of hosts used by this client. @@ -635,25 +595,18 @@ def set_hosts( zookeeper server cluster has undefined behavior. """ - # Change the client setting for randomization if specified + # Randomizing the list will be done at connect time if randomize_hosts is not None: self.randomize_hosts = randomize_hosts - # Randomizing the list will be done at connect time - self.hosts, chroot = collect_hosts(hosts) + self.hosts, chroot = self._collect_hosts(hosts) - if chroot: - new_chroot = normpath(chroot) - else: - new_chroot = "" - - if self.chroot is not None and new_chroot != self.chroot: + if chroot != self.chroot: raise ConfigurationError( "Changing chroot at runtime is not currently supported" ) - - self.chroot = new_chroot + self.chroot = chroot def add_listener(self, listener: ListenerFunc) -> None: """Add a function to be called for connection state changes. @@ -981,29 +934,12 @@ def _try_fetch() -> tuple[int, ...] | None: except ValueError: return None - def _is_valid(version: tuple[int, ...] | None) -> bool: - # All zookeeper versions should have at least major.minor - # version numbers; if we get one that doesn't it is likely not - # correct and was truncated... - if version and len(version) > 1: - return True - return False - - # FIXME A better way of doing this would be to put the initial - # _try_fetch in the loop and inline _is_valid but I want to minimise - # code changes - # Try 1 + retries amount of times to get a version that we know # will likely be acceptable... - version = _try_fetch() - if _is_valid(version): - # mypy doesn't recognise that _is_valid guarantees this - return version # type: ignore[return-value] - for _i in range(0, retries): + for _ in range(0, retries + 1): version = _try_fetch() - if _is_valid(version): - # mypy doesn't recognise that _is_valid guarantees this - return version # type: ignore[return-value] + if version is not None and len(version) > 1: + return version raise KazooException( "Unable to fetch useable server" " version after trying %s times" % (1 + max(0, retries)) @@ -1578,12 +1514,8 @@ def get_children_async( raise TypeError("Invalid type for 'include_data' (bool expected)") async_result = self.handler.async_result() - # FIXME? Do this as req = getc2 if include_data else getc - req: GetChildren | GetChildren2 - if include_data: - req = GetChildren2(_prefix_root(self.chroot, path), watch) - else: - req = GetChildren(_prefix_root(self.chroot, path), watch) + func = GetChildren2 if include_data else GetChildren + req = func(_prefix_root(self.chroot, path), watch) self._call(req, async_result) return async_result @@ -2099,9 +2031,7 @@ def _check_tx_state(self) -> None: raise ValueError("Transaction already committed") def _add( - self, - request: Any, - post_processor: Callable[[Any], Any] | None = None, + self, request: Any, post_processor: Callable[[Any], Any] | None = None ) -> None: self._check_tx_state() self.client.logger.log(BLATHER, "Added %r to %r", request, self) diff --git a/kazoo/handlers/eventlet.py b/kazoo/handlers/eventlet.py index c6f01d11f..08c5b8245 100644 --- a/kazoo/handlers/eventlet.py +++ b/kazoo/handlers/eventlet.py @@ -17,12 +17,12 @@ from kazoo.handlers import utils from kazoo.handlers.utils import selector_select +from kazoo.interfaces import IHandler, SpawnedFunc if TYPE_CHECKING: from kazoo.interfaces import ( Event, FdLike, - IHandler, Lockable, ReentrantLock, Socket, @@ -65,8 +65,7 @@ def __init__(self, handler: IHandler): ) -# FIXME This should inherit from IHandler -class SequentialEventletHandler: +class SequentialEventletHandler(IHandler): """Eventlet handler for sequentially executing callbacks. This handler executes callbacks in a sequential manner. A queue is @@ -96,15 +95,16 @@ class SequentialEventletHandler: """ name = "sequential_eventlet_handler" - queue_impl = green_queue.LightQueue + queue_impl = green_queue.LightQueue # type: ignore[assignment] queue_empty = green_queue.Empty + sleep_func = staticmethod(green_time.sleep) # type: ignore[assignment, attr-defined] def __init__(self) -> None: """Create a :class:`SequentialEventletHandler` instance""" - self.callback_queue = ( + self._callback_queue = ( self.queue_impl() # type: ignore[no-untyped-call] ) - self.completion_queue = ( + self._completion_queue = ( self.queue_impl() # type: ignore[no-untyped-call] ) self._workers: list[ # type: ignore[name-defined] @@ -115,10 +115,6 @@ def __init__(self) -> None: ] = [] self._started = False - @staticmethod - def sleep_func(wait: float) -> None: - green_time.sleep(wait) # type: ignore[attr-defined, no-untyped-call] - @property def running(self) -> bool: return self._started @@ -127,7 +123,7 @@ def running(self) -> bool: def _process_completion_queue(self) -> None: while True: - cb = self.completion_queue.get() # type: ignore[no-untyped-call] + cb = self._completion_queue.get() # type: ignore[no-untyped-call] if cb is _STOP: break try: @@ -143,7 +139,7 @@ def _process_completion_queue(self) -> None: def _process_callback_queue(self) -> None: while True: - cb = self.callback_queue.get() # type: ignore[no-untyped-call] + cb = self._callback_queue.get() # type: ignore[no-untyped-call] if cb is _STOP: break try: @@ -165,11 +161,11 @@ def start(self) -> None: w = eventlet.spawn( self._process_completion_queue # type: ignore[no-untyped-call] ) - self._workers.append((w, self.completion_queue)) + self._workers.append((w, self._completion_queue)) w = eventlet.spawn( self._process_callback_queue # type: ignore[no-untyped-call] ) - self._workers.append((w, self.callback_queue)) + self._workers.append((w, self._callback_queue)) self._started = True atexit.register(self.stop) @@ -236,6 +232,18 @@ def spawn( return t def dispatch_callback(self, callback: Callback) -> None: - self.callback_queue.put( # type: ignore[no-untyped-call] + self._callback_queue.put( # type: ignore[no-untyped-call] lambda: callback.func(*callback.args) ) + + def add_to_completion_queue(self, callback: SpawnedFunc) -> None: + """Add a method to callback on completion. + + :param callback: A :class:`~kazoo.protocol.states.Callback` + object to be called. + + The supplied method will be called whn an event completes. + + Note: This seems an odd place to store these. + """ + self._completion_queue.put(callback) # type: ignore[no-untyped-call] diff --git a/kazoo/handlers/gevent.py b/kazoo/handlers/gevent.py index 2f782fa28..7e8fe361c 100644 --- a/kazoo/handlers/gevent.py +++ b/kazoo/handlers/gevent.py @@ -3,33 +3,40 @@ from __future__ import annotations import atexit +import importlib.util import logging from typing import Any, Callable, Iterable, TYPE_CHECKING, cast +from typing_extensions import TypeAlias from kazoo.handlers.utils import selector_select from kazoo.handlers import utils - -# FIXME This is messy. -# We don't want to force the user to install gevent, so we need to handle the -# case where it's not available, but there should be a cleaner we of doing this -# than by importing all of gevent and ignoring all the import errors. -import gevent # type: ignore[import] -from gevent import socket # type: ignore[import] -import gevent.event # type: ignore[import] -import gevent.queue # type: ignore[import] - -import gevent.thread # type: ignore[import] -import gevent.selectors # type: ignore[import] -from gevent.lock import Semaphore, RLock # type: ignore[import] +from kazoo.interfaces import IAsyncResult, IHandler, SpawnedFunc if TYPE_CHECKING: - from kazoo.interfaces import FdLike, Lockable, Socket + from kazoo.interfaces import ( + Event, + FdLike, + Lockable, + ReentrantLock, + Socket, + Threadlike, + ) from kazoo.protocol.states import Callback - from gevent import Greenlet -_using_libevent = gevent.__version__.startswith("0.") + GEVENT_AVAILABLE: bool = True +else: + GEVENT_AVAILABLE = importlib.util.find_spec("gevent") is not None + +if GEVENT_AVAILABLE: + import gevent + from gevent import socket + import gevent.event + import gevent.queue + import gevent.thread + import gevent.selectors + from gevent.lock import Semaphore, RLock log = logging.getLogger(__name__) @@ -38,14 +45,11 @@ AsyncResult = gevent.event.AsyncResult -# The following would be great typenames, but python3.8 complains about type -# objects not being indexable. -# GCallback = Callable[..., None] -# Worker = Greenlet[..., Any] -# CallbackQueue = gevent.queue.Queue[Callable[..., None]] +Worker: TypeAlias = "gevent.Greenlet[..., Any]" +CallbackQueue: TypeAlias = "gevent.queue.Queue[Callable[..., None]]" -class SequentialGeventHandler: +class SequentialGeventHandler(IHandler): """Gevent handler for sequentially executing callbacks. This handler executes callbacks in a sequential manner. A queue is @@ -67,31 +71,52 @@ class SequentialGeventHandler: """ name = "sequential_gevent_handler" - queue_impl = gevent.queue.Queue + queue_impl = gevent.queue.Queue # type: ignore[assignment] queue_empty = gevent.queue.Empty - sleep_func = staticmethod(gevent.sleep) + sleep_func = staticmethod(gevent.sleep) # type: ignore[assignment] def __init__(self) -> None: """Create a :class:`SequentialGeventHandler` instance""" - self.callback_queue: gevent.queue.Queue[Callable[..., None]] = ( - self.queue_impl() - ) + self._callback_queue: CallbackQueue = self.queue_impl() + + def FakeQueue() -> CallbackQueue: + """Raise NotImplementedError() on put(). + + This is here because generally AyncResult is created with an + IHandler, but we don't actually return a kazoo AsyncResult, we + return a gevent one which behaves a little differently underneath. + """ + + class _FakeQueue: + def put(self, item: Any) -> None: + raise NotImplementedError( + "FakeQueue does not support put()" + ) + + return cast(CallbackQueue, _FakeQueue()) + + self._completion_queue = FakeQueue() + self._running = False self._async = None self._state_change = Semaphore() - self._workers: list[Greenlet[..., Any]] = [] + self._workers: list[Worker] = [] @property def running(self) -> bool: return self._running class timeout_exception(gevent.Timeout): + """Exception raised when a timeout occurs. + + Passing seconds=None to the constructor will create a timeout that + never expires. + """ + def __init__(self, msg: Any): - gevent.Timeout.__init__(self, exception=msg) + super().__init__(exception=msg) - def _create_greenlet_worker( - self, queue: gevent.queue.Queue[Callable[..., None]] - ) -> Greenlet[..., Any]: + def _create_greenlet_worker(self, queue: CallbackQueue) -> Worker: def greenlet_worker() -> None: while True: try: @@ -121,7 +146,7 @@ def start(self) -> None: # Spawn our worker greenlets, we have # - A callback worker for watch events to be called # FIXME Why the loop? - for queue in (self.callback_queue,): + for queue in (self._callback_queue,): w = self._create_greenlet_worker(queue) self._workers.append(w) atexit.register(self.stop) @@ -134,7 +159,7 @@ def stop(self) -> None: self._running = False - for queue in (self.callback_queue,): + for queue in (self._callback_queue,): queue.put(cast("Callable[..., None]", _STOP)) while self._workers: @@ -142,7 +167,7 @@ def stop(self) -> None: worker.join() # Clear the queues - self.callback_queue = self.queue_impl() # pragma: nocover + self._callback_queue = self.queue_impl() # pragma: nocover atexit.unregister(self.stop) @@ -167,7 +192,7 @@ def create_connection(self, *args: Any, **kwargs: Any) -> Socket: def create_socket_pair(self) -> tuple[Socket, Socket]: return utils.create_socket_pair(socket) - def event_object(self) -> gevent.event.Event: + def event_object(self) -> Event: """Create an appropriate Event object""" return gevent.event.Event() @@ -178,11 +203,11 @@ def lock_object(self) -> Lockable: gevent.thread.allocate_lock(), # type: ignore[no-untyped-call] ) - def rlock_object(self) -> RLock: + def rlock_object(self) -> ReentrantLock: """Create an appropriate RLock object""" return RLock() - def async_result(self) -> AsyncResult[Any]: + def async_result(self) -> IAsyncResult: """Create a :class:`AsyncResult` instance The :class:`AsyncResult` instance will have its completion @@ -193,9 +218,7 @@ def async_result(self) -> AsyncResult[Any]: """ return AsyncResult() - def spawn( - self, func: Any, *args: Any, **kwargs: Any - ) -> gevent.Greenlet[..., Any]: + def spawn(self, func: Any, *args: Any, **kwargs: Any) -> Threadlike: """Spawn a function to run asynchronously""" return gevent.spawn(func, *args, **kwargs) @@ -206,4 +229,16 @@ def dispatch_callback(self, callback: Callback) -> None: type as documented for the :class:`SequentialGeventHandler`. """ - self.callback_queue.put(lambda: callback.func(*callback.args)) + self._callback_queue.put(lambda: callback.func(*callback.args)) + + def add_to_completion_queue(self, callback: SpawnedFunc) -> None: + """Add a method to callback on completion. + + :param callback: A :class:`~kazoo.protocol.states.Callback` + object to be called. + + The supplied method will be called whn an event completes. + + Note: This seems an odd place to store these. + """ + raise NotImplementedError() diff --git a/kazoo/handlers/threading.py b/kazoo/handlers/threading.py index f816652fb..20758c24d 100644 --- a/kazoo/handlers/threading.py +++ b/kazoo/handlers/threading.py @@ -113,7 +113,7 @@ def __init__(self) -> None: self.callback_queue: queue.Queue[Callable[..., None]] = ( self.queue_impl() ) - self.completion_queue: queue.Queue[Callable[..., None]] = ( + self._completion_queue: queue.Queue[Callable[..., None]] = ( self.queue_impl() ) self._running = False @@ -155,7 +155,7 @@ def start(self) -> None: # Spawn our worker threads, we have # - A callback worker for watch events to be called # - A completion worker for completion events to be called - for work_queue in (self.completion_queue, self.callback_queue): + for work_queue in (self._completion_queue, self.callback_queue): w = self._create_thread_worker(work_queue) self._workers.append(w) self._running = True @@ -169,7 +169,7 @@ def stop(self) -> None: self._running = False - for work_queue in (self.completion_queue, self.callback_queue): + for work_queue in (self._completion_queue, self.callback_queue): work_queue.put(cast("Callable[..., None]", _STOP)) self._workers.reverse() @@ -179,7 +179,7 @@ def stop(self) -> None: # Clear the queues self.callback_queue = self.queue_impl() - self.completion_queue = self.queue_impl() + self._completion_queue = self.queue_impl() atexit.unregister(self.stop) def select( @@ -202,9 +202,7 @@ def event_object(self) -> Event: def lock_object(self) -> Lockable: """Create a lock object""" - # Note: This is not ideal, but the ContextManager Protocol seems to - # think you should return an object of the same type. - return cast("Lockable", threading.Lock()) + return threading.Lock() def rlock_object(self) -> ReentrantLock: """Create an appropriate RLock object""" @@ -230,3 +228,15 @@ def dispatch_callback(self, callback: Callback) -> None: """ self.callback_queue.put(lambda: callback.func(*callback.args)) + + def add_to_completion_queue(self, callback: SpawnedFunc) -> None: + """Add a method to callback on completion. + + :param callback: A :class:`~kazoo.protocol.states.Callback` + object to be called. + + The supplied method will be called when an event completes. + + Note: This seems an odd place to store these. + """ + self._completion_queue.put(callback) diff --git a/kazoo/handlers/utils.py b/kazoo/handlers/utils.py index c7741087b..0950911cb 100644 --- a/kazoo/handlers/utils.py +++ b/kazoo/handlers/utils.py @@ -13,7 +13,7 @@ from typing import Any, Callable, Iterable, TypeVar, TYPE_CHECKING from typing_extensions import ParamSpec -from kazoo.interfaces import IAsyncResult, FdLike +from kazoo.interfaces import IAsyncResult, IHandler, FdLike if TYPE_CHECKING: from types import ModuleType @@ -26,14 +26,7 @@ HAS_FNCTL = False -# sentinel objects -# Note: This needs to be a unique object that is not None, as None is used to -# indicate a successful result in AsyncResult. -# This should probably be an Enum, it would certainly be cleaner, but don't -# want to change the code too much. -_NONE = object() - -CallbackFunc = Callable[..., None] +CallbackFunc = Callable[["AsyncResult"], None] class AsyncResult(IAsyncResult): @@ -41,12 +34,13 @@ class AsyncResult(IAsyncResult): def __init__( self, - handler: Any, + handler: IHandler, condition_factory: Callable[[], Any], timeout_factory: Callable[[], Any], ) -> None: self._handler = handler - self._exception: object | Exception | None = _NONE + self._exception: Exception | None = None + self._ready = False self._condition = condition_factory() self._callbacks: list[CallbackFunc] = [] self._timeout_factory = timeout_factory @@ -55,32 +49,33 @@ def __init__( def ready(self) -> bool: """Return true if and only if it holds a value or an exception""" - return self._exception is not _NONE + return self._ready def successful(self) -> bool: """Return true if and only if it is ready and holds a value""" - return self._exception is None + return self._ready and self._exception is None @property def exception(self) -> Exception | None: - if self._exception is not _NONE: - return self._exception # type: ignore[return-value] - return None + return self._exception def set(self, value: Any = None) -> None: """Store the value. Wake up the waiters.""" with self._condition: self.value = value - self._exception = None - self._do_callbacks() - self._condition.notify_all() + self._wakeup() + + def _wakeup(self) -> None: + """All done, set ready and wake up any waiters""" + self._ready = True + self._do_callbacks() + self._condition.notify_all() def set_exception(self, exception: Exception) -> None: """Store the exception. Wake up the waiters.""" with self._condition: self._exception = exception - self._do_callbacks() - self._condition.notify_all() + self._wakeup() def get(self, block: bool = True, timeout: float | None = None) -> Any: """Return the stored value or raise the exception. @@ -89,16 +84,16 @@ def get(self, block: bool = True, timeout: float | None = None) -> Any: """ with self._condition: - if self._exception is not _NONE: + if self._ready: if self._exception is None: return self.value - raise self._exception # type: ignore[misc] - elif block: + raise self._exception + if block: self._condition.wait(timeout) - if self._exception is not _NONE: + if self.ready(): if self._exception is None: return self.value - raise self._exception # type: ignore[misc] + raise self._exception # if we get to this point we timeout raise self._timeout_factory() @@ -116,7 +111,7 @@ def wait(self, timeout: float | None = None) -> bool: with self._condition: if not self.ready(): self._condition.wait(timeout) - return self._exception is not _NONE + return self._ready def rawlink(self, callback: CallbackFunc) -> None: """Register a callback to call when a value or an exception is @@ -147,7 +142,7 @@ def _do_callbacks(self) -> None: for callback in self._callbacks: if self._handler.running: - self._handler.completion_queue.put( + self._handler.add_to_completion_queue( functools.partial(callback, self) ) else: @@ -382,33 +377,6 @@ def captured_function( return capture -def fileobj_to_fd(fileobj: FdLike) -> int: - """Return a file descriptor from a file object. - - Parameters: - fileobj -- file object or file descriptor - - Returns: - corresponding file descriptor - - Raises: - TypeError if the object is invalid - """ - if isinstance(fileobj, int): - fd = fileobj - else: - # FIXME given the protocol I don't think the try/catch/int are - # required. - try: - fd = int(fileobj.fileno()) - except (AttributeError, TypeError, ValueError): - raise TypeError("Invalid file object: {!r}".format(fileobj)) - # FIXME Questionable, just let select deal with it. - if fd < 0: - raise TypeError("Invalid file descriptor: {}".format(fd)) - return fd - - def selector_select( rlist: Iterable[FdLike], wlist: Iterable[FdLike], @@ -434,7 +402,7 @@ def selector_select( for event, fileobjs in events_mapping.items(): for fileobj in fileobjs: - fd = fileobj_to_fd(fileobj) + fd = fileobj if isinstance(fileobj, int) else fileobj.fileno() fd_events[fd] |= event fd_fileobjs[fd].append(fileobj) diff --git a/kazoo/hosts.py b/kazoo/hosts.py index cda746a3a..675f734f4 100644 --- a/kazoo/hosts.py +++ b/kazoo/hosts.py @@ -3,9 +3,7 @@ import urllib.parse -def collect_hosts( - hosts: str | list[str], -) -> tuple[list[tuple[str, int]], str | None]: +def collect_hosts(hosts: str | list[str]) -> tuple[list[tuple[str, int]], str]: """ Collect a set of hosts and an optional chroot from a string or a list of strings. @@ -14,11 +12,12 @@ def collect_hosts( if hosts[-1].strip().startswith("/"): host_ports, chroot = hosts[:-1], hosts[-1] else: - host_ports, chroot = hosts, None + host_ports, chroot = hosts, "" else: host_ports_1, chroot = hosts.partition("/")[::2] host_ports = host_ports_1.split(",") - chroot = "/" + chroot if chroot else None + if chroot != "": + chroot = "/" + chroot result = [] for host_port in host_ports: @@ -28,7 +27,7 @@ def collect_hosts( host = res.hostname if host is None: raise ValueError("bad hostname") - port = int(res.port) if res.port else 2181 + port = 2181 if res.port is None else res.port result.append((host.strip(), port)) return result, chroot diff --git a/kazoo/interfaces.py b/kazoo/interfaces.py index b7b685bb0..c5f838935 100644 --- a/kazoo/interfaces.py +++ b/kazoo/interfaces.py @@ -10,13 +10,14 @@ from __future__ import annotations - from typing import ( Any, Callable, Iterable, + Literal, Protocol, Union, + overload, TYPE_CHECKING, ) @@ -70,11 +71,11 @@ def shutdown(self, flag: int) -> None: ... class Lockable(Protocol): """This is what threading.Lock implements. - In python 3.9+ it's available natively. Though given it has some - very odd typing, I wouldn't put money on it. + In python 3.9+ Lockable is meant to be available natively. Though I haven't + seen it yet. """ - def __enter__(self) -> None: ... + def __enter__(self) -> bool: ... def __exit__( self, @@ -98,7 +99,10 @@ class ReentrantLock(Protocol): In python 3.14+, it's the same as Lock, which adds to the fun. """ - def __enter__(self) -> None: ... + # Note: GreenLet Rlock returns a bool, but the threading.RLock returns + # None, unlike threading.Lock. So we have to use a union type here to + # satisfy both. This is a hack + def __enter__(self) -> bool | None: ... def __exit__( self, @@ -121,14 +125,25 @@ def set(self) -> None: ... def clear(self) -> None: ... + # This overload is for the case where timeout is None, which means wait + # indefinitely. gevent's Event.wait explicitly returns True in this case, + # so we reflect that in the type. + # The threading.Event.wait actually conforms to this overload as well, + # since it returns True if the event is set, and if timeout is None, it + # will always return True, but it doesn't explicitly say that in the docs, + # so we have to be a bit more conservative with the typing. + @overload + def wait(self, timeout: None = None) -> bool | Literal[True]: ... + + @overload + def wait(self, timeout: float) -> bool: ... + def wait(self, timeout: float | None = None) -> bool: ... class Threadlike(Protocol): """Protocol for something like a thread.""" - def is_alive(self) -> bool: ... - def join(self, timeout: float | None = None) -> None: ... @@ -181,6 +196,10 @@ def stop(self) -> None: """Stop the handler. Should block until the handler is safely stopped.""" + @property + def running(self) -> bool: + """Return True if the handler is running, False otherwise.""" + def select( self, rlist: Iterable[FdLike], @@ -240,6 +259,17 @@ def dispatch_callback(self, callback: Callback) -> None: """ + def add_to_completion_queue(self, callback: SpawnedFunc) -> None: + """Add a method to callback on completion. + + :param callback: A :class:`~kazoo.protocol.states.Callback` + object to be called. + + The supplied method will be called whn an event completes. + + Note: This seems an odd place to store these. + """ + class IAsyncResult(Protocol): """An Async Result object that can be queried for a value that has @@ -344,6 +374,10 @@ def unlink(self, callback: Callable[[IAsyncResult], None]) -> None: """ @property - def exception(self) -> Exception | None: + def exception(self) -> BaseException | None: """The exception set by :meth:`set_exception` or `None` if no - exception has been set""" + exception has been set + + gevent specifies BaseException here so we have to be compatible with + that, though it seems a little OTT. + """ diff --git a/kazoo/protocol/connection.py b/kazoo/protocol/connection.py index d35952be5..8b1214a48 100644 --- a/kazoo/protocol/connection.py +++ b/kazoo/protocol/connection.py @@ -5,6 +5,7 @@ from binascii import hexlify from contextlib import contextmanager import copy +import importlib.util import logging import random import select @@ -54,25 +55,19 @@ WatchedEvent, EVENT_TYPE_MAP, ) -from kazoo.retry import ( - ForceRetryError, - KazooRetry, - RetryFailedError, -) +from kazoo.retry import ForceRetryError, KazooRetry, RetryFailedError if TYPE_CHECKING: from kazoo.client import KazooClient, WatchFunc from kazoo.interfaces import Socket, Threadlike -# FIXME This is NOT pretty, but we don't want to force users to have to -# install puresasl. Can we avoid some of the type: ignore stuff? -try: - import puresasl # type: ignore[import-untyped,import-not-found] - import puresasl.client # type: ignore[import-untyped,import-not-found] + PURESASL_AVAILABLE: bool = True +else: + PURESASL_AVAILABLE = importlib.util.find_spec("puresasl") is not None - PURESASL_AVAILABLE = True -except ImportError: - PURESASL_AVAILABLE = False +if PURESASL_AVAILABLE: + import puresasl + import puresasl.client log = logging.getLogger(__name__) @@ -215,7 +210,7 @@ def __init__( self._connection_routine: Threadlike | None = None - self.sasl_options = sasl_options + self._sasl_options = sasl_options self.sasl_cli = None # This is instance specific to avoid odd thread bug issues in Python @@ -462,7 +457,7 @@ def _read_watch_event(self, buffer: bytes, offset: int) -> None: elif watch.type == CHILD_EVENT: watchers.extend(client._child_watchers.pop(path, [])) else: - self.logger.warn("Received unknown event %r", watch.type) + self.logger.warning("Received unknown event %r", watch.type) return # Strip the chroot if needed @@ -478,10 +473,7 @@ def _read_watch_event(self, buffer: bytes, offset: int) -> None: client.handler.dispatch_callback(Callback("watch", watch1, (ev,))) def _read_response( - self, - header: ReplyHeader, - buffer: bytes, - offset: int, + self, header: ReplyHeader, buffer: bytes, offset: int ) -> object | None: client = self.client request, async_object, xid = client._pending.popleft() @@ -489,9 +481,7 @@ def _read_response( client.last_zxid = header.zxid if header.xid != xid: exc = RuntimeError( - "xids do not match, expected %r received %r", - xid, - header.xid, + "xids do not match, expected %r received %r", xid, header.xid ) async_object.set_exception(exc) raise exc @@ -578,9 +568,7 @@ def _read_socket(self, read_timeout: float) -> object | None: return None def _send_request( - self, - read_timeout: float, - connect_timeout: float, + self, read_timeout: float, connect_timeout: float ) -> None: """Called when we have something to send out on the socket""" client = self.client @@ -711,11 +699,7 @@ def _connect_loop(self, retry: KazooRetry) -> object: raise ForceRetryError("Reconnecting") def _connect_attempt( - self, - host: str, - hostip: str, - port: int, - retry: KazooRetry, + self, host: str, hostip: str, port: int, retry: KazooRetry ) -> object: client = self.client KazooTimeoutError = self.handler.timeout_exception @@ -828,10 +812,7 @@ def _connect_attempt( return None def _connect( - self, - host: str, - hostip: str, - port: int, + self, host: str, hostip: str, port: int ) -> tuple[float, float]: client = self.client self.logger.info( @@ -912,8 +893,10 @@ def _connect( client._session_callback(KeeperState.CONNECTED) self._ro_mode = None - if self.sasl_options is not None: - self._authenticate_with_sasl(host, connect_timeout / 1000.0) + if self._sasl_options is not None: + self._authenticate_with_sasl( + self._sasl_options, host, connect_timeout / 1000.0 + ) # Get a copy of the auth data before iterating, in case it is # changed. @@ -927,36 +910,26 @@ def _connect( return read_timeout, connect_timeout - def _authenticate_with_sasl(self, host: str, timeout: float) -> None: + def _authenticate_with_sasl( + self, sasl_options: dict[str, str], host: str, timeout: float + ) -> None: """Establish a SASL authenticated connection to the server.""" if not PURESASL_AVAILABLE: raise SASLException("Missing SASL support") - # Although this can only be called if sasl_options is not None, we - # really should just have make self.sasl_options into an empty dict - # in the constructor. However, I want to avoid code changes in as - # much as possible. - if "service" not in self.sasl_options: # type: ignore[operator] - self.sasl_options["service"] = "zookeeper" # type: ignore[index] + if "service" not in sasl_options: + sasl_options["service"] = "zookeeper" # NOTE: Zookeeper hardcoded the domain for Digest authentication # instead of using the hostname. See # zookeeper/util/SecurityUtils.java#L74 and Server/Client # initializations. - if ( - self.sasl_options["mechanism"] # type: ignore[index] - == "DIGEST-MD5" - ): + if sasl_options["mechanism"] == "DIGEST-MD5": host = "zk-sasl-md5" - # I don't think the client.sasl_cli attribute is actually used - # anywhere else, so not sure why we need to set it on the client, - # but again, I want to avoid code changes as much as possible. - sasl_cli = self.client.sasl_cli = ( # type: ignore[attr-defined] - puresasl.client.SASLClient( # type: ignore[no-untyped-call] - host=host, - **self.sasl_options, # type: ignore[arg-type] - ) + # FIXME puresasl isn't properly type hinted. + sasl_cli = puresasl.client.SASLClient( # type: ignore + host=host, **sasl_options ) # Initialize the process with an empty challenge token diff --git a/kazoo/protocol/serialization.py b/kazoo/protocol/serialization.py index 4653c1e59..b989cb6aa 100644 --- a/kazoo/protocol/serialization.py +++ b/kazoo/protocol/serialization.py @@ -392,8 +392,36 @@ def serialize(self) -> bytearray: return b -# FIXME Transaction class should move after Create2 -Transaction_Types = Union[Create, "Create2", Delete, SetData, CheckVersion] +class Create2(namedtuple("Create2", "path data acl flags")): + path: str + data: bytes | None + acl: Sequence[ACL] + flags: int + + type: ClassVar[int] = 15 + + def serialize(self) -> bytearray: + b = bytearray() + b.extend(write_string(self.path)) + b.extend(write_buffer(self.data)) + b.extend(int_struct.pack(len(self.acl))) + for acl in self.acl: + b.extend( + int_struct.pack(acl.perms) + + write_string(acl.id.scheme) + + write_string(acl.id.id) + ) + b.extend(int_struct.pack(self.flags)) + return b + + @classmethod + def deserialize(cls, bytes: bytes, offset: int) -> tuple[str, ZnodeStat]: + path, offset = read_string(bytes, offset) + stat = ZnodeStat(*stat_struct.unpack_from(bytes, offset)) + return path, stat + + +Transaction_Types = Union[Create, Create2, Delete, SetData, CheckVersion] Transaction_Response = Union[str, bool, ZnodeStat, ZookeeperError, None] @@ -449,35 +477,6 @@ def unchroot( return resp -class Create2(namedtuple("Create2", "path data acl flags")): - path: str - data: bytes | None - acl: Sequence[ACL] - flags: int - - type: ClassVar[int] = 15 - - def serialize(self) -> bytearray: - b = bytearray() - b.extend(write_string(self.path)) - b.extend(write_buffer(self.data)) - b.extend(int_struct.pack(len(self.acl))) - for acl in self.acl: - b.extend( - int_struct.pack(acl.perms) - + write_string(acl.id.scheme) - + write_string(acl.id.id) - ) - b.extend(int_struct.pack(self.flags)) - return b - - @classmethod - def deserialize(cls, bytes: bytes, offset: int) -> tuple[str, ZnodeStat]: - path, offset = read_string(bytes, offset) - stat = ZnodeStat(*stat_struct.unpack_from(bytes, offset)) - return path, stat - - class Reconfig( namedtuple("Reconfig", "joining leaving new_members config_id") ): diff --git a/kazoo/recipe/cache.py b/kazoo/recipe/cache.py index 107d6849a..bfe259d45 100644 --- a/kazoo/recipe/cache.py +++ b/kazoo/recipe/cache.py @@ -187,9 +187,6 @@ def get_children( does not exist. :raises ValueError: If the path is outside of this subtree. :returns: The :class:`frozenset` which including children names. - - # FIXME the default return value should be an empty frozenset, - # returning None is confusing. """ node = self._find_node(path) return default if node is None else frozenset(node._children) diff --git a/kazoo/recipe/counter.py b/kazoo/recipe/counter.py index 77d68cb55..bde47e57b 100644 --- a/kazoo/recipe/counter.py +++ b/kazoo/recipe/counter.py @@ -107,25 +107,15 @@ def _ensure_node(self) -> None: def _value(self) -> tuple[Number, int]: self._ensure_node() - # FIXME: This is astonishingly hard to follow... - # Should probably be refactored to be more clear. - # val, state = ... - # if val == b"": - # old = self.default - # elif self.support_curator: - # old = struct.unpack(">i", val)[0] - # else: - # old = val.decode("ascii") - # maybe (not sure it does anything for the messy type though) - old: Union[bytes, str, Number] - old, stat = self.client.get(self.path) - if self.support_curator: - old = struct.unpack(">i", old)[0] if old != b"" else self.default + old: Union[str, Number] + val, stat = self.client.get(self.path) + if val == b"": + old = self.default + elif self.support_curator: + old = int(struct.unpack(">i", val)[0]) else: - old = old.decode("ascii") if old != b"" else self.default - version = stat.version - data = self.default_type(old) - return data, version + old = val.decode("ascii") + return self.default_type(old), stat.version @property def value(self) -> Number: diff --git a/kazoo/recipe/lock.py b/kazoo/recipe/lock.py index 1737802b7..8a44ff9d0 100644 --- a/kazoo/recipe/lock.py +++ b/kazoo/recipe/lock.py @@ -20,12 +20,7 @@ import re import time import uuid -from typing import ( - Iterable, - Literal, - Pattern, - TYPE_CHECKING, -) +from typing import Iterable, Literal, Pattern, TYPE_CHECKING from kazoo.exceptions import ( CancelledError, @@ -34,11 +29,7 @@ NoNodeError, ) from kazoo.protocol.states import KazooState, WatchedEvent -from kazoo.retry import ( - ForceRetryError, - KazooRetry, - RetryFailedError, -) +from kazoo.retry import ForceRetryError, KazooRetry, RetryFailedError if TYPE_CHECKING: from types import TracebackType @@ -48,26 +39,13 @@ class _Watch: def __init__(self, duration: float | None = None): self.duration = duration - self.started_at: float | None = None - - def start(self) -> None: self.started_at = time.monotonic() def leftover(self) -> float | None: if self.duration is None: return None - else: - # We should probably set started_at to either 0 or - # time.monotonic() in __init__ to avoid the type ignore - # here, but this is a private class and it's pretty clear - # that start() should be called before leftover() so I'm - # not sure it's worth it. - # FIXME raise an exception if start() hasn't been called yet - # i.e. self.started_at is None - elapsed = ( - time.monotonic() - self.started_at # type: ignore[operator] - ) - return max(0, self.duration - elapsed) + elapsed = time.monotonic() - self.started_at + return max(0, self.duration - elapsed) class Lock: @@ -242,10 +220,7 @@ def _watch_session(self, state: KazooState) -> bool: return True def _inner_acquire( - self, - blocking: bool, - timeout: float | None, - ephemeral: bool = True, + self, blocking: bool, timeout: float | None, ephemeral: bool = True ) -> bool: # wait until it's our chance to get it.. if self.is_acquired: @@ -607,9 +582,7 @@ def cancel(self) -> None: self.wake_event.set() def acquire( - self, - blocking: bool = True, - timeout: float | None = None, + self, blocking: bool = True, timeout: float | None = None ) -> bool: """Acquire the semaphore. By defaults blocks and waits forever. @@ -649,9 +622,7 @@ def acquire( return self.is_acquired def _inner_acquire( - self, - blocking: bool, - timeout: float | None = None, + self, blocking: bool, timeout: float | None = None ) -> bool: """Inner loop that runs from the top anytime a command hits a retryable Zookeeper exception.""" @@ -666,7 +637,6 @@ def _inner_acquire( return True w = _Watch(duration=timeout) - w.start() # FIXME This is passing bytes data, but self.client.Lock expects a str, # which I think is a bug in this code. However, I don't want to # change any code at this point, so we just ignore the type error here. diff --git a/kazoo/tests/test_async_result.py b/kazoo/tests/test_async_result.py new file mode 100644 index 000000000..d41b52b7c --- /dev/null +++ b/kazoo/tests/test_async_result.py @@ -0,0 +1,413 @@ +from __future__ import annotations + +import threading +import unittest + +# from typing import Any, Callable, Iterable +from unittest.mock import Mock + +import pytest + +from kazoo.handlers.utils import capture_exceptions, wrap, AsyncResult +from kazoo.interfaces import IHandler, SpawnedFunc + + +class Mock_Condition: + def __init__(self) -> None: + self._event = threading.Event() + + def __enter__(self) -> Mock_Condition: + return self + + def __exit__(self, *exc_info: object) -> None: + pass + + def wait(self, timeout: float | None = None) -> bool: + return self._event.wait(timeout) + + def notify_all(self) -> None: + self._event.set() + + +class Mock_Timeout(BaseException): + pass + + +class Mock_Handler(Mock): + _running: bool + + @property + def running(self) -> bool: + return self._running + + +class TestAsyncResult(unittest.TestCase): + # FIXME There seems no need to pass a timeout_factory to AsyncResult, + # because it comes from the IHandler, which is already passed to + # AsyncResult. + # The timeout_factory is only used in the get() method of AsyncResult, and + # it is not used anywhere else. So we can remove the timeout_factory from + # the AsyncResult constructor and just use the IHandler's timeout_factory + # in the get() method. + def _makeOne(self, handler: IHandler) -> AsyncResult: + def condition_factory() -> Mock_Condition: + return Mock_Condition() + + def timeout_factory() -> Mock_Timeout: + return Mock_Timeout() + + return AsyncResult(handler, condition_factory, timeout_factory) + + def _makeHandler(self) -> Mock_Handler: + handler = Mock_Handler(spec=IHandler) + handler._running = False + + def start() -> None: + handler._running = True + + def stop() -> None: + handler._running = False + + def add_to_completion_queue(callback: SpawnedFunc) -> None: + callback() + + handler.start.side_effect = start + handler.stop.side_effect = stop + handler.add_to_completion_queue.side_effect = add_to_completion_queue + return handler + + def test_ready(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + assert async_result.ready() is False + async_result.set(None) + assert async_result.ready() is True + assert async_result.successful() is True + assert async_result.exception is None + assert async_result.get() is None + + def test_ready_value(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + assert async_result.ready() is False + async_result.set("value") + assert async_result.ready() is True + assert async_result.successful() is True + assert async_result.exception is None + assert async_result.get() == "value" + + def test_callback_queued(self) -> None: + mock_handler = self._makeHandler() + mock_handler.start() + async_result = self._makeOne(mock_handler) + + def add_on(async_result: AsyncResult) -> None: + pass + + async_result.rawlink(add_on) + async_result.set("val") + + mock_handler.add_to_completion_queue.assert_called_once() + + def test_set_exception(self) -> None: + mock_handler = self._makeHandler() + mock_handler.start() + async_result = self._makeOne(mock_handler) + + def add_on(async_result: AsyncResult) -> None: + pass + + async_result.rawlink(add_on) + async_result.set_exception(ImportError("Error occurred")) + + assert async_result.ready() is True + assert async_result.successful() is False + assert isinstance(async_result.exception, ImportError) + mock_handler.add_to_completion_queue.assert_called_once() + + def test_get_wait_while_setting(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + lst = [] + bv = threading.Event() + cv = threading.Event() + + def wait_for_val() -> None: + bv.set() + val = async_result.get() + lst.append(val) + cv.set() + + th = threading.Thread(target=wait_for_val) + th.start() + bv.wait() + + async_result.set("fred") + cv.wait() + assert lst == ["fred"] + th.join() + + def test_get_with_nowait(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + timeout = Mock_Timeout + + with pytest.raises(timeout): + async_result.get(block=False) + + with pytest.raises(timeout): + async_result.get_nowait() + + def test_get_with_exception(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + lst = [] + bv = threading.Event() + cv = threading.Event() + + def wait_for_val() -> None: + bv.set() + try: + val = async_result.get() + except ImportError: + lst.append("oops") + else: + lst.append(val) + cv.set() + + th = threading.Thread(target=wait_for_val) + th.start() + bv.wait() + + async_result.set_exception(ImportError()) + cv.wait() + assert lst == ["oops"] + th.join() + + def test_wait(self) -> None: + mock_handler = Mock() + async_result = self._makeOne(mock_handler) + + lst: list[bool | str] = [] + bv = threading.Event() + cv = threading.Event() + + def wait_for_val() -> None: + bv.set() + try: + val = async_result.wait(10) + except ImportError: + lst.append("oops") + else: + lst.append(val) + cv.set() + + th = threading.Thread(target=wait_for_val) + th.start() + bv.wait(10) + + async_result.set("fred") + cv.wait(15) + assert lst == [True] + th.join() + + def test_wait_race(self) -> None: + """Test that there is no race condition in `IAsyncResult.wait()`. + + Guards against the reappearance of: + https://github.com/python-zk/kazoo/issues/485 + """ + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + async_result.set("immediate") + + cv = threading.Event() + + def wait_for_val() -> None: + # NB: should not sleep + async_result.wait(20) + cv.set() + + th = threading.Thread(target=wait_for_val) + th.daemon = True + th.start() + + # if the wait() didn't sleep (correctly), cv will be set quickly + # if it did sleep, the cv will not be set yet and this will timeout + cv.wait(10) + assert cv.is_set() is True + th.join() + + def test_set_before_wait(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + lst = [] + cv = threading.Event() + async_result.set("fred") + + def wait_for_val() -> None: + val = async_result.get() + lst.append(val) + cv.set() + + th = threading.Thread(target=wait_for_val) + th.start() + cv.wait() + assert lst == ["fred"] + th.join() + + def test_set_exc_before_wait(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + lst = [] + cv = threading.Event() + async_result.set_exception(ImportError()) + + def wait_for_val() -> None: + try: + val = async_result.get() + except ImportError: + lst.append("ooops") + else: + lst.append(val) + cv.set() + + th = threading.Thread(target=wait_for_val) + th.start() + cv.wait() + assert lst == ["ooops"] + th.join() + + def test_linkage(self) -> None: + mock_handler = self._makeHandler() + mock_handler.start() + async_result = self._makeOne(mock_handler) + cv = threading.Event() + + lst = [] + + def add_on(async_result: AsyncResult) -> None: + lst.append(True) + + def wait_for_val() -> None: + async_result.get() + cv.set() + + th = threading.Thread(target=wait_for_val) + th.start() + + async_result.rawlink(add_on) + async_result.set(b"fred") + mock_handler.add_to_completion_queue.assert_called_once() + async_result.unlink(add_on) + cv.wait() + assert async_result.value == b"fred" + th.join() + + def test_linkage_not_ready(self) -> None: + mock_handler = self._makeHandler() + mock_handler.start() + async_result = self._makeOne(mock_handler) + + lst = [] + + def add_on(async_result: AsyncResult) -> None: + lst.append(True) + + async_result.set("fred") + mock_handler.add_to_completion_queue.assert_not_called() + async_result.rawlink(add_on) + mock_handler.add_to_completion_queue.assert_called_once() + + def test_link_and_unlink(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + lst = [] + + def add_on(async_result: AsyncResult) -> None: + lst.append(True) + + async_result.rawlink(add_on) + mock_handler.add_to_completion_queue.assert_not_called() + async_result.unlink(add_on) + async_result.set("fred") + mock_handler.add_to_completion_queue.assert_not_called() + + def test_captured_exception(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + @capture_exceptions(async_result) + def exceptional_function() -> float: + return 1 / 0 + + exceptional_function() + + with pytest.raises(ZeroDivisionError): + async_result.get() + + def test_no_capture_exceptions(self) -> None: + mock_handler = self._makeHandler() + async_result = self._makeOne(mock_handler) + + lst = [] + + def add_on(async_result: AsyncResult) -> None: + lst.append(True) + + async_result.rawlink(add_on) + + @capture_exceptions(async_result) + def regular_function() -> bool: + return True + + regular_function() + + mock_handler.add_to_completion_queue.assert_not_called() + + def test_wraps(self) -> None: + mock_handler = Mock() + async_result = self._makeOne(mock_handler) + + lst = [] + + def add_on(result: AsyncResult) -> None: + lst.append(result.get()) + + async_result.rawlink(add_on) + + @wrap(async_result) + def regular_function() -> str: + return "hello" + + assert regular_function() == "hello" + mock_handler.add_to_completion_queue.assert_called_once() + assert async_result.get() == "hello" + + def test_multiple_callbacks(self) -> None: + mockback1 = Mock(name="mockback1") + mockback2 = Mock(name="mockback2") + handler = self._makeHandler() + handler.start() + + async_result = self._makeOne(handler) + async_result.rawlink(mockback1) + async_result.rawlink(mockback2) + print("set") + async_result.set("howdy") + print("wait") + async_result.wait() + print("stop") + handler.stop() + print("done") + + mockback2.assert_called_once_with(async_result) + mockback1.assert_called_once_with(async_result) diff --git a/kazoo/tests/test_cache.py b/kazoo/tests/test_cache.py index 15a672018..1007d1782 100644 --- a/kazoo/tests/test_cache.py +++ b/kazoo/tests/test_cache.py @@ -4,7 +4,8 @@ import importlib import sys import uuid -from typing import Any, TYPE_CHECKING +from contextlib import contextmanager +from typing import Any, Iterator, TYPE_CHECKING from unittest.mock import patch, call, Mock import pytest @@ -53,7 +54,7 @@ def choose_an_installed_handler( try: mod = importlib.import_module(handler_module) cls = getattr(mod, handler_class) - except ImportError: + except (ImportError, NameError): continue else: return cls() # type: ignore[no-any-return] @@ -66,7 +67,6 @@ def setUp(self) -> None: self._event_queue: Queue[TreeEvent] = self.client.handler.queue_impl() self._error_queue = self.client.handler.queue_impl() self._path: str | None = None - self._cache: TreeCache | None = None def tearDown(self) -> None: if not self._error_queue.empty(): @@ -74,29 +74,22 @@ def tearDown(self) -> None: raise self._error_queue.get() except FakeException: pass - if self._cache is not None: - self._cache.close() - self._cache = None super().tearDown() - def make_cache(self) -> TreeCache: - if self._cache is None: - self._path = "/" + uuid.uuid4().hex - self._cache = TreeCache(self.client, self.path) - self._cache.listen(lambda event: self._event_queue.put(event)) - self._cache.listen_fault( - lambda error: self._error_queue.put(error) - ) - self._cache.start() - return self._cache + @contextmanager + def make_cache(self) -> Iterator[TreeCache]: + self._path = "/" + uuid.uuid4().hex + assert self.count_tree_node() == 0 + cache = TreeCache(self.client, self.path) + assert self.count_tree_node() == 1 + cache.listen(lambda event: self._event_queue.put(event)) + cache.listen_fault(lambda error: self._error_queue.put(error)) + cache.start() - # FIXME This is entirely for the purpose of minimising code changes. - # Calling make_cache twice should be an error and the return value - # should be used, not stored. - @property - def cache(self) -> TreeCache: - assert self._cache is not None - return self._cache + try: + yield cache + finally: + cache.close() @property def path(self) -> str: @@ -129,7 +122,7 @@ def _wait_gc(self) -> None: # trigger switching on some coroutine handlers self.client.handler.sleep_func(0.1) - completion_queue = getattr(self.handler, "completion_queue", None) + completion_queue = getattr(self.handler, "_completion_queue", None) if completion_queue is not None: while not completion_queue.empty(): self.client.handler.sleep_func(0.1) @@ -149,317 +142,337 @@ def count_tree_node(self) -> int: raise RuntimeError("could not count refs exactly") def test_start(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) - stat = self.client.exists(self.path) - assert stat is not None - assert stat.version == 0 + stat = self.client.exists(self.path) + assert stat is not None + assert stat.version == 0 - assert self.cache._state == TreeCache.STATE_STARTED - assert self.cache._root._state == TreeNode.STATE_LIVE + assert cache._state == TreeCache.STATE_STARTED + assert cache._root._state == TreeNode.STATE_LIVE def test_start_started(self) -> None: - self.make_cache() - with pytest.raises(KazooException): - self.cache.start() + with self.make_cache() as cache: + with pytest.raises(KazooException): + cache.start() def test_start_closed(self) -> None: - self.make_cache() - self.cache.close() - with pytest.raises(KazooException): - self.cache.start() + with self.make_cache() as cache: + cache.close() + with pytest.raises(KazooException): + cache.start() def test_close(self) -> None: - assert self.count_tree_node() == 0 + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) + self.client.create(self.path + "/foo/bar/baz", makepath=True) + for _ in range(3): + self.wait_cache(TreeEvent.NODE_ADDED) + + # setup stub watchers which are outside of tree cache + stub_data_watcher = Mock(spec=lambda event: None) + stub_child_watcher = Mock(spec=lambda event: None) + self.client.get(self.path + "/foo", stub_data_watcher) + self.client.get_children(self.path + "/foo", stub_child_watcher) + + # watchers inside tree cache should be here + root_path = self.client.chroot + self.path + assert len(self.client._data_watchers[root_path + "/foo"]) == 2 + assert len(self.client._data_watchers[root_path + "/foo/bar"]) == 1 + assert ( + len(self.client._data_watchers[root_path + "/foo/bar/baz"]) + == 1 + ) + assert len(self.client._child_watchers[root_path + "/foo"]) == 2 + assert ( + len(self.client._child_watchers[root_path + "/foo/bar"]) == 1 + ) + assert ( + len(self.client._child_watchers[root_path + "/foo/bar/baz"]) + == 1 + ) - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - self.client.create(self.path + "/foo/bar/baz", makepath=True) - for _ in range(3): - self.wait_cache(TreeEvent.NODE_ADDED) + cache.close() + + # nothing should be published since tree closed + assert self._event_queue.empty() + + # tree should be empty + assert cache._root._children == {} + assert cache._root._data is None + assert cache._state == TreeCache.STATE_CLOSED + + # node state should not be changed + assert cache._root._state != TreeNode.STATE_DEAD + + # watchers should be reset + assert len(self.client._data_watchers[root_path + "/foo"]) == 1 + assert len(self.client._data_watchers[root_path + "/foo/bar"]) == 0 + assert ( + len(self.client._data_watchers[root_path + "/foo/bar/baz"]) + == 0 + ) + assert len(self.client._child_watchers[root_path + "/foo"]) == 1 + assert ( + len(self.client._child_watchers[root_path + "/foo/bar"]) == 0 + ) + assert ( + len(self.client._child_watchers[root_path + "/foo/bar/baz"]) + == 0 + ) - # setup stub watchers which are outside of tree cache - stub_data_watcher = Mock(spec=lambda event: None) - stub_child_watcher = Mock(spec=lambda event: None) - self.client.get(self.path + "/foo", stub_data_watcher) - self.client.get_children(self.path + "/foo", stub_child_watcher) - - # watchers inside tree cache should be here - root_path = self.client.chroot + self.path - assert len(self.client._data_watchers[root_path + "/foo"]) == 2 - assert len(self.client._data_watchers[root_path + "/foo/bar"]) == 1 - assert len(self.client._data_watchers[root_path + "/foo/bar/baz"]) == 1 - assert len(self.client._child_watchers[root_path + "/foo"]) == 2 - assert len(self.client._child_watchers[root_path + "/foo/bar"]) == 1 - assert ( - len(self.client._child_watchers[root_path + "/foo/bar/baz"]) == 1 - ) - - self.cache.close() - - # nothing should be published since tree closed - assert self._event_queue.empty() - - # tree should be empty - assert self.cache._root._children == {} - assert self.cache._root._data is None - assert self.cache._state == TreeCache.STATE_CLOSED - - # node state should not be changed - assert self.cache._root._state != TreeNode.STATE_DEAD - - # watchers should be reset - assert len(self.client._data_watchers[root_path + "/foo"]) == 1 - assert len(self.client._data_watchers[root_path + "/foo/bar"]) == 0 - assert len(self.client._data_watchers[root_path + "/foo/bar/baz"]) == 0 - assert len(self.client._child_watchers[root_path + "/foo"]) == 1 - assert len(self.client._child_watchers[root_path + "/foo/bar"]) == 0 - assert ( - len(self.client._child_watchers[root_path + "/foo/bar/baz"]) == 0 - ) - - # outside watchers should not be deleted - assert ( - list(self.client._data_watchers[root_path + "/foo"])[0] - == stub_data_watcher - ) - assert ( - list(self.client._child_watchers[root_path + "/foo"])[0] - == stub_child_watcher - ) - - # FIXME This looks pointless at best. - self._cache = None + # outside watchers should not be deleted + assert ( + list(self.client._data_watchers[root_path + "/foo"])[0] + == stub_data_watcher + ) + assert ( + list(self.client._child_watchers[root_path + "/foo"])[0] + == stub_child_watcher + ) # should not be any leaked memory (tree node) here + cache = None # type: ignore assert self.count_tree_node() == 0 def test_delete_operation(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - - assert self.count_tree_node() == 1 + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) - self.client.create(self.path + "/foo/bar/baz", makepath=True) - for _ in range(3): - self.wait_cache(TreeEvent.NODE_ADDED) + self.client.create(self.path + "/foo/bar/baz", makepath=True) + for _ in range(3): + self.wait_cache(TreeEvent.NODE_ADDED) - self.client.delete(self.path + "/foo", recursive=True) - for _ in range(3): - self.wait_cache(TreeEvent.NODE_REMOVED) + self.client.delete(self.path + "/foo", recursive=True) + for _ in range(3): + self.wait_cache(TreeEvent.NODE_REMOVED) - # tree should be empty - assert self.cache._root._children == {} + # tree should be empty + assert cache._root._children == {} - # watchers should be reset - root_path = self.client.chroot + self.path - assert self.client._data_watchers[root_path + "/foo"] == set() - assert self.client._data_watchers[root_path + "/foo/bar"] == set() - assert self.client._data_watchers[root_path + "/foo/bar/baz"] == set() - assert self.client._child_watchers[root_path + "/foo"] == set() - assert self.client._child_watchers[root_path + "/foo/bar"] == set() - assert self.client._child_watchers[root_path + "/foo/bar/baz"] == set() + # watchers should be reset + root_path = self.client.chroot + self.path + assert self.client._data_watchers[root_path + "/foo"] == set() + assert self.client._data_watchers[root_path + "/foo/bar"] == set() + assert ( + self.client._data_watchers[root_path + "/foo/bar/baz"] == set() + ) + assert self.client._child_watchers[root_path + "/foo"] == set() + assert self.client._child_watchers[root_path + "/foo/bar"] == set() + assert ( + self.client._child_watchers[root_path + "/foo/bar/baz"] + == set() + ) - # should not be any leaked memory (tree node) here - assert self.count_tree_node() == 1 + # This should be the only tree left + assert self.count_tree_node() == 1 def test_children_operation(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - - self.client.create(self.path + "/test_children", b"test_children_1") - event = self.wait_cache(TreeEvent.NODE_ADDED) - assert event is not None - assert event.event_type == TreeEvent.NODE_ADDED - assert event.event_data.path == self.path + "/test_children" - assert event.event_data.data == b"test_children_1" - assert event.event_data.stat.version == 0 - - self.client.set(self.path + "/test_children", b"test_children_2") - event = self.wait_cache(TreeEvent.NODE_UPDATED) - assert event is not None - assert event.event_type == TreeEvent.NODE_UPDATED - assert event.event_data.path == self.path + "/test_children" - assert event.event_data.data == b"test_children_2" - assert event.event_data.stat.version == 1 - - self.client.delete(self.path + "/test_children") - event = self.wait_cache(TreeEvent.NODE_REMOVED) - assert event is not None - assert event.event_type == TreeEvent.NODE_REMOVED - assert event.event_data.path == self.path + "/test_children" - assert event.event_data.data == b"test_children_2" - assert event.event_data.stat.version == 1 - - def test_subtree_operation(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) + with self.make_cache(): + self.wait_cache(since=TreeEvent.INITIALIZED) - self.client.create(self.path + "/foo/bar/baz", makepath=True) - for relative_path in ("/foo", "/foo/bar", "/foo/bar/baz"): + self.client.create( + self.path + "/test_children", b"test_children_1" + ) event = self.wait_cache(TreeEvent.NODE_ADDED) assert event is not None assert event.event_type == TreeEvent.NODE_ADDED - assert event.event_data.path == self.path + relative_path - assert event.event_data.data == b"" + assert event.event_data.path == self.path + "/test_children" + assert event.event_data.data == b"test_children_1" assert event.event_data.stat.version == 0 - self.client.delete(self.path + "/foo", recursive=True) - for relative_path in ("/foo/bar/baz", "/foo/bar", "/foo"): + self.client.set(self.path + "/test_children", b"test_children_2") + event = self.wait_cache(TreeEvent.NODE_UPDATED) + assert event is not None + assert event.event_type == TreeEvent.NODE_UPDATED + assert event.event_data.path == self.path + "/test_children" + assert event.event_data.data == b"test_children_2" + assert event.event_data.stat.version == 1 + + self.client.delete(self.path + "/test_children") event = self.wait_cache(TreeEvent.NODE_REMOVED) assert event is not None assert event.event_type == TreeEvent.NODE_REMOVED - assert event.event_data.path == self.path + relative_path + assert event.event_data.path == self.path + "/test_children" + assert event.event_data.data == b"test_children_2" + assert event.event_data.stat.version == 1 + + def test_subtree_operation(self) -> None: + with self.make_cache(): + self.wait_cache(since=TreeEvent.INITIALIZED) + + self.client.create(self.path + "/foo/bar/baz", makepath=True) + for relative_path in ("/foo", "/foo/bar", "/foo/bar/baz"): + event = self.wait_cache(TreeEvent.NODE_ADDED) + assert event is not None + assert event.event_type == TreeEvent.NODE_ADDED + assert event.event_data.path == self.path + relative_path + assert event.event_data.data == b"" + assert event.event_data.stat.version == 0 + + self.client.delete(self.path + "/foo", recursive=True) + for relative_path in ("/foo/bar/baz", "/foo/bar", "/foo"): + event = self.wait_cache(TreeEvent.NODE_REMOVED) + assert event is not None + assert event.event_type == TreeEvent.NODE_REMOVED + assert event.event_data.path == self.path + relative_path def test_get_data(self) -> None: - cache = self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - self.client.create(self.path + "/foo/bar/baz", b"@", makepath=True) - self.wait_cache(TreeEvent.NODE_ADDED) - self.wait_cache(TreeEvent.NODE_ADDED) - self.wait_cache(TreeEvent.NODE_ADDED) - - with patch.object(cache, "_client"): # disable any remote operation - node = cache.get_data(self.path) - assert node is not None - assert node.data == b"" - assert node.stat.version == 0 - - node = cache.get_data(self.path + "foo") - assert node is not None - assert node.data == b"" - assert node.stat.version == 0 - - node = cache.get_data(self.path + "foo/bar") - assert node is not None - assert node.data == b"" - assert node.stat.version == 0 - - node = cache.get_data(self.path + "foo/bar/baz") - assert node is not None - assert node.data == b"@" - assert node.stat.version == 0 + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) + self.client.create(self.path + "/foo/bar/baz", b"@", makepath=True) + self.wait_cache(TreeEvent.NODE_ADDED) + self.wait_cache(TreeEvent.NODE_ADDED) + self.wait_cache(TreeEvent.NODE_ADDED) + + # disable any remote operations + with patch.object(cache, "_client"): + node = cache.get_data(self.path) + assert node is not None + assert node.data == b"" + assert node.stat.version == 0 + + node = cache.get_data(self.path + "foo") + assert node is not None + assert node.data == b"" + assert node.stat.version == 0 + + node = cache.get_data(self.path + "foo/bar") + assert node is not None + assert node.data == b"" + assert node.stat.version == 0 + + node = cache.get_data(self.path + "foo/bar/baz") + assert node is not None + assert node.data == b"@" + assert node.stat.version == 0 def test_get_children(self) -> None: - cache = self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - self.client.create(self.path + "/foo/bar/baz", b"@", makepath=True) - self.wait_cache(TreeEvent.NODE_ADDED) - self.wait_cache(TreeEvent.NODE_ADDED) - self.wait_cache(TreeEvent.NODE_ADDED) - - with patch.object(cache, "_client"): # disable any remote operation - assert ( - cache.get_children(self.path + "/foo/bar/baz") == frozenset() - ) - assert cache.get_children(self.path + "/foo/bar") == frozenset( - ["baz"] - ) - assert cache.get_children(self.path + "/foo") == frozenset(["bar"]) - assert cache.get_children(self.path) == frozenset(["foo"]) + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) + self.client.create(self.path + "/foo/bar/baz", b"@", makepath=True) + self.wait_cache(TreeEvent.NODE_ADDED) + self.wait_cache(TreeEvent.NODE_ADDED) + self.wait_cache(TreeEvent.NODE_ADDED) + + # Disable any remote operations + with patch.object(cache, "_client"): + assert ( + cache.get_children(self.path + "/foo/bar/baz") + == frozenset() + ) + assert cache.get_children(self.path + "/foo/bar") == frozenset( + ["baz"] + ) + assert cache.get_children(self.path + "/foo") == frozenset( + ["bar"] + ) + assert cache.get_children(self.path) == frozenset(["foo"]) def test_get_data_out_of_tree(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - with pytest.raises(ValueError): - self.cache.get_data("/out_of_tree") + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) + with pytest.raises(ValueError): + cache.get_data("/out_of_tree") def test_get_children_out_of_tree(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - with pytest.raises(ValueError): - self.cache.get_children("/out_of_tree") + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) + with pytest.raises(ValueError): + cache.get_children("/out_of_tree") def test_get_data_no_node(self) -> None: - cache = self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) - with patch.object(cache, "_client"): # disable any remote operation - assert cache.get_data(self.path + "/non_exists") is None + with patch.object(cache, "_client"): + assert cache.get_data(self.path + "/non_exists") is None def test_get_children_no_node(self) -> None: - cache = self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) - with patch.object(cache, "_client"): # disable any remote operation - assert cache.get_children(self.path + "/non_exists") is None + with patch.object(cache, "_client"): + assert cache.get_children(self.path + "/non_exists") is None def test_session_reconnected(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - - self.client.create(self.path + "/foo") - event = self.wait_cache(TreeEvent.NODE_ADDED) - assert event is not None - assert event.event_data.path == self.path + "/foo" - - with self.spy_client("get_async") as get_data: - with self.spy_client("get_children_async") as get_children: - # session suspended - self.lose_connection(self.client.handler.event_object) - self.wait_cache(TreeEvent.CONNECTION_SUSPENDED) - - # There are a serial refreshing operation here. But NODE_ADDED - # events will not be raised because the zxid of nodes are the - # same during reconnecting. - - # connection restore - self.wait_cache(TreeEvent.CONNECTION_RECONNECTED) - - # wait for outstanding operations - while self.cache._outstanding_ops > 0: - self.client.handler.sleep_func(0.1) - - # inspect in-memory nodes - _node_root = self.cache._root - _node_foo = self.cache._root._children["foo"] - - # make sure that all nodes are refreshed - get_data.assert_has_calls( - [ - call(self.path, watch=_node_root._process_watch), - call( - self.path + "/foo", watch=_node_foo._process_watch - ), - ], - any_order=True, - ) - get_children.assert_has_calls( - [ - call(self.path, watch=_node_root._process_watch), - call( - self.path + "/foo", watch=_node_foo._process_watch - ), - ], - any_order=True, - ) + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) + + self.client.create(self.path + "/foo") + event = self.wait_cache(TreeEvent.NODE_ADDED) + assert event is not None + assert event.event_data.path == self.path + "/foo" + + with self.spy_client("get_async") as get_data: + with self.spy_client("get_children_async") as get_children: + # session suspended + self.lose_connection(self.client.handler.event_object) + self.wait_cache(TreeEvent.CONNECTION_SUSPENDED) + + # There are a serial refreshing operation here. But + # NODE_ADDED events will not be raised because the zxid of + # nodes are the same during reconnecting. + + # connection restore + self.wait_cache(TreeEvent.CONNECTION_RECONNECTED) + + # wait for outstanding operations + while cache._outstanding_ops > 0: + self.client.handler.sleep_func(0.1) + + # inspect in-memory nodes + _node_root = cache._root + _node_foo = cache._root._children["foo"] + + # make sure that all nodes are refreshed + get_data.assert_has_calls( + [ + call(self.path, watch=_node_root._process_watch), + call( + self.path + "/foo", + watch=_node_foo._process_watch, + ), + ], + any_order=True, + ) + get_children.assert_has_calls( + [ + call(self.path, watch=_node_root._process_watch), + call( + self.path + "/foo", + watch=_node_foo._process_watch, + ), + ], + any_order=True, + ) def test_root_recreated(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) - - # remove root node - self.client.delete(self.path) - event = self.wait_cache(TreeEvent.NODE_REMOVED) - assert event is not None - assert event.event_type == TreeEvent.NODE_REMOVED - assert event.event_data.data == b"" - assert event.event_data.path == self.path - assert event.event_data.stat.version == 0 - - # re-create root node - self.client.ensure_path(self.path) - event = self.wait_cache(TreeEvent.NODE_ADDED) - assert event is not None - assert event.event_type == TreeEvent.NODE_ADDED - assert event.event_data.data == b"" - assert event.event_data.path == self.path - assert event.event_data.stat.version == 0 - - assert self.cache._outstanding_ops >= 0, ( - "unexpected outstanding ops %r" % self.cache._outstanding_ops - ) + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) + + # remove root node + self.client.delete(self.path) + event = self.wait_cache(TreeEvent.NODE_REMOVED) + assert event is not None + assert event.event_type == TreeEvent.NODE_REMOVED + assert event.event_data.data == b"" + assert event.event_data.path == self.path + assert event.event_data.stat.version == 0 + + # re-create root node + self.client.ensure_path(self.path) + event = self.wait_cache(TreeEvent.NODE_ADDED) + assert event is not None + assert event.event_type == TreeEvent.NODE_ADDED + assert event.event_data.data == b"" + assert event.event_data.path == self.path + assert event.event_data.stat.version == 0 + + assert cache._outstanding_ops >= 0, ( + "unexpected outstanding ops %r" % cache._outstanding_ops + ) def test_exception_handler(self) -> None: error_value = FakeException() @@ -468,30 +481,30 @@ def test_exception_handler(self) -> None: with patch.object(TreeNode, "on_deleted") as on_deleted: on_deleted.side_effect = [error_value] - self.make_cache() - self.cache.listen_fault(error_handler) + with self.make_cache() as cache: + cache.listen_fault(error_handler) - self.cache.close() - error_handler.assert_called_once_with(error_value) + cache.close() + error_handler.assert_called_once_with(error_value) def test_exception_suppressed(self) -> None: - self.make_cache() - self.wait_cache(since=TreeEvent.INITIALIZED) + with self.make_cache() as cache: + self.wait_cache(since=TreeEvent.INITIALIZED) - # stoke up ConnectionClosedError - self.client.stop() - self.client.close() - self.client.handler.start() # keep the async completion - self.wait_cache(since=TreeEvent.CONNECTION_LOST) + # stoke up ConnectionClosedError + self.client.stop() + self.client.close() + self.client.handler.start() # keep the async completion + self.wait_cache(since=TreeEvent.CONNECTION_LOST) - with patch.object(TreeNode, "on_created") as on_created: - self.cache._root._call_client("exists", "/") - self.cache._root._call_client("get", "/") - self.cache._root._call_client("get_children", "/") + with patch.object(TreeNode, "on_created") as on_created: + cache._root._call_client("exists", "/") + cache._root._call_client("get", "/") + cache._root._call_client("get_children", "/") - self.wait_cache(since=TreeEvent.INITIALIZED) - on_created.assert_not_called() - assert self.cache._outstanding_ops == 0 + self.wait_cache(since=TreeEvent.INITIALIZED) + on_created.assert_not_called() + assert cache._outstanding_ops == 0 class FakeException(Exception): diff --git a/kazoo/tests/test_client.py b/kazoo/tests/test_client.py index c0263c1ae..527e052bd 100644 --- a/kazoo/tests/test_client.py +++ b/kazoo/tests/test_client.py @@ -4,7 +4,6 @@ import socket import tempfile import threading -import time import uuid import unittest @@ -42,7 +41,6 @@ make_digest_acl_credential, CREATOR_ALL_ACL, make_digest_acl, - ACL, OPEN_ACL_UNSAFE, ) @@ -85,12 +83,16 @@ def listener(state: KazooState) -> None: class TestClientConstructor(unittest.TestCase): - def _makeOne(self, *args: Any, **kw: Any) -> KazooClient: + _makeOne = KazooClient + + def _old_makeOne(self, *args: Any, **kw: Any) -> KazooClient: + # This is a hack to so that test_invalid_handler doesn't generate a + # mypy error. return KazooClient(*args, **kw) def test_invalid_handler(self) -> None: with pytest.raises(ConfigurationError): - self._makeOne(handler=SequentialThreadingHandler) + self._old_makeOne(handler=SequentialThreadingHandler) def test_chroot(self) -> None: assert self._makeOne(hosts="127.0.0.1:2181/").chroot == "" @@ -127,6 +129,7 @@ def test_another_invalid_hostname(self) -> None: self._makeOne(hosts="/nosuchhost/a") def test_retry_options_dict(self) -> None: + # Deprecated API, but still supported for backwards compatibility client = self._makeOne( command_retry=dict(max_tries=99), connection_retry=dict(delay=99) ) @@ -135,17 +138,46 @@ def test_retry_options_dict(self) -> None: assert client._retry.max_tries == 99 assert client._conn_retry.delay == 99 + def test_retry_options_one_retry(self) -> None: + # Deprecated API, but still supported for backwards compatibility + client = self._makeOne( + command_retry=KazooRetry(max_tries=98), + connection_retry=dict(delay=97), + ) + assert type(client._conn_retry) is KazooRetry + assert type(client._retry) is KazooRetry + assert client._retry.max_tries == 98 + assert client._conn_retry.delay == 97 -class TestAuthentication(KazooTestCase): - def _makeAuth(self, *args: Any, **kwargs: Any) -> ACL: - return make_digest_acl(*args, **kwargs) + def test_retry_options_other_retry(self) -> None: + # Deprecated API, but still supported for backwards compatibility + client = self._makeOne( + command_retry=dict(max_tries=88), + connection_retry=KazooRetry(delay=87), + ) + assert type(client._conn_retry) is KazooRetry + assert type(client._retry) is KazooRetry + assert client._retry.max_tries == 88 + assert client._conn_retry.delay == 87 + + def test_retry_options_both_retry(self) -> None: + client = self._makeOne( + command_retry=KazooRetry(max_tries=96), + connection_retry=KazooRetry(delay=95), + ) + assert type(client._conn_retry) is KazooRetry + assert type(client._retry) is KazooRetry + assert client._retry.max_tries == 96 + assert client._conn_retry.delay == 95 + +class TestAuthentication(KazooTestCase): def test_auth(self) -> None: username = uuid.uuid4().hex password = uuid.uuid4().hex digest_auth = "%s:%s" % (username, password) - acl = self._makeAuth(username, password, all=True) + acl = make_digest_acl(username, password, all=True) client = self._get_client() client.start() @@ -177,12 +209,11 @@ def test_auth(self) -> None: eve.close() def test_connect_auth(self) -> None: - username = uuid.uuid4().hex password = uuid.uuid4().hex digest_auth = "%s:%s" % (username, password) - acl = self._makeAuth(username, password, all=True) + acl = make_digest_acl(username, password, all=True) client = self._get_client(auth_data=[("digest", digest_auth)]) client.start() @@ -206,7 +237,7 @@ def test_unicode_auth(self) -> None: username = r"xe4/\hm" password = r"/\xe4hm" digest_auth = "%s:%s" % (username, password) - acl = self._makeAuth(username, password, all=True) + acl = make_digest_acl(username, password, all=True) client = self._get_client() client.start() @@ -236,6 +267,8 @@ def test_unicode_auth(self) -> None: eve.close() def test_invalid_auth(self) -> None: + # Fixes deprecated warning for add_auth() with a tuple instead of a + # string client = self._get_client() client.start() @@ -340,7 +373,6 @@ def watch_events(event: KazooState) -> None: assert not cv.is_set() def test_state_listener(self) -> None: - states = [] condition = self.make_condition() @@ -366,7 +398,6 @@ def test_invalid_listener(self) -> None: self.client.add_listener(15) # type: ignore[arg-type] def test_listener_only_called_on_real_state_change(self) -> None: - assert self.client.state == KazooState.CONNECTED called = [False] condition = self.make_event() @@ -479,8 +510,7 @@ def test_watch(event: WatchedEvent) -> None: class TestClient(KazooTestCase): - def _makeOne(self, *args: Any) -> SequentialThreadingHandler: - return SequentialThreadingHandler(*args) + _makeOne = SequentialThreadingHandler def test_server_version_retries_fail(self) -> None: @@ -1250,10 +1280,7 @@ def test_basic_delete(self) -> None: assert results[0] is True def test_bad_deletes(self) -> None: - args_list = [ - (True,), - ("/smith", "woops"), - ] + args_list = [(True,), ("/smith", "woops")] for args in args_list: with pytest.raises(TypeError): diff --git a/kazoo/tests/test_connection.py b/kazoo/tests/test_connection.py index 7f9021fcc..1f5d6c8be 100644 --- a/kazoo/tests/test_connection.py +++ b/kazoo/tests/test_connection.py @@ -14,11 +14,7 @@ import pytest from kazoo.exceptions import ConnectionLoss, NotReadOnlyCallError -from kazoo.protocol.serialization import ( - Connect, - int_struct, - write_string, -) +from kazoo.protocol.serialization import Connect, int_struct, write_string from kazoo.protocol.states import KazooState, KeeperState from kazoo.protocol.connection import _CONNECTION_DROP from kazoo.testing import KazooTestCase @@ -400,11 +396,14 @@ def log_exception(*args: Any) -> None: args, exc_info = error_stack[-1] assert args == ("Unhandled exception in connection loop",) - assert exc_info[0] == RuntimeError + assert exc_info[0] is RuntimeError self.client.handler.sleep_func(0.2) assert self.connection_routine is not None - assert not self.connection_routine.is_alive() + # This vvvv doesn't work for greenlet handlers as the greenlet doesn't + # implement is_alive... + if (alive := getattr(self.connection_routine, "is_alive")) is not None: + assert not alive() assert len(self._zk_loop_errors) == 1 assert self._zk_loop_errors[0] == exc_info[1] diff --git a/kazoo/tests/test_eventlet_handler.py b/kazoo/tests/test_eventlet_handler.py index e410496a3..c0d980a12 100644 --- a/kazoo/tests/test_eventlet_handler.py +++ b/kazoo/tests/test_eventlet_handler.py @@ -9,6 +9,7 @@ from kazoo.handlers.utils import create_tcp_socket from kazoo.handlers import utils +from kazoo.interfaces import IAsyncResult from kazoo.protocol import states as kazoo_states if TYPE_CHECKING: @@ -65,10 +66,10 @@ def cb() -> None: assert len(captures) == 1 def test_async_link(self) -> None: - captures: list[SequentialEventletHandler] = [] + captures: list[IAsyncResult] = [] - def cb(handler: SequentialEventletHandler) -> None: - captures.append(handler) + def cb(async_result: IAsyncResult) -> None: + captures.append(async_result) with start_stop_one() as handler: r = handler.async_result() @@ -85,14 +86,14 @@ def test_timeout_raising(self) -> None: raise handler.timeout_exception("This is a timeout") def test_async_ok(self) -> None: - captures: list[Literal[1] | SequentialEventletHandler] = [] + captures: list[Literal[1] | IAsyncResult] = [] def delayed() -> Literal[1]: captures.append(1) return 1 - def after_delayed(handler: SequentialEventletHandler) -> None: - captures.append(handler) + def after_delayed(async_result: IAsyncResult) -> None: + captures.append(async_result) with start_stop_one() as handler: r = handler.async_result() diff --git a/kazoo/tests/test_gevent_handler.py b/kazoo/tests/test_gevent_handler.py index eda2505e4..ecfa94747 100644 --- a/kazoo/tests/test_gevent_handler.py +++ b/kazoo/tests/test_gevent_handler.py @@ -1,32 +1,34 @@ from __future__ import annotations +import importlib.util import unittest import sys -from typing import Any, Type +from typing import Type import pytest from kazoo.exceptions import NoNodeError from kazoo.handlers.utils import create_tcp_socket +from kazoo.interfaces import IAsyncResult from kazoo.protocol.states import Callback, KazooState, ZnodeStat from kazoo.testing import KazooTestCase -try: - import gevent # noqa: F401 +if importlib.util.find_spec("gevent") is None: + # The double test here stops complaints about imports not being at the top + # of the file. + pytest.importorskip("gevent", reason="gevent not available") +else: from gevent.event import Event from gevent.queue import Empty from gevent import socket from kazoo.handlers.gevent import AsyncResult, SequentialGeventHandler -except ImportError: - pytestmark = pytest.mark.skip(reason="gevent not available") @pytest.mark.skipif(sys.platform == "win32", reason="does not run on windows") class TestGeventHandler(unittest.TestCase): - def _makeOne(self, *args: Any) -> SequentialGeventHandler: - return SequentialGeventHandler(*args) + _makeOne = SequentialGeventHandler - def _getAsync(self) -> Type[AsyncResult[Any]]: + def _getAsync(self) -> Type[IAsyncResult]: return AsyncResult def _getEvent(self) -> Type[Event]: @@ -78,12 +80,11 @@ def func() -> None: @pytest.mark.skipif(sys.platform == "win32", reason="does not run on windows") class TestBasicGeventClient(KazooTestCase): + _makeOne = SequentialGeventHandler + def setUp(self) -> None: KazooTestCase.setUp(self) - def _makeOne(self, *args: Any) -> SequentialGeventHandler: - return SequentialGeventHandler(*args) - def _getEvent(self) -> Type[Event]: return Event diff --git a/kazoo/tests/test_hosts.py b/kazoo/tests/test_hosts.py index 80517d5d7..6cb603725 100644 --- a/kazoo/tests/test_hosts.py +++ b/kazoo/tests/test_hosts.py @@ -16,7 +16,7 @@ def test_ipv4(self) -> None: ("192.168.1.2", 2181), ("132.254.111.10", 2181), ] - assert chroot is None + assert chroot == "" hosts, chroot = collect_hosts( ["127.0.0.1:2181", "192.168.1.2:2181", "132.254.111.10:2181"] @@ -26,26 +26,26 @@ def test_ipv4(self) -> None: ("192.168.1.2", 2181), ("132.254.111.10", 2181), ] - assert chroot is None + assert chroot == "" def test_ipv6(self) -> None: hosts, chroot = collect_hosts("[fe80::200:5aee:feaa:20a2]:2181") assert hosts == [("fe80::200:5aee:feaa:20a2", 2181)] - assert chroot is None + assert chroot == "" hosts, chroot = collect_hosts(["[fe80::200:5aee:feaa:20a2]:2181"]) assert hosts == [("fe80::200:5aee:feaa:20a2", 2181)] - assert chroot is None + assert chroot == "" def test_hosts_list(self) -> None: hosts, chroot = collect_hosts("zk01:2181, zk02:2181, zk03:2181") expected1 = [("zk01", 2181), ("zk02", 2181), ("zk03", 2181)] assert hosts == expected1 - assert chroot is None + assert chroot == "" hosts, chroot = collect_hosts(["zk01:2181", "zk02:2181", "zk03:2181"]) assert hosts == expected1 - assert chroot is None + assert chroot == "" expected2 = "/test" hosts, chroot = collect_hosts("zk01:2181, zk02:2181, zk03:2181/test") diff --git a/kazoo/tests/test_lock.py b/kazoo/tests/test_lock.py index d03c2e193..a91a48fc6 100644 --- a/kazoo/tests/test_lock.py +++ b/kazoo/tests/test_lock.py @@ -341,7 +341,7 @@ def _acquire() -> None: attempts.append(int(lock.acquire(blocking=False))) threads = [] - for _i in range(0, self.thread_count): + for _ in range(0, self.thread_count): t = self.make_thread(target=_acquire) threads.append(t) t.start() @@ -374,7 +374,7 @@ def _acquire() -> None: differences.append(end_count - starting_count) threads = [] - for _i in range(0, self.thread_count): + for _ in range(0, self.thread_count): t = self.make_thread(target=_acquire) threads.append(t) t.start() diff --git a/kazoo/tests/test_retry.py b/kazoo/tests/test_retry.py index 9e67c6270..05ffa6edf 100644 --- a/kazoo/tests/test_retry.py +++ b/kazoo/tests/test_retry.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Any +from functools import partial from unittest import mock import pytest @@ -8,19 +8,7 @@ from kazoo import exceptions as ke from kazoo import retry as kr - -def _make_retry(*args: Any, **kwargs: Any) -> kr.KazooRetry: - """Return a KazooRetry instance with a dummy sleep function.""" - - def _sleep_func(_time: float) -> None: - pass - - # FIXME better way of doing this? Use TypedDict perhaps? - return kr.KazooRetry( - *args, - sleep_func=_sleep_func, - **kwargs, # type: ignore[misc] - ) +_make_retry = partial(kr.KazooRetry, sleep_func=lambda _time: None) def _make_try_func(times: int = 1) -> mock.Mock: @@ -28,7 +16,7 @@ def _make_try_func(times: int = 1) -> mock.Mock: returning None. """ callmock = mock.Mock( - side_effect=[kr.ForceRetryError("Failed!")] * times + [None], + side_effect=[kr.ForceRetryError("Failed!")] * times + [None] ) return callmock diff --git a/kazoo/tests/test_sasl.py b/kazoo/tests/test_sasl.py index 411777053..d04b77451 100644 --- a/kazoo/tests/test_sasl.py +++ b/kazoo/tests/test_sasl.py @@ -1,5 +1,6 @@ from __future__ import annotations +import importlib.util import os import subprocess import time @@ -7,17 +8,12 @@ import pytest from kazoo.testing import KazooTestHarness -from kazoo.exceptions import ( - AuthFailedError, - NoAuthError, -) +from kazoo.exceptions import AuthFailedError, NoAuthError class TestLegacySASLDigestAuthentication(KazooTestHarness): def setUp(self) -> None: - try: - import puresasl # NOQA - except ImportError: + if importlib.util.find_spec("puresasl") is None: pytest.skip("PureSASL not available.") os.environ["ZOOKEEPER_JAAS_AUTH"] = "digest" @@ -58,9 +54,7 @@ def test_invalid_sasl_auth(self) -> None: class TestSASLDigestAuthentication(KazooTestHarness): def setUp(self) -> None: - try: - import puresasl # NOQA - except ImportError: + if importlib.util.find_spec("puresasl") is None: pytest.skip("PureSASL not available.") os.environ["ZOOKEEPER_JAAS_AUTH"] = "digest" @@ -112,13 +106,9 @@ def test_invalid_sasl_auth(self) -> None: class TestSASLGSSAPIAuthentication(KazooTestHarness): def setUp(self) -> None: # puresasl isn't available under windows, so we can't do this test. - try: - import puresasl - except ImportError: + if importlib.util.find_spec("puresasl") is None: pytest.skip("PureSASL not available.") - try: - import kerberos # type: ignore[import-not-found,import-untyped] - except ImportError: + if importlib.util.find_spec("kerberos") is None: pytest.skip("Kerberos support not available.") if not os.environ.get("KRB5_TEST_ENV"): pytest.skip("Test Kerberos environ not setup.") diff --git a/kazoo/tests/test_selectors_select.py b/kazoo/tests/test_selectors_select.py index 0d9e77da5..84abb3bbb 100644 --- a/kazoo/tests/test_selectors_select.py +++ b/kazoo/tests/test_selectors_select.py @@ -10,7 +10,7 @@ import sys import unittest -from typing import cast, TYPE_CHECKING +from typing import TYPE_CHECKING from kazoo.handlers.utils import selector_select @@ -24,17 +24,8 @@ (sys.platform[:3] == "win"), "can't easily test on this system" ) class SelectTestCase(unittest.TestCase): - class Nope: - pass - - class Almost: - def fileno(self) -> str: - return "fileno" - def test_error_conditions(self) -> None: self.assertRaises(TypeError, select, 1, 2, 3) - self.assertRaises(TypeError, select, [self.Nope()], [], []) - self.assertRaises(TypeError, select, [self.Almost()], [], []) self.assertRaises(TypeError, select, [], [], [], "not a number") self.assertRaises(ValueError, select, [], [], [], -1) @@ -59,23 +50,14 @@ def test_returned_list_identity(self) -> None: def test_select(self) -> None: cmd = "for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done" with subprocess.Popen( - cmd, - shell=True, - stdout=subprocess.PIPE, - text=True, + cmd, shell=True, stdout=subprocess.PIPE, text=True ) as process: assert process.stdout is not None for tout in (0, 1, 2, 4, 8, 16) + (None,) * 10: - rfd, wfd, xfd = select( - [cast("HasFileNo", process.stdout)], [], [], tout - ) + rfd, wfd, xfd = select([process.stdout], [], [], tout) if (rfd, wfd, xfd) == ([], [], []): continue - if (rfd, wfd, xfd) == ( - [cast("HasFileNo", process.stdout)], - [], - [], - ): + if (rfd, wfd, xfd) == ([process.stdout], [], []): line = process.stdout.readline() if not line: break @@ -101,7 +83,3 @@ def fileno(self) -> int: finally: s1.close() s2.close() - - -if __name__ == "__main__": - unittest.main() diff --git a/kazoo/tests/test_threading_handler.py b/kazoo/tests/test_threading_handler.py index f4ac5d434..10c066554 100644 --- a/kazoo/tests/test_threading_handler.py +++ b/kazoo/tests/test_threading_handler.py @@ -5,7 +5,6 @@ import unittest from typing import Any, Type -from unittest.mock import Mock import pytest @@ -23,9 +22,7 @@ def _getAsync(self) -> Type[AsyncResult]: def test_proper_threading(self) -> None: h = self._makeOne() h.start() - # In Python 3.3 _Event is gone, before Event is function - event_class = getattr(threading, "_Event", threading.Event) - assert isinstance(h.event_object(), event_class) + assert isinstance(h.event_object(), threading.Event) def test_matching_async(self) -> None: h = self._makeOne() @@ -69,326 +66,3 @@ def test_huge_file_descriptor(self) -> None: h.stop() for sock in socks: sock.close() - - -class TestThreadingAsync(unittest.TestCase): - def _makeOne(self, *args: Any) -> AsyncResult: - return AsyncResult(*args) - - def _makeHandler(self) -> SequentialThreadingHandler: - return SequentialThreadingHandler() - - def test_ready(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - assert async_result.ready() is False - async_result.set("val") - assert async_result.ready() is True - assert async_result.successful() is True - assert async_result.exception is None - - def test_callback_queued(self) -> None: - mock_handler = Mock() - mock_handler.completion_queue = Mock() - async_result = self._makeOne(mock_handler) - - async_result.rawlink(lambda a: a) - async_result.set("val") - - assert mock_handler.completion_queue.put.called - - def test_set_exception(self) -> None: - mock_handler = Mock() - mock_handler.completion_queue = Mock() - async_result = self._makeOne(mock_handler) - async_result.rawlink(lambda a: a) - async_result.set_exception(ImportError("Error occurred")) - - assert isinstance(async_result.exception, ImportError) - assert mock_handler.completion_queue.put.called - - def test_get_wait_while_setting(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - bv = threading.Event() - cv = threading.Event() - - def wait_for_val() -> None: - bv.set() - val = async_result.get() - lst.append(val) - cv.set() - - th = threading.Thread(target=wait_for_val) - th.start() - bv.wait() - - async_result.set("fred") - cv.wait() - assert lst == ["fred"] - th.join() - - def test_get_with_nowait(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - timeout = self._makeHandler().timeout_exception - - with pytest.raises(timeout): - async_result.get(block=False) - - with pytest.raises(timeout): - async_result.get_nowait() - - def test_get_with_exception(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - bv = threading.Event() - cv = threading.Event() - - def wait_for_val() -> None: - bv.set() - try: - val = async_result.get() - except ImportError: - lst.append("oops") - else: - lst.append(val) - cv.set() - - th = threading.Thread(target=wait_for_val) - th.start() - bv.wait() - - async_result.set_exception(ImportError()) - cv.wait() - assert lst == ["oops"] - th.join() - - def test_wait(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst: list[bool | str] = [] - bv = threading.Event() - cv = threading.Event() - - def wait_for_val() -> None: - bv.set() - try: - val = async_result.wait(10) - except ImportError: - lst.append("oops") - else: - lst.append(val) - cv.set() - - th = threading.Thread(target=wait_for_val) - th.start() - bv.wait(10) - - async_result.set("fred") - cv.wait(15) - assert lst == [True] - th.join() - - def test_wait_race(self) -> None: - """Test that there is no race condition in `IAsyncResult.wait()`. - - Guards against the reappearance of: - https://github.com/python-zk/kazoo/issues/485 - """ - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - async_result.set("immediate") - - cv = threading.Event() - - def wait_for_val() -> None: - # NB: should not sleep - async_result.wait(20) - cv.set() - - th = threading.Thread(target=wait_for_val) - th.daemon = True - th.start() - - # if the wait() didn't sleep (correctly), cv will be set quickly - # if it did sleep, the cv will not be set yet and this will timeout - cv.wait(10) - assert cv.is_set() is True - th.join() - - def test_set_before_wait(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - cv = threading.Event() - async_result.set("fred") - - def wait_for_val() -> None: - val = async_result.get() - lst.append(val) - cv.set() - - th = threading.Thread(target=wait_for_val) - th.start() - cv.wait() - assert lst == ["fred"] - th.join() - - def test_set_exc_before_wait(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - cv = threading.Event() - async_result.set_exception(ImportError()) - - def wait_for_val() -> None: - try: - val = async_result.get() - except ImportError: - lst.append("ooops") - else: - lst.append(val) - cv.set() - - th = threading.Thread(target=wait_for_val) - th.start() - cv.wait() - assert lst == ["ooops"] - th.join() - - def test_linkage(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - cv = threading.Event() - - lst = [] - - def add_on() -> None: - lst.append(True) - - def wait_for_val() -> None: - async_result.get() - cv.set() - - th = threading.Thread(target=wait_for_val) - th.start() - - async_result.rawlink(add_on) - async_result.set(b"fred") - assert mock_handler.completion_queue.put.called - async_result.unlink(add_on) - cv.wait() - assert async_result.value == b"fred" - th.join() - - def test_linkage_not_ready(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - - def add_on() -> None: - lst.append(True) - - async_result.set("fred") - assert not mock_handler.completion_queue.called - async_result.rawlink(add_on) - assert mock_handler.completion_queue.put.called - - def test_link_and_unlink(self) -> None: - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - - def add_on() -> None: - lst.append(True) - - async_result.rawlink(add_on) - assert not mock_handler.completion_queue.put.called - async_result.unlink(add_on) - async_result.set("fred") - assert not mock_handler.completion_queue.put.called - - def test_captured_exception(self) -> None: - from kazoo.handlers.utils import capture_exceptions - - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - @capture_exceptions(async_result) - def exceptional_function() -> float: - return 1 / 0 - - exceptional_function() - - with pytest.raises(ZeroDivisionError): - async_result.get() - - def test_no_capture_exceptions(self) -> None: - from kazoo.handlers.utils import capture_exceptions - - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - - def add_on() -> None: - lst.append(True) - - async_result.rawlink(add_on) - - @capture_exceptions(async_result) - def regular_function() -> bool: - return True - - regular_function() - - assert not mock_handler.completion_queue.put.called - - def test_wraps(self) -> None: - from kazoo.handlers.utils import wrap - - mock_handler = Mock() - async_result = self._makeOne(mock_handler) - - lst = [] - - def add_on(result: AsyncResult) -> None: - lst.append(result.get()) - - async_result.rawlink(add_on) - - @wrap(async_result) - def regular_function() -> str: - return "hello" - - assert regular_function() == "hello" - assert mock_handler.completion_queue.put.called - assert async_result.get() == "hello" - - def test_multiple_callbacks(self) -> None: - mockback1 = Mock(name="mockback1") - mockback2 = Mock(name="mockback2") - handler = self._makeHandler() - handler.start() - - async_result = self._makeOne(handler) - async_result.rawlink(mockback1) - async_result.rawlink(mockback2) - async_result.set("howdy") - async_result.wait() - handler.stop() - - mockback2.assert_called_once_with(async_result) - mockback1.assert_called_once_with(async_result) diff --git a/pyproject.toml b/pyproject.toml index 9455bf328..5c53c55df 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,11 +57,14 @@ test-extras = [ "pyjks==20.0.0", "unittest-parametrize==1.8.0", ] +coverage = [ + "coverage==7.10.7" +] pytest = [ "pytest==8.4.2", ] test = [ - "coverage==7.10.7", + {include-group = "coverage"}, {include-group = "pytest"}, "pytest-cov==7.1.0", "pytest-timeout==2.4.0", diff --git a/run_failure.py b/run_failure.py deleted file mode 100644 index 7bc1dac2d..000000000 --- a/run_failure.py +++ /dev/null @@ -1,28 +0,0 @@ -import os -import sys - - -def test(arg): - return os.system("bin/pytest -v %s" % arg) - - -def main(args): - if not args: - print( - "Run as bin/python run_failure.py , for example: \n" - "bin/python run_failure.py " - "kazoo.tests.test_watchers:KazooChildrenWatcherTests" - ) - return - arg = args[0] - i = 0 - while 1: - i += 1 - print("Run number: %s" % i) - ret = test(arg) - if ret != 0: - break - - -if __name__ == "__main__": - main(sys.argv[1:]) diff --git a/tox.ini b/tox.ini index 88995960e..23069dd17 100644 --- a/tox.ini +++ b/tox.ini @@ -35,10 +35,16 @@ commands = bash \ sasl: {toxinidir}/init_krb5.sh {envtmpdir}/kerberos \ {toxinidir}/ensure-zookeeper-env.sh \ - pytest {posargs: -ra -v --cov-report=xml --cov=kazoo kazoo/tests} + pytest {posargs: -ra -v --cov-append --cov-report=xml --cov=kazoo kazoo/tests} [testenv:build] +[testenv:clean] +dependency_groups = + coverage +skip_install = true +commands = coverage erase + [testenv:docs] dependency_groups = test-extras