Fix test in CI.

.github/workflows/ci.yml

@@ -60,7 +60,7 @@ jobs:
         run: uv run pytest -m "not integration"
 
   test-integration:
-    name: Integration tests (testcontainers)
+    name: Integration tests
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
@@ -77,5 +77,16 @@ jobs:
       - name: Sync dev dependencies
         run: uv sync --group dev --frozen
 
+      # Integration tests talk to postgres/redis/rabbitmq/minio on localhost
+      # (see tests/integration/conftest.py). The default compose profile starts
+      # only infra; the app runs in-process via create_app(). `--wait` blocks
+      # until every healthcheck passes before pytest runs.
+      - name: Start infrastructure (postgres/redis/rabbitmq/minio)
+        run: docker compose up -d --wait
+
       - name: Run integration tests
         run: uv run pytest -m integration
+
+      - name: Teardown infrastructure
+        if: always()
+        run: docker compose down -v

tests/integration/test_b2b_api.py

@@ -5,6 +5,8 @@ Run against the Docker Compose infrastructure (`docker compose up -d`).
 
 from __future__ import annotations
 
+import asyncio
+import time
 import uuid
 from pathlib import Path
 
@@ -105,11 +107,16 @@ async def test_b2b_analyze_upload_returns_202_and_enqueues(
         queue = await channel.get_queue("extract.q", ensure=False)
         deadline = 10.0
         found = False
-        import time
-
         while deadline > 0:
             start = time.monotonic()
-            message = await queue.get(timeout=deadline)
+            try:
+                message = await queue.get(timeout=deadline)
+            except aio_pika.exceptions.QueueEmpty:
+                # basic_get is non-blocking; the routed message may not be
+                # visible yet. Treat as transient and retry until the deadline.
+                await asyncio.sleep(0.25)
+                deadline -= time.monotonic() - start
+                continue
             await message.ack()
             msg_body = message.body.decode("utf-8")
             if str(document_id) in msg_body:

tests/integration/test_extract_worker.py

@@ -12,6 +12,7 @@ Scenarios:
 
 from __future__ import annotations
 
+import asyncio
 import json
 import time
 import uuid
@@ -163,7 +164,14 @@ async def test_extract_worker_pdf_uploads_text_and_publishes_analyze(
         found = False
         while deadline > 0:
             start = time.monotonic()
-            message = await queue.get(timeout=deadline)
+            try:
+                message = await queue.get(timeout=deadline)
+            except aio_pika.exceptions.QueueEmpty:
+                # basic_get is non-blocking; the routed message may not be
+                # visible yet. Treat as transient and retry until the deadline.
+                await asyncio.sleep(0.25)
+                deadline -= time.monotonic() - start
+                continue
             await message.ack()
             body = json.loads(message.body.decode("utf-8"))
             if body["document_id"] == str(document_id):