From d33c312d32f13e5adbeea66142919a9c83387ef2 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 13:39:19 +0000 Subject: [PATCH 1/8] Give SSLCredentials a usable default TLS purpose `SSLCredentials.__init__` defaulted `purpose` to None and passed it straight into `ssl.create_default_context(purpose=...)`, which begins if not isinstance(purpose, _ASN1Object): raise TypeError(purpose) So `SSLCredentials()` and `SSLCredentials(cafile=...)` -- any call that does not supply an explicit `context` -- raised `TypeError: None`. The class could only ever be constructed by handing it a context built elsewhere, which defeats the cafile/capath/cadata parameters entirely. Default to `ssl.Purpose.SERVER_AUTH`: the same default `create_default_context()` itself applies, and the correct one for a client verifying a broker. It implies `check_hostname=True` and `verify_mode=CERT_REQUIRED`, which the tests assert. An explicitly passed `purpose` is still forwarded unchanged. Found by the type checker in #758 and marked `XXX` there. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- faust/auth.py | 11 +++-------- tests/unit/test_auth.py | 25 +++++++++++++++++++++++++ 2 files changed, 28 insertions(+), 8 deletions(-) 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/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) From 8bdaf3e8260a8e6fdcc88a7000b7e67da689e5d4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 13:39:19 +0000 Subject: [PATCH 2/8] Parse faust.version_info into the integers it declares `VersionInfo` declares `major`, `minor` and `micro` as ints, but the module splatted the regex groups `(prefix, version, suffix)` into them positionally: VersionInfo(major=None, minor='0.11.5', micro='') So `faust.version_info.major` was the `'v'` prefix or None, `.minor` was the whole version string, `.micro` was the suffix, and `.releaselevel` was always None -- every field wrong except by accident. Parse the dotted numbers properly instead, putting any non-numeric tail (`dev1+g1234`, `rc1`, a local segment) into `releaselevel`. Missing components pad with zero, and an unparsable version degrades to `VersionInfo(0, 0, 0, ...)` rather than raising, so `import faust` can never fail on the version string -- the old `RuntimeError('THIS IS A BROKEN RELEASE!')` branch is gone with it. This changes a public value, which is why #758 left it marked `XXX` rather than fixing it: code reading `faust.version_info.minor` as the version *string* must switch to `faust.__version__`, which is unchanged. Nothing in the repo consumes it and no docs reference it. `_parse_version` is injected into the lazy module's `__dict__` so it is reachable for testing; it is private and stays out of `__all__` and `dir()`. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- faust/__init__.py | 54 +++++++++++++++++++++++++++++----------- tests/unit/test_faust.py | 50 +++++++++++++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 15 deletions(-) diff --git a/faust/__init__.py b/faust/__init__.py index 132e784a5..e0b402063 100644 --- a/faust/__init__.py +++ b/faust/__init__.py @@ -28,7 +28,7 @@ else: from importlib.metadata import version -from typing import Any, Mapping, NamedTuple, Optional, Sequence, Tuple +from typing import Any, List, Mapping, NamedTuple, Optional, Sequence, Tuple __version__ = version("faust-streaming") __author__ = "Robinhood Markets, Inc." @@ -52,20 +52,43 @@ class VersionInfo(NamedTuple): # 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 +_VERSION_RE = re.compile(r"^v?(?P[^+]*)(?P.*)$") +_VERSION_PART_RE = re.compile(r"^(\d+)(.*)$") + + +def _parse_version(version_string: str) -> VersionInfo: + """Parse a version string into a :class:`VersionInfo` tuple. + + The leading dotted numbers become ``major``, ``minor`` and ``micro`` + (missing components default to ``0``), and whatever remains -- such as + the ``dev1+g1234`` of ``0.11.5.dev1+g1234`` -- becomes ``releaselevel``. + + Never raises: an unparsable version simply ends up as the + ``releaselevel`` of ``VersionInfo(0, 0, 0)``. + """ + match = _VERSION_RE.match(version_string) + if match is None: # pragma: no cover + version, suffix = version_string, "" + else: + version, suffix = match.group("version"), match.group("suffix") + numbers: List[int] = [] + parts = version.split(".") + while parts and len(numbers) < 3: + part_match = _VERSION_PART_RE.match(parts[0]) + if part_match is None: + break + numbers.append(int(part_match.group(1))) + parts.pop(0) + trailing = part_match.group(2) + if trailing: # e.g. the ``rc1`` of ``0.11.5rc1`` + parts.insert(0, trailing) + break + major, minor, micro = (numbers + [0, 0, 0])[:3] + releaselevel = ".".join(parts) + suffix + return VersionInfo(major, minor, micro, releaselevel or None) + + +VERSION = version_info = _parse_version(__version__) del re @@ -307,5 +330,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/tests/unit/test_faust.py b/tests/unit/test_faust.py index b1b8ac608..f67cafbd6 100644 --- a/tests/unit/test_faust.py +++ b/tests/unit/test_faust.py @@ -1,3 +1,53 @@ +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)), + ("0.11.5.dev1+g1234", VersionInfo(0, 11, 5, "dev1+g1234", None)), + ("0.11.5rc1", VersionInfo(0, 11, 5, "rc1", None)), + ("0.11.5+local.1", VersionInfo(0, 11, 5, "+local.1", None)), + ("1.2.3.4", VersionInfo(1, 2, 3, "4", None)), + # fewer than three components + ("1.2", VersionInfo(1, 2, 0, None, None)), + ("2", VersionInfo(2, 0, 0, None, None)), + # non-numeric components must not raise + ("", VersionInfo(0, 0, 0, None, None)), + ("nonsense", VersionInfo(0, 0, 0, "nonsense", None)), + ("1.x.3", VersionInfo(1, 0, 0, "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 + ) + + +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 From c87ee2d8e9c6d56ea693a9d846599eb46d0a5343 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 13:39:35 +0000 Subject: [PATCH 3/8] Look up window keys the way they are stored `Collection._del_old_keys` read `_partition_timestamp_keys` with self._partition_timestamp_keys.get((partition, window_range)) where `window_range` is the whole `(start, end)` tuple. The map is keyed `(partition, range_end)` -- an `(int, float)` pair -- written that way by `_maybe_set_key_ttl` and read that way by `_maybe_del_key_ttl`. So the lookup could never hit. `triggered_windows` was always `[None, ...]`, `window_data` stayed empty, and `on_window_close` was only ever handed the raw per-key value instead of the aggregated window data it exists to receive. Use `(partition, window_range[1])`, matching the writer. This is user-visible: applications with an `on_window_close` handler will start receiving the aggregated data the API always promised. That is why #758 marked it `XXX` instead of fixing it. Two existing tests relied on `mock_ranges` returning bare floats, which is not what `_window_ranges` yields; they now pass real `(start, end)` tuples. Their assertions are unchanged and the ranges still match nothing in `_partition_timestamp_keys`, so they continue to cover the untriggered path. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- faust/tables/base.py | 19 ++------- tests/unit/tables/test_base.py | 72 ++++++++++++++++++++++++++++++++-- 2 files changed, 71 insertions(+), 20 deletions(-) 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/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 From d25e81cec4172e05aa05199cb195e1d9332dae74 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 13:39:35 +0000 Subject: [PATCH 4/8] Stop applying changelog events to the wrong table Two bugs in `Recovery`, both found by the type checker in #758. `_slurp_changelogs` classifies each event's TP as active or standby and binds `table`, `offsets` and `bufsize` accordingly. The `else:` branch for a TP that is neither only logged `"recovery unknown topic"` and fell through -- so the event was applied using the *previous* iteration's bindings: written into an unrelated table's buffer and offset map, and passed to that table's `on_changelog_event`. On the first event of the loop there is nothing bound yet, so it raised `UnboundLocalError` instead. Skip applying an event for an untracked TP. Note a bare `continue` would be wrong: the statements at the bottom of the loop body -- `_maybe_signal_recovery_end()` and the standby-ready bookkeeping -- must keep running on every iteration, or recovery-end signalling loses a trigger. Only the event-application block is skipped. `detect_aborted_tx` compared `await self.app.consumer.position(tp) >= highwater` unguarded. `ConsumerT.position` is `Optional[int]` and does return None when a partition has no position yet, so that raised TypeError -- swallowed by the caller's `except Exception`, which then silently skipped the aborted-transaction fixup for every remaining partition in the loop. Skip a TP with no position. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- faust/tables/recovery.py | 112 +++++++++++++---------------- tests/unit/tables/test_recovery.py | 87 +++++++++++++++++++++- 2 files changed, 135 insertions(+), 64 deletions(-) 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/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", [ From 59339ca45b7ae1fd52b6a9f7bcff6ab33eaa0f8e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 13:39:51 +0000 Subject: [PATCH 5/8] Let require_app = False commands run without an app `_app_from_str` returns None for a `require_app = False` command invoked without `-A` -- that is the documented escape hatch, and `faust completion` is exactly such a command. `_finalize_app` handed that None straight back, and `AppCommand.__init__` then did self.key_serializer = key_serializer or self.app.conf.key_serializer unconditionally, so the command died with `AttributeError: 'NoneType' object has no attribute 'conf'`. The escape hatch was unusable: `faust completion` could not run without the `-A` it is written not to need. Make `AppCommand` tolerate having no app. `self.app` becomes a property over an `Optional[AppT]`, the serializer defaults fall back to None when there is no app, and `on_stop` and `blocking_timeout` no longer assume one. Behaviour with an app present is unchanged, and a command with `require_app = True` still gets the same `UsageError` from `_app_from_str` as before. Found by the type checker in #758 and marked `XXX` there. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- faust/cli/base.py | 59 +++++++++++++++++++++++++++---------- tests/unit/cli/test_base.py | 52 ++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 16 deletions(-) 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/tests/unit/cli/test_base.py b/tests/unit/cli/test_base.py index 8b81b1c04..3b2af06d3 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,43 @@ 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_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" From f3b2097579f38368988de6b71b34bf99293147d2 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 13:39:51 +0000 Subject: [PATCH 6/8] Fix two AttributeErrors in the confluent driver `Producer.key_partition` called `self._producer_thread.producer.list_topics()`. `ProducerThread.producer` is the *Faust* Producer; the confluent_kafka handle is `ProducerThread._producer`. Faust producers have no `list_topics`, so the method raised AttributeError and was dead on arrival. Read the confluent producer instead. `Consumer.verify_event_path` delegated to `self._thread.verify_event_path(...)`, but neither `ConsumerThread` nor `ConfluentConsumerThread` defines it, so the commit-livelock detector (`_commit_livelock_detector` -> `verify_all_partitions_active`) raised AttributeError on every tick. Add the no-op to `ConfluentConsumerThread`, matching the documented no-op stub the base `faust.transport.consumer.Consumer.verify_event_path` already is. This makes livelock detection inert for this driver rather than raising -- a real implementation is separate work. Both were found by the type checker in #758 and marked `XXX` there. Note these tests do not run here or in CI: tests/unit/transport/drivers/test_confluent.py starts with `pytest.importorskip("confluent_kafka")`, and confluent-kafka is the optional `faust[ckafka]` extra, which the CI test environment does not install. The fixes are verified by reading the class definitions, not by an executed test. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- faust/transport/drivers/confluent.py | 34 ++++++++--------- .../unit/transport/drivers/test_confluent.py | 37 +++++++++++++++++-- 2 files changed, 49 insertions(+), 22 deletions(-) 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/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() From ff8e5ff9aeb0c61e3bb5202872dada55c40552d7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 17:55:23 +0000 Subject: [PATCH 7/8] Parse the version with packaging instead of hand-rolled regexes `_parse_version` carried two regexes and a hand-written loop to pull major/minor/micro off `__version__`. That is a PEP 440 parser, and there is no reason for this repo to maintain one: * the stdlib has no version parser -- `importlib.metadata.version()` returns the string only, which is why the parsing existed in the first place; * `distutils.version.LooseVersion`, the historical answer, was removed from the stdlib in Python 3.12, and this package supports 3.10 through 3.14; * `packaging.version.Version` is the PyPA reference implementation, the same parser pip and setuptools use. `packaging` is not a new install for anyone: aiokafka, a core dependency, already requires it unconditionally. It was only listed in `requirements/dist.txt` though, so it is added to `requirements.txt` -- faust imports it directly now and must not rely on a transitive dependency staying put. The floor is 20.0, where `Version.major`/`.minor`/`.micro` landed. Behaviour changes, all in the direction of the field names: * A pre/dev/post segment is now split across `releaselevel` and `serial` instead of being concatenated into `releaselevel`. `0.11.5rc1` gives `releaselevel='rc', serial='1'`; it gave `releaselevel='rc1', serial=None`. `VersionInfo` mirrors `sys.version_info`, where those two fields mean exactly this, and `serial` was previously never populated at all. * A local segment and any fourth component are dropped, because `VersionInfo` has no field for either. `0.11.5.dev1+g1234` gives `VersionInfo(0, 11, 5, 'dev', '1')`; the `+g1234` is still available in full on `faust.__version__`, which is untouched. * A string PEP 440 cannot parse degrades to `VersionInfo(0, 0, 0)` carrying the raw string, so `import faust` still cannot fail on a bad version -- same guarantee as before, now via `except InvalidVersion`. One edge case moves: `1.x.3` gave `VersionInfo(1, 0, 0, 'x.3')` and now gives `VersionInfo(0, 0, 0, '1.x.3')`, since packaging rejects it outright rather than salvaging the leading number. Folded into this PR rather than sent separately because #760 already changes `version_info` from a broken value to a correct one; doing both at once moves the public shape once instead of twice. Net: -2 regexes, -18 lines, and the `re` import drops out of `faust/__init__.py`. Verified: `mypy -p faust` clean (packaging ships py.typed, so these are real types rather than Any), pinned flake8/isort/black clean, suites 2234 -> 2237 passed, 4 skipped. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- faust/__init__.py | 62 +++++++++++++++++------------------ requirements/requirements.txt | 6 ++++ tests/unit/test_faust.py | 26 +++++++++++---- 3 files changed, 56 insertions(+), 38 deletions(-) diff --git a/faust/__init__.py b/faust/__init__.py index e0b402063..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 @@ -28,7 +27,9 @@ else: from importlib.metadata import version -from typing import Any, List, Mapping, NamedTuple, Optional, Sequence, Tuple +from typing import Any, Mapping, NamedTuple, Optional, Sequence, Tuple + +from packaging.version import InvalidVersion, Version __version__ = version("faust-streaming") __author__ = "Robinhood Markets, Inc." @@ -50,46 +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. -_VERSION_RE = re.compile(r"^v?(?P[^+]*)(?P.*)$") -_VERSION_PART_RE = re.compile(r"^(\d+)(.*)$") - - def _parse_version(version_string: str) -> VersionInfo: """Parse a version string into a :class:`VersionInfo` tuple. - The leading dotted numbers become ``major``, ``minor`` and ``micro`` - (missing components default to ``0``), and whatever remains -- such as - the ``dev1+g1234`` of ``0.11.5.dev1+g1234`` -- becomes ``releaselevel``. + 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: an unparsable version simply ends up as the + Never raises: a version :mod:`packaging` cannot parse ends up as the ``releaselevel`` of ``VersionInfo(0, 0, 0)``. """ - match = _VERSION_RE.match(version_string) - if match is None: # pragma: no cover - version, suffix = version_string, "" + 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: - version, suffix = match.group("version"), match.group("suffix") - numbers: List[int] = [] - parts = version.split(".") - while parts and len(numbers) < 3: - part_match = _VERSION_PART_RE.match(parts[0]) - if part_match is None: - break - numbers.append(int(part_match.group(1))) - parts.pop(0) - trailing = part_match.group(2) - if trailing: # e.g. the ``rc1`` of ``0.11.5rc1`` - parts.insert(0, trailing) - break - major, minor, micro = (numbers + [0, 0, 0])[:3] - releaselevel = ".".join(parts) + suffix - return VersionInfo(major, minor, micro, releaselevel or None) + releaselevel, serial = None, None + return VersionInfo(parsed.major, parsed.minor, parsed.micro, releaselevel, serial) VERSION = version_info = _parse_version(__version__) -del re # This is here to support setting the --datadir argument 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/test_faust.py b/tests/unit/test_faust.py index f67cafbd6..0eb07e789 100644 --- a/tests/unit/test_faust.py +++ b/tests/unit/test_faust.py @@ -14,17 +14,25 @@ [ ("0.11.5", VersionInfo(0, 11, 5, None, None)), ("v0.11.5", VersionInfo(0, 11, 5, None, None)), - ("0.11.5.dev1+g1234", VersionInfo(0, 11, 5, "dev1+g1234", None)), - ("0.11.5rc1", VersionInfo(0, 11, 5, "rc1", None)), - ("0.11.5+local.1", VersionInfo(0, 11, 5, "+local.1", None)), - ("1.2.3.4", VersionInfo(1, 2, 3, "4", 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)), - # non-numeric components must not raise + # 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(1, 0, 0, "x.3", None)), + ("1.x.3", VersionInfo(0, 0, 0, "1.x.3", None)), ], ) def test_parse_version(version_string, expected): @@ -40,6 +48,12 @@ def test_version_info_is_numeric(): 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(): From a05057cccb106fee4d9b87858849d97270669a4e Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 18:02:10 +0000 Subject: [PATCH 8/8] Cover the AppCommand.app setter Codecov flagged the one line of this PR's diff that no test reaches: `faust/cli/base.py:874`, the body of the `app` setter. It matters more than a coverage percentage suggests. `app` was a plain writable attribute until this PR turned it into a property over an `Optional[AppT]`; the setter exists purely so `command.app = ...` keeps working for anything that assigned to it. Nothing else in the suite exercises it, so a later refactor could drop the setter and turn every such assignment into `AttributeError` with the tests still green. `test_app__raises_when_missing` already covers the getter's raise path; this covers the round trip -- assignment lands on `_app`, and the property reads it back. faust/cli/base.py patch coverage 96% -> 100%; the file's remaining misses (86, 480-489) are pre-existing and outside this PR's diff. Suites 2237 -> 2238 passed, 4 skipped; `mypy -p faust` clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7 --- tests/unit/cli/test_base.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/tests/unit/cli/test_base.py b/tests/unit/cli/test_base.py index 3b2af06d3..01c316ba5 100644 --- a/tests/unit/cli/test_base.py +++ b/tests/unit/cli/test_base.py @@ -541,6 +541,16 @@ 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