Compare commits
6 Commits
012b44d5f2
...
3d9269e54f
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3d9269e54f | ||
|
|
7c1fe79f1c | ||
|
|
1b873e5a6f | ||
|
|
133f565704 | ||
|
|
c7a5b69c0a | ||
|
|
2c72688440 |
24
CLAUDE.md
24
CLAUDE.md
@@ -27,9 +27,16 @@ Also working: `tenant_domains` plus `/v1/domains` (`src/application/domains/`),
|
||||
a strict per-tenant allowlist — `POST /v1/files` rejects an unregistered or
|
||||
disabled `domain` with `400` before anything is written, and domain management
|
||||
sits behind its own `domains:read`/`domains:write` scopes, never `files:write`.
|
||||
Not built yet: `/v1/points` CRUD and keyword search (plan 002), and
|
||||
`src/agent/`. That maps to plan 001 Phases 1-5 done, Phase 6 (runbook, e2e
|
||||
tests, Compose smoke test) not started.
|
||||
Also working: the operator runbook (`docs/runbook.md`), tenant/API-key/domain
|
||||
provisioning (`uv run python -m src.cli.provision_tenant` — the third deployment
|
||||
step, since nothing over HTTP can create the first tenant), a Testcontainers
|
||||
e2e suite in the default pytest run (`tests/e2e/test_ingestion_slice.py`:
|
||||
duplicate upload, retry after failure, tenant isolation, capacity, timeout,
|
||||
parse and Qdrant failure), and the one Compose-based test — `scripts/smoke.sh`
|
||||
driving `tests/e2e/test_compose_smoke.py` against a real uvicorn process, which
|
||||
skips itself unless `SMOKE_BASE_URL` is set. Not built yet: `/v1/points` CRUD
|
||||
and keyword search (plan 002), and `src/agent/`. That maps to plan 001 Phases
|
||||
1-6 done.
|
||||
|
||||
Architecture decisions live in `docs/adr/` (18 ADRs plus the 0000 template;
|
||||
0001–0004 are `Accepted` — 0004 amended by 0018; 0014 is `Superseded by 0017`;
|
||||
@@ -285,9 +292,14 @@ belongs, not the row-level function underneath it.
|
||||
- Test naming: `test_<unit>_<scenario>_<outcome>`, Arrange–Act–Assert.
|
||||
- Layout mirrors architecture: `tests/unit/{application,agent}`,
|
||||
`tests/integration/{postgres,minio,qdrant}`, `tests/e2e/`.
|
||||
- Integration tests use **Testcontainers** (never a developer's local
|
||||
services or Langfuse-owned storage/credentials) — this is the standard
|
||||
automated mechanism, not Docker Compose. Isolate data per test via unique
|
||||
- Integration **and e2e** tests use **Testcontainers** (never a developer's
|
||||
local services or Langfuse-owned storage/credentials) — this is the standard
|
||||
automated mechanism, not Docker Compose. Compose is reserved for exactly one
|
||||
thing: the serialized operational smoke test of the *running web process*
|
||||
(`scripts/smoke.sh`), which is gated out of `uv run pytest`. Shared container
|
||||
fixtures live in `tests/support/containers.py`, registered from the root
|
||||
`tests/conftest.py` via `pytest_plugins` (a non-root conftest cannot declare
|
||||
it). Isolate data per test via unique
|
||||
keys/queue/collection names; parallel integration execution is disabled
|
||||
until fixture isolation is proven safe.
|
||||
- Pytest never calls a live/paid model provider in routine runs — that's
|
||||
|
||||
15
README.md
15
README.md
@@ -2,6 +2,9 @@
|
||||
|
||||
Architecture decisions live in [`docs/adr`](docs/adr). The first implementation
|
||||
milestone is documented in the [ingestion vertical-slice plan](docs/plans/001-ingestion-vertical-slice.md).
|
||||
Day-to-day operation — tuning the ingestion bounds, the proxy timeout
|
||||
requirement, and how to investigate or retry a failed upload — is the
|
||||
[operator runbook](docs/runbook.md).
|
||||
|
||||
## Provisioning the datastores
|
||||
|
||||
@@ -16,6 +19,14 @@ uv run python -m src.cli.qdrant_bootstrap # the `chunks` collection
|
||||
uv run fastapi dev src/main.py
|
||||
```
|
||||
|
||||
Nothing over HTTP can create the first tenant — every `/v1` route needs an API
|
||||
key, and a key cannot exist before its tenant. One command issues both, plus any
|
||||
domains, printing the key once (only its hash is stored):
|
||||
|
||||
```bash
|
||||
uv run python -m src.cli.provision_tenant --slug acme --domain fire
|
||||
```
|
||||
|
||||
Before a tenant can upload, its domains must be registered — `POST /v1/files`
|
||||
rejects an unregistered or disabled `domain` with `400`. The calling backend
|
||||
manages them over `/v1/domains` using a key with the `domains:write` scope:
|
||||
@@ -31,6 +42,10 @@ Both bootstrap commands are idempotent and safe to re-run. `qdrant_bootstrap` ve
|
||||
existing collection against the pinned schema and exits non-zero on a mismatch,
|
||||
rather than leaving a silently degraded sparse index in place.
|
||||
|
||||
`./scripts/smoke.sh` verifies the whole path — Compose up, both deployment
|
||||
steps, provisioning, an upload through the running web process to indexed Qdrant
|
||||
points. See the [runbook](docs/runbook.md#12-verifying-a-deployment).
|
||||
|
||||
## Local Langfuse
|
||||
|
||||
This repo includes a root-level development Compose file for Langfuse:
|
||||
|
||||
@@ -289,17 +289,28 @@ retrying the upload produces a correct final state without duplicate chunks.
|
||||
|
||||
1. Add an operator runbook covering local startup, migrations, MinIO bucket
|
||||
setup, the run command, ingestion-bound tuning, the proxy/client timeout
|
||||
requirement, and how to retry a failed ingestion.
|
||||
requirement, and how to retry a failed ingestion. — `docs/runbook.md`.
|
||||
2. Add a serialized Compose-based operational smoke test covering upload through
|
||||
indexed points against the running web process. Testcontainers remains the
|
||||
standard pytest mechanism for individual adapter integration tests.
|
||||
standard pytest mechanism for individual adapter integration tests. —
|
||||
`scripts/smoke.sh` driving `tests/e2e/test_compose_smoke.py`, which skips
|
||||
itself unless `SMOKE_BASE_URL` is set so `uv run pytest` never invokes
|
||||
Compose.
|
||||
3. Add end-to-end tests for duplicate upload, retrying a failed upload, tenant
|
||||
isolation, capacity/timeout rejection, and failed parser/Qdrant behavior.
|
||||
isolation, capacity/timeout rejection, and failed parser/Qdrant behavior. —
|
||||
`tests/e2e/test_ingestion_slice.py`, on Testcontainers, in the default suite.
|
||||
4. Add health/readiness checks that distinguish process health from dependency
|
||||
readiness.
|
||||
readiness. — `/healthz` and `/readyz`; `/readyz` additionally requires the
|
||||
`chunks` collection to exist, since a reachable but unbootstrapped Qdrant
|
||||
would `502` on the first upload.
|
||||
5. Update the README with local-start instructions and links to ADRs, this plan,
|
||||
and the operations runbook.
|
||||
|
||||
Provisioning a tenant and its first API key turned out to be a prerequisite for
|
||||
1 and 2 rather than a separate milestone: nothing over HTTP can create the first
|
||||
tenant, so `src/cli/provision_tenant.py` was added alongside the other two
|
||||
deployment-step commands.
|
||||
|
||||
**Exit criteria:** a new developer can start the stack, apply migrations, upload a
|
||||
a document, observe the job through completion, and understand how to investigate or
|
||||
retry a failure.
|
||||
|
||||
291
docs/runbook.md
Normal file
291
docs/runbook.md
Normal file
@@ -0,0 +1,291 @@
|
||||
# Operator runbook
|
||||
|
||||
How to start this service, configure its ingestion bounds, and investigate or
|
||||
retry a failed upload. Architecture rationale lives in [`docs/adr/`](adr/); the
|
||||
implementation milestone is
|
||||
[plan 001](plans/001-ingestion-vertical-slice.md). This document covers
|
||||
operating what those describe.
|
||||
|
||||
The service is a **single process with no background work**. `POST /v1/files`
|
||||
parses, chunks, embeds, and indexes inline and returns a terminal result
|
||||
(ADR-0017). There is no queue, no worker, and no automatic retry — the caller
|
||||
owns the retry decision, which makes the request's duration a deployment
|
||||
constraint. That fact drives most of this document.
|
||||
|
||||
## 1. Prerequisites and local startup
|
||||
|
||||
Docker, and [`uv`](https://docs.astral.sh/uv/) with Python 3.13.
|
||||
|
||||
```bash
|
||||
cp .env.example .env # non-secret local defaults; .env is gitignored
|
||||
uv sync
|
||||
docker compose up -d --wait
|
||||
```
|
||||
|
||||
`docker-compose.yml` runs Postgres (`127.0.0.1:5433`), MinIO
|
||||
(`127.0.0.1:9100`, console `9101`), and Qdrant (`127.0.0.1:6343`). It is the
|
||||
local development stack and says so in its header — it is not a production
|
||||
deployment.
|
||||
|
||||
## 2. Deployment steps
|
||||
|
||||
Two schema steps run **before** the application, never at startup: FastAPI
|
||||
performs no DDL, for Postgres (ADR-0009) or for Qdrant (ADR-0001, "Collection
|
||||
provisioning"). Both commands and their reasoning are in the README's
|
||||
[Provisioning the datastores](../README.md#provisioning-the-datastores)
|
||||
section:
|
||||
|
||||
```bash
|
||||
uv run alembic upgrade head # Postgres schema
|
||||
uv run python -m src.cli.qdrant_bootstrap # the `chunks` collection
|
||||
```
|
||||
|
||||
Both are idempotent. `qdrant_bootstrap` verifies an existing collection against
|
||||
the pinned schema and **exits non-zero on a mismatch** rather than leaving a
|
||||
silently degraded sparse index in place — the `sparse` vector's
|
||||
`modifier="idf"` and the pinned dense dimensions (768 / 3072) fail silently if
|
||||
wrong, which is why they are checked rather than assumed.
|
||||
|
||||
Run both again after every deploy that ships a migration or a collection-schema
|
||||
change.
|
||||
|
||||
## 3. MinIO bucket
|
||||
|
||||
Under Compose the bucket already exists: the `app-minio` service's entrypoint
|
||||
runs `mkdir -p /data/${MINIO_BUCKET:-chatbot-source-files}` before starting the
|
||||
server, so first boot creates it. Nothing else needs to be done locally.
|
||||
|
||||
Outside Compose, create the bucket named by `MINIO_BUCKET` before the first
|
||||
upload — the application never creates it. It must stay **private**; ADR-0013
|
||||
keeps source bytes non-public and this slice ships no download API or presigned
|
||||
URLs.
|
||||
|
||||
## 4. Provisioning a tenant, an API key, and its domains
|
||||
|
||||
Nothing over HTTP can bootstrap a tenant: every `/v1` route needs an API key,
|
||||
and a key cannot exist before its tenant. So the first key is issued by an
|
||||
operator command:
|
||||
|
||||
```bash
|
||||
uv run python -m src.cli.provision_tenant \
|
||||
--slug acme --domain fire --domain life --scopes files:write,domains:read
|
||||
```
|
||||
|
||||
It prints `api_key=sk_...` **once**. Postgres stores only its SHA-256 hash
|
||||
(ADR-0009), so a lost key is reissued by re-running the command, never
|
||||
recovered. Structured logs carry only the non-secret `key_prefix` — a plaintext
|
||||
key must never reach a log sink (ADR-0011).
|
||||
|
||||
Re-running with the same `--slug` reuses the tenant and any domains it already
|
||||
has, and issues an **additional** key. Both keys stay valid; this adds a key, it
|
||||
does not rotate one.
|
||||
|
||||
Scopes are the security boundary between uploading and managing the allowlist.
|
||||
Give an upload client `files:write` only. `domains:write` lets its holder create
|
||||
new domains, which is exactly what the allowlist exists to prevent an upload key
|
||||
from doing.
|
||||
|
||||
### Domains after the first one
|
||||
|
||||
`POST /v1/files` rejects an unregistered or disabled `domain` with `400`
|
||||
(`unknown_domain`) before anything is written. Ongoing domain management is the
|
||||
`/v1/domains` API, under `domains:read` / `domains:write`:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:8000/v1/domains \
|
||||
-H "Authorization: Bearer $API_KEY" -H 'Content-Type: application/json' \
|
||||
-d '{"domain": "fire", "display_name": "Fire insurance"}'
|
||||
```
|
||||
|
||||
The `domain` key itself is immutable — it is denormalized into every Qdrant
|
||||
point payload and into `source_files`, so renaming it is a migration, not an
|
||||
edit (ADR-0009). Disabling a domain blocks new uploads; it does not delete
|
||||
existing points.
|
||||
|
||||
## 5. Running the service
|
||||
|
||||
```bash
|
||||
uv run fastapi dev src/main.py # local, reload
|
||||
uv run uvicorn src.main:app --host 0.0.0.0 --port 8000 # deployed shape
|
||||
```
|
||||
|
||||
Run more than one worker/replica only after reading §6: ingestion bounds are
|
||||
**per process**, so `INGESTION_MAX_CONCURRENCY` multiplies by the number of
|
||||
processes.
|
||||
|
||||
## 6. Ingestion bounds and tuning
|
||||
|
||||
Every bound is enforced server-side and maps to a status code. All are in
|
||||
`.env.example`. Ingestion is CPU- and network-bound in the request, so these are
|
||||
the numbers that decide whether the service degrades gracefully or falls over.
|
||||
|
||||
| Setting | Bounds | On breach | Size it against |
|
||||
|---|---|---|---|
|
||||
| `INGESTION_MAX_CONCURRENCY` | Ingestions in flight **per process** | `503` + `Retry-After` | Memory per in-flight upload (whole file plus its chunks and vectors are resident) and the embedder's capacity. Rejecting is deliberate: ADR-0017 refuses rather than queues. |
|
||||
| `INGESTION_THREAD_POOL_SIZE` | Threads for blocking work (parse, chunk, hash, BM25, the sync `minio` SDK) | — (waits) | CPU cores. It exists to stop ingestion exhausting Starlette's own thread pool, so keep it below the total thread budget. |
|
||||
| `INGESTION_TIMEOUT_SECONDS` | The whole work phase | `504`, job marked `failed` | The slowest legitimate document, plus headroom. See §7 — this must stay under every read timeout in front of it. |
|
||||
| `INGESTION_MAX_UPLOAD_SIZE_MB` | Bytes accepted | `413` | Memory: the upload is read fully into the process before any work starts. |
|
||||
| `INGESTION_MAX_CHUNKS_PER_FILE` | Chunks per file, checked before embedding | `413` | Embedder cost/time per chunk × `INGESTION_TIMEOUT_SECONDS`. This is the real defence against one pathological file eating a slot. |
|
||||
| `INGESTION_EMBED_BATCH_SIZE` | Texts per embedder request | `502` on embedder failure | The provider's per-request limits. Batch before parallelizing. |
|
||||
| `INGESTION_EMBED_CONCURRENCY` | Concurrent embed batches | `502` | Provider rate limits and the self-hosted embedder's throughput. Never unbounded. |
|
||||
| `QDRANT_UPSERT_BATCH_SIZE` / `_CONCURRENCY` | Points per upsert and concurrent upserts | `502` (`index_error`) | Qdrant's ingest capacity; the batch size stays in ADR-0001's 64–256 band. |
|
||||
|
||||
Two settings that look like tuning knobs but are not:
|
||||
|
||||
- **`EMBEDDING_NOMIC_KEEP_ALIVE`** holds the self-hosted model resident. A cold
|
||||
load of `nomic-embed-text-v2-moe` takes over 150 s — longer than any sane
|
||||
`INGESTION_TIMEOUT_SECONDS` — so an idle period followed by an upload would
|
||||
otherwise `504`. The lifespan also warms both dense embedders at startup for
|
||||
the same reason.
|
||||
- **The BM25 analyzer and weights** (`EMBEDDING_SPARSE_*`) are a measured
|
||||
artifact ported from the `emet` evaluation lab, verified token-for-token
|
||||
against it (ADR-0005). Re-benchmark; do not tune them in place.
|
||||
|
||||
## 7. The proxy and client read-timeout requirement
|
||||
|
||||
**Every read timeout in front of this service must exceed
|
||||
`INGESTION_TIMEOUT_SECONDS`.** That includes the reverse proxy / ingress, any
|
||||
load balancer, and the calling backend's own HTTP client.
|
||||
|
||||
If a proxy times out first, the client gets that proxy's error, the upload keeps
|
||||
running in the process, and the caller learns nothing about the outcome from the
|
||||
response. The job row still reaches a terminal status, so
|
||||
`GET /v1/files/{file_id}` remains the way to find out what happened — but the
|
||||
response contract is broken for that request. ADR-0017 names this the main cost
|
||||
of inline ingestion.
|
||||
|
||||
A workable local ordering: client read timeout > proxy read timeout >
|
||||
`INGESTION_TIMEOUT_SECONDS`.
|
||||
|
||||
## 8. Health and readiness
|
||||
|
||||
| Endpoint | Question it answers | Use for |
|
||||
|---|---|---|
|
||||
| `GET /healthz` | Is the process alive? | Liveness probes / restart policy. Never depends on Postgres, MinIO, or Qdrant. |
|
||||
| `GET /readyz` | Can it actually serve? | Load-balancer admission and post-deploy gating. `200` with each dependency `true`, `503` if any is `false`. |
|
||||
|
||||
`/readyz` checks Postgres, MinIO, and Qdrant reachability **and** that the
|
||||
`chunks` collection exists. A reachable-but-unbootstrapped Qdrant reports
|
||||
`{"qdrant": false}` on purpose: uploads to it would fail with `502`, so it is
|
||||
not ready, and this is how a skipped `qdrant_bootstrap` surfaces at deploy time
|
||||
instead of on a user's first upload.
|
||||
|
||||
## 9. Investigating a failure
|
||||
|
||||
Logs are structured (`structlog`, JSON in production) with stable event names —
|
||||
grep the event name, not prose (ADR-0011). Set `LOG_FILE_PATH` for a local
|
||||
JSON file sink alongside the console renderer; leave it unset in production,
|
||||
where stdout collection is preferred.
|
||||
|
||||
**Correlate by `request_id`.** Every request has one, echoed in the
|
||||
`X-Request-Id` response header and included in every error envelope, and bound
|
||||
into every log line emitted while handling that request. A client reporting a
|
||||
failed upload should quote it. `tenant_id`, `file_id`, and `ingestion_job_id`
|
||||
are the other join keys.
|
||||
|
||||
Events worth knowing:
|
||||
|
||||
| Event | Level | Means |
|
||||
|---|---|---|
|
||||
| `ingestion.job.started` | info | Txn A committed; work phase beginning. Carries `tenant_id`, `ingestion_job_id`, `file_id`, `domain`, `source_type`. |
|
||||
| `ingestion.job.completed` | info | Terminal success, with `chunks_parsed`, `points_upserted`, `points_soft_deleted`. |
|
||||
| `ingestion.job.failed` | warning | Terminal failure. **`error_code` says which stage**: `storage_upload_failed`, `parse_failed`, `chunk_limit_exceeded`, `embedding_failed`, `index_failed`, `timeout`. |
|
||||
| `files.upload.duplicate` | info | Identical content already ingested; the existing file/job was returned and nothing was re-ingested. |
|
||||
| `domain.rejected` | warning | Upload refused before any row was written; `reason` is `unregistered` or `disabled`. |
|
||||
| `auth.failed` | warning | `reason` is `malformed_key`, `unknown_key`, `key_inactive`, `key_expired`, or `tenant_inactive`. Never contains key material. |
|
||||
| `auth.succeeded` | info | Carries `tenant_id`, `api_key_id`, `actor_type`. |
|
||||
| `lifespan.embedder.warm_failed` | warning | An embedder was unreachable at startup. Boot continues by design — `/readyz` and the first upload are where this bites. |
|
||||
| `qdrant.bootstrap.schema_mismatch` | error | The existing collection diverges from the pinned schema. The bootstrap exits non-zero; do not start the app against it. |
|
||||
| `api.unhandled_exception` | error | A bug: an exception with no mapping to the error envelope. Always worth a look. |
|
||||
|
||||
A `503` (`ingestion_at_capacity`) is rejected before a job row exists, so it
|
||||
appears in the access log and metrics, not in `ingestion_jobs`.
|
||||
|
||||
### The durable record
|
||||
|
||||
Logs may roll; `ingestion_jobs` and `ingestion_job_events` do not. For one file:
|
||||
|
||||
```sql
|
||||
SELECT id, status, error_code, error_message, points_created, points_soft_deleted,
|
||||
created_at, updated_at
|
||||
FROM ingestion_jobs
|
||||
WHERE tenant_id = :tenant_id AND source_file_id = :file_id
|
||||
ORDER BY created_at DESC;
|
||||
|
||||
SELECT stage, level, message, details, created_at
|
||||
FROM ingestion_job_events
|
||||
WHERE tenant_id = :tenant_id AND ingestion_job_id = :ingestion_job_id
|
||||
ORDER BY created_at;
|
||||
```
|
||||
|
||||
Recent failures across a tenant:
|
||||
|
||||
```sql
|
||||
SELECT error_code, count(*), max(created_at)
|
||||
FROM ingestion_jobs
|
||||
WHERE tenant_id = :tenant_id AND status = 'failed' AND created_at > now() - interval '1 day'
|
||||
GROUP BY error_code ORDER BY 2 DESC;
|
||||
```
|
||||
|
||||
`GET /v1/files/{file_id}` reports the same terminal status over HTTP, scoped to
|
||||
the owning tenant — a file belonging to another tenant returns `404`, not `403`.
|
||||
|
||||
## 10. Retrying a failed ingestion
|
||||
|
||||
**Re-upload the same bytes.** There is no retry endpoint and no automatic retry;
|
||||
the client owns that decision (ADR-0017).
|
||||
|
||||
What that guarantees:
|
||||
|
||||
- Identical content with a **succeeded** job is recognized by
|
||||
`(tenant_id, domain, content_sha256)` and returned as-is with `200` — no
|
||||
re-ingestion, no duplicate points.
|
||||
- Identical content whose last job **failed** starts a fresh job against the
|
||||
same `source_files` row. A terminal job is never moved back to `running`.
|
||||
- Point ids are deterministic from `file_id` + `chunk_index` (ADR-0001), so the
|
||||
retry **overwrites in place** — it cannot duplicate chunks.
|
||||
- A failed attempt never empties or partially deletes a working index: the
|
||||
soft-delete sweep that retires a shortened file's leftover points runs only
|
||||
after every upsert has succeeded. An interrupted attempt can leave a prefix
|
||||
updated; a retry converges (ADR-0017, "Re-running an ingestion stays safe").
|
||||
|
||||
Fix the cause first — the `error_code` says where to look:
|
||||
|
||||
| `error_code` | Usual cause |
|
||||
|---|---|
|
||||
| `parse_failed` | The file is corrupt or is not really the type its extension claims. Retrying identical bytes will fail identically. |
|
||||
| `chunk_limit_exceeded` | The file is genuinely too large for one inline ingestion. Split it, or raise `INGESTION_MAX_CHUNKS_PER_FILE` knowing what §6 says about the timeout. |
|
||||
| `embedding_failed` | The embedder is down, rate-limiting, or unauthenticated. Fix it, then retry — this one usually succeeds unchanged. |
|
||||
| `index_failed` | Qdrant is down, or the collection is missing (run `qdrant_bootstrap`). |
|
||||
| `timeout` | The work exceeded `INGESTION_TIMEOUT_SECONDS`. Check whether the embedder was cold (see `lifespan.embedder.warm_failed` and `KEEP_ALIVE`) before raising the bound. |
|
||||
| `storage_upload_failed` | MinIO is unreachable or the bucket is missing (§3). |
|
||||
|
||||
## 11. What to alert on
|
||||
|
||||
ADR-0017's own triggers for moving ingestion back off the request path. These
|
||||
are the numbers that say the inline design has stopped fitting:
|
||||
|
||||
- **p95 ingestion duration** approaching `INGESTION_TIMEOUT_SECONDS`.
|
||||
- **`503` and `504` rates** ceasing to be negligible.
|
||||
- **Jobs stuck in `running` past the timeout** — every handled failure writes a
|
||||
terminal status, so a non-zero count here means the process died mid-request:
|
||||
|
||||
```sql
|
||||
SELECT count(*) FROM ingestion_jobs
|
||||
WHERE status = 'running' AND created_at < now() - interval '5 minutes';
|
||||
```
|
||||
|
||||
Also worth alerting: any `qdrant.bootstrap.schema_mismatch`, a sustained
|
||||
`/readyz` `503`, and any `api.unhandled_exception`.
|
||||
|
||||
## 12. Verifying a deployment
|
||||
|
||||
```bash
|
||||
./scripts/smoke.sh
|
||||
```
|
||||
|
||||
Brings up Compose, runs both deployment steps, provisions a throwaway tenant,
|
||||
starts the web process, and drives an upload through to indexed Qdrant points
|
||||
against the **running process** — including asserting the structured log output
|
||||
from §9. It is the only Compose-based test; everything else runs on
|
||||
Testcontainers under `uv run pytest` (ADR-0016). Run it before a release.
|
||||
98
scripts/smoke.sh
Executable file
98
scripts/smoke.sh
Executable file
@@ -0,0 +1,98 @@
|
||||
#!/usr/bin/env bash
|
||||
# Serialized operational smoke test of the running web process (ADR-0016, plan
|
||||
# 001 Phase 6).
|
||||
#
|
||||
# ./scripts/smoke.sh
|
||||
#
|
||||
# Brings up the Compose stack, runs both deployment steps for real, provisions
|
||||
# a tenant, starts uvicorn, and drives `tests/e2e/test_compose_smoke.py`
|
||||
# against it over a socket. This is the only Compose-based test: every other
|
||||
# test uses Testcontainers and an in-process ASGI transport, which is exactly
|
||||
# what makes this one worth having -- it is the only thing that exercises the
|
||||
# deployment steps, the real logging configuration, and a real HTTP server.
|
||||
#
|
||||
# Not part of `uv run pytest`: the smoke test skips itself unless SMOKE_BASE_URL
|
||||
# is set, so this script is the only way it runs. Run it before a release.
|
||||
#
|
||||
# Leaves the Compose stack running (it is the local dev stack); only the uvicorn
|
||||
# process and the temporary log file are cleaned up.
|
||||
set -euo pipefail
|
||||
|
||||
cd "$(dirname "${BASH_SOURCE[0]}")/.."
|
||||
|
||||
PORT="${SMOKE_PORT:-8021}"
|
||||
SLUG="smoke-$(date +%s)"
|
||||
DOMAIN="smoke"
|
||||
LOG_FILE="$(mktemp -t smoke-app-log.XXXXXX.jsonl)"
|
||||
CONSOLE_LOG="$(mktemp -t smoke-app-console.XXXXXX.log)"
|
||||
APP_PID=""
|
||||
|
||||
cleanup() {
|
||||
if [[ -n "${APP_PID}" ]] && kill -0 "${APP_PID}" 2>/dev/null; then
|
||||
kill "${APP_PID}" 2>/dev/null || true
|
||||
wait "${APP_PID}" 2>/dev/null || true
|
||||
fi
|
||||
rm -f "${LOG_FILE}" "${CONSOLE_LOG}"
|
||||
}
|
||||
trap cleanup EXIT
|
||||
|
||||
if [[ ! -f .env ]]; then
|
||||
echo "no .env found; copy .env.example first (see docs/runbook.md)" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "==> starting Postgres, MinIO, Qdrant"
|
||||
docker compose up -d --wait
|
||||
|
||||
echo "==> applying deployment steps"
|
||||
uv run alembic upgrade head
|
||||
uv run python -m src.cli.qdrant_bootstrap
|
||||
|
||||
echo "==> provisioning tenant '${SLUG}'"
|
||||
PROVISION_OUTPUT="$(uv run python -m src.cli.provision_tenant \
|
||||
--slug "${SLUG}" --domain "${DOMAIN}" --scopes files:write 2>/dev/null)"
|
||||
API_KEY="$(printf '%s\n' "${PROVISION_OUTPUT}" | sed -n 's/^api_key=//p')"
|
||||
if [[ -z "${API_KEY}" ]]; then
|
||||
echo "provisioning did not return an api_key" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "==> starting the web process on port ${PORT}"
|
||||
# JSON to a file sink, because the smoke test asserts the real ADR-0011 log
|
||||
# output -- the one thing no in-process test can check.
|
||||
LOG_JSON_FORMAT=true LOG_FILE_PATH="${LOG_FILE}" \
|
||||
uv run python -m uvicorn src.main:app --host 127.0.0.1 --port "${PORT}" \
|
||||
>"${CONSOLE_LOG}" 2>&1 &
|
||||
APP_PID=$!
|
||||
|
||||
echo "==> waiting for /readyz"
|
||||
for _ in $(seq 1 60); do
|
||||
if curl -fsS "http://127.0.0.1:${PORT}/readyz" >/dev/null 2>&1; then
|
||||
break
|
||||
fi
|
||||
if ! kill -0 "${APP_PID}" 2>/dev/null; then
|
||||
echo "the web process exited before becoming ready:" >&2
|
||||
tail -20 "${CONSOLE_LOG}" >&2
|
||||
exit 1
|
||||
fi
|
||||
sleep 1
|
||||
done
|
||||
|
||||
if ! curl -fsS "http://127.0.0.1:${PORT}/readyz" >/dev/null 2>&1; then
|
||||
# Most often an unbootstrapped Qdrant or an unreachable embedder host; the
|
||||
# runbook's health/readiness section covers reading this.
|
||||
echo "the web process never became ready:" >&2
|
||||
tail -20 "${CONSOLE_LOG}" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "==> running the smoke test"
|
||||
SMOKE_BASE_URL="http://127.0.0.1:${PORT}" \
|
||||
SMOKE_API_KEY="${API_KEY}" \
|
||||
SMOKE_DOMAIN="${DOMAIN}" \
|
||||
SMOKE_LOG_PATH="${LOG_FILE}" \
|
||||
SMOKE_QDRANT_URL="${QDRANT_URL:-http://127.0.0.1:6343}" \
|
||||
SMOKE_QDRANT_COLLECTION="${QDRANT_COLLECTION:-chunks}" \
|
||||
uv run python -m pytest tests/e2e/test_compose_smoke.py -q
|
||||
|
||||
echo "==> smoke test passed"
|
||||
@@ -97,7 +97,7 @@ def register_exception_handlers(app: FastAPI) -> None:
|
||||
@app.exception_handler(RequestValidationError)
|
||||
def _validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||
content=_envelope(
|
||||
"validation_error",
|
||||
"request validation failed",
|
||||
|
||||
9
src/application/tenants/__init__.py
Normal file
9
src/application/tenants/__init__.py
Normal file
@@ -0,0 +1,9 @@
|
||||
"""Operator-run tenant provisioning (ADR-0008, ADR-0009)."""
|
||||
|
||||
from src.application.tenants.provisioning import (
|
||||
DEFAULT_SCOPES,
|
||||
ProvisionResult,
|
||||
provision_tenant,
|
||||
)
|
||||
|
||||
__all__ = ["DEFAULT_SCOPES", "ProvisionResult", "provision_tenant"]
|
||||
123
src/application/tenants/provisioning.py
Normal file
123
src/application/tenants/provisioning.py
Normal file
@@ -0,0 +1,123 @@
|
||||
"""Provision a tenant, its first API key, and its domains (ADR-0008, ADR-0009).
|
||||
|
||||
Nothing in the HTTP surface can bootstrap a tenant: every `/v1` route needs a
|
||||
key, and a key can only exist once a tenant does. That chicken-and-egg is why
|
||||
this is an operator-run deployment step (`src/cli/provision_tenant.py`) rather
|
||||
than an endpoint — the same reasoning that keeps `alembic upgrade head` and
|
||||
`qdrant_bootstrap` off the request path.
|
||||
|
||||
This is the package's only caller-facing entry point. It owns the whole
|
||||
composition — tenant reuse-or-create, key generation and hashing, domain
|
||||
registration, and the single transaction the three share — so a caller cannot
|
||||
get the order wrong or commit a key whose tenant never landed (CLAUDE.md,
|
||||
"prefer deep modules"). The plaintext key is returned exactly once and is never
|
||||
logged (ADR-0011 forbids plaintext keys in logs); only its non-secret
|
||||
`key_prefix` appears in the event.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
|
||||
import structlog
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from src.application.auth.keys import generate_api_key, hash_secret
|
||||
from src.infrastructure.postgres.repositories import api_keys as api_keys_repo
|
||||
from src.infrastructure.postgres.repositories import tenant_domains as domains_repo
|
||||
from src.infrastructure.postgres.repositories import tenants as tenants_repo
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
DEFAULT_SCOPES = ("files:write", "domains:read", "domains:write")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProvisionResult:
|
||||
tenant_id: uuid.UUID
|
||||
tenant_slug: str
|
||||
tenant_created: bool
|
||||
api_key_id: uuid.UUID
|
||||
api_key_prefix: str
|
||||
api_key: str
|
||||
"""The plaintext bearer token. Only ever returned here — never stored, never logged."""
|
||||
domains_created: tuple[str, ...]
|
||||
domains_existing: tuple[str, ...]
|
||||
|
||||
|
||||
async def provision_tenant(
|
||||
sessionmaker: async_sessionmaker[AsyncSession],
|
||||
*,
|
||||
slug: str,
|
||||
name: str | None = None,
|
||||
key_name: str = "bootstrap",
|
||||
scopes: tuple[str, ...] = DEFAULT_SCOPES,
|
||||
domains: tuple[str, ...] = (),
|
||||
actor_type: str = "backend",
|
||||
) -> ProvisionResult:
|
||||
"""Create (or reuse) the tenant, issue a key, and register `domains`.
|
||||
|
||||
Re-running with the same `slug` reuses the tenant and its existing domains
|
||||
rather than failing, so an operator can add a key to a live tenant with the
|
||||
same command they used to create it. A *new* key is issued on every run —
|
||||
keys are write-once by construction (only the hash is stored), so there is
|
||||
nothing to return for an existing one.
|
||||
"""
|
||||
key_prefix, secret, full_key = generate_api_key()
|
||||
|
||||
async with sessionmaker() as session:
|
||||
tenant = await tenants_repo.get_by_slug(session, slug)
|
||||
tenant_created = tenant is None
|
||||
if tenant is None:
|
||||
tenant = tenants_repo.create(session, slug=slug, name=name or slug)
|
||||
# `api_keys.tenant_id` and `tenant_domains.tenant_id` FK to this row
|
||||
# and the mapped classes carry no ORM relationship for the unit of
|
||||
# work to order by itself, so the insert has to land first.
|
||||
await session.flush()
|
||||
|
||||
api_key = api_keys_repo.create(
|
||||
session,
|
||||
tenant_id=tenant.id,
|
||||
name=key_name,
|
||||
key_prefix=key_prefix,
|
||||
key_hash=hash_secret(secret),
|
||||
scopes=list(scopes),
|
||||
actor_type=actor_type,
|
||||
created_by="cli:provision_tenant",
|
||||
)
|
||||
|
||||
created: list[str] = []
|
||||
existing: list[str] = []
|
||||
for domain in domains:
|
||||
if await domains_repo.get(session, tenant_id=tenant.id, domain=domain) is not None:
|
||||
existing.append(domain)
|
||||
continue
|
||||
domains_repo.create(session, tenant_id=tenant.id, domain=domain, display_name=domain)
|
||||
created.append(domain)
|
||||
|
||||
await session.flush()
|
||||
tenant_id, api_key_id, tenant_slug = tenant.id, api_key.id, tenant.slug
|
||||
await session.commit()
|
||||
|
||||
if tenant_created:
|
||||
logger.info("tenant.provisioned", tenant_id=str(tenant_id), tenant_slug=tenant_slug)
|
||||
for domain in created:
|
||||
logger.info("domain.created", tenant_id=str(tenant_id), domain=domain)
|
||||
logger.info(
|
||||
"api_key.provisioned",
|
||||
tenant_id=str(tenant_id),
|
||||
api_key_id=str(api_key_id),
|
||||
key_prefix=key_prefix,
|
||||
scopes=list(scopes),
|
||||
actor_type=actor_type,
|
||||
)
|
||||
|
||||
return ProvisionResult(
|
||||
tenant_id=tenant_id,
|
||||
tenant_slug=tenant_slug,
|
||||
tenant_created=tenant_created,
|
||||
api_key_id=api_key_id,
|
||||
api_key_prefix=key_prefix,
|
||||
api_key=full_key,
|
||||
domains_created=tuple(created),
|
||||
domains_existing=tuple(existing),
|
||||
)
|
||||
94
src/cli/provision_tenant.py
Normal file
94
src/cli/provision_tenant.py
Normal file
@@ -0,0 +1,94 @@
|
||||
"""Create a tenant, issue its first API key, and register its domains.
|
||||
|
||||
uv run python -m src.cli.provision_tenant --slug acme --domain fire
|
||||
|
||||
A deployment step, like `alembic upgrade head` and `src.cli.qdrant_bootstrap`.
|
||||
It exists because nothing over HTTP can bootstrap a tenant: every `/v1` route
|
||||
requires an API key, and a key cannot exist before its tenant does.
|
||||
|
||||
The plaintext key is printed to **stdout once** and never stored or logged —
|
||||
Postgres holds only its SHA-256 hash (ADR-0009), so a lost key is reissued by
|
||||
re-running this command, not recovered. Structured logs go to stderr/the log
|
||||
sink and carry only the non-secret `key_prefix` (ADR-0011).
|
||||
|
||||
Re-running with the same `--slug` reuses the tenant and any domains it already
|
||||
has, and issues an additional key.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import sys
|
||||
|
||||
import structlog
|
||||
|
||||
from src.application.tenants import DEFAULT_SCOPES, provision_tenant
|
||||
from src.config import Settings
|
||||
from src.infrastructure.observability.logging import configure_logging
|
||||
from src.infrastructure.postgres.database import create_engine, create_sessionmaker
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="python -m src.cli.provision_tenant",
|
||||
description="Create a tenant, issue an API key, and register domains.",
|
||||
)
|
||||
parser.add_argument("--slug", required=True, help="URL-safe tenant identifier, e.g. 'acme'")
|
||||
parser.add_argument("--name", default=None, help="Display name (defaults to --slug)")
|
||||
parser.add_argument("--key-name", default="bootstrap", help="Label for the issued API key")
|
||||
parser.add_argument(
|
||||
"--scopes",
|
||||
default=",".join(DEFAULT_SCOPES),
|
||||
help=f"Comma-separated scopes for the key (default: {','.join(DEFAULT_SCOPES)})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--domain",
|
||||
action="append",
|
||||
default=[],
|
||||
dest="domains",
|
||||
help="Domain to register; repeatable. Uploads reject an unregistered domain.",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
async def run(argv: list[str] | None = None, settings: Settings | None = None) -> int:
|
||||
args = _parse_args(argv)
|
||||
resolved = settings or Settings()
|
||||
configure_logging(resolved.logging, resolved.app)
|
||||
|
||||
scopes = tuple(scope.strip() for scope in args.scopes.split(",") if scope.strip())
|
||||
if not scopes:
|
||||
logger.error("tenant.provision.failed", reason="no_scopes")
|
||||
return 2
|
||||
|
||||
engine = create_engine(resolved.postgres)
|
||||
try:
|
||||
result = await provision_tenant(
|
||||
create_sessionmaker(engine),
|
||||
slug=args.slug,
|
||||
name=args.name,
|
||||
key_name=args.key_name,
|
||||
scopes=scopes,
|
||||
domains=tuple(args.domains),
|
||||
)
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
# stdout, not the logger: this is the one value the operator must copy, and
|
||||
# it must never reach a log sink (ADR-0011).
|
||||
print(f"tenant_id={result.tenant_id}")
|
||||
print(f"tenant_slug={result.tenant_slug}")
|
||||
print(f"api_key_id={result.api_key_id}")
|
||||
print(f"domains={','.join(result.domains_created + result.domains_existing)}")
|
||||
print(f"api_key={result.api_key}")
|
||||
print("Store the api_key now -- only its hash is persisted and it cannot be shown again.")
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> None:
|
||||
sys.exit(asyncio.run(run()))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,4 +1,4 @@
|
||||
"""API-key lookups (ADR-0008, ADR-0009).
|
||||
"""API-key lookups and issuance (ADR-0008, ADR-0009).
|
||||
|
||||
Plain functions over an `AsyncSession` the caller owns. No function here
|
||||
commits, rolls back, or closes the session (ADR-0012). Secret comparison
|
||||
@@ -6,6 +6,8 @@ happens in `src/application/auth`, not here — this module only fetches rows
|
||||
by their non-secret `key_prefix`.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
@@ -15,3 +17,29 @@ from src.infrastructure.postgres.models.api_key import ApiKey
|
||||
async def get_by_prefix(session: AsyncSession, key_prefix: str) -> ApiKey | None:
|
||||
result = await session.execute(select(ApiKey).where(ApiKey.key_prefix == key_prefix))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
def create(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
tenant_id: uuid.UUID,
|
||||
name: str,
|
||||
key_prefix: str,
|
||||
key_hash: str,
|
||||
scopes: list[str],
|
||||
actor_type: str = "backend",
|
||||
created_by: str | None = None,
|
||||
) -> ApiKey:
|
||||
"""Persist an issued key. The caller hashes the secret; this never sees it."""
|
||||
api_key = ApiKey(
|
||||
id=uuid.uuid4(),
|
||||
tenant_id=tenant_id,
|
||||
name=name,
|
||||
key_prefix=key_prefix,
|
||||
key_hash=key_hash,
|
||||
scopes=scopes,
|
||||
actor_type=actor_type,
|
||||
created_by=created_by,
|
||||
)
|
||||
session.add(api_key)
|
||||
return api_key
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Tenant lookups (ADR-0009).
|
||||
"""Tenant lookups and creation (ADR-0009).
|
||||
|
||||
Plain functions over an `AsyncSession` the caller owns. No function here
|
||||
commits, rolls back, or closes the session (ADR-0012).
|
||||
@@ -6,6 +6,7 @@ commits, rolls back, or closes the session (ADR-0012).
|
||||
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.infrastructure.postgres.models.tenant import Tenant
|
||||
@@ -13,3 +14,14 @@ from src.infrastructure.postgres.models.tenant import Tenant
|
||||
|
||||
async def get_by_id(session: AsyncSession, tenant_id: uuid.UUID) -> Tenant | None:
|
||||
return await session.get(Tenant, tenant_id)
|
||||
|
||||
|
||||
async def get_by_slug(session: AsyncSession, slug: str) -> Tenant | None:
|
||||
result = await session.execute(select(Tenant).where(Tenant.slug == slug))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
def create(session: AsyncSession, *, slug: str, name: str) -> Tenant:
|
||||
tenant = Tenant(id=uuid.uuid4(), slug=slug, name=name)
|
||||
session.add(tenant)
|
||||
return tenant
|
||||
|
||||
@@ -10,6 +10,12 @@ from httpx import ASGITransport, AsyncClient
|
||||
from src.config import Settings
|
||||
from src.main import create_app
|
||||
|
||||
# Disposable Postgres/MinIO/Qdrant containers (ADR-0016). Registered here
|
||||
# because `pytest_plugins` is only honoured in the root conftest, and both
|
||||
# `tests/integration/*/` and `tests/e2e/` need the same containers. The
|
||||
# fixtures are session-scoped but lazy, so unit runs still need no Docker.
|
||||
pytest_plugins = ["tests.support.containers"]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_structlog_after_test() -> Iterator[None]:
|
||||
|
||||
244
tests/e2e/conftest.py
Normal file
244
tests/e2e/conftest.py
Normal file
@@ -0,0 +1,244 @@
|
||||
"""The vertical slice against real Postgres, MinIO, and Qdrant (ADR-0016).
|
||||
|
||||
These are the ADR-0016 end-to-end tests: "a small vertical-slice acceptance
|
||||
contract through the real application composition". They run on
|
||||
Testcontainers, in the default `uv run pytest` run, because ADR-0016 makes
|
||||
Testcontainers "the standard automated integration-test resource mechanism"
|
||||
and reserves Docker Compose for manual validation and the separate serialized
|
||||
smoke test of the *running web process* (`tests/e2e/test_compose_smoke.py`).
|
||||
|
||||
What is real here: routing, auth, scopes, the domain allowlist, the ADR-0008
|
||||
error envelope, the two-transaction upload, MinIO object writes, BM25 sparse
|
||||
embedding, and Qdrant upserts into a per-test collection created by the
|
||||
production `ensure_chunks_collection`.
|
||||
|
||||
What is faked, and only this: the two dense embedders. They are paid/remote
|
||||
network calls, and ADR-0016 forbids routine runs from touching a live
|
||||
provider. Their fakes return the pinned 768/3072 dimensions, because the real
|
||||
collection rejects anything else.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Callable, Coroutine
|
||||
from contextlib import AsyncExitStack
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from asgi_lifespan import LifespanManager
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from qdrant_client import AsyncQdrantClient, models
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
||||
|
||||
from src.bootstrap.dependencies import get_dense_embedders, get_sessionmaker
|
||||
from src.config import MinioSettings, PostgresSettings, QdrantSettings, Settings
|
||||
from src.infrastructure.postgres.database import create_sessionmaker
|
||||
from src.infrastructure.qdrant.collection import (
|
||||
DENSE_NOMIC_DIMENSIONS,
|
||||
DENSE_OPENAI_DIMENSIONS,
|
||||
ensure_chunks_collection,
|
||||
)
|
||||
from src.main import create_app
|
||||
from tests.fakes import FakeDenseEmbedder
|
||||
from tests.support.factories import create_api_key, create_tenant, create_tenant_domain
|
||||
|
||||
CSV_BYTES = b"question,answer\nhow do I file a claim,call the branch\nwhat is covered,see policy\n"
|
||||
|
||||
# Past validation's OOXML magic-byte check, so the failure lands in the DOCX
|
||||
# parser rather than in upload validation -- the branch this exercises.
|
||||
CORRUPT_DOCX_BYTES = b"PK\x03\x04" + b"not really a docx" * 8
|
||||
|
||||
|
||||
@dataclass
|
||||
class E2EStack:
|
||||
"""One configured app instance plus the handles a test asserts against."""
|
||||
|
||||
client: AsyncClient
|
||||
collection: str
|
||||
dense_embedders: list[FakeDenseEmbedder]
|
||||
|
||||
|
||||
StackFactory = Callable[..., Coroutine[Any, Any, E2EStack]]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Principal:
|
||||
"""A provisioned tenant with a bearer token and a registered domain.
|
||||
|
||||
Plain values, not the ORM `Tenant`: tests re-read rows the app committed on
|
||||
a shared connection, and any expiry/refresh of a held ORM instance would
|
||||
lazy-load outside a greenlet context (`MissingGreenlet`).
|
||||
"""
|
||||
|
||||
tenant_id: uuid.UUID
|
||||
tenant_slug: str
|
||||
token: str
|
||||
domain: str
|
||||
|
||||
@property
|
||||
def headers(self) -> dict[str, str]:
|
||||
return {"Authorization": f"Bearer {self.token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def e2e_settings(
|
||||
postgres_settings: PostgresSettings,
|
||||
minio_settings: MinioSettings,
|
||||
qdrant_settings: QdrantSettings,
|
||||
) -> Settings:
|
||||
"""Settings wired to this test's containers.
|
||||
|
||||
Every nested settings object is passed explicitly. `src/config.py` cascades
|
||||
`.env` into each nested class, so omitting one would let a developer's real
|
||||
`EMBEDDING_NOMIC_BASE_URL` (a colleague's Ollama box) into a routine test
|
||||
run. The embedder URLs below point at a closed port for the same reason:
|
||||
the lifespan warms the *real* embedders before the fakes are substituted,
|
||||
and connection-refused is both instant and free.
|
||||
"""
|
||||
return Settings(
|
||||
postgres=postgres_settings,
|
||||
minio=minio_settings,
|
||||
qdrant=qdrant_settings,
|
||||
embedding={
|
||||
"nomic": {"base_url": "http://127.0.0.1:1/v1", "timeout_seconds": 0.5},
|
||||
"openai": {"base_url": "http://127.0.0.1:1/v1", "timeout_seconds": 0.5},
|
||||
},
|
||||
app={"readiness_check_timeout_seconds": 2.0},
|
||||
)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def make_stack(
|
||||
e2e_settings: Settings,
|
||||
db_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
qdrant_client: AsyncQdrantClient,
|
||||
) -> AsyncIterator[StackFactory]:
|
||||
"""Build an app under test, optionally with tightened ingestion bounds.
|
||||
|
||||
A factory rather than a fixture because the capacity and timeout tests need
|
||||
their own `INGESTION_*` values, and those are read at app construction.
|
||||
"""
|
||||
async with AsyncExitStack() as exit_stack:
|
||||
|
||||
async def _make(
|
||||
*,
|
||||
ingestion: dict[str, object] | None = None,
|
||||
collection: str | None = None,
|
||||
dense_delay_seconds: float = 0.0,
|
||||
bootstrap_collection: bool = True,
|
||||
sessionmaker: async_sessionmaker[AsyncSession] | None = None,
|
||||
) -> E2EStack:
|
||||
resolved_collection = collection or e2e_settings.qdrant.collection
|
||||
if bootstrap_collection:
|
||||
await ensure_chunks_collection(qdrant_client, collection=resolved_collection)
|
||||
|
||||
settings = e2e_settings.model_copy(
|
||||
update={
|
||||
"qdrant": e2e_settings.qdrant.model_copy(
|
||||
update={"collection": resolved_collection}
|
||||
),
|
||||
"ingestion": e2e_settings.ingestion.model_copy(update=ingestion or {}),
|
||||
}
|
||||
)
|
||||
|
||||
dense = [
|
||||
FakeDenseEmbedder(
|
||||
name="dense_nomic",
|
||||
dimensions=DENSE_NOMIC_DIMENSIONS,
|
||||
value=0.1,
|
||||
delay_seconds=dense_delay_seconds,
|
||||
),
|
||||
FakeDenseEmbedder(
|
||||
name="dense_openai",
|
||||
dimensions=DENSE_OPENAI_DIMENSIONS,
|
||||
value=0.2,
|
||||
delay_seconds=dense_delay_seconds,
|
||||
),
|
||||
]
|
||||
|
||||
app = create_app(settings)
|
||||
# The session factory is the test's, so factory-created tenants are
|
||||
# visible to the app and every write rolls back at teardown. Object
|
||||
# storage, point storage, and the sparse embedder stay real.
|
||||
resolved_sessionmaker = sessionmaker or db_sessionmaker
|
||||
app.dependency_overrides[get_sessionmaker] = lambda: resolved_sessionmaker
|
||||
app.dependency_overrides[get_dense_embedders] = lambda: dense
|
||||
|
||||
manager = await exit_stack.enter_async_context(LifespanManager(app))
|
||||
client = await exit_stack.enter_async_context(
|
||||
AsyncClient(
|
||||
transport=ASGITransport(app=manager.app),
|
||||
base_url="http://test",
|
||||
timeout=30.0,
|
||||
)
|
||||
)
|
||||
return E2EStack(client=client, collection=resolved_collection, dense_embedders=dense)
|
||||
|
||||
yield _make
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def committed_sessionmaker(
|
||||
postgres_engine: AsyncEngine,
|
||||
) -> async_sessionmaker[AsyncSession]:
|
||||
"""A session factory whose writes really commit, one connection each.
|
||||
|
||||
The default `db_sessionmaker` pins every session to a single connection
|
||||
inside one rolled-back transaction, which is what isolates a test's writes
|
||||
-- but savepoints on a shared connection cannot interleave, so any test
|
||||
that issues genuinely *concurrent* requests deadlocks or fails with
|
||||
`InvalidSavepointSpecificationError`. Those tests use this instead and rely
|
||||
on their uuid-keyed tenant for isolation (ADR-0016 allows either).
|
||||
"""
|
||||
return create_sessionmaker(postgres_engine)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def principal(db_session: AsyncSession) -> Principal:
|
||||
"""A tenant with an upload-scoped key and one registered domain.
|
||||
|
||||
Registering the domain is not optional: `POST /v1/files` rejects an
|
||||
unregistered one with `400 unknown_domain` before anything else runs
|
||||
(ADR-0009).
|
||||
"""
|
||||
return await provision_principal(db_session)
|
||||
|
||||
|
||||
async def provision_principal(
|
||||
session: AsyncSession, *, domain: str = "fire", scopes: list[str] | None = None
|
||||
) -> Principal:
|
||||
tenant = await create_tenant(session)
|
||||
_, token = await create_api_key(session, tenant=tenant, scopes=scopes or ["files:write"])
|
||||
await create_tenant_domain(session, tenant=tenant, domain=domain)
|
||||
await session.commit()
|
||||
return Principal(tenant_id=tenant.id, tenant_slug=tenant.slug, token=token, domain=domain)
|
||||
|
||||
|
||||
def upload_payload(
|
||||
*, domain: str, filename: str = "faq.csv", data: bytes = CSV_BYTES
|
||||
) -> dict[str, Any]:
|
||||
return {"files": {"file": (filename, data, "text/csv")}, "data": {"domain": domain}}
|
||||
|
||||
|
||||
async def count_active_points(
|
||||
client: AsyncQdrantClient, collection: str, *, tenant_id: uuid.UUID
|
||||
) -> int:
|
||||
"""Active points for one tenant, read through the raw client.
|
||||
|
||||
`PointStorage` is deliberately write-only (point reads are plan 002's
|
||||
`/v1/points`), so the verification read here uses the SDK directly.
|
||||
"""
|
||||
result = await client.count(
|
||||
collection_name=collection,
|
||||
count_filter=models.Filter(
|
||||
must=[
|
||||
models.FieldCondition(
|
||||
key="tenant_id", match=models.MatchValue(value=str(tenant_id))
|
||||
),
|
||||
models.FieldCondition(key="is_active", match=models.MatchValue(value=True)),
|
||||
]
|
||||
),
|
||||
exact=True,
|
||||
)
|
||||
return result.count
|
||||
143
tests/e2e/test_compose_smoke.py
Normal file
143
tests/e2e/test_compose_smoke.py
Normal file
@@ -0,0 +1,143 @@
|
||||
"""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"]
|
||||
329
tests/e2e/test_ingestion_slice.py
Normal file
329
tests/e2e/test_ingestion_slice.py
Normal file
@@ -0,0 +1,329 @@
|
||||
"""The ingestion vertical slice end to end (plan 001, Phase 6).
|
||||
|
||||
Every test here asserts a reliability invariant ADR-0016 lists as a reusable
|
||||
contract, not a happy path: the bound maps to its status code, the failure
|
||||
still writes a terminal job row, the retry converges, and the tenant filter
|
||||
holds. Failures additionally assert the ADR-0008 error envelope, because the
|
||||
status code alone is not the contract a client integrates against.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from httpx import Response
|
||||
from qdrant_client import AsyncQdrantClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from src.infrastructure.postgres.models.ingestion_job import IngestionJob
|
||||
from tests.e2e.conftest import (
|
||||
CORRUPT_DOCX_BYTES,
|
||||
Principal,
|
||||
StackFactory,
|
||||
count_active_points,
|
||||
provision_principal,
|
||||
upload_payload,
|
||||
)
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.e2e,
|
||||
pytest.mark.postgres,
|
||||
pytest.mark.minio,
|
||||
pytest.mark.qdrant,
|
||||
pytest.mark.asyncio(loop_scope="session"),
|
||||
]
|
||||
|
||||
|
||||
async def _latest_job(session: AsyncSession, *, tenant_id: uuid.UUID) -> IngestionJob:
|
||||
"""The tenant's most recent job row, read fresh.
|
||||
|
||||
`populate_existing` matters: this session shares a connection with the
|
||||
app's sessions, so without it the identity map can answer from a row
|
||||
loaded before the request under test committed its update.
|
||||
"""
|
||||
result = await session.execute(
|
||||
select(IngestionJob)
|
||||
.where(IngestionJob.tenant_id == tenant_id)
|
||||
.order_by(IngestionJob.created_at.desc(), IngestionJob.id)
|
||||
.execution_options(populate_existing=True)
|
||||
)
|
||||
return result.scalars().first() or pytest.fail("no ingestion job was written")
|
||||
|
||||
|
||||
def _envelope(response: Response) -> dict[str, object]:
|
||||
"""The ADR-0008 error body. Asserting the code here, not just the HTTP
|
||||
status, is the difference between testing the contract and testing the
|
||||
number."""
|
||||
error: dict[str, object] = response.json()["error"]
|
||||
assert error["request_id"], "every error envelope carries its correlation id"
|
||||
return error
|
||||
|
||||
|
||||
async def test_upload_succeeds_and_indexes_points_under_the_tenant_filter(
|
||||
make_stack: StackFactory, principal: Principal, qdrant_client: AsyncQdrantClient
|
||||
) -> None:
|
||||
stack = await make_stack()
|
||||
|
||||
response = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
body = response.json()
|
||||
assert body["status"] == "succeeded"
|
||||
assert body["chunks_indexed"] > 0
|
||||
indexed = await count_active_points(
|
||||
qdrant_client, stack.collection, tenant_id=principal.tenant_id
|
||||
)
|
||||
assert indexed == body["chunks_indexed"]
|
||||
|
||||
|
||||
async def test_get_file_reports_the_terminal_status_after_upload(
|
||||
make_stack: StackFactory, principal: Principal
|
||||
) -> None:
|
||||
stack = await make_stack()
|
||||
created = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
|
||||
status = await stack.client.get(
|
||||
f"/v1/files/{created.json()['file_id']}", headers=principal.headers
|
||||
)
|
||||
|
||||
assert status.status_code == 200
|
||||
assert status.json()["ingestion_status"] == "succeeded"
|
||||
assert status.json()["domain"] == principal.domain
|
||||
|
||||
|
||||
async def test_upload_duplicate_content_returns_the_existing_job_without_reingesting(
|
||||
make_stack: StackFactory, principal: Principal, qdrant_client: AsyncQdrantClient
|
||||
) -> None:
|
||||
stack = await make_stack()
|
||||
first = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
points_after_first = await count_active_points(
|
||||
qdrant_client, stack.collection, tenant_id=principal.tenant_id
|
||||
)
|
||||
|
||||
second = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
|
||||
assert first.status_code == 201
|
||||
# 200, not 201: nothing was created the second time (ADR-0017 idempotency).
|
||||
assert second.status_code == 200
|
||||
assert second.json()["file_id"] == first.json()["file_id"]
|
||||
assert second.json()["ingestion_job_id"] == first.json()["ingestion_job_id"]
|
||||
assert (
|
||||
await count_active_points(qdrant_client, stack.collection, tenant_id=principal.tenant_id)
|
||||
== points_after_first
|
||||
)
|
||||
|
||||
|
||||
async def test_upload_retried_after_a_failed_job_succeeds_without_duplicate_points(
|
||||
make_stack: StackFactory,
|
||||
principal: Principal,
|
||||
db_session: AsyncSession,
|
||||
qdrant_client: AsyncQdrantClient,
|
||||
) -> None:
|
||||
"""ADR-0017, "Re-running an ingestion stays safe": a failed attempt is
|
||||
retried by re-uploading the same bytes, and deterministic point ids make
|
||||
the retry converge instead of duplicating.
|
||||
"""
|
||||
stack = await make_stack()
|
||||
stack.dense_embedders[0].fail_next = True
|
||||
|
||||
failed = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
failed_job = await _latest_job(db_session, tenant_id=principal.tenant_id)
|
||||
retried = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
|
||||
assert failed.status_code == 502
|
||||
assert failed_job.status == "failed"
|
||||
assert failed_job.error_code == "embedding_failed"
|
||||
assert retried.status_code == 201
|
||||
assert retried.json()["status"] == "succeeded"
|
||||
# Same source file, new attempt -- a terminal job is never reused or
|
||||
# transitioned back to running.
|
||||
assert retried.json()["ingestion_job_id"] != str(failed_job.id)
|
||||
indexed = await count_active_points(
|
||||
qdrant_client, stack.collection, tenant_id=principal.tenant_id
|
||||
)
|
||||
assert indexed == retried.json()["chunks_indexed"]
|
||||
|
||||
|
||||
async def test_get_file_for_another_tenants_file_returns_404_not_403(
|
||||
make_stack: StackFactory, principal: Principal, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""404, never 403: a 403 would confirm the file exists to a stranger."""
|
||||
stack = await make_stack()
|
||||
created = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
other = await provision_principal(db_session)
|
||||
|
||||
response = await stack.client.get(
|
||||
f"/v1/files/{created.json()['file_id']}", headers=other.headers
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert _envelope(response)["code"] == "not_found"
|
||||
|
||||
|
||||
async def test_upload_to_an_unregistered_domain_is_rejected_before_any_job_is_written(
|
||||
make_stack: StackFactory, principal: Principal, db_session: AsyncSession
|
||||
) -> None:
|
||||
stack = await make_stack()
|
||||
|
||||
response = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain="never-registered"), headers=principal.headers
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert _envelope(response)["code"] == "unknown_domain"
|
||||
jobs = await db_session.execute(
|
||||
select(IngestionJob).where(IngestionJob.tenant_id == principal.tenant_id)
|
||||
)
|
||||
assert jobs.scalars().all() == []
|
||||
|
||||
|
||||
async def test_upload_at_capacity_is_rejected_with_503_and_retry_after(
|
||||
make_stack: StackFactory, committed_sessionmaker: async_sessionmaker[AsyncSession]
|
||||
) -> None:
|
||||
"""ADR-0017 rejects rather than queues, so the second concurrent upload
|
||||
must fail fast with a backoff hint instead of waiting for a slot.
|
||||
|
||||
This is the one test that needs two requests genuinely in flight at once,
|
||||
so it runs on `committed_sessionmaker` -- see that fixture for why the
|
||||
shared-connection default cannot serve concurrent sessions.
|
||||
"""
|
||||
async with committed_sessionmaker() as session:
|
||||
principal = await provision_principal(session)
|
||||
stack = await make_stack(
|
||||
ingestion={"max_concurrency": 1},
|
||||
dense_delay_seconds=0.4,
|
||||
sessionmaker=committed_sessionmaker,
|
||||
)
|
||||
|
||||
first, second = await asyncio.gather(
|
||||
stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
),
|
||||
stack.client.post(
|
||||
"/v1/files",
|
||||
**upload_payload(domain=principal.domain, filename="other.csv", data=b"a,b\n1,2\n"),
|
||||
headers=principal.headers,
|
||||
),
|
||||
)
|
||||
|
||||
statuses = sorted([first.status_code, second.status_code])
|
||||
assert statuses == [201, 503]
|
||||
rejected = first if first.status_code == 503 else second
|
||||
assert _envelope(rejected)["code"] == "ingestion_at_capacity"
|
||||
assert rejected.headers["Retry-After"] == "1"
|
||||
|
||||
|
||||
async def test_upload_exceeding_the_timeout_returns_504_and_writes_a_terminal_job(
|
||||
make_stack: StackFactory, principal: Principal, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""A timeout must never leave a job stuck in `running` (ADR-0016)."""
|
||||
stack = await make_stack(ingestion={"timeout_seconds": 0.05}, dense_delay_seconds=0.5)
|
||||
|
||||
response = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
|
||||
assert response.status_code == 504
|
||||
assert _envelope(response)["code"] == "ingestion_timeout"
|
||||
job = await _latest_job(db_session, tenant_id=principal.tenant_id)
|
||||
assert job.status == "failed"
|
||||
assert job.error_code == "timeout"
|
||||
|
||||
|
||||
async def test_upload_of_an_unparseable_document_returns_400_and_a_failed_job(
|
||||
make_stack: StackFactory, principal: Principal, db_session: AsyncSession
|
||||
) -> None:
|
||||
stack = await make_stack()
|
||||
|
||||
response = await stack.client.post(
|
||||
"/v1/files",
|
||||
**upload_payload(domain=principal.domain, filename="broken.docx", data=CORRUPT_DOCX_BYTES),
|
||||
headers=principal.headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert _envelope(response)["code"] == "validation_error"
|
||||
job = await _latest_job(db_session, tenant_id=principal.tenant_id)
|
||||
assert job.status == "failed"
|
||||
assert job.error_code == "parse_failed"
|
||||
|
||||
|
||||
async def test_upload_returns_502_when_qdrant_indexing_fails(
|
||||
make_stack: StackFactory, principal: Principal, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""A real Qdrant error, not a fake: the collection the deployment step
|
||||
should have created is missing, which is the failure an operator actually
|
||||
hits when `qdrant_bootstrap` was skipped.
|
||||
"""
|
||||
stack = await make_stack(
|
||||
collection=f"never_bootstrapped_{uuid.uuid4().hex}", bootstrap_collection=False
|
||||
)
|
||||
|
||||
response = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
|
||||
)
|
||||
|
||||
assert response.status_code == 502
|
||||
assert _envelope(response)["code"] == "index_error"
|
||||
job = await _latest_job(db_session, tenant_id=principal.tenant_id)
|
||||
assert job.status == "failed"
|
||||
assert job.error_code == "index_failed"
|
||||
|
||||
|
||||
async def test_upload_without_the_files_write_scope_is_forbidden(
|
||||
make_stack: StackFactory, db_session: AsyncSession
|
||||
) -> None:
|
||||
stack = await make_stack()
|
||||
reader = await provision_principal(db_session, scopes=["domains:read"])
|
||||
|
||||
response = await stack.client.post(
|
||||
"/v1/files", **upload_payload(domain=reader.domain), headers=reader.headers
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert _envelope(response)["code"] == "missing_scope"
|
||||
|
||||
|
||||
async def test_readyz_reports_every_dependency_ready_against_real_services(
|
||||
make_stack: StackFactory,
|
||||
) -> None:
|
||||
"""`/readyz` is dependency readiness, `/healthz` is process health."""
|
||||
stack = await make_stack()
|
||||
|
||||
ready = await stack.client.get("/readyz")
|
||||
healthy = await stack.client.get("/healthz")
|
||||
|
||||
assert healthy.status_code == 200
|
||||
assert ready.status_code == 200
|
||||
assert ready.json() == {"postgres": True, "minio": True, "qdrant": True}
|
||||
|
||||
|
||||
async def test_readyz_is_unready_when_the_chunks_collection_is_missing(
|
||||
make_stack: StackFactory,
|
||||
) -> None:
|
||||
"""A reachable but unbootstrapped Qdrant is deliberately not ready --
|
||||
uploads to it would 502 (see the indexing test above).
|
||||
"""
|
||||
stack = await make_stack(
|
||||
collection=f"never_bootstrapped_{uuid.uuid4().hex}", bootstrap_collection=False
|
||||
)
|
||||
|
||||
response = await stack.client.get("/readyz")
|
||||
|
||||
assert response.status_code == 503
|
||||
assert response.json()["qdrant"] is False
|
||||
@@ -32,6 +32,9 @@ class FakeDenseEmbedder:
|
||||
|
||||
name: str
|
||||
dimensions: int = 4
|
||||
value: float = 0.0
|
||||
"""Component value of every returned vector. Non-zero where a real Qdrant
|
||||
has to score the result, since a zero vector has no direction to compare."""
|
||||
model_version: str = "fake-dense-v1"
|
||||
calls: list[list[str]] = field(default_factory=list)
|
||||
fail_next: bool = False
|
||||
@@ -45,7 +48,7 @@ class FakeDenseEmbedder:
|
||||
if self.fail_next:
|
||||
self.fail_next = False
|
||||
raise RuntimeError("simulated embedder failure")
|
||||
return [[0.0] * self.dimensions for _ in texts]
|
||||
return [[self.value] * self.dimensions for _ in texts]
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
@@ -1,39 +0,0 @@
|
||||
from collections.abc import Iterator
|
||||
|
||||
import pytest
|
||||
from testcontainers.community.minio import MinioContainer
|
||||
|
||||
from src.config import MinioSettings
|
||||
from src.infrastructure.minio.client import create_client
|
||||
|
||||
_BUCKET = "test-source-files"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def minio_container() -> Iterator[MinioContainer]:
|
||||
with MinioContainer() as container:
|
||||
yield container
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def minio_settings(minio_container: MinioContainer) -> MinioSettings:
|
||||
config = minio_container.get_config()
|
||||
# Pinned to IPv4 for the same reason as postgres_url (see
|
||||
# tests/integration/postgres/conftest.py): `localhost` resolves to `::1`
|
||||
# first, but Docker only publishes the mapped port on IPv4, so the
|
||||
# connection hangs instead of failing.
|
||||
endpoint = config["endpoint"].replace("localhost:", "127.0.0.1:")
|
||||
return MinioSettings(
|
||||
endpoint=endpoint,
|
||||
access_key=config["access_key"],
|
||||
secret_key=config["secret_key"],
|
||||
secure=False,
|
||||
bucket=_BUCKET,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def _ensure_bucket(minio_settings: MinioSettings) -> None:
|
||||
client = create_client(minio_settings)
|
||||
if not client.bucket_exists(minio_settings.bucket):
|
||||
client.make_bucket(minio_settings.bucket)
|
||||
@@ -1,98 +0,0 @@
|
||||
from collections.abc import AsyncIterator, Iterator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from alembic.command import upgrade
|
||||
from alembic.config import Config
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
||||
from testcontainers.community.postgres import PostgresContainer
|
||||
|
||||
from src.config import PostgresSettings
|
||||
from src.infrastructure.postgres.database import create_engine
|
||||
|
||||
|
||||
def _alembic_config(database_url: str) -> Config:
|
||||
config = Config("alembic.ini")
|
||||
config.set_main_option("sqlalchemy.url", database_url)
|
||||
return config
|
||||
|
||||
|
||||
def _settings_from_url(url: str) -> PostgresSettings:
|
||||
# testcontainers returns postgresql+asyncpg://user:pass@host:port/db ;
|
||||
# PostgresSettings builds its own dsn from parts, so parse the parts back out.
|
||||
without_scheme = url.split("://", 1)[1]
|
||||
creds, hostpart = without_scheme.split("@", 1)
|
||||
user, password = creds.split(":", 1)
|
||||
hostport, db = hostpart.split("/", 1)
|
||||
host, port = hostport.split(":", 1)
|
||||
return PostgresSettings(host=host, port=int(port), user=user, password=password, db=db)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def postgres_container() -> Iterator[PostgresContainer]:
|
||||
with PostgresContainer("postgres:17", driver="asyncpg") as container:
|
||||
yield container
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def postgres_url(postgres_container: PostgresContainer) -> str:
|
||||
"""The container's URL, pinned to IPv4.
|
||||
|
||||
Testcontainers reports the host as `localhost`, which resolves to `::1`
|
||||
before `127.0.0.1`. Docker publishes the mapped port on IPv4 only, and the
|
||||
IPv6 SYN is dropped rather than refused, so asyncpg blocks on the first
|
||||
address until its connect timeout instead of falling back to the second --
|
||||
the connection does not fail, it hangs.
|
||||
"""
|
||||
return postgres_container.get_connection_url().replace("@localhost:", "@127.0.0.1:")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def migrated_postgres_url(postgres_url: str) -> str:
|
||||
"""The container's URL, after Alembic has created the schema on it once."""
|
||||
upgrade(_alembic_config(postgres_url), "head")
|
||||
return postgres_url
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||||
async def postgres_engine(migrated_postgres_url: str) -> AsyncIterator[AsyncEngine]:
|
||||
engine = create_engine(_settings_from_url(migrated_postgres_url))
|
||||
try:
|
||||
yield engine
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def db_sessionmaker(
|
||||
postgres_engine: AsyncEngine,
|
||||
) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
|
||||
"""A session *factory* per test, bound to a rolled-back outer transaction.
|
||||
|
||||
Every session it produces shares one connection/outer transaction, so
|
||||
writes `commit()`ed by one session are visible to the next -- needed for
|
||||
code under test that opens more than one session per operation (auth
|
||||
resolution, the ADR-0017 two-phase upload) -- while the whole test's
|
||||
writes still roll back together at teardown (ADR-0016: isolate data per
|
||||
test).
|
||||
"""
|
||||
async with postgres_engine.connect() as connection:
|
||||
outer_transaction = await connection.begin()
|
||||
sessionmaker = async_sessionmaker(
|
||||
bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
|
||||
)
|
||||
yield sessionmaker
|
||||
await outer_transaction.rollback()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def db_session(
|
||||
db_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
) -> AsyncIterator[AsyncSession]:
|
||||
"""One session per test, bound to a rolled-back outer transaction.
|
||||
|
||||
Isolates each test's writes (ADR-0016: isolate data per test) without
|
||||
needing a fresh container or unique keys per test.
|
||||
"""
|
||||
async with db_sessionmaker() as session:
|
||||
yield session
|
||||
95
tests/integration/postgres/test_provisioning.py
Normal file
95
tests/integration/postgres/test_provisioning.py
Normal file
@@ -0,0 +1,95 @@
|
||||
"""Tenant provisioning against real Postgres (ADR-0009).
|
||||
|
||||
The property worth a real database here is the one a fake cannot show: the
|
||||
issued key authenticates through the *production* auth path, and the row it
|
||||
authenticates against holds no plaintext.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from src.application.auth.service import resolve_auth_context
|
||||
from src.application.domains import ensure_domain_allowed
|
||||
from src.application.domains.errors import UnknownDomainError
|
||||
from src.application.tenants import provision_tenant
|
||||
from src.infrastructure.postgres.models.api_key import ApiKey
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.integration,
|
||||
pytest.mark.postgres,
|
||||
pytest.mark.asyncio(loop_scope="session"),
|
||||
]
|
||||
|
||||
|
||||
async def test_provision_tenant_issues_a_key_that_authenticates(
|
||||
db_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
result = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
|
||||
|
||||
auth = await resolve_auth_context(db_sessionmaker, result.api_key)
|
||||
|
||||
assert auth.tenant_id == result.tenant_id
|
||||
assert auth.tenant_slug == "acme"
|
||||
assert auth.api_key_id == result.api_key_id
|
||||
assert "files:write" in auth.scopes
|
||||
|
||||
|
||||
async def test_provision_tenant_persists_only_the_key_hash(
|
||||
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
||||
) -> None:
|
||||
"""A plaintext key in Postgres would make every later hashing decision moot."""
|
||||
result = await provision_tenant(db_sessionmaker, slug="acme")
|
||||
|
||||
stored = (
|
||||
await db_session.execute(select(ApiKey).where(ApiKey.id == result.api_key_id))
|
||||
).scalar_one()
|
||||
|
||||
assert result.api_key not in stored.key_hash
|
||||
assert stored.key_hash != result.api_key
|
||||
assert stored.key_prefix == result.api_key_prefix
|
||||
assert result.api_key.startswith(f"sk_{result.api_key_prefix}_")
|
||||
|
||||
|
||||
async def test_provision_tenant_registers_domains_so_uploads_are_allowed(
|
||||
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
||||
) -> None:
|
||||
result = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
|
||||
|
||||
await ensure_domain_allowed(db_session, tenant_id=result.tenant_id, domain="fire")
|
||||
with pytest.raises(UnknownDomainError):
|
||||
await ensure_domain_allowed(db_session, tenant_id=result.tenant_id, domain="life")
|
||||
|
||||
assert result.domains_created == ("fire",)
|
||||
|
||||
|
||||
async def test_provision_tenant_rerun_reuses_the_tenant_and_issues_a_new_key(
|
||||
db_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""Adding a key to a live tenant must not need a different command."""
|
||||
first = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
|
||||
|
||||
second = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire", "life"))
|
||||
|
||||
assert second.tenant_id == first.tenant_id
|
||||
assert second.tenant_created is False
|
||||
assert second.api_key_id != first.api_key_id
|
||||
assert second.domains_created == ("life",)
|
||||
assert second.domains_existing == ("fire",)
|
||||
# Both keys stay valid -- reprovisioning adds a key, it does not rotate one.
|
||||
assert (await resolve_auth_context(db_sessionmaker, first.api_key)).tenant_id == first.tenant_id
|
||||
assert (
|
||||
await resolve_auth_context(db_sessionmaker, second.api_key)
|
||||
).tenant_id == first.tenant_id
|
||||
|
||||
|
||||
async def test_provision_tenant_honours_requested_scopes(
|
||||
db_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
) -> None:
|
||||
"""A key scoped to uploads must not be able to manage the allowlist."""
|
||||
result = await provision_tenant(db_sessionmaker, slug="acme", scopes=("files:write",))
|
||||
|
||||
auth = await resolve_auth_context(db_sessionmaker, result.api_key)
|
||||
|
||||
assert auth.scopes == frozenset({"files:write"})
|
||||
assert not auth.has_scope("domains:write")
|
||||
@@ -1,52 +0,0 @@
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Iterator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from qdrant_client import AsyncQdrantClient
|
||||
from testcontainers.community.qdrant import QdrantContainer
|
||||
|
||||
from src.config import QdrantSettings
|
||||
from src.infrastructure.qdrant.client import create_client
|
||||
|
||||
# Pinned to match the `qdrant-client` major/minor in pyproject.toml. The
|
||||
# testcontainers default image trails it far enough that the client emits an
|
||||
# incompatibility warning, and testing against a version we do not deploy is
|
||||
# the wrong signal anyway.
|
||||
_QDRANT_IMAGE = "qdrant/qdrant:v1.19.0"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def qdrant_container() -> Iterator[QdrantContainer]:
|
||||
with QdrantContainer(image=_QDRANT_IMAGE) as container:
|
||||
yield container
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def qdrant_url(qdrant_container: QdrantContainer) -> str:
|
||||
"""The container's REST URL, pinned to IPv4.
|
||||
|
||||
Same gotcha as postgres_url/minio_settings (see their conftests):
|
||||
testcontainers reports the host as `localhost`, which resolves to `::1`
|
||||
first, but Docker publishes the mapped port on IPv4 only. The IPv6 SYN is
|
||||
dropped rather than refused, so the client hangs until its timeout instead
|
||||
of falling back to the second address -- the connection does not fail, it
|
||||
hangs.
|
||||
"""
|
||||
host = qdrant_container.get_container_host_ip().replace("localhost", "127.0.0.1")
|
||||
return f"http://{host}:{qdrant_container.get_exposed_port(6333)}"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def qdrant_settings(qdrant_url: str) -> QdrantSettings:
|
||||
"""Settings naming a collection unique to this test (ADR-0016 isolation)."""
|
||||
return QdrantSettings(url=qdrant_url, collection=f"chunks_{uuid.uuid4().hex}")
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def qdrant_client(qdrant_settings: QdrantSettings) -> AsyncIterator[AsyncQdrantClient]:
|
||||
client = create_client(qdrant_settings)
|
||||
try:
|
||||
yield client
|
||||
finally:
|
||||
await client.close()
|
||||
205
tests/support/containers.py
Normal file
205
tests/support/containers.py
Normal file
@@ -0,0 +1,205 @@
|
||||
"""Disposable real infrastructure for integration and e2e tests (ADR-0016).
|
||||
|
||||
Registered as a plugin from the root `tests/conftest.py` rather than living in
|
||||
a per-boundary conftest, because `tests/e2e/` needs the same Postgres, MinIO,
|
||||
and Qdrant containers that `tests/integration/*/` does, and `pytest_plugins`
|
||||
is only honoured in the root conftest.
|
||||
|
||||
Nothing here is autouse: the container fixtures are session-scoped but lazy,
|
||||
so `pytest -m unit` still runs without Docker.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Iterator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from alembic.command import upgrade
|
||||
from alembic.config import Config
|
||||
from qdrant_client import AsyncQdrantClient
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
||||
from testcontainers.community.minio import MinioContainer
|
||||
from testcontainers.community.postgres import PostgresContainer
|
||||
from testcontainers.community.qdrant import QdrantContainer
|
||||
|
||||
from src.config import MinioSettings, PostgresSettings, QdrantSettings
|
||||
from src.infrastructure.minio.client import create_client as create_minio_client
|
||||
from src.infrastructure.postgres.database import create_engine
|
||||
from src.infrastructure.qdrant.client import create_client as create_qdrant_client
|
||||
|
||||
_MINIO_BUCKET = "test-source-files"
|
||||
|
||||
# Pinned to match the `qdrant-client` major/minor in pyproject.toml. The
|
||||
# testcontainers default image trails it far enough that the client emits an
|
||||
# incompatibility warning, and testing against a version we do not deploy is
|
||||
# the wrong signal anyway.
|
||||
_QDRANT_IMAGE = "qdrant/qdrant:v1.19.0"
|
||||
|
||||
|
||||
# --- Postgres ---------------------------------------------------------------
|
||||
|
||||
|
||||
def _alembic_config(database_url: str) -> Config:
|
||||
config = Config("alembic.ini")
|
||||
config.set_main_option("sqlalchemy.url", database_url)
|
||||
return config
|
||||
|
||||
|
||||
def _postgres_settings_from_url(url: str) -> PostgresSettings:
|
||||
# testcontainers returns postgresql+asyncpg://user:pass@host:port/db ;
|
||||
# PostgresSettings builds its own dsn from parts, so parse the parts back out.
|
||||
without_scheme = url.split("://", 1)[1]
|
||||
creds, hostpart = without_scheme.split("@", 1)
|
||||
user, password = creds.split(":", 1)
|
||||
hostport, db = hostpart.split("/", 1)
|
||||
host, port = hostport.split(":", 1)
|
||||
return PostgresSettings(host=host, port=int(port), user=user, password=password, db=db)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def postgres_container() -> Iterator[PostgresContainer]:
|
||||
with PostgresContainer("postgres:17", driver="asyncpg") as container:
|
||||
yield container
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def postgres_url(postgres_container: PostgresContainer) -> str:
|
||||
"""The container's URL, pinned to IPv4.
|
||||
|
||||
Testcontainers reports the host as `localhost`, which resolves to `::1`
|
||||
before `127.0.0.1`. Docker publishes the mapped port on IPv4 only, and the
|
||||
IPv6 SYN is dropped rather than refused, so asyncpg blocks on the first
|
||||
address until its connect timeout instead of falling back to the second --
|
||||
the connection does not fail, it hangs.
|
||||
"""
|
||||
return postgres_container.get_connection_url().replace("@localhost:", "@127.0.0.1:")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def migrated_postgres_url(postgres_url: str) -> str:
|
||||
"""The container's URL, after Alembic has created the schema on it once."""
|
||||
upgrade(_alembic_config(postgres_url), "head")
|
||||
return postgres_url
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def postgres_settings(migrated_postgres_url: str) -> PostgresSettings:
|
||||
"""Settings an application component can be constructed from directly."""
|
||||
return _postgres_settings_from_url(migrated_postgres_url)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||||
async def postgres_engine(postgres_settings: PostgresSettings) -> AsyncIterator[AsyncEngine]:
|
||||
engine = create_engine(postgres_settings)
|
||||
try:
|
||||
yield engine
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def db_sessionmaker(
|
||||
postgres_engine: AsyncEngine,
|
||||
) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
|
||||
"""A session *factory* per test, bound to a rolled-back outer transaction.
|
||||
|
||||
Every session it produces shares one connection/outer transaction, so
|
||||
writes `commit()`ed by one session are visible to the next -- needed for
|
||||
code under test that opens more than one session per operation (auth
|
||||
resolution, the ADR-0017 two-phase upload) -- while the whole test's
|
||||
writes still roll back together at teardown (ADR-0016: isolate data per
|
||||
test).
|
||||
"""
|
||||
async with postgres_engine.connect() as connection:
|
||||
outer_transaction = await connection.begin()
|
||||
sessionmaker = async_sessionmaker(
|
||||
bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
|
||||
)
|
||||
yield sessionmaker
|
||||
await outer_transaction.rollback()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def db_session(
|
||||
db_sessionmaker: async_sessionmaker[AsyncSession],
|
||||
) -> AsyncIterator[AsyncSession]:
|
||||
"""One session per test, bound to a rolled-back outer transaction.
|
||||
|
||||
Isolates each test's writes (ADR-0016: isolate data per test) without
|
||||
needing a fresh container or unique keys per test.
|
||||
"""
|
||||
async with db_sessionmaker() as session:
|
||||
yield session
|
||||
|
||||
|
||||
# --- MinIO ------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def minio_container() -> Iterator[MinioContainer]:
|
||||
with MinioContainer() as container:
|
||||
yield container
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def minio_settings(minio_container: MinioContainer) -> MinioSettings:
|
||||
"""Settings for the container, with the bucket already created.
|
||||
|
||||
Bucket creation belongs here rather than in a separate autouse fixture:
|
||||
autouse in a *globally registered* plugin would pull a MinIO container
|
||||
into every unit test run.
|
||||
"""
|
||||
config = minio_container.get_config()
|
||||
# Pinned to IPv4 for the same reason as postgres_url above: `localhost`
|
||||
# resolves to `::1` first, but Docker only publishes the mapped port on
|
||||
# IPv4, so the connection hangs instead of failing.
|
||||
endpoint = config["endpoint"].replace("localhost:", "127.0.0.1:")
|
||||
settings = MinioSettings(
|
||||
endpoint=endpoint,
|
||||
access_key=config["access_key"],
|
||||
secret_key=config["secret_key"],
|
||||
secure=False,
|
||||
bucket=_MINIO_BUCKET,
|
||||
)
|
||||
client = create_minio_client(settings)
|
||||
if not client.bucket_exists(settings.bucket):
|
||||
client.make_bucket(settings.bucket)
|
||||
return settings
|
||||
|
||||
|
||||
# --- Qdrant -----------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def qdrant_container() -> Iterator[QdrantContainer]:
|
||||
with QdrantContainer(image=_QDRANT_IMAGE) as container:
|
||||
yield container
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def qdrant_url(qdrant_container: QdrantContainer) -> str:
|
||||
"""The container's REST URL, pinned to IPv4.
|
||||
|
||||
Same gotcha as postgres_url/minio_settings above: testcontainers reports
|
||||
the host as `localhost`, which resolves to `::1` first, but Docker
|
||||
publishes the mapped port on IPv4 only. The IPv6 SYN is dropped rather
|
||||
than refused, so the client hangs until its timeout instead of falling
|
||||
back to the second address -- the connection does not fail, it hangs.
|
||||
"""
|
||||
host = qdrant_container.get_container_host_ip().replace("localhost", "127.0.0.1")
|
||||
return f"http://{host}:{qdrant_container.get_exposed_port(6333)}"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def qdrant_settings(qdrant_url: str) -> QdrantSettings:
|
||||
"""Settings naming a collection unique to this test (ADR-0016 isolation)."""
|
||||
return QdrantSettings(url=qdrant_url, collection=f"chunks_{uuid.uuid4().hex}")
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="session")
|
||||
async def qdrant_client(qdrant_settings: QdrantSettings) -> AsyncIterator[AsyncQdrantClient]:
|
||||
client = create_qdrant_client(qdrant_settings)
|
||||
try:
|
||||
yield client
|
||||
finally:
|
||||
await client.close()
|
||||
0
tests/unit/cli/__init__.py
Normal file
0
tests/unit/cli/__init__.py
Normal file
34
tests/unit/cli/test_provision_tenant.py
Normal file
34
tests/unit/cli/test_provision_tenant.py
Normal file
@@ -0,0 +1,34 @@
|
||||
"""Argument parsing for `python -m src.cli.provision_tenant`.
|
||||
|
||||
Only the argv -> arguments mapping is covered here; the provisioning behaviour
|
||||
itself needs a real database and lives in
|
||||
`tests/integration/postgres/test_provisioning.py`.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.application.tenants import DEFAULT_SCOPES
|
||||
from src.cli.provision_tenant import _parse_args
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
||||
def test_parse_args_defaults_scopes_and_domains() -> None:
|
||||
args = _parse_args(["--slug", "acme"])
|
||||
|
||||
assert args.slug == "acme"
|
||||
assert args.name is None
|
||||
assert args.key_name == "bootstrap"
|
||||
assert tuple(args.scopes.split(",")) == DEFAULT_SCOPES
|
||||
assert args.domains == []
|
||||
|
||||
|
||||
def test_parse_args_collects_repeated_domain_flags() -> None:
|
||||
args = _parse_args(["--slug", "acme", "--domain", "fire", "--domain", "life"])
|
||||
|
||||
assert args.domains == ["fire", "life"]
|
||||
|
||||
|
||||
def test_parse_args_requires_a_slug() -> None:
|
||||
with pytest.raises(SystemExit):
|
||||
_parse_args(["--domain", "fire"])
|
||||
Reference in New Issue
Block a user