Add a docker-compose for integration test services

examples/docker/docker-compose.yml

@@ -11,6 +11,8 @@ services:
       - "5001:5001"  # exposed on the host for ad-hoc debugging (5001)
     environment:
       - DOCLING_SERVE_ENABLE_UI=1
+      # Allow remote VLM calls (picture descriptions via an external model API).
+      - DOCLING_SERVE_ENABLE_REMOTE_SERVICES=true
     restart: unless-stopped
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:5001/health"]
@@ -27,6 +29,8 @@ services:
       - "5002:5001"  # host 5002 → container 5001 (debug only)
     environment:
       - DOCLING_SERVE_ENABLE_UI=1
+      # Allow remote VLM calls (picture descriptions via an external model API).
+      - DOCLING_SERVE_ENABLE_REMOTE_SERVICES=true
     restart: unless-stopped
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:5001/health"]

tests/conftest.py

@@ -23,6 +23,8 @@ import pydantic_ai.models  # noqa: E402
 import pytest  # noqa: E402
 import yaml  # noqa: E402
 
+from .services import reachable  # noqa: E402
+
 if TYPE_CHECKING:
     from vcr import VCR
 
@@ -148,3 +150,42 @@ def doclaynet_first_page_pdf(tmp_path_factory) -> Path:
     finally:
         src.close()
     return out_path
+
+
+# --- external services for integration tests ---
+#
+# Integration tests (marked `integration`, excluded in CI via `-m "not
+# integration"`) need external services. Bring them up with
+#   docker compose -f tests/docker/docker-compose.yml up -d
+# These fixtures hand the test a connection URL when the service is reachable
+# (or when the matching env var points at an external instance), and skip the
+# test otherwise.
+
+_COMPOSE_HINT = (
+    "start it with `docker compose -f tests/docker/docker-compose.yml up -d`"
+)
+
+
+@pytest.fixture(scope="session")
+def postgres_dburi() -> str:
+    """A reachable Postgres queue URL. Uses HAIKU_RAG_TEST_PG_DBURI when set,
+    otherwise the docker-compose `postgres` service. Skips when neither is up."""
+    override = os.environ.get("HAIKU_RAG_TEST_PG_DBURI")
+    if override:
+        return override
+    if not reachable("localhost", 55432):
+        pytest.skip(f"Postgres not reachable on localhost:55432 — {_COMPOSE_HINT}")
+    return "postgresql+asyncpg://haiku:secret@localhost:55432/haiku_rag_test"
+
+
+@pytest.fixture(scope="session")
+def docling_serve_url() -> str:
+    """A reachable docling-serve base URL. Uses HAIKU_RAG_TEST_DOCLING_SERVE_URL
+    when set, otherwise the docker-compose `docling-serve` service. Skips when
+    neither is up."""
+    override = os.environ.get("HAIKU_RAG_TEST_DOCLING_SERVE_URL")
+    if override:
+        return override
+    if not reachable("localhost", 5001):
+        pytest.skip(f"docling-serve not reachable on localhost:5001 — {_COMPOSE_HINT}")
+    return "http://localhost:5001"

tests/docker/docker-compose.s3.yml

@@ -1,18 +0,0 @@
-services:
-  seaweedfs:
-    image: chrislusf/seaweedfs
-    ports:
-      - "8333:8333"
-    command: server -s3 -s3.config=/etc/seaweedfs/s3-config.json
-    volumes:
-      - ./s3-config.json:/etc/seaweedfs/s3-config.json:ro
-
-  createbucket:
-    image: chrislusf/seaweedfs
-    depends_on:
-      - seaweedfs
-    entrypoint: >
-      /bin/sh -c "
-      sleep 3 &&
-      echo 's3.bucket.create -name test-bucket' | weed shell -master seaweedfs:9333
-      "

tests/docker/docker-compose.yml

@@ -0,0 +1,50 @@
+# Services for the integration tests (marked `integration`, excluded in CI).
+# Start them all:
+#   docker compose -f tests/docker/docker-compose.yml up -d
+# Stop and clean up:
+#   docker compose -f tests/docker/docker-compose.yml down -v
+#
+# Tests skip automatically when a service isn't reachable, so you only need to
+# bring up the ones you care about. The docling-serve picture-description test
+# also needs Ollama running on the host with the `ministral-3` model.
+services:
+  postgres:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: haiku
+      POSTGRES_PASSWORD: secret
+      POSTGRES_DB: haiku_rag_test
+    ports:
+      - "55432:5432"
+    tmpfs:
+      - /var/lib/postgresql/data
+
+  docling-serve:
+    image: quay.io/docling-project/docling-serve:latest
+    ports:
+      - "5001:5001"
+    environment:
+      # Required for picture-description via a remote VLM (e.g. host Ollama).
+      - DOCLING_SERVE_ENABLE_REMOTE_SERVICES=true
+      - DOCLING_SERVE_ENABLE_UI=1
+    extra_hosts:
+      # Let the container reach the host's Ollama for VLM picture descriptions.
+      - "host.docker.internal:host-gateway"
+
+  seaweedfs:
+    image: chrislusf/seaweedfs
+    ports:
+      - "8333:8333"
+    command: server -s3 -s3.config=/etc/seaweedfs/s3-config.json
+    volumes:
+      - ./s3-config.json:/etc/seaweedfs/s3-config.json:ro
+
+  createbucket:
+    image: chrislusf/seaweedfs
+    depends_on:
+      - seaweedfs
+    entrypoint: >
+      /bin/sh -c "
+      sleep 3 &&
+      echo 's3.bucket.create -name test-bucket' | weed shell -master seaweedfs:9333
+      "

tests/ingester/test_queue_postgres.py

@@ -1,9 +1,10 @@
-"""Postgres-backed queue tests. Skipped unless HAIKU_RAG_TEST_PG_DBURI points
-at a reachable Postgres (a SQLAlchemy async URL, e.g.
-postgresql+asyncpg://user:pw@localhost:5432/haiku_rag_test). They exercise the
-dialect-specific SQL the SQLite suite cannot: ON CONFLICT, COALESCE upserts,
-the partial unique index, and FOR UPDATE SKIP LOCKED under real concurrent
-connections.
+"""Postgres-backed queue tests. Marked `integration` (excluded in CI). The
+`postgres_dburi` fixture connects to the docker-compose `postgres` service
+(`docker compose -f tests/docker/docker-compose.yml up -d`), or uses
+HAIKU_RAG_TEST_PG_DBURI when set, and skips when neither is reachable. They
+exercise the dialect-specific SQL the SQLite suite cannot: ON CONFLICT, COALESCE
+upserts, the partial unique index, and FOR UPDATE SKIP LOCKED under real
+concurrent connections.
 
 Each test owns its engine for its whole body. asyncpg binds connections to the
 loop that created them and pytest-asyncio uses a per-test loop, so opening the
@@ -12,7 +13,6 @@ avoids cross-loop fixture handoff.
 """
 
 import asyncio
-import os
 import uuid
 from contextlib import asynccontextmanager
 
@@ -25,22 +25,16 @@ from haiku.rag.ingester.queue.db import metadata
 from haiku.rag.ingester.queue.models import JobOp, JobStatus, SyncRow
 from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo
 
-PG_DBURI = os.environ.get("HAIKU_RAG_TEST_PG_DBURI")
-
-pytestmark = pytest.mark.skipif(
-    not PG_DBURI,
-    reason="Set HAIKU_RAG_TEST_PG_DBURI to run the Postgres queue tests",
-)
+pytestmark = pytest.mark.integration
 
 
 @asynccontextmanager
-async def queue_engine():
+async def queue_engine(dburi: str):
     """An engine scoped to a throwaway Postgres schema, so concurrent xdist
     workers each get their own isolated copy of the queue tables."""
-    assert PG_DBURI is not None  # guarded by the module skip marker
     schema = f"q_{uuid.uuid4().hex[:12]}"
     engine = create_async_engine(
-        make_url(PG_DBURI),
+        make_url(dburi),
         poolclass=NullPool,
         connect_args={"server_settings": {"search_path": schema}},
     )
@@ -56,10 +50,10 @@ async def queue_engine():
 
 
 @pytest.mark.asyncio
-async def test_enqueue_dedup_via_partial_unique_index():
+async def test_enqueue_dedup_via_partial_unique_index(postgres_dburi):
     """ON CONFLICT DO NOTHING against uq_jobs_live drops a second live job for
     the same (source_id, uri), regardless of op."""
-    async with queue_engine() as engine:
+    async with queue_engine(postgres_dburi) as engine:
         jobs = JobRepo(engine)
         first = await jobs.enqueue("s", "u", JobOp.UPSERT)
         second = await jobs.enqueue("s", "u", JobOp.DELETE)
@@ -68,10 +62,10 @@ async def test_enqueue_dedup_via_partial_unique_index():
 
 
 @pytest.mark.asyncio
-async def test_enqueue_after_terminal_succeeds():
+async def test_enqueue_after_terminal_succeeds(postgres_dburi):
     """Once a job is terminal it no longer satisfies the partial index, so a
     re-enqueue for the same URI is allowed."""
-    async with queue_engine() as engine:
+    async with queue_engine(postgres_dburi) as engine:
         jobs = JobRepo(engine)
         first = await jobs.enqueue("s", "u", JobOp.UPSERT)
         assert first is not None
@@ -83,11 +77,11 @@ async def test_enqueue_after_terminal_succeeds():
 
 
 @pytest.mark.asyncio
-async def test_skip_locked_claims_each_job_once():
+async def test_skip_locked_claims_each_job_once(postgres_dburi):
     """FOR UPDATE SKIP LOCKED: many concurrent claims over real Postgres
     connections each take a distinct job, with none claimed twice. Without
     SKIP LOCKED, concurrent transactions would grab the same row."""
-    async with queue_engine() as engine:
+    async with queue_engine(postgres_dburi) as engine:
         jobs = JobRepo(engine)
         enqueued = []
         for i in range(20):
@@ -104,9 +98,9 @@ async def test_skip_locked_claims_each_job_once():
 
 
 @pytest.mark.asyncio
-async def test_reap_stale_clamps_attempts():
+async def test_reap_stale_clamps_attempts(postgres_dburi):
     """The attempts-1 clamp (CASE) renders and runs on Postgres."""
-    async with queue_engine() as engine:
+    async with queue_engine(postgres_dburi) as engine:
         jobs = JobRepo(engine)
         job = await jobs.enqueue("s", "u", JobOp.UPSERT)
         assert job is not None
@@ -121,10 +115,10 @@ async def test_reap_stale_clamps_attempts():
 
 
 @pytest.mark.asyncio
-async def test_sync_state_upsert_coalesce_preserves_revision():
+async def test_sync_state_upsert_coalesce_preserves_revision(postgres_dburi):
     """ON CONFLICT DO UPDATE with COALESCE leaves an existing revision in place
     when a later upsert passes revision=None."""
-    async with queue_engine() as engine:
+    async with queue_engine(postgres_dburi) as engine:
         sync = SyncStateRepo(engine)
         await sync.upsert("s", "u", revision="v1", content_hash="h1")
         await sync.upsert("s", "u", revision=None, content_hash=None, ingested=True)
@@ -136,8 +130,8 @@ async def test_sync_state_upsert_coalesce_preserves_revision():
 
 
 @pytest.mark.asyncio
-async def test_sync_state_batch_upsert():
-    async with queue_engine() as engine:
+async def test_sync_state_batch_upsert(postgres_dburi):
+    async with queue_engine(postgres_dburi) as engine:
         sync = SyncStateRepo(engine)
         await sync.batch_upsert(
             [
@@ -149,8 +143,8 @@ async def test_sync_state_batch_upsert():
 
 
 @pytest.mark.asyncio
-async def test_prune_terminal_removes_old_rows():
-    async with queue_engine() as engine:
+async def test_prune_terminal_removes_old_rows(postgres_dburi):
+    async with queue_engine(postgres_dburi) as engine:
         jobs = JobRepo(engine)
         job = await jobs.enqueue("s", "u", JobOp.UPSERT)
         assert job is not None

tests/services.py

@@ -0,0 +1,11 @@
+import socket
+
+
+def reachable(host: str, port: int, timeout: float = 1.0) -> bool:
+    """True if a TCP connection to host:port succeeds within timeout. Used by
+    integration tests to skip when their backing service isn't running."""
+    try:
+        with socket.create_connection((host, port), timeout=timeout):
+            return True
+    except OSError:
+        return False

tests/test_converters.py

@@ -1339,6 +1339,21 @@ class TestDoclingServeConverterPictureDescription:
             assert "picture_description_api" not in data
 
 
+async def _skip_without_ollama_model(model: str) -> None:
+    """Skip when the host Ollama isn't serving `model`. docling-serve calls it
+    for VLM picture descriptions, so without it the test would fail with an
+    opaque error from inside the container rather than skip."""
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            response = await client.get("http://localhost:11434/api/tags")
+            response.raise_for_status()
+            names = [m["name"] for m in response.json().get("models", [])]
+    except Exception as exc:  # noqa: BLE001
+        pytest.skip(f"Ollama not reachable on localhost:11434 ({exc})")
+    if not any(name == model or name.startswith(f"{model}:") for name in names):
+        pytest.skip(f"Ollama model '{model}' not pulled (run `ollama pull {model}`)")
+
+
 class TestDoclingServeConverterIntegration:
     """Integration tests with real docling-serve recorded via VCR."""
 
@@ -1379,12 +1394,14 @@ class TestDoclingServeConverterIntegration:
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_picture_description_end_to_end(
-        self, config, doclaynet_first_page_pdf
+        self, config, docling_serve_url, doclaynet_first_page_pdf
     ):
         """End-to-end test: convert PDF with VLM picture descriptions via docling-serve.
 
         Note: Not using VCR because this test involves polling with changing task IDs.
         """
+        await _skip_without_ollama_model("ministral-3")
+        config.providers.docling_serve.base_url = docling_serve_url
         pdf_path = doclaynet_first_page_pdf
         config.processing.pictures = "description"
         config.processing.conversion_options.picture_description.model.provider = (

tests/test_s3_integration.py

@@ -1,9 +1,8 @@
-# Start SeaweedFS before running:
-#   docker compose -f tests/docker/docker-compose.s3.yml up -d
+# Start the services before running:
+#   docker compose -f tests/docker/docker-compose.yml up -d
 # Stop after:
-#   docker compose -f tests/docker/docker-compose.s3.yml down -v
+#   docker compose -f tests/docker/docker-compose.yml down -v
 
-import socket
 from uuid import uuid4
 
 import pytest
@@ -12,6 +11,7 @@ from haiku.rag.app import HaikuRAGApp
 from haiku.rag.client import HaikuRAG
 from haiku.rag.config.models import AppConfig, LanceDBConfig
 from haiku.rag.store.engine import Store
+from tests.services import reachable
 
 S3_ENDPOINT = "http://localhost:8333"
 S3_BUCKET = "test-bucket"
@@ -24,19 +24,11 @@ S3_STORAGE_OPTIONS = {
 }
 
 
-def _s3_available() -> bool:
-    try:
-        s = socket.create_connection(("localhost", 8333), timeout=1)
-        s.close()
-        return True
-    except OSError:
-        return False
-
-
 pytestmark = [
     pytest.mark.integration,
     pytest.mark.skipif(
-        not _s3_available(), reason="SeaweedFS not running on localhost:8333"
+        not reachable("localhost", 8333),
+        reason="SeaweedFS not running on localhost:8333",
     ),
 ]