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/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 d36665704..0f7173f4f 100644 --- a/tests/unit/transport/drivers/test_aiokafka.py +++ b/tests/unit/transport/drivers/test_aiokafka.py @@ -437,26 +437,30 @@ 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() - 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, ) -@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 +468,14 @@ 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="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 self._set_last_request(now - 5.0) @@ -501,7 +513,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 +520,14 @@ 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="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) self._set_last_response(now - cthread.tp_fetch_response_timeout_secs * 2) @@ -608,7 +627,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 +648,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 +676,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 +925,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 +937,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 +955,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 +968,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,46 +1062,41 @@ 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 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( {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 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.skip("Needs fixing") @pytest.mark.asyncio 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):