ci: add a dedicated confluent-driver test matrix leg#717
Merged
wbarnha merged 15 commits intoJul 21, 2026
Merged
Conversation
The `test_take` / `test_take_wit_timestamp*` tests asserted that an event consumed via `stream.take()` had been acked after exactly two `await asyncio.sleep(0)` yields. `take()` acks consumed events from a background task, and on slower interpreters (notably PyPy) that task hasn't run within two event-loop ticks, so `event.message.acked` was still False and the assertion flaked -- as seen on the PyPy CI leg. Replace the fixed sleeps with `wait_for_stream_ack()`, which polls for the ack (up to a 1s timeout) instead of assuming a fixed number of ticks. The existing assertions are unchanged, so a genuinely missing ack still fails the test with the same signal after the timeout. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
…stop The cleanup for Stream.take() / take_events() / take_with_timestamp() / noack_take() -- acking consumed events, restoring enable_acks, and detaching the buffering processor -- lives in `finally` blocks that only run when the generator is finalized. When a caller abandons the generator (`break` inside `async for`), CPython's reference counting finalizes it immediately, but PyPy defers finalization to a future GC cycle that may never come: acks are withheld indefinitely, the stream is left with acks disabled, and the buffering processor stays attached. (This is the root cause of the take() test failures on the PyPy CI leg, demonstrated on a real PyPy: the generator's finally never runs without an explicit gc.collect().) Track every buffering generator in a WeakSet and close leftovers explicitly in Stream.on_stop(), making cleanup deterministic on any interpreter. The WeakSet keeps CPython's prompt refcount finalization (and therefore existing behavior) fully intact; already-exhausted generators make aclose() a no-op, and a generator still running in another task is skipped (its owner cleans up). The new test holds a strong reference to the generator -- blocking finalization on every interpreter, mimicking PyPy -- and asserts the stream's stop path performs the cleanup; it fails without this fix on CPython as well. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
…laude/pypy-agen-cleanup
Extend the on_stop generator-cleanup mechanism to the main stream iterator (__aiter__ -> _py_aiter/_c_aiter), whose inner finally acks the last-yielded event and whose outer finally detaches from the channel -- previously only the take() family was tracked, so a plain `async for value in stream: break` still leaked its cleanup to GC timing on PyPy. Service.stop() sets _stopped before awaiting on_stop(), so the outer finally's re-entrant self.stop() is a guarded no-op. Make the test suite PyPy-clean and remove every skipif(PyPy) marker (21 decorators covering 30 tests, added wholesale against PyPy 3.9 in #530/#621 and never re-validated): * tests/conftest.py: override the event_loop fixture to run gc.collect() + loop.shutdown_asyncgens() before closing the loop on PyPy, so deferred async-generator finalizers run inside the owning test instead of landing on a dead loop and leaking tasks/warnings into later tests. * tests/meticulous: disable the 4s hypothesis deadline on PyPy -- JIT warm-up makes the first examples orders of magnitude slower than steady state, tripping DeadlineExceeded on correct code. * New regression test test_aiter_cleanup_on_stream_stop mirrors the take() one: holds a strong reference (blocking finalization on every interpreter, mimicking PyPy) and asserts stop-time cleanup acks the event. Verified to fail without the __aiter__ tracking. Verified on CPython: functional+unit+meticulous suites green with Cython enabled AND with NO_CYTHON=1 (the pure-Python paths PyPy uses), and green under a PyPy-semantics simulation (an asyncgen firstiter hook holding strong references to defeat refcount finalization). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
CopartitionedAssignor._assign_round_robin frees a slot on an exhausted client by evicting an arbitrary partition (set.pop) and re-queueing it. When that arbitrary victim is the partition's only possible home, two partitions ping-pong between the work queue and the same client forever. Whether it happens depends on set iteration order, so identical input (e.g. 4 clients / 16 partitions / 1 replica) completes in <1ms on CPython but livelocks indefinitely on PyPy -- the reason the copartitioned-assignor property tests were skipped there. Evict a victim that some other not-yet-exhausted client can immediately take, in deterministic (sorted) order, so the re-queued partition makes forward progress instead of bouncing straight back. This terminates on every interpreter and keeps the assignment valid. Unskip the three property tests on PyPy (they now pass with the full example budget in ~4s) and add a regression test asserting the standby path terminates for any replicas >= 1. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
…I timeout Replace the timeout-poll wait_for_stream_ack helper with an explicit generator aclose(): the take()/iterator finally blocks (which ack consumed events) only run when the generator is finalized, and abandoning it defers that to interpreter finalization -- immediate on CPython via refcounting, deferred to a maybe-never GC cycle on PyPy. Closing it runs the finally now, deterministically, on every interpreter. Callers hold the generator and close it instead of leaning on asyncio.sleep(0) yields. The unit aiter test drains via a bounded gc.collect() + sleep loop for the same reason (its agent-actor generator can't be reached to close directly). Bump the PyPy CI leg's timeout from 10 to 15 minutes: it now runs the formerly-skipped tests and PyPy is slower, and the old cap already clipped the run at ~96%. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
…r on stop
on_stop() closes the async generators the stream handed out, skipping any
that are still executing in another task (an agent actor iterating the
stream during a rebalance). That "still running" case raises RuntimeError
on CPython but ValueError ("async generator already executing") on PyPy, and
only RuntimeError was caught -- so on PyPy the ValueError propagated out of
on_stop, aborting the stream's stop mid-way and leaving stale entries in the
topic's active_partitions. That tripped the conductor's
`active_partitions.issubset(assigned)` assertion during isolated-partition
rebalances (test_agent_isolated_partitions_rebalancing).
Catch both exception types in on_stop and in the wait_for_stream_ack test
helper, so closing a running generator is always the intended no-op.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## claude/reenable-remaining-skipped-tests #717 +/- ##
========================================================================
Coverage 94.81% 94.81%
========================================================================
Files 104 104
Lines 11154 11154
Branches 1202 1202
========================================================================
Hits 10576 10576
Misses 482 482
Partials 96 96 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
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
8d6758f); 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
tests/unit/transport/drivers/test_confluent.py is skipped in CI because confluent-kafka is an optional extra (faust[ckafka]) that the test job never installs -- so the confluent driver ships with its unit tests effectively unrun. Give the test matrix an explicit `kafka-driver` axis: * aiokafka -- the default driver and a core dependency -- keeps the full Python x Cython grid (10 jobs, unchanged behaviour). * confluent -- a broker-less, pure-Python wrapper -- gets one dedicated leg per Python version (Cython off; the driver isn't Cython-accelerated). Each installs requirements/extras/ckafka.txt and runs just test_confluent.py, which now executes instead of skipping and adds coverage for the confluent driver. confluent-kafka 2.x publishes manylinux wheels for cp310-cp314, so every matrix Python is covered. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
The release wheel job pinned pypa/cibuildwheel@v2.21.3 (Oct 2024), which predates CPython 3.14. cibuildwheel only builds the interpreters its own release knows about, so `build = "cp3*"` expanded to at most cp313 and no 3.14 wheels were ever produced or uploaded to PyPI for 0.12.0/0.12.1 -- a local install on 3.14 has to compile from the sdist (issue #715). The test matrix separately tests 3.14 via setup-python, which is why the gap only showed up in the published wheels. Bump the action to v4.1.0, which builds CPython 3.14 by default (since cibuildwheel 3.1) and still supports cp310-cp313. Because free-threading is no longer experimental in 3.14, cibuildwheel would now also build free-threaded (cp314t) wheels by default; the Cython extension has not been validated under a no-GIL interpreter, so skip `cp31?t-*` for now. Verified with `cibuildwheel --print-build-identifiers` that the linux build set is now cp310..cp314 manylinux_x86_64 (no cp314t, no musllinux). The manylinux2014 image shorthand and the existing build/skip/archs/before-build options remain valid in 4.x. Fixes #715 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
The redis cache backend was broken against a real server: it maps schemes to the *synchronous* redis.StrictRedis / redis.RedisCluster clients but `_get`/ `_set`/`_delete`/`connect` await them, so `await self.client.get(...)` raised "object ... can't be used in 'await' expression". It only appeared to work because every test mocks the client. * Use the asyncio clients (redis.asyncio.StrictRedis / .RedisCluster), so the awaited calls are real coroutines. Verified end-to-end against a live redis (set/get/delete round-trip). * Close the client on stop (`on_stop` -> `await client.aclose()`); the async client holds real connections. * Bind `redis = None` (not just `aredis`) on ImportError so the "not installed" guard raises ImproperlyConfigured instead of NameError. * mocked_redis fixture: patch redis.asyncio.StrictRedis (the name the backend now uses) and give the mock `aclose`. * Re-enable Test_RedisScheme / the not-installed test, asserting the async client wiring. Add a real-redis CI leg: a `test-redis-integration` job with a redis service container runs new env-gated integration tests (tests/integration/cache) that exercise the backend against a live server -- the coverage the mocks can't give. Advisory (not in the required `check` gate), matching the Kafka job. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
wbarnha
force-pushed
the
claude/ci-confluent-driver-matrix
branch
from
July 21, 2026 00:02
cd41419 to
a958d64
Compare
wbarnha
changed the base branch from
master
to
claude/reenable-remaining-skipped-tests
July 21, 2026 00:02
wbarnha
force-pushed
the
claude/reenable-remaining-skipped-tests
branch
from
July 21, 2026 01:01
6692060 to
2cbafe2
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
tests/unit/transport/drivers/test_confluent.pyis silently skipped in CI:confluent-kafkais an optional extra (faust[ckafka],requirements/extras/ckafka.txt) that thetest-pytestjob never installs, so the confluent driver ships with its unit tests effectively unrun.Change
Give the test matrix an explicit
kafka-driveraxis with dedicated values for each driver:requirements/requirements.txt) — keeps the full Python × Cython grid (10 jobs, unchanged behaviour, still runsscripts/tests).confluent_kafka— gets one dedicated leg per Python version (Cython off; the driver isn't Cython-accelerated, so the Cython axis adds nothing). Each leg additionally installsrequirements/extras/ckafka.txtand runs justtest_confluent.py, which now executes instead of skipping — adding coverage for the confluent driver that was previously 0.Net matrix: 15 jobs (10 aiokafka + 5 confluent).
Notes
confluent-kafka2.x publishes manylinux x86_64 wheels forcp310–cp314, so every matrix Python is covered (verified against the 2.15.0 release).confluent_kafka'sConsumer/Producer, so no broker is required.pytest tests/unit/transport/drivers/test_confluent.py→ 57 passed withconfluent-kafkainstalled.🤖 Generated with Claude Code
Generated by Claude Code