From 09821ec2bee6a0b43c9ee177e7efc4e7e6c950f8 Mon Sep 17 00:00:00 2001 From: Thomas Tanner Date: Sat, 5 Sep 2026 09:06:15 +0100 Subject: [PATCH] chore(typing): clean up some of the workarounds from the initial PR --- kazoo/client.py | 173 +++----- kazoo/handlers/gevent.py | 30 +- kazoo/handlers/utils.py | 29 +- kazoo/hosts.py | 9 +- kazoo/interfaces.py | 1 - kazoo/protocol/connection.py | 54 +-- kazoo/protocol/serialization.py | 61 ++- kazoo/recipe/cache.py | 3 - kazoo/recipe/counter.py | 26 +- kazoo/recipe/lock.py | 18 +- kazoo/testing/common.py | 6 +- kazoo/tests/test_cache.py | 617 ++++++++++++++------------- kazoo/tests/test_client.py | 59 ++- kazoo/tests/test_connection.py | 2 +- kazoo/tests/test_gevent_handler.py | 14 +- kazoo/tests/test_hosts.py | 12 +- kazoo/tests/test_lock.py | 4 +- kazoo/tests/test_retry.py | 13 +- kazoo/tests/test_sasl.py | 19 +- kazoo/tests/test_selectors_select.py | 21 +- 20 files changed, 526 insertions(+), 645 deletions(-) diff --git a/kazoo/client.py b/kazoo/client.py index 3f2c3b94..0c437654 100644 --- a/kazoo/client.py +++ b/kazoo/client.py @@ -178,7 +178,6 @@ def __init__( ) -> None: ... - # FIXME This should be deprecated then killed @overload @deprecated( "Passing retry configuration parameters directly to the client" @@ -348,16 +347,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 @@ -397,74 +388,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": @@ -508,7 +479,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], @@ -517,14 +487,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) @@ -609,6 +572,12 @@ 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], @@ -637,25 +606,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) - - if chroot: - new_chroot = normpath(chroot) - else: - new_chroot = "" + self.hosts, chroot = self._collect_hosts(hosts) - 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" + "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. @@ -993,31 +955,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 - # and the next 2 suppress should include return-value - # but hound is broken - return version # type: ignore - 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 + 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)) @@ -1596,12 +1539,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 diff --git a/kazoo/handlers/gevent.py b/kazoo/handlers/gevent.py index d8e1a838..3cf56a32 100644 --- a/kazoo/handlers/gevent.py +++ b/kazoo/handlers/gevent.py @@ -3,6 +3,7 @@ from __future__ import annotations import atexit +import importlib.util import logging from typing import Any, Callable, Iterable, TYPE_CHECKING, cast @@ -11,26 +12,23 @@ 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] - - if TYPE_CHECKING: from kazoo.interfaces import FdLike, Lockable, Socket from kazoo.protocol.states import Callback - from gevent import Greenlet -_using_libevent = gevent.__version__.startswith("0.") + GEVENT_AVAILABLE: bool = True + from gevent import Greenlet +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__) diff --git a/kazoo/handlers/utils.py b/kazoo/handlers/utils.py index 097daf4a..d1a056b6 100644 --- a/kazoo/handlers/utils.py +++ b/kazoo/handlers/utils.py @@ -384,33 +384,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], @@ -436,7 +409,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 cda746a3..04bfa44b 100644 --- a/kazoo/hosts.py +++ b/kazoo/hosts.py @@ -5,7 +5,7 @@ def collect_hosts( hosts: str | list[str], -) -> tuple[list[tuple[str, int]], str | None]: +) -> 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 +14,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 +29,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 466067e3..65837b02 100644 --- a/kazoo/interfaces.py +++ b/kazoo/interfaces.py @@ -10,7 +10,6 @@ from __future__ import annotations - from typing import ( Any, Callable, diff --git a/kazoo/protocol/connection.py b/kazoo/protocol/connection.py index 33582720..7a9046ac 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 @@ -63,16 +64,13 @@ 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? -# NB Those should be ignore import but I don't trust hound. -try: - import puresasl # type: ignore - import puresasl.client # type: ignore + 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__) @@ -214,7 +212,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 +460,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 @@ -905,8 +903,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. @@ -920,36 +920,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 914540a8..fe448b89 100644 --- a/kazoo/protocol/serialization.py +++ b/kazoo/protocol/serialization.py @@ -393,8 +393,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] @@ -450,35 +478,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 1d361df8..95c40a20 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 77d68cb5..bde47e57 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 da30cf43..2715c9d1 100644 --- a/kazoo/recipe/lock.py +++ b/kazoo/recipe/lock.py @@ -48,26 +48,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: @@ -666,7 +653,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/testing/common.py b/kazoo/testing/common.py index 0da736b9..c423a6d2 100644 --- a/kazoo/testing/common.py +++ b/kazoo/testing/common.py @@ -176,7 +176,7 @@ def run(self) -> None: to_java_compatible_path(truststore_path), "\n".join(self.configuration_entries), ) - ) # NOQA + ) # setup a replicated setup if peers are specified if self.peers: @@ -220,7 +220,7 @@ def run(self) -> None: log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender log4j.appender.ROLLINGFILE.Threshold=DEBUG log4j.appender.ROLLINGFILE.File=""" - + to_java_compatible_path( # NOQA + + to_java_compatible_path( self.working_path + os.sep + "zookeeper.log\n" ) ) @@ -331,7 +331,7 @@ def stop(self) -> None: self.process.terminate() self.process.wait() if self.process.returncode != 0: - log.warn( + log.warning( "Zookeeper process %s failed to terminate with" " non-zero return code (it terminated with %s return" " code instead)", diff --git a/kazoo/tests/test_cache.py b/kazoo/tests/test_cache.py index 4a5441e9..29b4a3c6 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 @@ -67,7 +68,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(): @@ -75,29 +75,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: @@ -150,317 +143,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() @@ -469,30 +482,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 a031b1ff..2bc12243 100644 --- a/kazoo/tests/test_client.py +++ b/kazoo/tests/test_client.py @@ -42,7 +42,6 @@ make_digest_acl_credential, CREATOR_ALL_ACL, make_digest_acl, - ACL, OPEN_ACL_UNSAFE, ) @@ -86,12 +85,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 == "" @@ -128,6 +131,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) ) @@ -136,17 +140,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() @@ -178,12 +211,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() @@ -204,7 +236,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() @@ -234,6 +266,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() @@ -325,7 +359,6 @@ def watch_events(event: KazooState) -> None: assert not cv.is_set() def test_state_listener(self) -> None: - states = [] condition = self.make_condition() @@ -351,7 +384,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() @@ -464,8 +496,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: diff --git a/kazoo/tests/test_connection.py b/kazoo/tests/test_connection.py index 5439cd0b..1c40c9d8 100644 --- a/kazoo/tests/test_connection.py +++ b/kazoo/tests/test_connection.py @@ -411,7 +411,7 @@ 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 diff --git a/kazoo/tests/test_gevent_handler.py b/kazoo/tests/test_gevent_handler.py index fb897987..4d2dbaec 100644 --- a/kazoo/tests/test_gevent_handler.py +++ b/kazoo/tests/test_gevent_handler.py @@ -1,5 +1,6 @@ from __future__ import annotations +import importlib.util import unittest import sys @@ -11,20 +12,18 @@ from kazoo.protocol.states import Callback, KazooState, ZnodeStat from kazoo.testing import KazooTestCase -try: - import gevent # NOQA: +if importlib.util.find_spec("gevent") is None: + pytestmark = pytest.mark.skip(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]]: return AsyncResult @@ -81,8 +80,7 @@ class TestBasicGeventClient(KazooTestCase): def setUp(self) -> None: KazooTestCase.setUp(self) - def _makeOne(self, *args: Any) -> SequentialGeventHandler: - return SequentialGeventHandler(*args) + _makeOne = SequentialGeventHandler def _getEvent(self) -> Type[Event]: return Event diff --git a/kazoo/tests/test_hosts.py b/kazoo/tests/test_hosts.py index 80517d5d..6cb60372 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 1ba1327a..f9aa4c9f 100644 --- a/kazoo/tests/test_lock.py +++ b/kazoo/tests/test_lock.py @@ -343,7 +343,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() @@ -376,7 +376,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 1c3f9282..ace3cfd7 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 @@ -9,16 +9,7 @@ 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: diff --git a/kazoo/tests/test_sasl.py b/kazoo/tests/test_sasl.py index 78087002..cc1eb086 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 @@ -16,9 +17,7 @@ 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" @@ -66,9 +65,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" @@ -127,15 +124,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: - # FIXME Hound objects to import not found as it thinks it's a - # syntax error. I don't know why it thinks that. - import kerberos # type: ignore - 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 7b068e1d..7ea14a92 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) @@ -66,13 +57,11 @@ def test_select(self) -> None: ) 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)], + [process.stdout], [], [], ): @@ -97,7 +86,3 @@ def fileno(self) -> int: a[:] = [F()] * 10 self.assertEqual(select([], a, []), ([], a[:5], [])) - - -if __name__ == "__main__": - unittest.main()