chatbot_v3/tests/e2e/test_compose_smoke.py

"""Operational smoke test against the *running web process* (ADR-0016).

Different in kind from every other test in this repo. Everything else drives
the app in-process through `ASGITransport`; this drives a real uvicorn process
over a real socket, backed by the Docker Compose stack, after the two
deployment steps (`alembic upgrade head`, `src.cli.qdrant_bootstrap`) have run
for real. ADR-0016 reserves Compose for exactly this: "manual local validation
and a later, serialized operational smoke test of the running web process",
run "as a serialized pre-release or scheduled gate".

It is therefore **skipped unless `SMOKE_BASE_URL` is set**, so a plain
`uv run pytest` never touches Docker Compose. Run it through `scripts/smoke.sh`,
which brings the stack up, provisions a tenant, starts the process, and exports
the environment below.

Because the app runs in its own process, this is also the only place in the
suite where the real `configure_logging()` is in effect --
`tests/conftest.py::_no_real_logging_configuration` no-ops it everywhere else
to keep global structlog state out of other tests. So the ADR-0011 log sink is
asserted here and nowhere else.
"""

import json
import os
import uuid
from collections.abc import AsyncIterator
from pathlib import Path

import pytest
import pytest_asyncio
from httpx import AsyncClient
from qdrant_client import AsyncQdrantClient, models

BASE_URL = os.environ.get("SMOKE_BASE_URL")
API_KEY = os.environ.get("SMOKE_API_KEY", "")
DOMAIN = os.environ.get("SMOKE_DOMAIN", "smoke")
LOG_PATH = os.environ.get("SMOKE_LOG_PATH", "")
QDRANT_URL = os.environ.get("SMOKE_QDRANT_URL", "http://127.0.0.1:6343")
QDRANT_COLLECTION = os.environ.get("SMOKE_QDRANT_COLLECTION", "chunks")

pytestmark = [
    pytest.mark.e2e,
    pytest.mark.slow,
    pytest.mark.asyncio,
    pytest.mark.skipif(
        not BASE_URL,
        reason="SMOKE_BASE_URL is unset; run this through scripts/smoke.sh",
    ),
    # Well past the global 10s budget: this drives a real process over a
    # socket, and the upload does real embedding work end to end.
    pytest.mark.timeout(120),
]

# Unique per run so a re-run against the same persistent Compose volumes is a
# new file rather than an idempotent duplicate-hash hit (ADR-0017).
_CSV = f"question,answer\nsmoke run {uuid.uuid4().hex},indexed\n".encode()


@pytest_asyncio.fixture
async def client() -> AsyncIterator[AsyncClient]:
    async with AsyncClient(base_url=BASE_URL or "", timeout=120.0) as async_client:
        yield async_client


async def test_running_process_reports_healthy_and_ready() -> None:
    """Readiness must be green *after* the deployment steps, not before --
    an unbootstrapped Qdrant is deliberately unready.
    """
    async with AsyncClient(base_url=BASE_URL or "", timeout=30.0) as client:
        healthz = await client.get("/healthz")
        readyz = await client.get("/readyz")

    assert healthz.status_code == 200
    assert healthz.json() == {"status": "ok"}
    assert readyz.status_code == 200, readyz.text
    assert readyz.json() == {"postgres": True, "minio": True, "qdrant": True}


async def test_upload_through_the_running_process_indexes_retrievable_points(
    client: AsyncClient,
) -> None:
    """The whole slice against the deployed shape: HTTP in, Qdrant points out."""
    upload = await client.post(
        "/v1/files",
        files={"file": ("smoke.csv", _CSV, "text/csv")},
        data={"domain": DOMAIN},
        headers={"Authorization": f"Bearer {API_KEY}"},
    )
    assert upload.status_code == 201, upload.text
    body = upload.json()
    assert body["status"] == "succeeded"
    assert body["chunks_indexed"] > 0

    status = await client.get(
        f"/v1/files/{body['file_id']}", headers={"Authorization": f"Bearer {API_KEY}"}
    )
    assert status.status_code == 200
    assert status.json()["ingestion_status"] == "succeeded"

    qdrant = AsyncQdrantClient(url=QDRANT_URL)
    try:
        count = await qdrant.count(
            collection_name=QDRANT_COLLECTION,
            count_filter=models.Filter(
                must=[
                    models.FieldCondition(
                        key="file_id", match=models.MatchValue(value=body["file_id"])
                    ),
                    models.FieldCondition(key="is_active", match=models.MatchValue(value=True)),
                ]
            ),
            exact=True,
        )
    finally:
        await qdrant.close()

    assert count.count == body["chunks_indexed"]


async def test_running_process_writes_structured_ingestion_logs() -> None:
    """The real ADR-0011 sink, which only a separate process exercises.

    An upload that succeeds while emitting nothing parseable is an upload no
    operator can investigate -- and the runbook's failure procedure is written
    against these exact event names.
    """
    if not LOG_PATH:
        pytest.skip("SMOKE_LOG_PATH is unset; scripts/smoke.sh normally provides it")

    lines = Path(LOG_PATH).read_text("utf-8").splitlines()
    events = [json.loads(line) for line in lines if line.startswith("{")]

    by_name = {event.get("event") for event in events}
    assert "ingestion.job.started" in by_name
    assert "ingestion.job.completed" in by_name

    completed = next(event for event in events if event.get("event") == "ingestion.job.completed")
    assert completed["tenant_id"]
    assert completed["ingestion_job_id"]
    assert completed["points_upserted"] > 0
    # Static process context bound once at startup (ADR-0011).
    assert completed["env"]
    assert completed["service_version"]