Skip to content

test: re-enable recoverable skipped tests (+ fix latent LazySpan bug) #719

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wbarnha wants to merge 1 commit into from
Closed

test: re-enable recoverable skipped tests (+ fix latent LazySpan bug) #719

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion faust/transport/drivers/aiokafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 0 additions & 1 deletion tests/unit/agents/test_agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
47 changes: 31 additions & 16 deletions tests/unit/transport/drivers/test_aiokafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -456,14 +460,18 @@ 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)
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, _consumer, now, tp, logger):
assert cthread.verify_event_path(now, tp) is None
self._set_last_request(now - 5.0)
Expand Down Expand Up @@ -501,14 +509,18 @@ 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)
self._set_last_response(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_last_request(now - 10.0)
self._set_last_response(now - cthread.tp_fetch_response_timeout_secs * 2)
Expand Down Expand Up @@ -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
Expand All @@ -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):
Expand All @@ -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,
)


Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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")
Expand All @@ -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()
Expand Down