diff --git a/faust/__init__.py b/faust/__init__.py index 132e784a5..6db7a3c06 100644 --- a/faust/__init__.py +++ b/faust/__init__.py @@ -19,7 +19,6 @@ # faust/agents.py - Agents use all of the above. # --- ~~~~~ ~ ~ ~ ~ ~ ~ ~ import os -import re import sys import typing @@ -30,6 +29,8 @@ from typing import Any, Mapping, NamedTuple, Optional, Sequence, Tuple +from packaging.version import InvalidVersion, Version + __version__ = version("faust-streaming") __author__ = "Robinhood Markets, Inc." __contact__ = "schrohm@gmail.com, vpatki@wayfair.com, williambbarnhart@gmail.com" @@ -50,23 +51,43 @@ class VersionInfo(NamedTuple): version_info_t = VersionInfo # XXX compat -# bumpversion can only search for {current_version} -# so we have to parse the version here. -_match = re.match(r"^(?Pv)?(?P[^\+]+)(?P.*)?$", __version__) -if _match is None: # pragma: no cover - raise RuntimeError("THIS IS A BROKEN RELEASE!") -_temp = _match.groups() -# XXX This is broken and so is the public ``faust.version_info``: the regex -# yields the strings ``(prefix, version, suffix)``, which land positionally in -# the ``(major, minor, micro)`` int fields. So ``.major`` is the ``'v'`` -# prefix or :const:`None`, ``.minor`` is the entire version string and -# ``.micro`` is the suffix -- e.g. ``VersionInfo(major=None, -# minor='0.11.5', micro='')`` instead of ``(0, 11, 5)``. -# Left as-is because fixing it changes what ``faust.VERSION`` holds. -VERSION = version_info = VersionInfo(*_temp) # type: ignore[arg-type] -del _match -del _temp -del re +def _parse_version(version_string: str) -> VersionInfo: + """Parse a version string into a :class:`VersionInfo` tuple. + + The PEP 440 grammar is delegated to :mod:`packaging`, the same parser + pip and setuptools use, rather than maintained here. ``release`` + supplies ``major``/``minor``/``micro`` (missing components default to + ``0``), and a pre/dev/post segment supplies ``releaselevel`` and + ``serial`` separately -- ``0.11.5rc1`` gives ``releaselevel='rc'``, + ``serial='1'`` -- mirroring what those field names mean on + :data:`sys.version_info`. + + Only the parts :class:`VersionInfo` has fields for are kept. A local + segment (the ``+g1234`` of ``0.11.5.dev1+g1234``) and any component + past the third are dropped; :data:`faust.__version__` remains the + full, unmodified string. + + Never raises: a version :mod:`packaging` cannot parse ends up as the + ``releaselevel`` of ``VersionInfo(0, 0, 0)``. + """ + releaselevel: Optional[str] + serial: Optional[str] + try: + parsed = Version(version_string) + except InvalidVersion: + return VersionInfo(0, 0, 0, version_string or None) + if parsed.pre is not None: # 1.0rc1 / 1.0a2 / 1.0b3 + releaselevel, serial = parsed.pre[0], str(parsed.pre[1]) + elif parsed.dev is not None: # 1.0.dev1 + releaselevel, serial = "dev", str(parsed.dev) + elif parsed.post is not None: # 1.0.post2 + releaselevel, serial = "post", str(parsed.post) + else: + releaselevel, serial = None, None + return VersionInfo(parsed.major, parsed.minor, parsed.micro, releaselevel, serial) + + +VERSION = version_info = _parse_version(__version__) # This is here to support setting the --datadir argument @@ -307,5 +328,6 @@ def __dir__(self) -> Sequence[str]: "version_info_t": version_info_t, "version_info": version_info, "VERSION": VERSION, + "_parse_version": _parse_version, } ) diff --git a/faust/auth.py b/faust/auth.py index da78d6842..3c8b7df4b 100644 --- a/faust/auth.py +++ b/faust/auth.py @@ -130,15 +130,10 @@ def __init__( cadata: Optional[str] = None, ) -> None: if context is None: + if purpose is None: + purpose = ssl.Purpose.SERVER_AUTH context = ssl.create_default_context( - # XXX ``purpose`` defaults to None here, but - # ``ssl.create_default_context`` requires an ``ssl.Purpose`` - # and raises ``TypeError`` on None -- so ``SSLCredentials()`` - # with no explicit ``purpose`` cannot build a context at all. - # Real bug, kept as-is: fixing it changes the default TLS - # purpose of a security-relevant public API, which is out of - # scope for a typing pass. - purpose=purpose, # type: ignore[arg-type] + purpose=purpose, cafile=cafile, capath=capath, cadata=cadata, diff --git a/faust/cli/base.py b/faust/cli/base.py index 0b1c0c3db..21514b297 100644 --- a/faust/cli/base.py +++ b/faust/cli/base.py @@ -792,7 +792,10 @@ class AppCommand(Command): abstract: ClassVar[bool] = True - app: AppT + #: The app this command runs for, or :const:`None` when the command + #: sets ``require_app = False`` and was invoked without ``-A``. + #: Use the :attr:`app` property to access it when an app is required. + _app: Optional[AppT] = None require_app = True @@ -839,25 +842,44 @@ def __init__( ) -> None: super().__init__(ctx) - self.app = self._finalize_app(getattr(ctx.find_root(), "app", None)) + self._app = self._finalize_app(getattr(ctx.find_root(), "app", None)) self.args = args self.kwargs = kwargs - self.key_serializer = key_serializer or self.app.conf.key_serializer - self.value_serializer = value_serializer or self.app.conf.value_serializer + if self._app is not None: + conf = self._app.conf + self.key_serializer = key_serializer or conf.key_serializer + self.value_serializer = value_serializer or conf.value_serializer + else: + # ``require_app = False`` command invoked without ``-A``: + # there is no app to take the default codecs from. + self.key_serializer = key_serializer + self.value_serializer = value_serializer + + @property + def app(self) -> AppT: + """Return the app this command runs for. + + Raises: + UsageError: if the command was invoked without an app. + Only commands setting ``require_app = False`` can get + this far without one. + """ + app = self._app + if app is None: + raise self.UsageError("Need to specify app using -A parameter") + return app + + @app.setter + def app(self, app: AppT) -> None: + self._app = app - def _finalize_app(self, app: Optional[AppT]) -> AppT: + def _finalize_app(self, app: Optional[AppT]) -> Optional[AppT]: if app is not None: return self._finalize_concrete_app(app) else: - # XXX ``_app_from_str`` returns None for ``require_app = False`` - # commands (see faust/cli/completion.py) that were invoked without - # ``-A``, so this really can hand back None -- and - # ``AppCommand.__init__`` dereferences ``self.app.conf`` - # unconditionally right after, raising ``AttributeError: 'NoneType' - # object has no attribute 'conf'``. Real bug; fixing it means - # deciding what such commands should do without an app, which is - # out of scope for a typing pass. - return self._app_from_str(self.state.app) # type: ignore[return-value] + # Returns None for ``require_app = False`` commands + # (see faust/cli/completion.py) invoked without ``-A``. + return self._app_from_str(self.state.app) def _app_from_str(self, appstr: Optional[str] = None) -> Optional[AppT]: if appstr: @@ -894,7 +916,9 @@ def _detect_main_package(self, argv: List[str]) -> str: # pragma: no cover async def on_stop(self) -> None: """Call after command executed.""" await super().on_stop() - app = cast(_App, self.app) + if self._app is None: + return # command ran without an app: nothing to clean up. + app = cast(_App, self._app) # If command started the producer, we should also stop that # - this will flush any buffers before exiting. if app._producer is not None and app._producer.started: @@ -1009,7 +1033,10 @@ def abbreviate_fqdn(self, name: str, *, prefix: str = "") -> str: @property def blocking_timeout(self) -> float: """Return the blocking timeout used for this command.""" - return self._blocking_timeout or self.app.conf.blocking_timeout + if self._app is None: + # command ran without an app: use the Command default. + return self._blocking_timeout or 0.0 + return self._blocking_timeout or self._app.conf.blocking_timeout @blocking_timeout.setter def blocking_timeout(self, timeout: float) -> None: diff --git a/faust/tables/base.py b/faust/tables/base.py index 33a167d74..34bae9dc5 100644 --- a/faust/tables/base.py +++ b/faust/tables/base.py @@ -384,23 +384,10 @@ async def _del_old_keys(self) -> None: while timestamps and window.stale(timestamps[0], time.time()): timestamp = heappop(timestamps) triggered_windows = [ - # XXX bug: this lookup can never hit. # ``_partition_timestamp_keys`` is keyed by - # ``(partition, range_end)`` -- a ``(int, float)`` pair, - # written that way in ``_maybe_set_key_ttl`` and read that - # way in ``_maybe_del_key_ttl``. Here it is looked up by - # ``(partition, window_range)`` where ``window_range`` is - # the ``(start, end)`` tuple, so no key ever matches and - # ``triggered_windows`` is always ``[None, ...]``. The - # consequence is that ``window_data`` stays empty and - # ``on_window_close`` never receives the aggregated window - # data, only the raw per-key value. The correct key is - # ``(partition, window_range[1])``; that is a behaviour - # change, so it is not made here and the type error is - # only silenced. - self._partition_timestamp_keys.get( - (partition, window_range) # type: ignore[arg-type] - ) + # ``(partition, range_end)``, so look up the end of + # each ``(start, end)`` range. + self._partition_timestamp_keys.get((partition, window_range[1])) for window_range in self._window_ranges(timestamp) ] keys_to_remove = self._partition_timestamp_keys.pop( diff --git a/faust/tables/recovery.py b/faust/tables/recovery.py index 2ff38b736..61ad27a98 100644 --- a/faust/tables/recovery.py +++ b/faust/tables/recovery.py @@ -811,18 +811,10 @@ async def detect_aborted_tx() -> None: and offsets[tp] is not None and offsets[tp] < highwater ): - # XXX bug: ConsumerT.position is declared - # ``Optional[int]`` and really does return None when the - # partition has no position yet, so this comparison raises - # TypeError on that path (swallowed by the ``except - # Exception`` in the caller's loop, which then skips the - # aborted-tx fixup). Guarding for None would change which - # events reach the caller, so the behaviour is kept and - # only the type error is silenced. - if ( - await self.app.consumer.position(tp) # type: ignore[operator] - >= highwater - ): + # The partition may have no position yet, in which case + # there is nothing to compare against and we skip it. + position = await self.app.consumer.position(tp) + if position is not None and position >= highwater: logger.info(f"Aborted tx until highwater for {tp}") offsets[tp] = highwater @@ -848,58 +840,52 @@ async def detect_aborted_tx() -> None: tp = message.tp offset = message.offset logger.debug(f"Recovery message topic {tp} offset {offset}") - offsets: Counter[TP] - bufsize = buffer_sizes.get(tp) - is_active = False - if tp in active_tps: - is_active = True - table = tp_to_table[tp] - offsets = active_offsets - if bufsize is None: - bufsize = buffer_sizes[tp] = table.recovery_buffer_size - active_events_received_at[tp] = now - elif tp in standby_tps: - table = tp_to_table[tp] - offsets = standby_offsets - if bufsize is None: - bufsize = buffer_sizes[tp] = table.standby_buffer_size - standby_events_received_at[tp] = now - else: + if tp not in active_tps and tp not in standby_tps: + # Not a partition we are recovering: skip the event, + # but keep the bookkeeping below running. logger.warning(f"recovery unknown topic {tp} offset {offset}") - - seen_offset = offsets.get(tp, None) - logger.debug( - f"seen offset for {tp} is {seen_offset} message offset {offset}" - ) - if seen_offset is None or offset > seen_offset: - offsets[tp] = offset - buf = buffers[table] - buf.append(event) - await table.on_changelog_event(event) - # XXX bug: the ``else`` branch above only logs a warning - # and falls through, so an event for a TP that is neither - # active nor standby reaches here with ``table``, - # ``offsets`` and ``bufsize`` still holding the PREVIOUS - # iteration's values -- the event is then applied to an - # unrelated table (or raises UnboundLocalError if it is - # the first event of the loop). That is why ``bufsize`` - # is still ``Optional[int]`` here and the comparison can - # raise TypeError. Fixing it means skipping the untracked - # TP, which is a behaviour change, so it is left as is. - if len(buf) >= bufsize: # type: ignore[operator] - table.apply_changelog_batch(buf) - buf.clear() - self._last_flush_at = now - now_after = monotonic() - - if is_active: - last_processed_at = self._last_active_event_processed_at - if last_processed_at is not None: - processing_times.append(now_after - last_processed_at) - max_samples = self.num_samples_required_for_estimate - if len(processing_times) > max_samples: - processing_times.popleft() - self._last_active_event_processed_at = now_after + else: + offsets: Counter[TP] + bufsize = buffer_sizes.get(tp) + is_active = False + if tp in active_tps: + is_active = True + table = tp_to_table[tp] + offsets = active_offsets + if bufsize is None: + bufsize = buffer_sizes[tp] = table.recovery_buffer_size + active_events_received_at[tp] = now + else: + table = tp_to_table[tp] + offsets = standby_offsets + if bufsize is None: + bufsize = buffer_sizes[tp] = table.standby_buffer_size + standby_events_received_at[tp] = now + + seen_offset = offsets.get(tp, None) + logger.debug( + f"seen offset for {tp} is {seen_offset} " + f"message offset {offset}" + ) + if seen_offset is None or offset > seen_offset: + offsets[tp] = offset + buf = buffers[table] + buf.append(event) + await table.on_changelog_event(event) + if len(buf) >= bufsize: + table.apply_changelog_batch(buf) + buf.clear() + self._last_flush_at = now + now_after = monotonic() + + if is_active: + last_processed_at = self._last_active_event_processed_at + if last_processed_at is not None: + processing_times.append(now_after - last_processed_at) + max_samples = self.num_samples_required_for_estimate + if len(processing_times) > max_samples: + processing_times.popleft() + self._last_active_event_processed_at = now_after await _maybe_signal_recovery_end() diff --git a/faust/transport/drivers/confluent.py b/faust/transport/drivers/confluent.py index 2c04913dd..bd8e0886e 100644 --- a/faust/transport/drivers/confluent.py +++ b/faust/transport/drivers/confluent.py @@ -155,15 +155,8 @@ async def on_stop(self) -> None: await super().on_stop() def verify_event_path(self, now: float, tp: TP) -> None: - # XXX broken: neither ConsumerThread nor ConfluentConsumerThread - # implements verify_event_path, so this raises AttributeError on - # every tick of the commit livelock detector - # (faust.transport.consumer.Consumer._commit_livelock_detector -> - # verify_all_partitions_active). Livelock detection is therefore - # dead for this driver. Not fixed here: adding a no-op stub would - # change runtime behaviour, and a real implementation belongs in - # ConfluentConsumerThread. - return self._thread.verify_event_path(now, tp) # type: ignore[attr-defined] + """Verify the path of an event, if this is not working.""" + return self._thread.verify_event_path(now, tp) class AsyncConsumer: @@ -292,6 +285,15 @@ def _create_client_consumer( def close(self) -> None: ... + def verify_event_path(self, now: float, tp: TP) -> None: + """Verify the path of an event. + + Livelock detection is not implemented for this driver, so this is + a no-op, matching the stub in + :meth:`faust.transport.consumer.Consumer.verify_event_path`. + """ + return None + async def subscribe(self, topics: Iterable[str]) -> None: # XXX pattern does not work :/ await self.cast_thread( @@ -684,16 +686,10 @@ async def flush(self) -> None: def key_partition(self, topic: str, key: bytes) -> TP: """Return topic and partition destination for key.""" # Get the partition count for the topic - # XXX broken: ``ProducerThread.producer`` is the Faust Producer - # (i.e. ``self``), not the underlying confluent_kafka.Producer -- - # that one is ``ProducerThread._producer``. Faust producers have no - # ``list_topics``, so this raises AttributeError and - # ``Producer.key_partition`` is dead on arrival for this driver. - # Behaviour left untouched in this annotation-only pass; the fix is - # to read ``self._producer_thread._producer``. - metadata = self._producer_thread.producer.list_topics( # type: ignore[attr-defined] # noqa: E501 - topic - ) + _producer = self._producer_thread._producer + if _producer is None: + raise RuntimeError("Producer not started") + metadata = _producer.list_topics(topic) partition_count = len(metadata.topics[topic].partitions) # Calculate the partition number based on the key hash diff --git a/requirements/requirements.txt b/requirements/requirements.txt index 62c54daf9..ba542b67e 100644 --- a/requirements/requirements.txt +++ b/requirements/requirements.txt @@ -11,6 +11,12 @@ click>=6.7,<8.2 # keeps `scripts/check` reproducible, for the same reason mypy itself is pinned # in typecheck.txt. mode-streaming>=0.6.0 +# `faust/__init__.py` parses `__version__` with `packaging.version.Version` +# rather than hand-rolled regexes. aiokafka already requires packaging, so +# this is already installed everywhere faust is; declare it anyway, because +# faust imports it directly and must not depend on a transitive dep staying +# put. `Version.major`/`.minor`/`.micro` need >=20.0. +packaging>=20.0 terminaltables>=3.1,<4.0 yarl>=1.0,<2.0 croniter>=0.3.16 diff --git a/tests/unit/cli/test_base.py b/tests/unit/cli/test_base.py index 8b81b1c04..01c316ba5 100644 --- a/tests/unit/cli/test_base.py +++ b/tests/unit/cli/test_base.py @@ -21,6 +21,7 @@ find_app, option, ) +from faust.cli.completion import completion from faust.types._env import CONSOLE_PORT from tests.helpers import AsyncMock @@ -490,7 +491,21 @@ def test_console_port(self, *, command, ctx): assert command.console_port == CONSOLE_PORT +def test_appcommand__require_app_false__runs_without_app(): + # `faust completion` sets require_app = False, so it must run + # when no app was given with -A. + assert completion.require_app is False + with patch("faust.cli.completion.click_completion") as cc: + cc.get_code.return_value = "# completion code" + exitcode, stdout, stderr = call_command("completion") + assert exitcode == 0 + assert "# completion code" in stdout.getvalue() + + class Test_AppCommand: + class NoAppRequiredCommand(AppCommand): + require_app = False + @pytest.fixture() def ctx(self): return Mock(name="ctx") @@ -499,6 +514,53 @@ def ctx(self): def command(self, *, app, ctx): return AppCommand(app=app, ctx=ctx) + @pytest.fixture() + def no_app_command(self, *, ctx): + # a ``require_app = False`` command invoked without ``-A``. + ctx.find_root.return_value.app = None + ctx.ensure_object.return_value.app = None + return self.NoAppRequiredCommand(ctx=ctx) + + def test_init__no_app_when_not_required(self, *, no_app_command): + assert no_app_command._app is None + assert no_app_command.key_serializer is None + assert no_app_command.value_serializer is None + + def test_init__serializers_without_app(self, *, ctx): + ctx.find_root.return_value.app = None + ctx.ensure_object.return_value.app = None + command = self.NoAppRequiredCommand( + ctx=ctx, + key_serializer="raw", + value_serializer="json", + ) + assert command.key_serializer == "raw" + assert command.value_serializer == "json" + + def test_app__raises_when_missing(self, *, no_app_command): + with pytest.raises(no_app_command.UsageError): + no_app_command.app + + def test_app__is_still_writable(self, *, no_app_command, app): + # ``app`` was a plain writable attribute before it became a + # property, so the setter has to keep ``command.app = ...`` + # working for anything that assigned to it. + with pytest.raises(no_app_command.UsageError): + no_app_command.app + no_app_command.app = app + assert no_app_command._app is app + assert no_app_command.app is app + + def test_blocking_timeout__no_app(self, *, no_app_command): + no_app_command.blocking_timeout = None + assert no_app_command.blocking_timeout == 0.0 + no_app_command.blocking_timeout = 32.41 + assert no_app_command.blocking_timeout == 32.41 + + @pytest.mark.asyncio + async def test_on_stop__no_app(self, *, no_app_command): + await no_app_command.on_stop() + def test_finalize_app__str(self, *, command): command._app_from_str = Mock() command.state.app = "foo" diff --git a/tests/unit/tables/test_base.py b/tests/unit/tables/test_base.py index bcd6e5fd1..5a8aaed7b 100644 --- a/tests/unit/tables/test_base.py +++ b/tests/unit/tables/test_base.py @@ -11,7 +11,7 @@ from faust.stores.base import Store from faust.tables.base import Collection from faust.types import TP -from faust.windows import Window +from faust.windows import HoppingWindow, Window from tests.helpers import AsyncMock TP1 = TP("foo", 0) @@ -52,6 +52,11 @@ def as_ansitable(self, *args, **kwargs): class Test_Collection: + # _window_ranges yields (start, end) pairs. None of these pairs ends on + # a timestamp registered in _partition_timestamp_keys, so they trigger + # no windows. + UNTRIGGERED_RANGES = [(1.0, 1.1), (1.1, 1.2), (1.2, 1.3)] + @pytest.fixture def table(self, *, app): return MyTable(app, name="name") @@ -229,7 +234,7 @@ async def test_last_closed_window__mock_ranges(self, *, table): assert table.last_closed_window == 0.0 table.window = Mock(name="window") - self.mock_ranges(table) + self.mock_ranges(table, self.UNTRIGGERED_RANGES) table._data = { ("boo", (1.1, 1.4)): "BOO", ("moo", (1.4, 1.6)): "MOO", @@ -329,7 +334,7 @@ async def test_del_old_keys__mock_ranges(self, *, table): on_window_close = table._on_window_close = AsyncMock(name="on_window_close") table.window = Mock(name="window") - self.mock_ranges(table) + self.mock_ranges(table, self.UNTRIGGERED_RANGES) table._data = { ("boo", (1.1, 1.4)): "BOO", ("moo", (1.4, 1.6)): "MOO", @@ -440,7 +445,7 @@ async def test_del_old_keys_non_async_cb__mock_ranges(self, *, table): on_window_close = table._on_window_close = Mock(name="on_window_close") table.window = Mock(name="window") - self.mock_ranges(table) + self.mock_ranges(table, self.UNTRIGGERED_RANGES) table._data = { ("boo", (1.1, 1.4)): "BOO", ("moo", (1.4, 1.6)): "MOO", @@ -489,6 +494,65 @@ def is_stale(timestamp, latest_timestamp): assert not table.data + @pytest.mark.asyncio + async def test_del_old_keys__triggered_windows_are_aggregated(self, *, table): + # _partition_timestamp_keys is keyed by (partition, range_end), + # so every range overlapping the expiring timestamp must be found + # and its data handed to on_window_close. + on_window_close = table._on_window_close = AsyncMock(name="on_window_close") + + table.window = Mock(name="window") + table.window.stale.side_effect = lambda timestamp, latest: timestamp <= 20.0 + # the two windows that overlap the expiring timestamp (20.0) + self.mock_ranges(table, [(10.0, 20.0), (15.0, 25.0)]) + table._data = { + ("k", (10.0, 20.0)): ["e1", "e2"], + ("k", (15.0, 25.0)): ["e3"], + } + table._partition_timestamps = {TP1: [20.0]} + table._partition_timestamp_keys = { + (TP1, 20.0): {("k", (10.0, 20.0))}, + (TP1, 25.0): {("k", (15.0, 25.0))}, + } + + await table._del_old_keys() + + on_window_close.assert_called_once_with( + ("k", (10.0, 20.0)), + ["e1", "e2", "e3"], + ) + assert table.data == {("k", (15.0, 25.0)): ["e3"]} + assert table.last_closed_window == 10.0 + + @pytest.mark.asyncio + async def test_del_old_keys__aggregates_with_real_window(self, *, app): + # end to end: the keys written by _maybe_set_key_ttl must be the + # keys _del_old_keys looks up for the real window ranges. + on_window_close = AsyncMock(name="on_window_close") + window = HoppingWindow(size=10, step=5, expires=10) + table = MyTable( + app, + name="name", + window=window, + on_window_close=on_window_close, + ) + partition = 0 + first, second = window.ranges(20.0) + table._data = { + ("k", first): ["e1", "e2"], + ("k", second): ["e3"], + } + for window_range in (first, second): + table._maybe_set_key_ttl(("k", window_range), partition) + + await table._del_old_keys() + + assert not table.data + assert on_window_close.call_args_list == [ + call(("k", first), ["e1", "e2", "e3"]), + call(("k", second), ["e3"]), + ] + @pytest.mark.asyncio async def test_on_window_close__default(self, *, table): assert table._on_window_close is None diff --git a/tests/unit/tables/test_recovery.py b/tests/unit/tables/test_recovery.py index 81fcad8d5..8c274a6e5 100644 --- a/tests/unit/tables/test_recovery.py +++ b/tests/unit/tables/test_recovery.py @@ -1,3 +1,4 @@ +import asyncio from collections import Counter from unittest.mock import MagicMock, Mock @@ -5,7 +6,7 @@ from faust.tables.recovery import RebalanceAgain, Recovery, ServiceStopped from faust.types import TP -from tests.helpers import AsyncMock +from tests.helpers import AsyncMock, new_event TP1 = TP("foo", 6) TP2 = TP("bar", 3) @@ -13,6 +14,14 @@ TP4 = TP("xuz", 0) +class LoopBreak(BaseException): + """Raised by the fake changelog queue to break out of the slurp loop. + + Inherits from :class:`BaseException` so it is not swallowed by the + ``except Exception`` guarding the body of ``_slurp_changelogs``. + """ + + @pytest.fixture() def tables(): return Mock(name="tables") @@ -387,6 +396,82 @@ def test__is_changelog_tp(self, *, recovery, tables): assert recovery._is_changelog_tp(TP1) +class TestSlurpChangelogs: + @pytest.fixture() + def table(self): + return Mock( + name="table", + recovery_buffer_size=1000, + standby_buffer_size=1000, + on_changelog_event=AsyncMock(), + ) + + async def _slurp(self, recovery, tables, items): + # Feed the changelog queue with ``items`` (events, or exceptions to + # raise), then break out of the otherwise infinite loop. + tables.changelog_queue = Mock( + name="changelog_queue", + get=AsyncMock(side_effect=[*items, LoopBreak()]), + ) + with pytest.raises(LoopBreak): + await Recovery._slurp_changelogs(recovery) + + @pytest.mark.asyncio + async def test_unknown_tp_is_not_applied_to_previous_table( + self, *, recovery, tables, table, app + ): + recovery.active_tps.add(TP1) + recovery.tp_to_table[TP1] = table + known = new_event(app, topic=TP1.topic, partition=TP1.partition, offset=1) + unknown = new_event(app, topic=TP3.topic, partition=TP3.partition, offset=5) + + await self._slurp(recovery, tables, [known, unknown]) + + assert recovery.active_offsets[TP1] == 1 + # the event for the untracked partition must not end up in the + # offsets/buffer of the table handled in the previous iteration. + assert TP3 not in recovery.active_offsets + assert recovery.buffers[table] == [known] + table.on_changelog_event.assert_called_once_with(known) + + @pytest.mark.asyncio + async def test_unknown_tp_still_runs_end_of_loop(self, *, recovery, tables, app): + tables.standbys_ready = False + recovery.in_recovery = True + unknown = new_event(app, topic=TP3.topic, partition=TP3.partition, offset=5) + + await self._slurp(recovery, tables, [unknown]) + + # Everything after the "apply event" block must still run for an + # event on an untracked partition: recovery end is signalled... + assert not recovery.in_recovery + # ...and the standby bookkeeping happens. + tables.on_standbys_ready.assert_called_once_with() + + @pytest.mark.asyncio + async def test_detect_aborted_tx__partition_without_position( + self, *, recovery, tables, app + ): + positions = {TP1: None, TP2: 30} + app.in_transaction = True + app.consumer = Mock( + name="consumer", + position=AsyncMock(side_effect=lambda tp: positions[tp]), + ) + recovery.active_highwaters.update({TP1: 10, TP2: 20}) + recovery.active_offsets.update({TP1: 5, TP2: 5}) + + # two timeouts in a row trigger the aborted transaction detection. + await self._slurp( + recovery, tables, [asyncio.TimeoutError(), asyncio.TimeoutError()] + ) + + # TP1 has no position yet, so it is skipped, and that must not + # prevent TP2 from being fixed up to its highwater. + assert recovery.active_offsets[TP1] == 5 + assert recovery.active_offsets[TP2] == 20 + + @pytest.mark.parametrize( "highwaters,offsets,needs_recovery,total,remaining", [ diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py index 57ecd84d0..b2f49100f 100644 --- a/tests/unit/test_auth.py +++ b/tests/unit/test_auth.py @@ -148,6 +148,31 @@ def test_constructor(self): cadata="moo", ) + def test_constructor__default_purpose_is_SERVER_AUTH(self): + with patch("faust.auth.ssl.create_default_context") as cdc: + c = SSLCredentials(cafile="/foo/bar/ca.file") + assert c.context is cdc.return_value + cdc.assert_called_once_with( + purpose=ssl.Purpose.SERVER_AUTH, + cafile="/foo/bar/ca.file", + capath=None, + cadata=None, + ) + + def test_constructor__no_arguments_builds_verifying_context(self): + # Not mocked: builds a real context, so this exercises the + # ``ssl.create_default_context`` argument validation. + c = SSLCredentials() + assert isinstance(c.context, ssl.SSLContext) + # SERVER_AUTH means we verify the broker's certificate. + assert c.context.check_hostname is True + assert c.context.verify_mode == ssl.CERT_REQUIRED + + def test_constructor__purpose_is_overridable(self): + c = SSLCredentials(purpose=ssl.Purpose.CLIENT_AUTH) + assert isinstance(c.context, ssl.SSLContext) + assert c.context.check_hostname is False + def test_having_context(self): context = Mock(name="context") c = SSLCredentials(context) diff --git a/tests/unit/test_faust.py b/tests/unit/test_faust.py index b1b8ac608..0eb07e789 100644 --- a/tests/unit/test_faust.py +++ b/tests/unit/test_faust.py @@ -1,3 +1,67 @@ +from itertools import takewhile + +import pytest + +import faust import faust.exceptions # noqa: F401 import faust.transport.base # noqa: F401 import faust.transport.drivers.aiokafka # noqa: F401 +from faust import VersionInfo + + +@pytest.mark.parametrize( + "version_string,expected", + [ + ("0.11.5", VersionInfo(0, 11, 5, None, None)), + ("v0.11.5", VersionInfo(0, 11, 5, None, None)), + # pre/dev/post segments land in releaselevel + serial separately, + # the way the same fields work on sys.version_info. + ("0.11.5.dev1+g1234", VersionInfo(0, 11, 5, "dev", "1")), + ("0.11.5rc1", VersionInfo(0, 11, 5, "rc", "1")), + ("0.11.5a2", VersionInfo(0, 11, 5, "a", "2")), + ("0.11.5b3", VersionInfo(0, 11, 5, "b", "3")), + ("0.11.5.post2", VersionInfo(0, 11, 5, "post", "2")), + # VersionInfo has no field for a local segment or a fourth + # component, so they are dropped; faust.__version__ keeps them. + ("0.11.5+local.1", VersionInfo(0, 11, 5, None, None)), + ("1.2.3.4", VersionInfo(1, 2, 3, None, None)), + # fewer than three components + ("1.2", VersionInfo(1, 2, 0, None, None)), + ("2", VersionInfo(2, 0, 0, None, None)), + # anything PEP 440 cannot parse must not raise: it degrades to + # VersionInfo(0, 0, 0) carrying the raw string. + ("", VersionInfo(0, 0, 0, None, None)), + ("nonsense", VersionInfo(0, 0, 0, "nonsense", None)), + ("1.x.3", VersionInfo(0, 0, 0, "1.x.3", None)), + ], +) +def test_parse_version(version_string, expected): + assert faust._parse_version(version_string) == expected + + +def test_version_info_is_numeric(): + assert faust.VERSION is faust.version_info + assert isinstance(faust.version_info, VersionInfo) + assert isinstance(faust.version_info.major, int) + assert isinstance(faust.version_info.minor, int) + assert isinstance(faust.version_info.micro, int) + assert faust.version_info.releaselevel is None or isinstance( + faust.version_info.releaselevel, str + ) + assert faust.version_info.serial is None or isinstance( + faust.version_info.serial, str + ) + # serial is only meaningful alongside a releaselevel. + if faust.version_info.serial is not None: + assert faust.version_info.releaselevel is not None + + +def test_version_info_matches_version_string(): + leading = faust.__version__.lstrip("v").split("+")[0].split(".")[:3] + expected = [int(part) for part in takewhile(str.isdigit, leading)] + actual = [ + faust.version_info.major, + faust.version_info.minor, + faust.version_info.micro, + ] + assert actual[: len(expected)] == expected diff --git a/tests/unit/transport/drivers/test_confluent.py b/tests/unit/transport/drivers/test_confluent.py index 2b58d0d87..45e2c01ba 100644 --- a/tests/unit/transport/drivers/test_confluent.py +++ b/tests/unit/transport/drivers/test_confluent.py @@ -196,6 +196,15 @@ def test_verify_event_path(self, *, consumer): consumer.verify_event_path(303.3, TP1) consumer._thread.verify_event_path.assert_called_once_with(303.3, TP1) + def test_verify_event_path__real_thread_is_a_noop(self, *, consumer, cthread): + # Regression: the commit livelock detector calls verify_event_path + # on every tick (Consumer._commit_livelock_detector -> + # verify_all_partitions_active). With the real thread in place -- + # not a Mock -- this used to raise AttributeError because neither + # ConsumerThread nor ConfluentConsumerThread defined the method. + consumer._thread = cthread + assert consumer.verify_event_path(303.3, TP1) is None + class TestAsyncConsumer: @pytest.fixture() @@ -383,6 +392,11 @@ def test_key_partition(self, *, cthread, _consumer): partition = cthread.key_partition("topic", b"key") assert 0 <= partition < 3 + def test_verify_event_path__is_a_noop(self, *, cthread): + # Livelock detection is not implemented for this driver, but the + # method must exist -- Consumer.verify_event_path delegates to it. + assert cthread.verify_event_path(303.3, TP1) is None + def test_topic_partitions(self, *, cthread): assert cthread.topic_partitions("topic") is None @@ -441,17 +455,34 @@ async def test_create_topic__is_noop(self, *, producer): # create_topic short-circuits (XXX) -- must not raise. assert await producer.create_topic("topic", 3, 1) is None - def test_key_partition(self, *, producer): + def test_key_partition(self, *, producer, app): + # Regression: key_partition used to read + # ``self._producer_thread.producer``, which is the *Faust* producer + # and has no ``list_topics``. Use a real ProducerThread here so + # that mistake raises AttributeError instead of being absorbed by a + # Mock. + thread = ProducerThread(producer, loop=app.loop, beacon=producer.beacon) metadata = Mock(name="metadata") topic_meta = Mock() topic_meta.partitions = {0: 1, 1: 1} metadata.topics = {"topic": topic_meta} - producer._producer_thread.producer = Mock() - producer._producer_thread.producer.list_topics.return_value = metadata + thread._producer = Mock(name="confluent_kafka.Producer") + thread._producer.list_topics.return_value = metadata + producer._producer_thread = thread + tp = producer.key_partition("topic", b"key") + + thread._producer.list_topics.assert_called_once_with("topic") assert tp.topic == "topic" assert 0 <= tp.partition < 2 + def test_key_partition__not_started(self, *, producer, app): + thread = ProducerThread(producer, loop=app.loop, beacon=producer.beacon) + thread._producer = None + producer._producer_thread = thread + with pytest.raises(RuntimeError): + producer.key_partition("topic", b"key") + class TestProducerThread: @pytest.fixture()