From 01cdb41027ec5d4247d6ab0fd9496c9b58c56c6c 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/utils.py | 29 +- kazoo/hosts.py | 9 +- kazoo/interfaces.py | 1 - kazoo/protocol/serialization.py | 61 ++- kazoo/recipe/cache.py | 3 - kazoo/recipe/counter.py | 26 +- kazoo/recipe/lock.py | 18 +- kazoo/tests/test_cache.py | 617 ++++++++++++++------------- kazoo/tests/test_hosts.py | 12 +- kazoo/tests/test_selectors_select.py | 21 +- 11 files changed, 426 insertions(+), 544 deletions(-) diff --git a/kazoo/client.py b/kazoo/client.py index 3f2c3b94..fffd456b 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 _i 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/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/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/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_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_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()