From d2af1a2e62e127901f0935f7929d4cf767efb79e Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Mon, 7 Sep 2026 14:42:03 +0300 Subject: [PATCH] Wait for drained jobs instead of sleeping in the breaker isolation test test_breaker_isolates_sources snapshotted job status after a fixed 0.2s sleep and failed on a loaded runner while two good jobs were still in flight. Poll for the drained jobs under a 5s deadline. Assert the paused source's jobs are unattempted: the queued-uri assertion alone passes with source isolation disabled, since a retried job returns to QUEUED. --- tests/ingester/test_workers.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index 82036372..4ececf47 100644 --- a/tests/ingester/test_workers.py +++ b/tests/ingester/test_workers.py @@ -1054,16 +1054,23 @@ async def test_breaker_isolates_sources(client, jobs, sync): for _ in range(10): pool._breaker_for("bad").record_failure() + async def _good_jobs_drained(): + while True: + done = await jobs.list_jobs(status=JobStatus.SUCCEEDED, limit=50) + if len(done) == 3: + return done + await asyncio.sleep(0.02) + await pool.start() try: - await asyncio.sleep(0.2) - succeeded = await jobs.list_jobs(status=JobStatus.SUCCEEDED, limit=50) + succeeded = await asyncio.wait_for(_good_jobs_drained(), timeout=5.0) queued = await jobs.list_jobs(status=JobStatus.QUEUED, limit=50) finally: await pool.stop() assert {j.uri for j in succeeded} == {"g0", "g1", "g2"} assert {j.uri for j in queued} == {"b0", "b1", "b2"} + assert [j.attempts for j in queued] == [0, 0, 0] @pytest.mark.asyncio