From 835ce656307592400491e2c0cba1ed49dfe35283 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 18:48:13 +0000 Subject: [PATCH 1/2] test: re-enable recoverable skipped tests; fix latent LazySpan bug Sweep the suite's legacy `@pytest.mark.skip("Needs fixing")` tests and re-enable the ones that can be made green, fixing a real bug found along the way. * fix(aiokafka): bind the lazy span's replacement `finish`. `LazySpan.finish` was assigned to `span.finish` as a raw (unbound) function, so the later `span.finish()` call raised "missing 1 required positional argument: 'self'". Lazy spans are used whenever tracing is enabled, so this was a latent crash; the skipped `test_transform_span_*` tests were hiding it. Re-enables the four transform-span tests. * test(aiokafka): the two stream-idle VEP tests called the 2-arg `_make_slow_processing_error`; it now takes `setting`/`current_value` and bakes the explanation into the message, so `log.error` no longer receives those as kwargs. Updated the expectations and re-enabled them. * test(agent): `test_execute_actor__cancelled_running` (was "Fix is TBD") passes as-is now -- skip rot -- re-enabled. The remaining previously-skipped tests need per-test work beyond a re-enable and stay skipped, but with precise reasons replacing "Needs fixing": the fetch-timeout VEP asserts (driver behaviour drift) and the `_commit` mock tests in test_aiokafka.py. Tests needing live infrastructure or a dependency API refresh (redis cache backend, rocksdb `Options`, the transaction-manager mocks, the CLI compat-option test, tests/consistency's stale aiokafka import) are left as they were. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- faust/transport/drivers/aiokafka.py | 9 +++- tests/unit/agents/test_agent.py | 1 - tests/unit/transport/drivers/test_aiokafka.py | 47 ++++++++++++------- 3 files changed, 39 insertions(+), 18 deletions(-) diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py index 3ebaf6b45..2e4799f58 100644 --- a/faust/transport/drivers/aiokafka.py +++ b/faust/transport/drivers/aiokafka.py @@ -629,7 +629,14 @@ class LazySpan(cls): def finish(self) -> None: consumer._span_finish(span) - span._real_finish, span.finish = span.finish, LazySpan.finish + # Bind the replacement to ``span`` -- assigning the raw + # ``LazySpan.finish`` function leaves it unbound, so the later + # ``span.finish()`` call raises "missing 1 required positional + # argument: 'self'". + span._real_finish, span.finish = ( + span.finish, + LazySpan.finish.__get__(span), + ) def _span_finish(self, span: opentracing.Span) -> None: assert self._consumer is not None diff --git a/tests/unit/agents/test_agent.py b/tests/unit/agents/test_agent.py index 54f881f68..c9206cff2 100644 --- a/tests/unit/agents/test_agent.py +++ b/tests/unit/agents/test_agent.py @@ -469,7 +469,6 @@ async def test_execute_actor__cancelled_stopped(self, *, agent): await agent._execute_actor(coro, Mock(name="aref", autospec=Actor)) coro.assert_awaited() - @pytest.mark.skip(reason="Fix is TBD") @pytest.mark.asyncio async def test_execute_actor__cancelled_running(self, *, agent): agent._on_error = AsyncMock(name="on_error") diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py index d36665704..fc089095b 100644 --- a/tests/unit/transport/drivers/test_aiokafka.py +++ b/tests/unit/transport/drivers/test_aiokafka.py @@ -437,13 +437,17 @@ def test_log_slow_processing_stream( ) -@pytest.mark.skip("Needs fixing") class Test_VEP_no_fetch_since_start(Test_verify_event_path_base): def test_just_started(self, *, cthread, now, tp, logger): self._set_started(now - 2.0) assert cthread.verify_event_path(now, tp) is None logger.error.assert_not_called() + @pytest.mark.skip( + reason="verify_event_path no longer emits the fetch-timeout " + "error() under this condition; predates that behaviour change and " + "needs re-deriving against the current driver" + ) def test_timed_out(self, *, cthread, now, tp, logger): self._set_started( now - cthread.tp_fetch_request_timeout_secs * 2, @@ -456,7 +460,6 @@ def test_timed_out(self, *, cthread, now, tp, logger): ) -@pytest.mark.skip("Needs fixing") class Test_VEP_no_response_since_start(Test_verify_event_path_base): def test_just_started(self, *, cthread, _consumer, now, tp, logger): self._set_last_request(now - 5.0) @@ -464,6 +467,11 @@ def test_just_started(self, *, cthread, _consumer, now, tp, logger): assert cthread.verify_event_path(now, tp) is None logger.error.assert_not_called() + @pytest.mark.skip( + reason="verify_event_path no longer emits the fetch-timeout " + "error() under this condition; predates that behaviour change and " + "needs re-deriving against the current driver" + ) def test_timed_out(self, *, cthread, _consumer, now, tp, logger): assert cthread.verify_event_path(now, tp) is None self._set_last_request(now - 5.0) @@ -501,7 +509,6 @@ def test_timed_out(self, *, cthread, now, tp, logger, _consumer): ) -@pytest.mark.skip("Needs fixing") class Test_VEP_no_recent_response(Test_verify_event_path_base): def test_recent_response(self, *, cthread, now, tp, logger): self._set_last_request(now - 10.0) @@ -509,6 +516,11 @@ def test_recent_response(self, *, cthread, now, tp, logger): assert cthread.verify_event_path(now, tp) is None logger.error.assert_not_called() + @pytest.mark.skip( + reason="verify_event_path no longer emits the fetch-timeout " + "error() under this condition; predates that behaviour change and " + "needs re-deriving against the current driver" + ) def test_timed_out(self, *, cthread, now, tp, logger): self._set_last_request(now - 10.0) self._set_last_response(now - cthread.tp_fetch_response_timeout_secs * 2) @@ -608,7 +620,6 @@ def test_main(self, *, cthread, now, tp, logger): logger.error.assert_not_called() -@pytest.mark.skip("Needs fixing") class Test_VEP_stream_idle_highwater_no_inbound(Test_verify_event_path_base): highwater = 20 committed_offset = 10 @@ -630,13 +641,13 @@ def test_timed_out_since_start(self, *, app, cthread, now, tp, logger): expected_message = cthread._make_slow_processing_error( mod.SLOW_PROCESSING_STREAM_IDLE_SINCE_START, [mod.SLOW_PROCESSING_CAUSE_STREAM, mod.SLOW_PROCESSING_CAUSE_AGENT], + "stream_processing_timeout", + app.conf.stream_processing_timeout, ) logger.error.assert_called_once_with( expected_message, tp, ANY, - setting="stream_processing_timeout", - current_value=app.conf.stream_processing_timeout, ) def test_has_inbound(self, *, app, cthread, now, tp, logger): @@ -658,13 +669,13 @@ def test_inbound_timed_out(self, *, app, cthread, now, tp, logger): expected_message = cthread._make_slow_processing_error( mod.SLOW_PROCESSING_STREAM_IDLE, [mod.SLOW_PROCESSING_CAUSE_STREAM, mod.SLOW_PROCESSING_CAUSE_AGENT], + "stream_processing_timeout", + app.conf.stream_processing_timeout, ) logger.error.assert_called_once_with( expected_message, tp, ANY, - setting="stream_processing_timeout", - current_value=app.conf.stream_processing_timeout, ) @@ -907,7 +918,6 @@ def test__start_span(self, *, cthread, app): def test_trace_category(self, *, cthread, app): assert cthread.trace_category == f"{app.conf.name}-_aiokafka" - @pytest.mark.skip("Needs fixing") def test_transform_span_lazy(self, *, cthread, app, tracer): cthread._consumer = Mock(name="_consumer") cthread._consumer._coordinator.generation = -1 @@ -920,7 +930,6 @@ def test_transform_span_lazy(self, *, cthread, app, tracer): cthread.on_generation_id_known() assert not pending - @pytest.mark.skip("Needs fixing") def test_transform_span_flush_spans(self, *, cthread, app, tracer): cthread._consumer = Mock(name="_consumer") cthread._consumer._coordinator.generation = -1 @@ -939,7 +948,6 @@ def test_span_without_operation_name(self, *, cthread): assert cthread._on_span_cancelled_early(span) is None - @pytest.mark.skip("Needs fixing") def test_transform_span_lazy_no_consumer(self, *, cthread, app, tracer): cthread._consumer = Mock(name="_consumer") cthread._consumer._coordinator.generation = -1 @@ -953,7 +961,6 @@ def test_transform_span_lazy_no_consumer(self, *, cthread, app, tracer): span = pending.popleft() cthread._on_span_generation_known(span) - @pytest.mark.skip("Needs fixing") def test_transform_span_eager(self, *, cthread, app, tracer): cthread._consumer = Mock(name="_consumer") cthread._consumer._coordinator.generation = 10 @@ -1048,8 +1055,11 @@ async def test_commit(self, *, cthread, _consumer): with self.assert_calls_thread(cthread, _consumer, cthread._commit, offsets): await cthread.commit(offsets) - @pytest.mark.skip("Needs fixing") @pytest.mark.asyncio + @pytest.mark.skip( + reason="stale mock: _commit no longer forwards offsets to the " + "wrapped AIOKafkaConsumer.commit in this harness; needs a mock refresh" + ) async def test__commit(self, *, cthread, _consumer): offsets = {TP1: 1001} cthread._consumer = _consumer @@ -1059,15 +1069,17 @@ async def test__commit(self, *, cthread, _consumer): {TP1: OffsetAndMetadata(1001, "")}, ) - @pytest.mark.skip("Needs fixing") @pytest.mark.asyncio async def test__commit__already_rebalancing(self, *, cthread, _consumer): cthread._consumer = _consumer _consumer.commit.side_effect = CommitFailedError("already rebalanced") assert not (await cthread._commit({TP1: 1001})) - @pytest.mark.skip("Needs fixing") @pytest.mark.asyncio + @pytest.mark.skip( + reason="stale mock: _commit no longer forwards offsets to the " + "wrapped AIOKafkaConsumer.commit in this harness; needs a mock refresh" + ) async def test__commit__CommitFailedError(self, *, cthread, _consumer): cthread._consumer = _consumer exc = _consumer.commit.side_effect = CommitFailedError("xx") @@ -1077,8 +1089,11 @@ async def test__commit__CommitFailedError(self, *, cthread, _consumer): cthread.crash.assert_called_once_with(exc) cthread.supervisor.wakeup.assert_called_once() - @pytest.mark.skip("Needs fixing") @pytest.mark.asyncio + @pytest.mark.skip( + reason="stale mock: _commit no longer forwards offsets to the " + "wrapped AIOKafkaConsumer.commit in this harness; needs a mock refresh" + ) async def test__commit__IllegalStateError(self, *, cthread, _consumer): cthread._consumer = _consumer cthread.assignment = Mock() From db3c72db35ca9802f5a9219acbd0522cc09f39d8 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 19:26:32 +0000 Subject: [PATCH 2/2] test: re-enable remaining recoverable skipped tests (aiokafka + CLI) Continues the skipped-test sweep from the parent branch. aiokafka (tests/unit/transport/drivers/test_aiokafka.py): * Test_VEP_no_fetch_since_start::test_timed_out -- the base _consumer fixture stamps a current poll timestamp, so the NO_FETCH_SINCE_START branch (fires only when the partition has no poll timestamp yet) could never trigger. Set state_value().timestamp = None (the real "no fetch request ever sent" condition) and tighten the asserted tp. Now green. * test__commit / __CommitFailedError / __IllegalStateError -- _commit filters offsets by self.assignment() before building the payload; the fixture's assignment didn't contain TP1, so every offset was filtered out (commit({})). Point the mocked assignment at TP1 so offsets flow through. Also dropped the stale supervisor.wakeup() assertions -- those calls were intentionally removed from _commit's error handlers in 2020 (commit 8d6758f1); the crash(exc)/returns-False assertions are kept. CLI (tests/unit/cli/test_base.py): * test_compat_option -- the old test asserted on a Mock (option(ctx)._callback is a Mock attribute, never 33). compat_option wires the real callback via the click `callback=` kwarg; grab it from opt.kwargs["callback"] and assert its actual behaviour against faust/cli/base.py: ensure_object(State), value returned unchanged, stored on the State only when the previous value is None and the value differs from the option default. Two response-staleness VEP tests stay skipped, now with an accurate reason: SLOW_PROCESSING_NO_RESPONSE_SINCE_START and SLOW_PROCESSING_NO_RECENT_RESPONSE are defined in the driver but never emitted (the broker-response-staleness tracking was dropped), so they cannot pass without re-adding that source feature -- out of scope for a test-only change. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- tests/unit/cli/test_base.py | 30 ++++++++--- tests/unit/transport/drivers/test_aiokafka.py | 54 +++++++++---------- 2 files changed, 48 insertions(+), 36 deletions(-) diff --git a/tests/unit/cli/test_base.py b/tests/unit/cli/test_base.py index 5fd3aeab0..8b81b1c04 100644 --- a/tests/unit/cli/test_base.py +++ b/tests/unit/cli/test_base.py @@ -13,6 +13,7 @@ from faust.cli import AppCommand, Command, call_command from faust.cli.base import ( DEFAULT_LOGLEVEL, + State, _Group, _prepare_cli, argument, @@ -80,18 +81,35 @@ def test_call_command__custom_ins(): assert stderr is o_err -@pytest.mark.skip("Needs fixing") def test_compat_option(): - option = compat_option("--foo", default=1, state_key="foo") + # compat_option returns our `option` wrapper; the real callback is + # wired into click.option through the `callback` kwarg (it is a + # closure, not an attribute on the decorated function). + opt = compat_option("--foo", default=1, state_key="foo") + callback = opt.kwargs["callback"] + ctx = Mock(name="ctx") param = Mock(name="param") + param.default = 1 state = ctx.ensure_object.return_value + + # The callback fetches (or creates) the State object from the ctx. state.foo = 33 - print(dir(option(ctx))) - assert option(ctx)._callback(ctx, param, None) == 33 - assert option(ctx)._callback(ctx, param, 44) == 44 + assert callback(ctx, param, None) is None + ctx.ensure_object.assert_called_once_with(State) + # A previously set state value is left untouched. + assert state.foo == 33 + + # When the state value is unset and a non-default value is passed, + # the value is stored on the state and returned unchanged. + state.foo = None + assert callback(ctx, param, 44) == 44 + assert state.foo == 44 + + # A value equal to the option default is returned but not stored. state.foo = None - assert option(ctx)._callback(ctx, param, 44) == 44 + assert callback(ctx, param, 1) == 1 + assert state.foo is None def test_find_app(): diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py index fc089095b..0f7173f4f 100644 --- a/tests/unit/transport/drivers/test_aiokafka.py +++ b/tests/unit/transport/drivers/test_aiokafka.py @@ -443,19 +443,20 @@ def test_just_started(self, *, cthread, now, tp, logger): assert cthread.verify_event_path(now, tp) is None logger.error.assert_not_called() - @pytest.mark.skip( - reason="verify_event_path no longer emits the fetch-timeout " - "error() under this condition; predates that behaviour change and " - "needs re-deriving against the current driver" - ) - def test_timed_out(self, *, cthread, now, tp, logger): + def test_timed_out(self, *, cthread, now, tp, logger, _consumer): + # No fetch request has ever been sent: aiokafka has not stamped a + # poll timestamp on the partition state yet. + state = ( + _consumer._fetcher._subscriptions.subscription.assignment.state_value.return_value # noqa: E501 + ) + state.timestamp = None self._set_started( now - cthread.tp_fetch_request_timeout_secs * 2, ) assert cthread.verify_event_path(now, tp) is None logger.error.assert_called_with( mod.SLOW_PROCESSING_NO_FETCH_SINCE_START, - ANY, + tp, ANY, ) @@ -468,9 +469,12 @@ def test_just_started(self, *, cthread, _consumer, now, tp, logger): logger.error.assert_not_called() @pytest.mark.skip( - reason="verify_event_path no longer emits the fetch-timeout " - "error() under this condition; predates that behaviour change and " - "needs re-deriving against the current driver" + reason="SOURCE BUG: faust/transport/drivers/aiokafka.py defines " + "SLOW_PROCESSING_NO_RESPONSE_SINCE_START but never emits it. " + "_verify_aiokafka_event_path only logs NO_FETCH_SINCE_START and " + "NO_RECENT_FETCH from the aiokafka poll timestamp; the " + "response-since-start check was dropped and the constant is now " + "dead code, so no condition can make verify_event_path log it." ) def test_timed_out(self, *, cthread, _consumer, now, tp, logger): assert cthread.verify_event_path(now, tp) is None @@ -517,9 +521,12 @@ def test_recent_response(self, *, cthread, now, tp, logger): logger.error.assert_not_called() @pytest.mark.skip( - reason="verify_event_path no longer emits the fetch-timeout " - "error() under this condition; predates that behaviour change and " - "needs re-deriving against the current driver" + reason="SOURCE BUG: faust/transport/drivers/aiokafka.py defines " + "SLOW_PROCESSING_NO_RECENT_RESPONSE but never emits it. " + "_verify_aiokafka_event_path only tracks the aiokafka poll " + "timestamp (request side) and logs NO_RECENT_FETCH; the " + "broker-response-staleness check was dropped and the constant is " + "now dead code, so no condition can make verify_event_path log it." ) def test_timed_out(self, *, cthread, now, tp, logger): self._set_last_request(now - 10.0) @@ -1056,13 +1063,11 @@ async def test_commit(self, *, cthread, _consumer): await cthread.commit(offsets) @pytest.mark.asyncio - @pytest.mark.skip( - reason="stale mock: _commit no longer forwards offsets to the " - "wrapped AIOKafkaConsumer.commit in this harness; needs a mock refresh" - ) async def test__commit(self, *, cthread, _consumer): offsets = {TP1: 1001} cthread._consumer = _consumer + # _commit only commits offsets for partitions in the assignment. + _consumer.assignment.return_value = {TP1} await cthread._commit(offsets) _consumer.commit.assert_called_once_with( @@ -1076,33 +1081,22 @@ async def test__commit__already_rebalancing(self, *, cthread, _consumer): assert not (await cthread._commit({TP1: 1001})) @pytest.mark.asyncio - @pytest.mark.skip( - reason="stale mock: _commit no longer forwards offsets to the " - "wrapped AIOKafkaConsumer.commit in this harness; needs a mock refresh" - ) async def test__commit__CommitFailedError(self, *, cthread, _consumer): cthread._consumer = _consumer + _consumer.assignment.return_value = {TP1} exc = _consumer.commit.side_effect = CommitFailedError("xx") cthread.crash = AsyncMock() - cthread.supervisor = Mock(name="supervisor") assert not (await cthread._commit({TP1: 1001})) cthread.crash.assert_called_once_with(exc) - cthread.supervisor.wakeup.assert_called_once() @pytest.mark.asyncio - @pytest.mark.skip( - reason="stale mock: _commit no longer forwards offsets to the " - "wrapped AIOKafkaConsumer.commit in this harness; needs a mock refresh" - ) async def test__commit__IllegalStateError(self, *, cthread, _consumer): cthread._consumer = _consumer - cthread.assignment = Mock() + cthread.assignment = Mock(return_value={TP1}) exc = _consumer.commit.side_effect = IllegalStateError("xx") cthread.crash = AsyncMock() - cthread.supervisor = Mock(name="supervisor") assert not (await cthread._commit({TP1: 1001})) cthread.crash.assert_called_once_with(exc) - cthread.supervisor.wakeup.assert_called_once() @pytest.mark.asyncio async def test_position(self, *, cthread, _consumer):