From ca0e1d41bd7f53fb97d3c469720e48cd284b7f33 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 18:48:13 +0000 Subject: [PATCH] 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()