Compare commits
1 Commits
chore/loca
...
us/US07-04
| Author | SHA1 | Date | |
|---|---|---|---|
| c4e64d2b9a |
9
.gitignore
vendored
9
.gitignore
vendored
@@ -20,12 +20,3 @@ _IGNORE/
|
||||
|
||||
# Test failure evidence (US07-04)
|
||||
.artifacts/
|
||||
|
||||
# Any dotenv, not only the default name.
|
||||
*.env
|
||||
|
||||
# Local virtualenv for running the app.
|
||||
.venv/
|
||||
|
||||
# setuptools editable-install metadata.
|
||||
*.egg-info/
|
||||
|
||||
217
README.md
217
README.md
@@ -7,49 +7,16 @@ archive workflow. Planning lives in `INTEGRATED_PIPELINE_CONCEPT.md` and
|
||||
## Application (`photo_pipeline`)
|
||||
|
||||
The target application lives in `photo_pipeline/` (FastAPI + SQLAlchemy + Alembic).
|
||||
Install it into a virtualenv once:
|
||||
Run it with:
|
||||
|
||||
```bash
|
||||
python3.12 -m venv .venv
|
||||
.venv/bin/pip install -e ".[vision]" # drop [vision] for a review-only install
|
||||
python -m photo_pipeline migrate # apply database migrations
|
||||
python -m photo_pipeline serve # start the API + static review UI (127.0.0.1:8000)
|
||||
```
|
||||
|
||||
Then run the two processes:
|
||||
|
||||
```bash
|
||||
.venv/bin/python -m photo_pipeline migrate # apply database migrations
|
||||
.venv/bin/python -m photo_pipeline serve # API + review UI at 127.0.0.1:8000/app/
|
||||
.venv/bin/python -m photo_pipeline worker # second terminal: runs the jobs
|
||||
```
|
||||
|
||||
The server enqueues work and serves the UI; nothing actually scans, scores,
|
||||
analyses, uploads, or archives without a worker. `work_item/scripts/python` is the
|
||||
*helper's* launcher — it prefers Conda base and falls back to a bare system
|
||||
interpreter, so it is not how the application is run.
|
||||
|
||||
Configuration comes from `PHOTO_PIPELINE_*` environment variables (see
|
||||
`photo_pipeline/config.py`); secrets are referenced, never logged.
|
||||
|
||||
### Configuration file
|
||||
|
||||
`.env` in the working directory is read at startup, or any path named by
|
||||
`PHOTO_PIPELINE_ENV_FILE`. It is parsed, never executed: `KEY=value` lines,
|
||||
`#` comments, optional quotes — no interpolation and no `export`. **Anything already
|
||||
exported wins**, so the file is the standing configuration and the shell is the
|
||||
override for one run.
|
||||
|
||||
The archived CLI's variable names still work, so an existing `photo_analyzer.env`
|
||||
can be used as-is:
|
||||
|
||||
| in the file | applied as |
|
||||
|---|---|
|
||||
| `LLM_API_KEY` / `GEMINI_API_KEY` | `OPENAI_API_KEY` |
|
||||
| `LLM_BASE_URL` | `OPENAI_BASE_URL` |
|
||||
| `LIBRARY` | `PHOTO_PIPELINE_LIBRARY_ROOTS` |
|
||||
|
||||
`.env` and `*.env` are gitignored and denied by the work-item safety checks: the
|
||||
file holds a real key and must never be committed.
|
||||
|
||||
### API access (US07-02)
|
||||
|
||||
The app listens on loopback, so its attacker is another page in the same browser.
|
||||
@@ -314,184 +281,6 @@ file in the temporary library are copied to `.artifacts/<test id>/` before pytes
|
||||
deletes the directory. Point `PHOTO_PIPELINE_TEST_ARTIFACTS` elsewhere to collect
|
||||
them from CI.
|
||||
|
||||
## Release gate (US07-07)
|
||||
|
||||
One command runs every suite in an isolated stack and keeps the evidence:
|
||||
|
||||
```bash
|
||||
work_item/scripts/python -m photo_pipeline release-gate --output data/release/$(date -u +%Y%m%dT%H%M%SZ)
|
||||
```
|
||||
|
||||
It fails — and exits non-zero — when any stage fails, when a suite skips a test for
|
||||
a reason that is not a documented environment limit (`exiftool not installed`,
|
||||
`root ignores directory permissions`), or when the story matrix has a hole. The
|
||||
evidence directory holds `release-report.json` (revision, per-stage result, timings,
|
||||
summaries), `logs/<stage>.log`, and `CHECKSUMS.sha256` over both.
|
||||
|
||||
**The story matrix** lives in `tests/story_traceability.json`: every story under
|
||||
`delivery_backlog/stories/` is either mapped to test files that exist, or listed in
|
||||
`planned` as an accepted but unimplemented story. A story that is neither, or a
|
||||
mapping to a file that has been deleted, fails the gate.
|
||||
|
||||
**The journey** (`tests/e2e/test_release_journey.py`) takes one fresh library through
|
||||
discovery, duplicate review, safety, analysis, EXIF verification, album proposal,
|
||||
guarded rename, rescan, upload with server-side verification, archive, offline
|
||||
deduplication, and restore — over HTTP against real server and worker processes,
|
||||
with a full restart in the middle and at the end.
|
||||
|
||||
### Real-library dry run and approval
|
||||
|
||||
Before the application is pointed at photos that cannot be replaced:
|
||||
|
||||
```bash
|
||||
work_item/scripts/python -m photo_pipeline dry-run --output dry-run.json
|
||||
work_item/scripts/python -m photo_pipeline approve-dry-run dry-run.json --approver "$(whoami)"
|
||||
```
|
||||
|
||||
The dry run is strictly read-only: it opens no file for writing, writes no database
|
||||
row, and reports what it found — file counts by extension, folders, bytes, unreadable
|
||||
files, excluded directories, and a reconciliation against what the database already
|
||||
knows (already registered, new, recorded but absent). Set
|
||||
`PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL=1` and **every mutating API request is
|
||||
refused with `403 dry_run_not_approved`** until a report for exactly those library
|
||||
roots has been approved. Reading stays open — you have to be able to see what was
|
||||
found in order to approve it — and so does taking a backup. Change the library roots
|
||||
and the approval no longer applies: it approves that reconciliation, not the idea of
|
||||
mutating.
|
||||
|
||||
## Performance budgets (US07-06)
|
||||
|
||||
Budgets are measured, not asserted in prose. `python -m photo_pipeline benchmark`
|
||||
builds a synthetic library of a stated size, runs the same scenarios every time,
|
||||
writes a machine-readable report, and **exits non-zero when a budget is breached**.
|
||||
|
||||
```bash
|
||||
work_item/scripts/python -m photo_pipeline benchmark --profile smoke # ~2 s, runs in CI
|
||||
work_item/scripts/python -m photo_pipeline benchmark --profile short # 25k assets
|
||||
work_item/scripts/python -m photo_pipeline benchmark --profile full \
|
||||
--output data/benchmarks/full.json # 25k + 100k
|
||||
work_item/scripts/python -m photo_pipeline benchmark --profile huge \
|
||||
--soak-seconds 3600 --output data/benchmarks/soak.json # 500k + soak
|
||||
```
|
||||
|
||||
| Metric | Budget | Why |
|
||||
|---|---|---|
|
||||
| `latency_p95_ms` | 250 ms | a list or search page must feel immediate |
|
||||
| `latency_max_ms` | 2 000 ms | no single page may stall the review flow |
|
||||
| `rss_growth_bytes` | 400 MB | a run must not leak the library |
|
||||
| `open_files` | 256 | file descriptors are a hard operating-system limit |
|
||||
| `wal_bytes` | 200 MB | a growing write-ahead log means checkpoints are starving |
|
||||
| `queue_depth` | 1 000 | an unbounded queue is an out-of-memory in waiting |
|
||||
| `cache_over_quota_bytes` | 0 | the thumbnail cache has to respect its quota |
|
||||
|
||||
Measured on the reference machine (Apple Silicon, SQLite WAL), p95 per scenario:
|
||||
|
||||
| Scenario | 25k | 100k |
|
||||
|---|---|---|
|
||||
| `inventory_page` | 0.5 ms | 0.6 ms |
|
||||
| `library_search` | 4.8 ms | 17.1 ms |
|
||||
| `library_stats` | 56.8 ms | 197.4 ms |
|
||||
| `workflow_readiness` | 52.6 ms | 235.4 ms |
|
||||
| `duplicate_cluster_list` | 0.6 ms | 0.5 ms |
|
||||
| `duplicate_cluster_page` | 1.7 ms | 1.7 ms |
|
||||
|
||||
CI runs the `smoke` profile through `tests/integration/test_performance_budgets.py`;
|
||||
the 25k/100k/500k matrix and the multi-hour soak belong to scheduled infrastructure,
|
||||
because minutes of build time do not belong in the suite that runs on every change.
|
||||
|
||||
**Exceptions.** A budget that cannot be met is not a warning to ignore: it goes into
|
||||
`APPROVED_EXCEPTIONS` in `photo_pipeline/services/benchmarks.py` with its raised
|
||||
limit, who approved it, why, and a review date. Every report lists the exceptions it
|
||||
applied, so a release review sees them.
|
||||
|
||||
Approved today, both for the 500k `huge` profile only, review by 2027-02-17:
|
||||
|
||||
| Scenario | Measured at 500k | Raised limit |
|
||||
|---|---|---|
|
||||
| `library_stats` | 1.08 s p95 · 3.2 s max | 1.5 s p95 · 4 s max |
|
||||
| `workflow_readiness` | 1.40 s p95 · 3.3 s max | 1.8 s p95 · 4 s max |
|
||||
|
||||
Both are library-wide aggregates — the current safety decision of every asset, and
|
||||
the album/tag/year breakdown of every analysis row — and both meet the 250 ms budget
|
||||
at the 100k rows the concept sets it for. Beyond that they are linear against one
|
||||
SQLite writer; the fix is denormalized totals or the planned PostgreSQL transition,
|
||||
not a query tweak. Everything else at 500k is inside budget, and a soak at that size
|
||||
grows neither resident memory nor the job queue.
|
||||
|
||||
## Backup and recovery (US07-05)
|
||||
|
||||
Backups go through SQLite's online backup API, never a file copy: with WAL enabled
|
||||
the `.db` file alone is missing every committed page still in the write-ahead log.
|
||||
Each backup is a directory under `data/backups/` holding the snapshot and a
|
||||
`manifest.json` describing it — schema revision, SHA-256, row counts, the archive
|
||||
media the library depends on, and which configuration was set. Secrets are recorded
|
||||
as `configured`, never as values, so a manifest is safe to attach to a bug report.
|
||||
|
||||
```bash
|
||||
work_item/scripts/python -m photo_pipeline backup --reason before-upgrade --keep 7
|
||||
work_item/scripts/python -m photo_pipeline verify-backup data/backups/<name>
|
||||
work_item/scripts/python -m photo_pipeline diagnostics
|
||||
```
|
||||
|
||||
The same is available at `GET /api/v1/diagnostics`, `GET|POST /api/v1/backups`,
|
||||
`GET /api/v1/backups/{name}/verify`, and `POST /api/v1/backups/prune`. **Restore is
|
||||
not an endpoint** — it replaces the state of an installation, so it belongs to a
|
||||
stopped one and a person at a terminal.
|
||||
|
||||
### Integrity check
|
||||
|
||||
`verify-backup` runs `PRAGMA integrity_check` (structure) *and*
|
||||
`PRAGMA foreign_key_check` (references), compares the snapshot's SHA-256 with the
|
||||
manifest, and re-counts every table the manifest recorded. Any mismatch — bit rot, a
|
||||
truncated copy, a "repaired" snapshot — fails the check, and `restore` refuses a
|
||||
backup that does not verify.
|
||||
|
||||
### Restore drill
|
||||
|
||||
1. Stop the server and the worker.
|
||||
2. `python -m photo_pipeline verify-backup data/backups/<name>` — never restore an
|
||||
unverified snapshot.
|
||||
3. `python -m photo_pipeline restore data/backups/<name> --into /path/to/fresh-data`
|
||||
(a target that already holds a database is refused; recovering in place means
|
||||
moving the old data directory aside first).
|
||||
4. Point `PHOTO_PIPELINE_DATA_DIR` at the restored directory and run
|
||||
`python -m photo_pipeline migrate`.
|
||||
5. Run an inventory scan so paths are reconciled against the real library.
|
||||
6. Mount every archive location named in the manifest before archiving again — the
|
||||
database records where archived originals are, but it does not contain them.
|
||||
|
||||
Practise this against a copy before you need it; the drill is exercised
|
||||
automatically by `tests/integration/test_backup_recovery.py`.
|
||||
|
||||
### Failed migration
|
||||
|
||||
A pending schema upgrade is snapshotted first (`reason: pre-migration`), by both the
|
||||
API startup and `python -m photo_pipeline migrate`. If a migration fails, the error
|
||||
log names the backup directory: stop everything and run the restore drill against
|
||||
it. An up-to-date database is not backed up again on every start.
|
||||
|
||||
### Archive media
|
||||
|
||||
Archived originals live on their medium, not in the backup. The manifest lists every
|
||||
archive location with its `media_id` and whether it was mounted when the backup was
|
||||
taken. Keep one copy of each medium off-site, and remount a location before
|
||||
restoring assets from it.
|
||||
|
||||
### Retention and disk
|
||||
|
||||
`--keep N` (default 7) prunes the oldest backups and never the newest.
|
||||
`diagnostics` reports the database, write-ahead log, thumbnail cache, uploader
|
||||
reports, backups, and logs separately, with free space and warnings for low disk
|
||||
(`disk_low`, `disk_critical`), a cache over its quota, a write-ahead log outgrowing
|
||||
its database, and a legacy CLI writing the library.
|
||||
|
||||
### Process locking
|
||||
|
||||
`serve` and `worker` take a JSON lock in the data directory (`api.lock.json`,
|
||||
`worker.lock.json`). A second worker exits `2` and names the holder; a lock whose
|
||||
process is gone is taken over. If the frozen CLI's state files are being written,
|
||||
both refuse with exit `3` — `--allow-legacy` overrides, and you own the outcome.
|
||||
|
||||
## Legacy CLI archive
|
||||
|
||||
The command-line tools this application was extracted from are frozen in
|
||||
|
||||
@@ -1,32 +0,0 @@
|
||||
# E08 — Container Deployment
|
||||
|
||||
Concept phase: none. This epic is a delivery-format addition on top of the concept:
|
||||
the same application, same safety invariants, packaged as a Docker image and deployed
|
||||
continuously from Gitea Actions instead of being started by hand from a working copy.
|
||||
|
||||
It does not change the product scope in
|
||||
[`INTEGRATED_PIPELINE_CONCEPT.md`](../INTEGRATED_PIPELINE_CONCEPT.md). SQLite stays the
|
||||
store, one worker stays the writer, the library process lock stays authoritative, and
|
||||
no path outside the configured library roots becomes reachable because the process now
|
||||
runs in a container.
|
||||
|
||||
One decision does extend the concept and is made here explicitly: the application may
|
||||
be reached through a reverse proxy under a real hostname, not only over loopback. That
|
||||
requires a configurable trust boundary and an authentication gate, because the
|
||||
loopback-only checks of US07-02 are what currently stand in for authentication.
|
||||
|
||||
## Stories
|
||||
|
||||
1. [US08-01 — Make the trust boundary configurable and authenticated](stories/US08-01-trusted-hosts-auth.md)
|
||||
2. [US08-02 — Build a reproducible application image](stories/US08-02-container-image.md)
|
||||
3. [US08-03 — Compose the runtime and mount the library safely](stories/US08-03-compose-runtime.md)
|
||||
4. [US08-04 — Publish and deploy from Gitea Actions](stories/US08-04-gitea-cicd.md)
|
||||
5. [US08-05 — Automate container deployment acceptance](stories/US08-05-container-e2e.md)
|
||||
|
||||
## Epic outcome
|
||||
|
||||
A tagged image built from `main` runs the API and the worker as separate containers
|
||||
against a mounted library and a persistent data volume, is published to the Gitea
|
||||
registry, is redeployed by webhook, survives restart and upgrade with its database and
|
||||
journals intact, and refuses every request that a loopback deployment would have
|
||||
refused.
|
||||
@@ -2,13 +2,11 @@
|
||||
|
||||
This backlog decomposes the phases in
|
||||
[`INTEGRATED_PIPELINE_CONCEPT.md`](../INTEGRATED_PIPELINE_CONCEPT.md) into seven
|
||||
epics and small, independently verifiable user stories, plus one delivery-format
|
||||
epic (E08) that packages the released application as a deployable container.
|
||||
epics and small, independently verifiable user stories.
|
||||
|
||||
## Numbering and file naming
|
||||
|
||||
- Epics: `E01` through `E07`, matching concept Phases A through G; `E08` has no
|
||||
concept phase and must not change product scope.
|
||||
- Epics: `E01` through `E07`, matching concept Phases A through G.
|
||||
- Stories: `US<epic>-<sequence>`, for example `US03-02`.
|
||||
- Epic files: `E01-<slug>.md`.
|
||||
- Story files: `stories/US01-01-<slug>.md`.
|
||||
@@ -38,7 +36,6 @@ epic (E08) that packages the released application as a deployable container.
|
||||
5. [E05 — Immich upload](E05-immich-upload.md)
|
||||
6. [E06 — Archive lifecycle](E06-archive-lifecycle.md)
|
||||
7. [E07 — Hardening and release](E07-hardening-release.md)
|
||||
8. [E08 — Container deployment](E08-container-deployment.md)
|
||||
|
||||
## Shared definition of done
|
||||
|
||||
|
||||
@@ -1,42 +0,0 @@
|
||||
# US08-01 — Make the Trust Boundary Configurable and Authenticated
|
||||
|
||||
Epic: [E08](../E08-container-deployment.md)
|
||||
|
||||
As an operator, I want to reach the application through my own hostname without
|
||||
weakening it, so a container behind a reverse proxy is as safe as the loopback
|
||||
deployment it replaces.
|
||||
|
||||
## Context
|
||||
|
||||
`photo_pipeline/api/security.py` refuses any request whose `Host` or `Origin` is not
|
||||
loopback. That check is the current stand-in for authentication: whoever can reach
|
||||
`127.0.0.1:8000` is the owner. Behind a proxy the hostname is no longer loopback, so
|
||||
relaxing the check without adding an authentication gate would publish the library.
|
||||
|
||||
## Acceptance criteria
|
||||
|
||||
- Allowed hosts and origins come from configuration (`PHOTO_PIPELINE_*`), default to
|
||||
the current loopback set, and an unset configuration behaves exactly as today.
|
||||
- Whenever a non-loopback host is configured, startup requires an access secret and
|
||||
refuses to serve without one; loopback-only deployments keep working with no secret.
|
||||
- The secret is exchanged for the existing session cookie and CSRF token through the
|
||||
bootstrap endpoint; every protected route keeps its current session and CSRF
|
||||
requirements unchanged.
|
||||
- Forwarded headers (`X-Forwarded-Proto`, `X-Forwarded-Host`) are honored only from a
|
||||
configured trusted proxy and ignored otherwise, so a client cannot forge its origin.
|
||||
- Cookies are marked `Secure` when the effective external scheme is HTTPS.
|
||||
- Failed authentication is rate-limited and logged without the secret, the session id,
|
||||
or any request body.
|
||||
- Health endpoints stay reachable without the secret; nothing else does.
|
||||
|
||||
## Automated tests
|
||||
|
||||
- Unit tests for host/origin evaluation across loopback default, configured host,
|
||||
unconfigured host, forged forwarded headers, and trusted-proxy forwarded headers.
|
||||
- Integration tests: startup refusal without a secret, successful exchange, wrong
|
||||
secret, replay of an old session, cross-site request, and unauthenticated access to
|
||||
every route class.
|
||||
|
||||
## Dependencies
|
||||
|
||||
- US07-02
|
||||
@@ -1,41 +0,0 @@
|
||||
# US08-02 — Build a Reproducible Application Image
|
||||
|
||||
Epic: [E08](../E08-container-deployment.md)
|
||||
|
||||
As an operator, I want one image that can run either application role, so deployment is
|
||||
a pull instead of a Python environment I have to reproduce by hand.
|
||||
|
||||
## Context
|
||||
|
||||
The application shells out to `exiftool` and `immich-go`, writes into the library as a
|
||||
normal filesystem user, and serves a static frontend from `frontend/`. All three have to
|
||||
be true inside the image, or the container starts and then fails on the first real
|
||||
operation.
|
||||
|
||||
## Acceptance criteria
|
||||
|
||||
- A `Dockerfile` builds from a pinned Python base, installs the project and its runtime
|
||||
dependencies, and contains no test, playwright, or build-only tooling in the final
|
||||
layer.
|
||||
- `exiftool` and `immich-go` are present at pinned versions, and their versions are
|
||||
recorded in the image and reported by `python -m photo_pipeline diagnostics`.
|
||||
- The image runs as a non-root user whose UID/GID are build-time arguments, so files
|
||||
the application renames or writes keep the ownership the host library expects.
|
||||
- One entrypoint selects the role: `serve` or `worker`, passing through the existing
|
||||
CLI arguments; no supervisor runs two roles in one container.
|
||||
- `serve` containers declare a `HEALTHCHECK` against `/api/v1/health/ready`, so an
|
||||
unmigrated or misconfigured database is not reported healthy.
|
||||
- The image contains no secrets, no library data, no database, and no `.git`; the build
|
||||
context is constrained by `.dockerignore`.
|
||||
- Image build is reproducible from a clean checkout and documented in `README.md`.
|
||||
|
||||
## Automated tests
|
||||
|
||||
- A build-and-run test asserts the image starts, reports ready, serves the frontend
|
||||
index, and returns the pinned `exiftool` and `immich-go` versions.
|
||||
- A test asserts the container refuses to run as UID 0 and that a file created by the
|
||||
container is owned by the configured UID/GID.
|
||||
|
||||
## Dependencies
|
||||
|
||||
- US07-05
|
||||
@@ -1,47 +0,0 @@
|
||||
# US08-03 — Compose the Runtime and Mount the Library Safely
|
||||
|
||||
Epic: [E08](../E08-container-deployment.md)
|
||||
|
||||
As an operator, I want a single compose file that runs the API and the worker against my
|
||||
real library, so a deployment is one command and the safety invariants survive it.
|
||||
|
||||
## Context
|
||||
|
||||
The library process lock (US07-05) assumes both roles see the same lock file, and SQLite
|
||||
in WAL mode assumes a real local filesystem. Container path policy is the same problem
|
||||
as host path policy with a new failure mode: the configured library roots must name the
|
||||
in-container mount paths, not the host paths.
|
||||
|
||||
## Acceptance criteria
|
||||
|
||||
- `docker-compose.yml` runs exactly one `serve` and one `worker` container from the same
|
||||
image and the same data volume, and a second worker is refused by the existing lock
|
||||
rather than by convention.
|
||||
- The library is a bind mount; `PHOTO_PIPELINE_LIBRARY_ROOTS` names the container-side
|
||||
paths, and a mismatch between mounted and configured roots fails at startup with a
|
||||
clear message instead of at the first write.
|
||||
- The data volume holds the database, WAL, thumbnail cache, and backups on a local
|
||||
filesystem; the composition documents that a network mount is unsupported for it.
|
||||
- Migrations run before `serve` and `worker` accept work, using the existing backup-then-
|
||||
migrate path, and an upgrade that fails leaves the previous database intact.
|
||||
- Configuration and secrets come from the environment, never from the image or a
|
||||
committed file; a `.env.example` lists every `PHOTO_PIPELINE_*` variable with safe
|
||||
defaults and no values.
|
||||
- The API port is published to host loopback by default; exposing it publicly requires
|
||||
the configured hostname and access secret from US08-01.
|
||||
- Containers restart automatically, and a restart mid-job resumes exactly as a host
|
||||
restart does today.
|
||||
- Backup, verify-backup, restore, and diagnostics are documented as container commands
|
||||
and work against the mounted volumes.
|
||||
|
||||
## Automated tests
|
||||
|
||||
- An integration test brings the composition up against a temporary fixture library,
|
||||
runs a job, restarts both containers, and asserts the job resumes and the database is
|
||||
intact.
|
||||
- Tests for: second worker refused, library-root mismatch refused at startup, failed
|
||||
migration leaving the previous database restorable.
|
||||
|
||||
## Dependencies
|
||||
|
||||
- US08-01, US08-02
|
||||
@@ -1,40 +0,0 @@
|
||||
# US08-04 — Publish and Deploy from Gitea Actions
|
||||
|
||||
Epic: [E08](../E08-container-deployment.md)
|
||||
|
||||
As a release owner, I want `main` to build, publish, and redeploy the image
|
||||
automatically, so deployment is the same reproducible path every time.
|
||||
|
||||
## Context
|
||||
|
||||
The workflow is adapted from the `crowdsec-admin` deployment workflow
|
||||
(`.gitea/workflows/deploy.yml` in that repository): build, log in to the Gitea registry,
|
||||
push, trigger a Portainer webhook, prune. This project needs the same shape plus a test
|
||||
gate, because unlike that project it has a required suite that must not be skipped.
|
||||
|
||||
## Acceptance criteria
|
||||
|
||||
- `.gitea/workflows/` contains a test workflow that runs on pull requests and on `main`,
|
||||
executing the configured required suites, and a deploy workflow that runs only after
|
||||
the tests pass on `main` and on manual dispatch.
|
||||
- The deploy workflow publishes to `git.domverse-berlin.eu` under this project's own
|
||||
image path, tagged `latest` and the commit SHA, so a rollback is a tag change.
|
||||
- Registry credentials and the Portainer webhook come from repository secrets; runtime
|
||||
secrets (vision key, Immich key, access secret) stay in the Portainer stack and never
|
||||
enter the repository or the image.
|
||||
- Redeploy is triggered by webhook and the workflow fails when the webhook call fails.
|
||||
- Dangling images are pruned; published tags are not.
|
||||
- A concurrency guard prevents two deploys of different commits overlapping.
|
||||
- `README.md` documents the required secrets, the image path, the rollback procedure,
|
||||
and that the stack is managed by Portainer from git.
|
||||
|
||||
## Automated tests
|
||||
|
||||
- Workflow files are validated (syntax and required job/step names) by a repository test
|
||||
so a rename cannot silently disable the test gate.
|
||||
- A dry-run job builds and pushes to a scratch tag on manual dispatch without touching
|
||||
`latest` or triggering a redeploy.
|
||||
|
||||
## Dependencies
|
||||
|
||||
- US08-02, US08-03
|
||||
@@ -1,30 +0,0 @@
|
||||
# US08-05 — Automate Container Deployment Acceptance
|
||||
|
||||
Epic: [E08](../E08-container-deployment.md)
|
||||
|
||||
As a release owner, I want one automated gate that proves the deployed container, so the
|
||||
packaged application is verified the same way the host application is.
|
||||
|
||||
## Acceptance criteria
|
||||
|
||||
- One documented command provisions the composition from the built image against a
|
||||
temporary fixture library and an isolated data volume, and destroys it afterwards.
|
||||
- A browser journey against the containerized application covers discovery, duplicate
|
||||
review, analysis, album proposal, rename, upload preflight, and archive views.
|
||||
- An upgrade journey runs the previous published image, then the new one, and asserts
|
||||
migrations, journals, jobs, and the thumbnail cache survive.
|
||||
- A restart journey kills both containers mid-job and asserts resume without duplicate
|
||||
side effects.
|
||||
- Security gates run against the deployed instance: unauthenticated access refused,
|
||||
forged forwarded headers refused, paths outside the mounted library roots refused, and
|
||||
no secret in container logs.
|
||||
- Evidence is retained per run and the gate fails on any skipped required check.
|
||||
|
||||
## Automated tests
|
||||
|
||||
- The container acceptance suite runs on a `phase_h` marker in CI on `main` and before a
|
||||
published deploy; earlier epic suites keep running unchanged.
|
||||
|
||||
## Dependencies
|
||||
|
||||
- US08-01 through US08-04
|
||||
@@ -74,10 +74,8 @@ export const api = {
|
||||
request("/inventory/assets?" + new URLSearchParams(params).toString(), opts),
|
||||
listClusters: (params = {}, opts = {}) =>
|
||||
request("/duplicates/clusters?" + new URLSearchParams(params).toString(), opts),
|
||||
getCluster: (id, params = {}, opts = {}) => {
|
||||
const query = new URLSearchParams(params).toString();
|
||||
return request(`/duplicates/clusters/${encodeURIComponent(id)}${query ? `?${query}` : ""}`, opts);
|
||||
},
|
||||
getCluster: (id, opts = {}) =>
|
||||
request(`/duplicates/clusters/${encodeURIComponent(id)}`, opts),
|
||||
decide: (id, payload, opts = {}) =>
|
||||
request(`/duplicates/clusters/${encodeURIComponent(id)}/decision`, {
|
||||
method: "POST",
|
||||
|
||||
@@ -193,7 +193,7 @@ async function renderClusters(params) {
|
||||
" ",
|
||||
el("span", { class: `badge ${cluster.state}` }, cluster.state)
|
||||
),
|
||||
el("div", { class: "muted" }, `${cluster.member_total ?? cluster.members.length} members · confidence ${cluster.confidence}`)
|
||||
el("div", { class: "muted" }, `${cluster.members.length} members · confidence ${cluster.confidence}`)
|
||||
)
|
||||
);
|
||||
|
||||
@@ -207,11 +207,8 @@ async function renderClusters(params) {
|
||||
async function renderClusterDetail(id, extra = {}) {
|
||||
setActiveNav("duplicates");
|
||||
let cluster;
|
||||
// A cluster can hold thousands of members, so the server pages them; the page
|
||||
// asks for as many as it is currently showing (US07-06).
|
||||
const shown = extra.shown || 0;
|
||||
try {
|
||||
cluster = await api.getCluster(id, shown ? { limit: shown } : {});
|
||||
cluster = await api.getCluster(id);
|
||||
} catch (error) {
|
||||
show(errorBanner(`Failed to load cluster: ${error.message}`));
|
||||
return;
|
||||
@@ -301,19 +298,6 @@ async function renderClusterDetail(id, extra = {}) {
|
||||
nodes.push(decisionBar);
|
||||
if (extra.pending) nodes.push(confirmPanel(cluster, extra.pending));
|
||||
nodes.push(el("div", { class: "cluster-grid" }, ...members));
|
||||
const total = cluster.member_total ?? cluster.members.length;
|
||||
if (cluster.members.length < total) {
|
||||
nodes.push(
|
||||
el(
|
||||
"button",
|
||||
{
|
||||
"data-testid": "show-more-members",
|
||||
onclick: () => renderClusterDetail(id, { ...extra, shown: cluster.members.length + 100 }),
|
||||
},
|
||||
`Show more (${cluster.members.length} of ${total})`
|
||||
)
|
||||
);
|
||||
}
|
||||
show(...nodes);
|
||||
}
|
||||
|
||||
|
||||
@@ -1,45 +0,0 @@
|
||||
"""Indexes the large-library read paths need (US07-06).
|
||||
|
||||
Revision ID: 0016_performance_indexes
|
||||
Revises: 0015_exif_projections
|
||||
Create Date: 2026-08-17
|
||||
|
||||
Measured, not guessed. At 100k assets the workflow home and the safety queue both
|
||||
resolve "the current decision per asset" with a window function over
|
||||
``safety_reviews``; a plain ``asset_id`` index makes SQLite sort every partition by
|
||||
hand. Ordering the index by ``(asset_id, created_at DESC)`` halves that query.
|
||||
|
||||
``duplicate_members(cluster_id, asset_id)`` serves the paged member list of a
|
||||
cluster with thousands of members, which is the other page that stopped being cheap.
|
||||
"""
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision = "0016_performance_indexes"
|
||||
down_revision = "0015_exif_projections"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_index(
|
||||
"ix_safety_reviews_asset_created",
|
||||
"safety_reviews",
|
||||
["asset_id", "created_at"],
|
||||
)
|
||||
op.create_index(
|
||||
"ix_duplicate_members_cluster_asset",
|
||||
"duplicate_members",
|
||||
["cluster_id", "asset_id"],
|
||||
)
|
||||
op.create_index(
|
||||
"ix_analysis_results_approx_year",
|
||||
"analysis_results",
|
||||
["approx_year"],
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index("ix_analysis_results_approx_year", table_name="analysis_results")
|
||||
op.drop_index("ix_duplicate_members_cluster_asset", table_name="duplicate_members")
|
||||
op.drop_index("ix_safety_reviews_asset_created", table_name="safety_reviews")
|
||||
@@ -1,37 +1,21 @@
|
||||
"""Application management CLI:
|
||||
``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores,backup,verify-backup,restore,diagnostics}``.
|
||||
|
||||
``serve`` and ``worker`` take the library process lock for their role (US07-05):
|
||||
two workers, or the frozen CLI running beside the app, would each be safe on their
|
||||
own and destructive together. ``restore`` is here rather than in the API because it
|
||||
replaces the state of an installation and belongs to a stopped one.
|
||||
"""
|
||||
"""Application management CLI: ``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores}``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Sequence
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.services.app_lock import LegacyProcessActive, LibraryLock, LockHeld
|
||||
from photo_pipeline.services.backup import BackupError, BackupService, migrate_with_backup
|
||||
from photo_pipeline.db import run_migrations
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(prog="photo_pipeline")
|
||||
commands = parser.add_subparsers(dest="command", required=True)
|
||||
serve_cmd = commands.add_parser("serve", help="Run the API server")
|
||||
commands.add_parser("serve", help="Run the API server")
|
||||
commands.add_parser("migrate", help="Upgrade the database to the latest revision")
|
||||
worker_cmd = commands.add_parser("worker", help="Run a durable-job worker")
|
||||
worker_cmd.add_argument("--id", default="worker-1", help="Worker id (lease owner)")
|
||||
for locked in (serve_cmd, worker_cmd):
|
||||
locked.add_argument(
|
||||
"--allow-legacy",
|
||||
action="store_true",
|
||||
help="Start even though a legacy CLI looks active (unsafe; you own the outcome)",
|
||||
)
|
||||
import_cmd = commands.add_parser(
|
||||
"import-legacy-scores",
|
||||
help="Import the archived CLI's nsfw_scores.csv into the database (US07-01)",
|
||||
@@ -43,139 +27,22 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
import_cmd.add_argument(
|
||||
"--dry-run", action="store_true", help="Report what would happen and change nothing"
|
||||
)
|
||||
|
||||
backup_cmd = commands.add_parser("backup", help="Take an online database backup")
|
||||
backup_cmd.add_argument("--reason", default="manual", help="Why (part of the directory name)")
|
||||
backup_cmd.add_argument("--keep", type=int, default=7, help="How many backups to retain")
|
||||
verify_cmd = commands.add_parser("verify-backup", help="Check a backup is intact and readable")
|
||||
verify_cmd.add_argument("backup", help="Path to the backup directory")
|
||||
restore_cmd = commands.add_parser(
|
||||
"restore", help="Restore a verified backup into a fresh data directory"
|
||||
)
|
||||
restore_cmd.add_argument("backup", help="Path to the backup directory")
|
||||
restore_cmd.add_argument("--into", required=True, help="Fresh data directory to restore into")
|
||||
commands.add_parser("diagnostics", help="Report sizes, disk headroom, locks, and warnings")
|
||||
bench_cmd = commands.add_parser(
|
||||
"benchmark", help="Measure latency and resource use against agreed budgets (US07-06)"
|
||||
)
|
||||
bench_cmd.add_argument("--profile", default="smoke", help="smoke | short | full | huge")
|
||||
bench_cmd.add_argument(
|
||||
"--soak-seconds", type=float, default=0.0, help="Also run a soak of this length"
|
||||
)
|
||||
bench_cmd.add_argument("--output", help="Write the JSON report here as well as to stdout")
|
||||
|
||||
gate_cmd = commands.add_parser(
|
||||
"release-gate", help="Run every suite in an isolated stack and keep the evidence"
|
||||
)
|
||||
gate_cmd.add_argument("--output", help="Evidence directory (default: data/release/<stamp>)")
|
||||
dry_cmd = commands.add_parser(
|
||||
"dry-run", help="Read-only reconciliation of the configured library (US07-07)"
|
||||
)
|
||||
dry_cmd.add_argument("--output", help="Write the report here as well as to stdout")
|
||||
approve_cmd = commands.add_parser(
|
||||
"approve-dry-run", help="Approve a dry-run report, which is what enables mutation"
|
||||
)
|
||||
approve_cmd.add_argument("report", help="Path to the dry-run report")
|
||||
approve_cmd.add_argument("--approver", required=True, help="Who is accepting this")
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
config = Config.from_env()
|
||||
config.database_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if args.command == "migrate":
|
||||
manifest = migrate_with_backup(config)
|
||||
if manifest:
|
||||
print(json.dumps({"pre_migration_backup": manifest["name"]}, indent=2))
|
||||
return 0
|
||||
|
||||
if args.command == "backup":
|
||||
try:
|
||||
manifest = BackupService(config).create(reason=args.reason, keep=args.keep)
|
||||
except BackupError as error:
|
||||
print(str(error))
|
||||
return 1
|
||||
print(json.dumps(manifest, indent=2))
|
||||
return 0
|
||||
|
||||
if args.command == "verify-backup":
|
||||
result = BackupService(config).verify(args.backup)
|
||||
print(json.dumps(result.as_dict(), indent=2))
|
||||
return 0 if result.ok else 1
|
||||
|
||||
if args.command == "restore":
|
||||
try:
|
||||
report = BackupService(config).restore(args.backup, args.into)
|
||||
except BackupError as error:
|
||||
print(str(error))
|
||||
return 1
|
||||
print(json.dumps(report, indent=2))
|
||||
return 0
|
||||
|
||||
if args.command == "benchmark":
|
||||
from photo_pipeline.services import benchmarks
|
||||
|
||||
try:
|
||||
report = benchmarks.run(
|
||||
config,
|
||||
profile=args.profile,
|
||||
soak_seconds=args.soak_seconds,
|
||||
output=args.output,
|
||||
)
|
||||
except ValueError as error:
|
||||
print(str(error))
|
||||
return 1
|
||||
print(json.dumps({k: v for k, v in report.items() if k != "runs"}, indent=2))
|
||||
# A breached budget is a failed run, so a scheduled job notices without
|
||||
# anyone reading the JSON.
|
||||
return 0 if report["ok"] else 1
|
||||
|
||||
if args.command == "release-gate":
|
||||
from photo_pipeline.services import release
|
||||
|
||||
report = release.run_gate(config, output=args.output)
|
||||
print(
|
||||
json.dumps(
|
||||
{k: v for k, v in report.items() if k not in ("stages", "matrix")}, indent=2
|
||||
)
|
||||
)
|
||||
return 0 if report["ok"] else 1
|
||||
|
||||
if args.command == "dry-run":
|
||||
from photo_pipeline.services import release
|
||||
|
||||
try:
|
||||
report = release.dry_run(config)
|
||||
except release.ReleaseError as error:
|
||||
print(str(error))
|
||||
return 1
|
||||
if args.output:
|
||||
Path(args.output).write_text(json.dumps(report, indent=2))
|
||||
print(json.dumps(report, indent=2))
|
||||
return 0
|
||||
|
||||
if args.command == "approve-dry-run":
|
||||
from photo_pipeline.services import release
|
||||
|
||||
try:
|
||||
record = release.approve(config, args.report, approver=args.approver)
|
||||
except (release.ReleaseError, OSError, ValueError) as error:
|
||||
print(str(error))
|
||||
return 1
|
||||
print(json.dumps(record, indent=2))
|
||||
return 0
|
||||
|
||||
if args.command == "diagnostics":
|
||||
from photo_pipeline.services import diagnostics
|
||||
|
||||
print(json.dumps(diagnostics.report(config), indent=2))
|
||||
run_migrations(config.database_url)
|
||||
return 0
|
||||
|
||||
if args.command == "import-legacy-scores":
|
||||
import json
|
||||
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory
|
||||
from photo_pipeline.services.legacy_import import LegacyImportService, write_report
|
||||
|
||||
migrate_with_backup(config)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
service = LegacyImportService(create_session_factory(engine))
|
||||
report = service.import_nsfw_scores(
|
||||
@@ -196,50 +63,18 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
import photo_pipeline.jobs.domain_handlers # noqa: F401
|
||||
from photo_pipeline.jobs.worker import Worker
|
||||
|
||||
lock = LibraryLock(config, "worker")
|
||||
if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
|
||||
return held
|
||||
try:
|
||||
migrate_with_backup(config)
|
||||
engine = create_db_engine(config.database_url)
|
||||
Worker(
|
||||
create_session_factory(engine), worker_id=args.id, config=config
|
||||
).run_forever()
|
||||
finally:
|
||||
lock.release()
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
Worker(create_session_factory(engine), worker_id=args.id, config=config).run_forever()
|
||||
return 0
|
||||
|
||||
import uvicorn
|
||||
|
||||
from photo_pipeline.api.app import create_app
|
||||
|
||||
lock = LibraryLock(config, "api")
|
||||
if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
|
||||
return held
|
||||
try:
|
||||
uvicorn.run(create_app(config), host=config.host, port=config.port)
|
||||
finally:
|
||||
lock.release()
|
||||
uvicorn.run(create_app(config), host=config.host, port=config.port)
|
||||
return 0
|
||||
|
||||
|
||||
def _acquire(lock: LibraryLock, *, allow_legacy: bool) -> int | None:
|
||||
"""Take the lock, or explain on stderr why this process must not start.
|
||||
|
||||
Returns an exit code to return, or ``None`` when the lock was acquired.
|
||||
"""
|
||||
import sys
|
||||
|
||||
try:
|
||||
lock.acquire(allow_legacy=allow_legacy)
|
||||
except LockHeld as error:
|
||||
print(str(error), file=sys.stderr)
|
||||
return 2
|
||||
except LegacyProcessActive as error:
|
||||
print(f"{error} (override with --allow-legacy)", file=sys.stderr)
|
||||
return 3
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
||||
@@ -27,7 +27,6 @@ from photo_pipeline.api.routes import (
|
||||
inventory,
|
||||
jobs,
|
||||
library,
|
||||
operations,
|
||||
renames,
|
||||
safety,
|
||||
session as session_routes,
|
||||
@@ -40,8 +39,7 @@ from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Ses
|
||||
# Registers the safety_score / analysis job handlers on import.
|
||||
import photo_pipeline.jobs.domain_handlers # noqa: F401
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory
|
||||
from photo_pipeline.services.backup import migrate_with_backup
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.logging import configure_logging
|
||||
from photo_pipeline.services.thumbnails import ThumbnailService
|
||||
from photo_pipeline.services.upload_batches import UploadBatchService
|
||||
@@ -91,9 +89,7 @@ def create_app(config: Config | None = None) -> FastAPI:
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
config.database_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
# A schema upgrade is snapshotted first, so a migration that fails halfway
|
||||
# leaves a restorable database behind rather than a damaged one (US07-05).
|
||||
migrate_with_backup(config)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
app.state.config = config
|
||||
app.state.engine = engine
|
||||
@@ -130,7 +126,6 @@ def create_app(config: Config | None = None) -> FastAPI:
|
||||
app.include_router(renames.router, prefix="/api/v1")
|
||||
app.include_router(uploads.router, prefix="/api/v1")
|
||||
app.include_router(archives.router, prefix="/api/v1")
|
||||
app.include_router(operations.router, prefix="/api/v1")
|
||||
# Static single-page app (hash-routed). Mounted last so /api/v1 wins.
|
||||
if FRONTEND_DIR.is_dir():
|
||||
app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="app")
|
||||
|
||||
@@ -11,13 +11,7 @@ from fastapi import APIRouter, Query, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from photo_pipeline.schemas import DecisionRequest
|
||||
from photo_pipeline.services.duplicates import (
|
||||
MAX_MEMBER_PAGE,
|
||||
MEMBER_PAGE,
|
||||
ConflictError,
|
||||
DuplicateError,
|
||||
DuplicateService,
|
||||
)
|
||||
from photo_pipeline.services.duplicates import ConflictError, DuplicateError, DuplicateService
|
||||
|
||||
router = APIRouter(tags=["duplicates"])
|
||||
|
||||
@@ -48,13 +42,8 @@ def list_clusters(
|
||||
|
||||
|
||||
@router.get("/duplicates/clusters/{cluster_id}")
|
||||
def get_cluster(
|
||||
cluster_id: str,
|
||||
request: Request,
|
||||
limit: int = Query(MEMBER_PAGE, ge=1, le=MAX_MEMBER_PAGE),
|
||||
offset: int = Query(0, ge=0),
|
||||
):
|
||||
detail = _service(request).get_cluster(cluster_id, limit=limit, offset=offset)
|
||||
def get_cluster(cluster_id: str, request: Request):
|
||||
detail = _service(request).get_cluster(cluster_id)
|
||||
if detail is None:
|
||||
return _error(404, "not_found", f"unknown cluster {cluster_id}")
|
||||
return detail
|
||||
|
||||
@@ -1,70 +0,0 @@
|
||||
"""Operational endpoints: diagnostics and backups (US07-05).
|
||||
|
||||
Backups can be taken and verified here because both are safe, additive, and the
|
||||
operator needs them from the same screen that shows the disk filling up.
|
||||
|
||||
**Restore is deliberately not an endpoint.** It replaces the state of the running
|
||||
application with an older one, so it belongs to a stopped installation and a person
|
||||
at a terminal: ``python -m photo_pipeline restore``. An HTTP call that can silently
|
||||
roll the library back to last week is a hole, not a feature.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Query, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel
|
||||
|
||||
from photo_pipeline.services import diagnostics
|
||||
from photo_pipeline.services.backup import DEFAULT_KEEP, BackupError, BackupService
|
||||
|
||||
router = APIRouter(tags=["operations"])
|
||||
|
||||
|
||||
class CreateBackupRequest(BaseModel):
|
||||
reason: str = "manual"
|
||||
keep: int = DEFAULT_KEEP
|
||||
|
||||
|
||||
def _service(request: Request) -> BackupService:
|
||||
return BackupService(request.app.state.config)
|
||||
|
||||
|
||||
def _error(status: int, code: str, message: str) -> JSONResponse:
|
||||
return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
|
||||
|
||||
|
||||
@router.get("/diagnostics")
|
||||
def read_diagnostics(request: Request) -> dict:
|
||||
return diagnostics.report(request.app.state.config)
|
||||
|
||||
|
||||
@router.get("/backups")
|
||||
def list_backups(request: Request) -> dict:
|
||||
return {"backups": _service(request).list()}
|
||||
|
||||
|
||||
@router.post("/backups", status_code=201)
|
||||
def create_backup(body: CreateBackupRequest, request: Request):
|
||||
try:
|
||||
return _service(request).create(reason=body.reason, keep=body.keep)
|
||||
except BackupError as error:
|
||||
return _error(422, "backup_failed", str(error))
|
||||
|
||||
|
||||
@router.get("/backups/{name}/verify")
|
||||
def verify_backup(name: str, request: Request):
|
||||
service = _service(request)
|
||||
# The name comes from the browser, so it names a backup — it is never joined
|
||||
# into a path until it has been matched against one that exists (US07-02).
|
||||
if name not in {entry["name"] for entry in service.list()}:
|
||||
return _error(404, "not_found", f"unknown backup {name}")
|
||||
return {"name": name, **service.verify(service.root / name).as_dict()}
|
||||
|
||||
|
||||
@router.post("/backups/prune")
|
||||
def prune_backups(request: Request, keep: int = Query(DEFAULT_KEEP, ge=1)):
|
||||
try:
|
||||
return {"removed": _service(request).prune(keep=keep)}
|
||||
except BackupError as error:
|
||||
return _error(422, "invalid_retention", str(error))
|
||||
@@ -43,9 +43,6 @@ PUBLIC_PATHS = frozenset(
|
||||
{f"{API_PREFIX}/health/live", f"{API_PREFIX}/health/ready", f"{API_PREFIX}/session"}
|
||||
)
|
||||
LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"})
|
||||
# Mutating endpoints that must stay reachable while mutation itself is gated: the
|
||||
# backup a careful operator takes first, and its retention (US07-07).
|
||||
MUTATION_EXEMPT_PATHS = frozenset({f"{API_PREFIX}/backups", f"{API_PREFIX}/backups/prune"})
|
||||
|
||||
# Applied to every response. No inline script/style is used by the frontend, so the
|
||||
# policy can stay strict; `frame-ancestors 'none'` and CORP keep other pages from
|
||||
@@ -148,7 +145,6 @@ class SecurityMiddleware:
|
||||
def __init__(self, app, *, session: Session, config) -> None:
|
||||
self.app = app
|
||||
self.session = session
|
||||
self.config = config
|
||||
self.max_request_bytes = config.max_request_bytes
|
||||
self.allowed_hosts = frozenset(LOOPBACK_HOSTS | {str(config.host).lower()})
|
||||
|
||||
@@ -169,8 +165,6 @@ class SecurityMiddleware:
|
||||
allowed_hosts=self.allowed_hosts,
|
||||
max_request_bytes=self.max_request_bytes,
|
||||
)
|
||||
if refusal is None:
|
||||
refusal = self._mutation_refusal(scope)
|
||||
if refusal is not None:
|
||||
response = JSONResponse(
|
||||
status_code=refusal.status,
|
||||
@@ -189,26 +183,6 @@ class SecurityMiddleware:
|
||||
|
||||
await self.app(scope, receive, send_with_headers)
|
||||
|
||||
def _mutation_refusal(self, scope) -> Refusal | None:
|
||||
"""Refuse every mutating request while the library's dry run is unapproved.
|
||||
|
||||
One choke point for the whole API: every mutation the browser can start is a
|
||||
non-safe method under ``/api/v1``. Reading stays open — an operator has to be
|
||||
able to look at what the application found in order to approve it (US07-07).
|
||||
"""
|
||||
method = scope.get("method", "GET").upper()
|
||||
path = scope.get("path", "/")
|
||||
if method in SAFE_METHODS or not path.startswith(API_PREFIX):
|
||||
return None
|
||||
if path in MUTATION_EXEMPT_PATHS:
|
||||
return None
|
||||
from photo_pipeline.services.release import mutation_blockers
|
||||
|
||||
blockers = mutation_blockers(self.config)
|
||||
if not blockers:
|
||||
return None
|
||||
return Refusal(403, blockers[0]["code"], blockers[0]["message"])
|
||||
|
||||
|
||||
def _cookie(header: str, name: str) -> str:
|
||||
for part in header.split(";"):
|
||||
|
||||
@@ -18,61 +18,6 @@ from typing import Mapping
|
||||
from pydantic import BaseModel, ConfigDict, SecretStr
|
||||
|
||||
ENV_PREFIX = "PHOTO_PIPELINE_"
|
||||
ENV_FILE_VAR = f"{ENV_PREFIX}ENV_FILE"
|
||||
DEFAULT_ENV_FILE = Path(".env")
|
||||
|
||||
# The archived CLI's variable names, so the configuration file an operator already
|
||||
# has keeps working. The vision provider reads the OpenAI SDK's names, and the
|
||||
# library root is configuration here rather than a bare path (US07-01 donor).
|
||||
LEGACY_ALIASES = {
|
||||
"LLM_API_KEY": "OPENAI_API_KEY",
|
||||
"GEMINI_API_KEY": "OPENAI_API_KEY",
|
||||
"LLM_BASE_URL": "OPENAI_BASE_URL",
|
||||
"LIBRARY": f"{ENV_PREFIX}LIBRARY_ROOTS",
|
||||
}
|
||||
|
||||
|
||||
def parse_env_file(text: str) -> dict[str, str]:
|
||||
"""``KEY=value`` lines into a mapping. Comments, blanks, and quotes handled.
|
||||
|
||||
Deliberately not a shell: no interpolation, no ``export``, no multi-line values.
|
||||
A configuration file that can run code is a configuration file that can be a
|
||||
vulnerability.
|
||||
"""
|
||||
values: dict[str, str] = {}
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, _, raw = line.partition("=")
|
||||
key = key.strip()
|
||||
if not key or key.startswith("#"):
|
||||
continue
|
||||
value = raw.strip().strip('"').strip("'")
|
||||
values[key] = value
|
||||
alias = LEGACY_ALIASES.get(key)
|
||||
if alias:
|
||||
values.setdefault(alias, value)
|
||||
return values
|
||||
|
||||
|
||||
def load_env_file(path: Path | str | None = None) -> dict[str, str]:
|
||||
"""Load ``PHOTO_PIPELINE_ENV_FILE`` (or ``./.env``) into the environment.
|
||||
|
||||
Anything already exported wins: a file is the standing configuration, the shell
|
||||
is what you meant *this time*. Returns what it applied, which is what the CLI
|
||||
prints — names only, never values.
|
||||
"""
|
||||
candidate = path or os.environ.get(ENV_FILE_VAR) or DEFAULT_ENV_FILE
|
||||
candidate = Path(candidate)
|
||||
if not candidate.is_file():
|
||||
return {}
|
||||
applied = {}
|
||||
for key, value in parse_env_file(candidate.read_text()).items():
|
||||
if key not in os.environ:
|
||||
os.environ[key] = value
|
||||
applied[key] = value
|
||||
return applied
|
||||
|
||||
|
||||
class Config(BaseModel):
|
||||
@@ -97,12 +42,6 @@ class Config(BaseModel):
|
||||
# Free space an archive destination must keep beyond the transfer itself.
|
||||
archive_free_space_reserve_bytes: int = 1_000_000_000
|
||||
|
||||
# Refuse every mutating request until a read-only dry run of the configured
|
||||
# library has been produced and explicitly approved (US07-07). Off by default so
|
||||
# a development setup is unchanged; turn it on before pointing the application at
|
||||
# a library whose photos cannot be replaced.
|
||||
require_dry_run_approval: bool = False
|
||||
|
||||
vision_api_key: SecretStr | None = None
|
||||
immich_api_key: SecretStr | None = None
|
||||
immich_server_url: str = ""
|
||||
@@ -123,9 +62,6 @@ class Config(BaseModel):
|
||||
@classmethod
|
||||
def from_env(cls, environ: Mapping[str, str] | None = None) -> "Config":
|
||||
env = os.environ if environ is None else environ
|
||||
if environ is None:
|
||||
load_env_file() # a file never overrides what the shell already set
|
||||
env = os.environ
|
||||
data: dict = {}
|
||||
for name in cls.model_fields:
|
||||
raw = env.get(ENV_PREFIX + name.upper())
|
||||
|
||||
@@ -41,41 +41,12 @@ def create_session_factory(engine: Engine) -> sessionmaker:
|
||||
return sessionmaker(bind=engine, expire_on_commit=False, future=True)
|
||||
|
||||
|
||||
def _alembic_config(url: str):
|
||||
def run_migrations(url: str) -> None:
|
||||
"""Upgrade the database at ``url`` to the latest revision."""
|
||||
from alembic import command
|
||||
from alembic.config import Config as AlembicConfig
|
||||
|
||||
cfg = AlembicConfig(str(_REPO_ROOT / "alembic.ini"))
|
||||
cfg.set_main_option("script_location", str(_REPO_ROOT / "migrations"))
|
||||
cfg.set_main_option("sqlalchemy.url", url)
|
||||
return cfg
|
||||
|
||||
|
||||
def run_migrations(url: str) -> None:
|
||||
"""Upgrade the database at ``url`` to the latest revision."""
|
||||
from alembic import command
|
||||
|
||||
command.upgrade(_alembic_config(url), "head")
|
||||
|
||||
|
||||
def head_revision() -> str | None:
|
||||
"""The revision this code expects. ``None`` if the scripts cannot be read."""
|
||||
from alembic.script import ScriptDirectory
|
||||
|
||||
try:
|
||||
return ScriptDirectory.from_config(_alembic_config("sqlite://")).get_current_head()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def current_revision(url: str) -> str | None:
|
||||
"""The revision a database is actually at, or ``None`` for an unstamped one."""
|
||||
engine = create_db_engine(url)
|
||||
try:
|
||||
with engine.connect() as connection:
|
||||
from alembic.runtime.migration import MigrationContext
|
||||
|
||||
return MigrationContext.configure(connection).get_current_revision()
|
||||
except Exception:
|
||||
return None
|
||||
finally:
|
||||
engine.dispose()
|
||||
command.upgrade(cfg, "head")
|
||||
|
||||
@@ -18,6 +18,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Protocol
|
||||
|
||||
@@ -25,9 +26,9 @@ from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline import path_policy
|
||||
from photo_pipeline.models import AnalysisResult, Asset
|
||||
from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
|
||||
from photo_pipeline.services import exif_checkpoint
|
||||
from photo_pipeline.services.safety import SFW, latest_reviews
|
||||
from photo_pipeline.services.safety import SFW
|
||||
|
||||
MODEL = "gemini-2.5-flash"
|
||||
PROMPT_VERSION = "1"
|
||||
@@ -71,26 +72,11 @@ class AnalysisService:
|
||||
|
||||
def _sfw_asset_ids(self, session) -> set[str]:
|
||||
"""Asset ids whose latest safety decision is ``sfw`` — the ONLY assets that
|
||||
may reach the provider.
|
||||
|
||||
The "latest row wins" rule is applied in SQL (US07-06); loading every review
|
||||
to fold it in Python made the gate cost grow with the review history rather
|
||||
than with the work being gated.
|
||||
"""
|
||||
latest = latest_reviews().subquery()
|
||||
return set(
|
||||
session.scalars(select(latest.c.asset_id).where(latest.c.decision == SFW))
|
||||
)
|
||||
|
||||
def _sfw_count(self, session) -> int:
|
||||
"""How many assets the gate currently allows, without listing them."""
|
||||
latest = latest_reviews().subquery()
|
||||
return int(
|
||||
session.scalar(
|
||||
select(func.count()).select_from(latest).where(latest.c.decision == SFW)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
may reach the provider."""
|
||||
latest: dict[str, str | None] = {}
|
||||
for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
|
||||
latest[review.asset_id] = review.decision
|
||||
return {aid for aid, decision in latest.items() if decision == SFW}
|
||||
|
||||
def _is_still_sfw(self, asset_id: str) -> bool:
|
||||
"""Re-read the current safety decision straight from the database."""
|
||||
@@ -118,12 +104,9 @@ class AnalysisService:
|
||||
)
|
||||
return [a.id for a in assets if a.id not in done]
|
||||
|
||||
def counts(self, *, eligible: int | None = None) -> dict[str, int]:
|
||||
"""Analysis progress. ``eligible`` may be passed by a caller that has just
|
||||
counted confirmed-SFW assets, so the workflow home does not resolve the
|
||||
latest decision of every asset twice on one page load (US07-06)."""
|
||||
def counts(self) -> dict[str, int]:
|
||||
with self._session_factory() as session:
|
||||
eligible = self._sfw_count(session) if eligible is None else eligible
|
||||
sfw = self._sfw_asset_ids(session)
|
||||
rows = dict(
|
||||
session.execute(
|
||||
select(AnalysisResult.status, func.count()).group_by(AnalysisResult.status)
|
||||
@@ -132,10 +115,10 @@ class AnalysisService:
|
||||
analyzed = int(rows.get("analyzed", 0))
|
||||
errored = int(rows.get("error", 0))
|
||||
return {
|
||||
"eligible": eligible,
|
||||
"eligible": len(sfw),
|
||||
"analyzed": analyzed,
|
||||
"error": errored,
|
||||
"pending": max(eligible - analyzed - errored, 0),
|
||||
"pending": max(len(sfw) - analyzed - errored, 0),
|
||||
}
|
||||
|
||||
def run(self, asset_ids: list[str] | None = None) -> dict:
|
||||
@@ -264,11 +247,8 @@ class AnalysisService:
|
||||
asset = session.get(Asset, asset_id)
|
||||
if asset is not None and checkpoint.sha256:
|
||||
# The bytes changed when the container was rewritten; upload must use
|
||||
# the hash of what is actually on disk now (concept §3), and the
|
||||
# recorded size has to move with it (US07-07).
|
||||
# the hash of what is actually on disk now (concept §3).
|
||||
asset.current_sha256 = checkpoint.sha256
|
||||
if checkpoint.byte_size is not None:
|
||||
asset.byte_size = checkpoint.byte_size
|
||||
session.commit()
|
||||
|
||||
def get(self, asset_id: str) -> dict | None:
|
||||
|
||||
@@ -1,221 +0,0 @@
|
||||
"""Library-level process lock, and detection of an incompatible legacy run
|
||||
(US07-05, concept §15 "migration and operational risks").
|
||||
|
||||
Every safety this application has — durable job leases, rename journals, archive
|
||||
manifests — assumes that one installation owns the library. Two workers, or the
|
||||
frozen CLI running beside the app, break that assumption *below* the level those
|
||||
mechanisms can see: the second process simply does not know the first one's
|
||||
database exists.
|
||||
|
||||
So mutation requires a file lock in the data directory, shaped as JSON so any
|
||||
future or migrated entry point can read and honour it without importing this
|
||||
package:
|
||||
|
||||
{"lock_version": 1, "role": "worker", "pid": 4242, "host": "...",
|
||||
"started_at": "...", "library_roots": ["..."]}
|
||||
|
||||
One holder per role: an API and a worker are designed to run together, a second
|
||||
worker is not. A lock whose process is gone is stale and is taken over with the
|
||||
takeover recorded — refusing to start because of a crashed predecessor would turn
|
||||
one outage into two.
|
||||
|
||||
Legacy detection is deliberately a heuristic, not a promise: the archived CLI has
|
||||
no lock of its own, so what can be observed is its state files being written right
|
||||
now. Recent writes to them mean something else is mutating this library, and every
|
||||
mutating stage should refuse until it stops.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
|
||||
LOCK_VERSION = 1
|
||||
LOCK_SUFFIX = ".lock.json"
|
||||
# State files only the archived CLIs write. Their presence is history; a *recent*
|
||||
# modification is a running process.
|
||||
# ponytail: the real fix is a lock the migrated CLI paths take too — this catches
|
||||
# the frozen archive, which has no lock and cannot be changed (US07-01).
|
||||
LEGACY_ARTIFACTS = (
|
||||
"photo_analyzer.db",
|
||||
"nsfw_scores.csv",
|
||||
"photo_analyzer_history.jsonl",
|
||||
"photo_analyzer.log",
|
||||
"photo_analyzer_debug.log",
|
||||
)
|
||||
LEGACY_ACTIVE_SECONDS = 300
|
||||
|
||||
|
||||
class LockHeld(RuntimeError):
|
||||
"""Another live process of the same role owns this library."""
|
||||
|
||||
def __init__(self, holder: "Holder") -> None:
|
||||
super().__init__(
|
||||
f"{holder.role} is already running for this library "
|
||||
f"(pid {holder.pid} on {holder.host}, since {holder.started_at})"
|
||||
)
|
||||
self.holder = holder
|
||||
|
||||
|
||||
class LegacyProcessActive(RuntimeError):
|
||||
"""A legacy CLI appears to be mutating the same library right now."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Holder:
|
||||
role: str
|
||||
pid: int
|
||||
host: str
|
||||
started_at: str
|
||||
lock_version: int = LOCK_VERSION
|
||||
library_roots: tuple[str, ...] = ()
|
||||
|
||||
@property
|
||||
def alive(self) -> bool:
|
||||
"""Whether the recorded process still exists on this host.
|
||||
|
||||
A lock from another host cannot be probed, so it is believed: assuming a
|
||||
remote holder is dead is how two machines end up renaming the same folder.
|
||||
"""
|
||||
if self.host != socket.gethostname():
|
||||
return True
|
||||
try:
|
||||
os.kill(self.pid, 0)
|
||||
except ProcessLookupError:
|
||||
return False
|
||||
except PermissionError:
|
||||
return True # exists, owned by someone else
|
||||
return True
|
||||
|
||||
def as_dict(self) -> dict:
|
||||
return {
|
||||
"lock_version": self.lock_version,
|
||||
"role": self.role,
|
||||
"pid": self.pid,
|
||||
"host": self.host,
|
||||
"started_at": self.started_at,
|
||||
"library_roots": list(self.library_roots),
|
||||
"alive": self.alive,
|
||||
}
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def legacy_activity(config: Config) -> dict:
|
||||
"""Legacy state files written within the activity window, if any."""
|
||||
seen: list[dict] = []
|
||||
cutoff = _now().timestamp() - LEGACY_ACTIVE_SECONDS
|
||||
roots = [Path(root) for root in config.library_roots] + [Path(config.data_dir)]
|
||||
for root in roots:
|
||||
for name in LEGACY_ARTIFACTS:
|
||||
path = root / name
|
||||
try:
|
||||
modified = path.stat().st_mtime
|
||||
except OSError:
|
||||
continue
|
||||
if modified >= cutoff:
|
||||
seen.append(
|
||||
{
|
||||
"path": str(path),
|
||||
"modified_at": datetime.fromtimestamp(modified, timezone.utc).isoformat(),
|
||||
}
|
||||
)
|
||||
return {"active": bool(seen), "artifacts": seen, "window_seconds": LEGACY_ACTIVE_SECONDS}
|
||||
|
||||
|
||||
class LibraryLock:
|
||||
"""One holder per role for one library. Used as a context manager."""
|
||||
|
||||
def __init__(self, config: Config, role: str = "worker") -> None:
|
||||
self._config = config
|
||||
self.role = role
|
||||
self.path = Path(config.data_dir) / f"{role}{LOCK_SUFFIX}"
|
||||
self._acquired = False
|
||||
|
||||
# ── inspection ────────────────────────────────────────────────────────────
|
||||
|
||||
def holder(self) -> Holder | None:
|
||||
try:
|
||||
payload = json.loads(self.path.read_text())
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
try:
|
||||
return Holder(
|
||||
role=payload["role"],
|
||||
pid=int(payload["pid"]),
|
||||
host=payload["host"],
|
||||
started_at=payload["started_at"],
|
||||
lock_version=int(payload.get("lock_version", LOCK_VERSION)),
|
||||
library_roots=tuple(payload.get("library_roots", ())),
|
||||
)
|
||||
except (KeyError, TypeError, ValueError):
|
||||
# An unreadable lock is not an absent lock: something wrote it.
|
||||
return Holder(role=self.role, pid=-1, host="unknown", started_at="unknown")
|
||||
|
||||
# ── acquire / release ─────────────────────────────────────────────────────
|
||||
|
||||
def acquire(self, *, allow_legacy: bool = False) -> Holder:
|
||||
"""Take the lock for this role, or explain who has it.
|
||||
|
||||
Raises ``LockHeld`` when a live process of the same role owns the library,
|
||||
and ``LegacyProcessActive`` when the archived CLI looks like it is running
|
||||
against it.
|
||||
"""
|
||||
if not allow_legacy:
|
||||
legacy = legacy_activity(self._config)
|
||||
if legacy["active"]:
|
||||
raise LegacyProcessActive(
|
||||
"a legacy CLI is writing this library "
|
||||
f"({', '.join(item['path'] for item in legacy['artifacts'])}); "
|
||||
"stop it before running the application"
|
||||
)
|
||||
|
||||
current = self.holder()
|
||||
if current is not None:
|
||||
if current.alive:
|
||||
raise LockHeld(current)
|
||||
# Stale: its process is gone. Take over, and say so.
|
||||
self.path.unlink(missing_ok=True)
|
||||
|
||||
mine = Holder(
|
||||
role=self.role,
|
||||
pid=os.getpid(),
|
||||
host=socket.gethostname(),
|
||||
started_at=_now().isoformat(),
|
||||
library_roots=tuple(str(root) for root in self._config.library_roots),
|
||||
)
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
payload = {k: v for k, v in mine.as_dict().items() if k != "alive"}
|
||||
# Exclusive create, so two processes racing here cannot both believe they won.
|
||||
try:
|
||||
with open(self.path, "x", encoding="utf-8") as handle:
|
||||
json.dump(payload, handle, indent=2)
|
||||
except FileExistsError:
|
||||
winner = self.holder()
|
||||
raise LockHeld(winner or mine) from None
|
||||
self._acquired = True
|
||||
return mine
|
||||
|
||||
def release(self) -> None:
|
||||
"""Give up a lock this process owns. Another holder's lock is left alone."""
|
||||
if not self._acquired:
|
||||
return
|
||||
current = self.holder()
|
||||
if current is not None and current.pid == os.getpid():
|
||||
self.path.unlink(missing_ok=True)
|
||||
self._acquired = False
|
||||
|
||||
def __enter__(self) -> "LibraryLock":
|
||||
self.acquire()
|
||||
return self
|
||||
|
||||
def __exit__(self, *_) -> None:
|
||||
self.release()
|
||||
@@ -1,413 +0,0 @@
|
||||
"""Online backups, verification, retention, and restore drills (US07-05).
|
||||
|
||||
A backup taken by copying a live SQLite file is not a backup: with WAL enabled the
|
||||
file on disk is missing every committed page still in the write-ahead log, and a
|
||||
writer mid-transaction makes the copy inconsistent. So every backup here goes
|
||||
through SQLite's online backup API, which takes a consistent snapshot of a database
|
||||
that is still being used (concept §3).
|
||||
|
||||
A backup directory holds exactly two things:
|
||||
|
||||
photo_pipeline.db the snapshot
|
||||
manifest.json what it is, what it came from, and how to check it
|
||||
|
||||
The manifest is what makes the snapshot restorable by someone who was not there
|
||||
when it was taken: the schema revision, the snapshot's SHA-256, the row counts it
|
||||
should still have, the archive locations whose media the library depends on, and
|
||||
which configuration values were set — **names and non-secret values only**. A
|
||||
secret is recorded as "configured", never as its value, so a manifest can be
|
||||
attached to a bug report.
|
||||
|
||||
Restore never writes into a live installation: it refuses a target that already
|
||||
holds a database, because the one thing worse than a lost library is a half-merged
|
||||
one. The drill is documented in README ("Backup and recovery").
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import shutil
|
||||
import sqlite3
|
||||
from contextlib import closing
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
DB_NAME = "photo_pipeline.db"
|
||||
MANIFEST_NAME = "manifest.json"
|
||||
# How many backups the retention helper keeps by default. Small on purpose: a
|
||||
# backup is a snapshot of state that is itself recoverable from the library, and
|
||||
# the disk it lives on is the same one the low-disk warning watches.
|
||||
DEFAULT_KEEP = 7
|
||||
# Tables whose row counts are worth proving after a restore. Not the whole schema —
|
||||
# these are the ones whose loss would be silent.
|
||||
COUNTED_TABLES = (
|
||||
"assets",
|
||||
"asset_paths",
|
||||
"safety_reviews",
|
||||
"analysis_results",
|
||||
"exif_projections",
|
||||
"upload_batches",
|
||||
"upload_items",
|
||||
"archive_locations",
|
||||
"archive_plans",
|
||||
"archive_operations",
|
||||
"rename_plans",
|
||||
"rename_operations",
|
||||
)
|
||||
|
||||
|
||||
class BackupError(RuntimeError):
|
||||
"""The backup could not be created, read, verified, or restored."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VerifyResult:
|
||||
ok: bool
|
||||
issues: tuple[str, ...] = ()
|
||||
revision: str | None = None
|
||||
counts: dict | None = None
|
||||
|
||||
def as_dict(self) -> dict:
|
||||
return {
|
||||
"ok": self.ok,
|
||||
"issues": list(self.issues),
|
||||
"revision": self.revision,
|
||||
"counts": self.counts,
|
||||
}
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _revision(database: Path) -> str | None:
|
||||
with closing(sqlite3.connect(database)) as connection:
|
||||
try:
|
||||
row = connection.execute("SELECT version_num FROM alembic_version").fetchone()
|
||||
except sqlite3.Error:
|
||||
return None
|
||||
return row[0] if row else None
|
||||
|
||||
|
||||
def _counts(database: Path) -> dict:
|
||||
counts: dict[str, int] = {}
|
||||
with closing(sqlite3.connect(database)) as connection:
|
||||
for table in COUNTED_TABLES:
|
||||
try:
|
||||
counts[table] = connection.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
|
||||
except sqlite3.Error:
|
||||
continue # a table this revision does not have yet
|
||||
return counts
|
||||
|
||||
|
||||
def _integrity(database: Path) -> tuple[str, list[str]]:
|
||||
"""``PRAGMA integrity_check`` plus ``foreign_key_check`` — structure and links.
|
||||
|
||||
Structural soundness is not referential soundness: a database can pass
|
||||
``integrity_check`` and still hold an upload item pointing at an asset that
|
||||
is gone.
|
||||
"""
|
||||
issues: list[str] = []
|
||||
with closing(sqlite3.connect(database)) as connection:
|
||||
try:
|
||||
result = connection.execute("PRAGMA integrity_check").fetchone()[0]
|
||||
if result != "ok":
|
||||
issues.append(f"integrity_check: {result}")
|
||||
violations = connection.execute("PRAGMA foreign_key_check").fetchall()
|
||||
if violations:
|
||||
issues.append(f"foreign_key_check: {len(violations)} violation(s)")
|
||||
except sqlite3.DatabaseError as error:
|
||||
issues.append(f"unreadable: {error}")
|
||||
return "error", issues
|
||||
return "ok" if not issues else "damaged", issues
|
||||
|
||||
|
||||
def configuration_references(config: Config) -> dict:
|
||||
"""Which configuration a restore has to reproduce — never the secrets themselves.
|
||||
|
||||
Paths and URLs are recorded because a restore into a fresh root has to be told
|
||||
where the library and the Immich server were; API keys are recorded as
|
||||
``configured`` so an operator knows one is required without the manifest ever
|
||||
carrying it.
|
||||
"""
|
||||
return {
|
||||
"data_dir": str(config.data_dir),
|
||||
"database_path": str(config.database_path),
|
||||
"library_roots": [str(root) for root in config.library_roots],
|
||||
"thumbnail_cache_dir": str(config.thumbnail_cache_dir),
|
||||
"immich_server_url": config.immich_server_url,
|
||||
"immich_go_binary": config.immich_go_binary,
|
||||
"secrets": {
|
||||
"immich_api_key": "configured" if config.immich_api_key else "unset",
|
||||
"vision_api_key": "configured" if config.vision_api_key else "unset",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def migrate_with_backup(config: Config) -> dict | None:
|
||||
"""Upgrade the schema, with a snapshot first when there is state to lose.
|
||||
|
||||
A migration is the one routine operation that can damage every record at once,
|
||||
and Alembic's own transaction does not cover SQLite DDL reliably. So a pending
|
||||
upgrade is preceded by an online backup, and a failed upgrade names it in the
|
||||
error: recovery is "restore that directory", not "reconstruct the library".
|
||||
Returns the manifest of the backup it took, or ``None`` when none was needed.
|
||||
"""
|
||||
import logging
|
||||
|
||||
from photo_pipeline.db import run_migrations
|
||||
|
||||
service = BackupService(config)
|
||||
manifest = service.pre_migration() if service.migration_pending() else None
|
||||
try:
|
||||
run_migrations(config.database_url)
|
||||
except Exception:
|
||||
if manifest is not None:
|
||||
logging.getLogger(__name__).error(
|
||||
"migration failed; restore the pre-migration backup at %s",
|
||||
service.root / manifest["name"],
|
||||
)
|
||||
raise
|
||||
return manifest
|
||||
|
||||
|
||||
class BackupService:
|
||||
def __init__(self, config: Config) -> None:
|
||||
self._config = config
|
||||
|
||||
@property
|
||||
def root(self) -> Path:
|
||||
return self._config.data_dir / "backups"
|
||||
|
||||
# ── create ────────────────────────────────────────────────────────────────
|
||||
|
||||
def create(self, *, reason: str = "manual", keep: int | None = DEFAULT_KEEP) -> dict:
|
||||
"""Take an online snapshot and describe it. Returns the manifest."""
|
||||
source = self._config.database_path
|
||||
if not source.exists():
|
||||
raise BackupError(f"no database at {source}")
|
||||
|
||||
stamp = _now().strftime("%Y%m%dT%H%M%SZ")
|
||||
safe_reason = "".join(c for c in reason if c.isalnum() or c in "-_") or "manual"
|
||||
directory = self.root / f"{stamp}-{safe_reason}"
|
||||
if directory.exists(): # same second, same reason
|
||||
directory = self.root / f"{stamp}-{safe_reason}-{len(list(self.root.iterdir()))}"
|
||||
directory.mkdir(parents=True)
|
||||
|
||||
target = directory / DB_NAME
|
||||
try:
|
||||
with closing(sqlite3.connect(source)) as src, closing(sqlite3.connect(target)) as dst:
|
||||
src.backup(dst) # the online backup API, not a file copy
|
||||
except (sqlite3.Error, OSError) as error:
|
||||
shutil.rmtree(directory, ignore_errors=True)
|
||||
raise BackupError(f"backup failed: {error}") from error
|
||||
|
||||
state, issues = _integrity(target)
|
||||
manifest = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"name": directory.name,
|
||||
"created_at": _now().isoformat(),
|
||||
"reason": reason,
|
||||
"revision": _revision(target),
|
||||
"database": {
|
||||
"name": DB_NAME,
|
||||
"bytes": target.stat().st_size,
|
||||
"sha256": sha256_file(target),
|
||||
"integrity": state,
|
||||
"issues": issues,
|
||||
},
|
||||
"counts": _counts(target),
|
||||
"archive_locations": self._archive_locations(),
|
||||
"configuration": configuration_references(self._config),
|
||||
"retention": {
|
||||
"keep": keep,
|
||||
"guidance": (
|
||||
"Keep the newest snapshot on a different disk than data_dir, and one "
|
||||
"off-site copy per archive medium. A backup only covers the database: "
|
||||
"the photos themselves live in the library and archive locations named "
|
||||
"above, which need their own copies."
|
||||
),
|
||||
},
|
||||
}
|
||||
(directory / MANIFEST_NAME).write_text(json.dumps(manifest, indent=2))
|
||||
if keep is not None:
|
||||
manifest["pruned"] = self.prune(keep=keep)
|
||||
return manifest
|
||||
|
||||
def migration_pending(self) -> bool:
|
||||
"""True when the database exists and is not at the revision this code wants."""
|
||||
from photo_pipeline.db import current_revision, head_revision
|
||||
|
||||
if not self._config.database_path.exists():
|
||||
return False
|
||||
return current_revision(self._config.database_url) != head_revision()
|
||||
|
||||
def pre_migration(self) -> dict | None:
|
||||
"""Snapshot before a schema change, when there is something to lose.
|
||||
|
||||
Returns ``None`` when the database does not exist yet (a fresh install has
|
||||
no state a failed migration could damage).
|
||||
"""
|
||||
if not self._config.database_path.exists():
|
||||
return None
|
||||
return self.create(reason="pre-migration")
|
||||
|
||||
def _archive_locations(self) -> list[dict]:
|
||||
"""The media the library's archived originals live on.
|
||||
|
||||
A restored database still points at these; if they are not restored too,
|
||||
the pictures are gone even though every record survived.
|
||||
"""
|
||||
engine = create_db_engine(self._config.database_url)
|
||||
try:
|
||||
factory = create_session_factory(engine)
|
||||
with factory() as session:
|
||||
rows = session.execute(
|
||||
text("SELECT id, name, root, media_id, state FROM archive_locations")
|
||||
).mappings().all()
|
||||
except Exception:
|
||||
return []
|
||||
finally:
|
||||
engine.dispose()
|
||||
return [
|
||||
{
|
||||
"id": row["id"],
|
||||
"name": row["name"],
|
||||
"root": row["root"],
|
||||
"media_id": row["media_id"],
|
||||
"last_state": row["state"],
|
||||
"mounted": Path(row["root"]).is_dir(),
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
# ── inspect ───────────────────────────────────────────────────────────────
|
||||
|
||||
def list(self) -> list[dict]:
|
||||
"""Every backup, newest first, with what is known about it."""
|
||||
if not self.root.is_dir():
|
||||
return []
|
||||
entries = []
|
||||
for directory in sorted(self.root.iterdir(), reverse=True):
|
||||
if not directory.is_dir():
|
||||
continue
|
||||
manifest = self.manifest(directory)
|
||||
database = directory / DB_NAME
|
||||
entries.append(
|
||||
{
|
||||
"name": directory.name,
|
||||
"path": str(directory),
|
||||
"created_at": (manifest or {}).get("created_at"),
|
||||
"reason": (manifest or {}).get("reason"),
|
||||
"revision": (manifest or {}).get("revision"),
|
||||
"bytes": database.stat().st_size if database.exists() else 0,
|
||||
"complete": bool(manifest) and database.exists(),
|
||||
}
|
||||
)
|
||||
return entries
|
||||
|
||||
def manifest(self, directory: Path) -> dict | None:
|
||||
path = Path(directory) / MANIFEST_NAME
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
return json.loads(path.read_text())
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
def verify(self, directory: Path | str) -> VerifyResult:
|
||||
"""Prove a snapshot is still the one that was taken and still readable."""
|
||||
directory = Path(directory)
|
||||
if not directory.is_dir():
|
||||
return VerifyResult(False, (f"no backup at {directory}",))
|
||||
manifest = self.manifest(directory)
|
||||
if manifest is None:
|
||||
return VerifyResult(False, ("manifest is missing or unreadable",))
|
||||
database = directory / manifest["database"]["name"]
|
||||
if not database.exists():
|
||||
return VerifyResult(False, ("the snapshot file is missing",), manifest.get("revision"))
|
||||
|
||||
issues: list[str] = []
|
||||
if sha256_file(database) != manifest["database"]["sha256"]:
|
||||
# Bit rot, a truncated copy, or an edited snapshot: all three mean the
|
||||
# bytes are not the ones that were verified when the backup was made.
|
||||
issues.append("sha256 does not match the manifest")
|
||||
state, structural = _integrity(database)
|
||||
issues.extend(structural)
|
||||
counts = _counts(database) if state != "error" else None
|
||||
if counts is not None and manifest.get("counts") and counts != manifest["counts"]:
|
||||
issues.append(f"row counts changed: {manifest['counts']} -> {counts}")
|
||||
return VerifyResult(not issues, tuple(issues), manifest.get("revision"), counts)
|
||||
|
||||
# ── retention ─────────────────────────────────────────────────────────────
|
||||
|
||||
def prune(self, *, keep: int = DEFAULT_KEEP) -> list[str]:
|
||||
"""Delete the oldest backups beyond ``keep``. Never deletes the newest one."""
|
||||
if keep < 1:
|
||||
raise BackupError("retention must keep at least one backup")
|
||||
removed = []
|
||||
for entry in self.list()[keep:]:
|
||||
shutil.rmtree(entry["path"], ignore_errors=True)
|
||||
removed.append(entry["name"])
|
||||
return removed
|
||||
|
||||
# ── restore ───────────────────────────────────────────────────────────────
|
||||
|
||||
def restore(self, directory: Path | str, target_data_dir: Path | str) -> dict:
|
||||
"""Restore a verified snapshot into a **fresh** data directory.
|
||||
|
||||
Refuses a target that already holds a database. Restoring on top of a live
|
||||
installation would merge two histories that disagree about which files were
|
||||
renamed, uploaded, and archived — the one failure this whole story exists to
|
||||
prevent. Recovering in place is: stop everything, move the old data
|
||||
directory aside, restore into a new one.
|
||||
"""
|
||||
directory = Path(directory)
|
||||
result = self.verify(directory)
|
||||
if not result.ok:
|
||||
raise BackupError(f"refusing to restore an unverified backup: {result.issues}")
|
||||
|
||||
target = Path(target_data_dir)
|
||||
target.mkdir(parents=True, exist_ok=True)
|
||||
destination = target / DB_NAME
|
||||
if destination.exists():
|
||||
raise BackupError(
|
||||
f"{destination} already exists; restore into a fresh data directory"
|
||||
)
|
||||
shutil.copy2(directory / DB_NAME, destination)
|
||||
# The write-ahead log of the *source* installation must not travel with a
|
||||
# snapshot: the backup API already folded every committed page into it.
|
||||
for leftover in (target / f"{DB_NAME}-wal", target / f"{DB_NAME}-shm"):
|
||||
leftover.unlink(missing_ok=True)
|
||||
|
||||
restored = _integrity(destination)
|
||||
return {
|
||||
"backup": directory.name,
|
||||
"restored_to": str(destination),
|
||||
"revision": result.revision,
|
||||
"counts": _counts(destination),
|
||||
"integrity": restored[0],
|
||||
"issues": restored[1],
|
||||
"next_steps": [
|
||||
"point PHOTO_PIPELINE_DATA_DIR at the restored directory",
|
||||
"run `python -m photo_pipeline migrate` to reach the current revision",
|
||||
"run an inventory scan so paths are reconciled against the real library",
|
||||
"mount every archive location listed in the manifest before archiving again",
|
||||
],
|
||||
}
|
||||
@@ -1,502 +0,0 @@
|
||||
"""Load, soak, and resource-budget harness (US07-06, concept §17 and §18).
|
||||
|
||||
Performance here is not "it felt fast on my library". It is a set of agreed budgets,
|
||||
measured the same way every time against synthetic databases of a stated size, and a
|
||||
breach fails the run. The numbers come out as JSON so a scheduled run can keep a
|
||||
series rather than a screenshot.
|
||||
|
||||
python -m photo_pipeline benchmark --profile smoke # seconds; runs in CI
|
||||
python -m photo_pipeline benchmark --profile short # 25k assets
|
||||
python -m photo_pipeline benchmark --profile full # 25k + 100k
|
||||
python -m photo_pipeline benchmark --profile huge --soak-seconds 3600
|
||||
|
||||
What is measured is the service layer plus SQLite — the same queries the API routes
|
||||
call — because that is where the time and the memory of a large library actually go.
|
||||
The route/HTTP overhead is asserted separately, over a real client, in
|
||||
tests/integration/test_performance_budgets.py.
|
||||
|
||||
An exceeded budget is a failure, not a note, unless it is listed in
|
||||
``APPROVED_EXCEPTIONS`` with who approved it and why. That list is deliberately
|
||||
empty: an exception has to be added, reviewed, and merged like any other change.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import gc
|
||||
import json
|
||||
import os
|
||||
import resource
|
||||
import statistics
|
||||
import sqlite3
|
||||
import time
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import func, insert, select
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import (
|
||||
AnalysisResult,
|
||||
Asset,
|
||||
DuplicateCluster,
|
||||
DuplicateMember,
|
||||
Job,
|
||||
JobEvent,
|
||||
SafetyReview,
|
||||
)
|
||||
from photo_pipeline.services.duplicates import DuplicateService
|
||||
from photo_pipeline.services.inventory import InventoryService
|
||||
from photo_pipeline.services.jobs import ACTIVE_STATES, JobService
|
||||
from photo_pipeline.services.library import LibraryService
|
||||
from photo_pipeline.services.workflow import WorkflowService
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
# ── profiles ─────────────────────────────────────────────────────────────────
|
||||
|
||||
PROFILES: dict[str, dict] = {
|
||||
# Small enough to run on every change, large enough that an O(n) mistake in a
|
||||
# list query still shows up.
|
||||
"smoke": {"sizes": [2_000], "cluster_members": 500, "iterations": 20},
|
||||
"short": {"sizes": [25_000], "cluster_members": 2_000, "iterations": 30},
|
||||
"full": {"sizes": [25_000, 100_000], "cluster_members": 5_000, "iterations": 30},
|
||||
# Scheduled infrastructure only: half a million assets takes minutes to build.
|
||||
"huge": {"sizes": [500_000], "cluster_members": 5_000, "iterations": 20},
|
||||
}
|
||||
|
||||
# ── budgets ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Budget:
|
||||
metric: str
|
||||
limit: float
|
||||
unit: str
|
||||
why: str
|
||||
|
||||
|
||||
BUDGETS: tuple[Budget, ...] = (
|
||||
Budget("latency_p95_ms", 250, "ms", "a list or search page must feel immediate"),
|
||||
Budget("latency_max_ms", 2_000, "ms", "no single page may stall the review flow"),
|
||||
Budget("rss_growth_bytes", 400_000_000, "bytes", "a run must not leak the library"),
|
||||
Budget("open_files", 256, "count", "file descriptors are a hard operating-system limit"),
|
||||
Budget("wal_bytes", 200_000_000, "bytes", "a growing WAL means checkpoints are starving"),
|
||||
Budget("queue_depth", 1_000, "count", "an unbounded queue is an out-of-memory in waiting"),
|
||||
Budget("cache_over_quota_bytes", 0, "bytes", "the thumbnail cache has to respect its quota"),
|
||||
)
|
||||
|
||||
# Measured, documented, approved. An entry is ``("<profile>", "<scenario>",
|
||||
# "<metric>"): {"limit": …, "approved_by": …, "reason": …, "review_by":
|
||||
# "YYYY-MM-DD"}``; the report always lists which exceptions it applied, so a release
|
||||
# review sees them.
|
||||
#
|
||||
# The two below are the half-million-asset scale point. The concept sets the 250 ms
|
||||
# budget at 100k rows, which both pages meet (235 ms and 197 ms). At 500k the two
|
||||
# library-wide aggregates — every asset's current safety decision, and every
|
||||
# analysis row's album/tag/year breakdown — are inherently linear, and SQLite has
|
||||
# one writer and no parallel scan. Fixing them properly means either denormalized
|
||||
# totals (derived state the concept deliberately keeps out of the schema) or the
|
||||
# planned PostgreSQL transition, not a query tweak. Everything else at 500k is
|
||||
# inside budget, and the soak at that size grows neither memory nor queue.
|
||||
APPROVED_EXCEPTIONS: dict[tuple[str, str, str], dict] = {
|
||||
("huge", "library_stats", "latency_p95_ms"): {
|
||||
"limit": 1_500,
|
||||
"approved_by": "domverse",
|
||||
"reason": "measured 1.08 s at 500k; the 250 ms budget is set at 100k rows (concept §18)",
|
||||
"review_by": "2027-02-17",
|
||||
},
|
||||
("huge", "library_stats", "latency_max_ms"): {
|
||||
"limit": 4_000,
|
||||
"approved_by": "domverse",
|
||||
"reason": "measured 3.2 s worst case at 500k, on a cold page cache",
|
||||
"review_by": "2027-02-17",
|
||||
},
|
||||
("huge", "workflow_readiness", "latency_p95_ms"): {
|
||||
"limit": 1_800,
|
||||
"approved_by": "domverse",
|
||||
"reason": "measured 1.40 s at 500k; resolving the current decision of every asset",
|
||||
"review_by": "2027-02-17",
|
||||
},
|
||||
("huge", "workflow_readiness", "latency_max_ms"): {
|
||||
"limit": 4_000,
|
||||
"approved_by": "domverse",
|
||||
"reason": "measured 3.3 s worst case at 500k, on a cold page cache",
|
||||
"review_by": "2027-02-17",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
# ── resource sampling ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def rss_bytes() -> int:
|
||||
"""Resident set size of this process, without a psutil dependency."""
|
||||
usage = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
|
||||
# Linux reports kilobytes, BSD/macOS bytes.
|
||||
return usage if usage > 1 << 32 or os.uname().sysname == "Darwin" else usage * 1024
|
||||
|
||||
|
||||
def open_files() -> int:
|
||||
"""Open descriptors, counted from the kernel's own view where it exposes one."""
|
||||
for directory in ("/proc/self/fd", "/dev/fd"):
|
||||
try:
|
||||
return len(os.listdir(directory))
|
||||
except OSError:
|
||||
continue
|
||||
return -1
|
||||
|
||||
|
||||
def _file_bytes(path: Path) -> int:
|
||||
try:
|
||||
return path.stat().st_size
|
||||
except OSError:
|
||||
return 0
|
||||
|
||||
|
||||
def _tree_bytes(path: Path) -> int:
|
||||
if not path.is_dir():
|
||||
return 0
|
||||
return sum(p.stat().st_size for p in path.rglob("*") if p.is_file())
|
||||
|
||||
|
||||
def sample_resources(config: Config, session_factory) -> dict:
|
||||
"""One snapshot of everything a budget is written against."""
|
||||
database = config.database_path
|
||||
with session_factory() as session:
|
||||
queue_depth = int(
|
||||
session.scalar(select(func.count()).select_from(Job).where(Job.state.in_(ACTIVE_STATES)))
|
||||
or 0
|
||||
)
|
||||
events = int(session.scalar(select(func.count()).select_from(JobEvent)) or 0)
|
||||
cache_bytes = _tree_bytes(config.thumbnail_cache_dir)
|
||||
return {
|
||||
"at": _now().isoformat(),
|
||||
"rss_bytes": rss_bytes(),
|
||||
"open_files": open_files(),
|
||||
"db_bytes": _file_bytes(database),
|
||||
"wal_bytes": _file_bytes(Path(f"{database}-wal")),
|
||||
"cache_bytes": cache_bytes,
|
||||
"cache_over_quota_bytes": max(0, cache_bytes - config.thumbnail_cache_quota_bytes),
|
||||
"queue_depth": queue_depth,
|
||||
"event_rows": events,
|
||||
}
|
||||
|
||||
|
||||
# ── synthetic library ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def synthesize(config: Config, *, assets: int, cluster_members: int, batch: int = 5_000) -> dict:
|
||||
"""Build a database of ``assets`` rows and one cluster of ``cluster_members``.
|
||||
|
||||
Rows only — no image files. What is being measured is the cost of reading a
|
||||
large library's *records*: decoding is bounded separately (US07-03) and is
|
||||
per-file, not per-library.
|
||||
"""
|
||||
config.database_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
started = time.monotonic()
|
||||
root = config.library_roots[0] if config.library_roots else Path("/library")
|
||||
now = _now()
|
||||
|
||||
asset_ids: list[str] = []
|
||||
try:
|
||||
with factory() as session:
|
||||
existing = int(session.scalar(select(func.count()).select_from(Asset)) or 0)
|
||||
for start in range(existing, assets, batch):
|
||||
rows = []
|
||||
reviews = []
|
||||
analyses = []
|
||||
for index in range(start, min(start + batch, assets)):
|
||||
asset_id = f"asset-{index:08d}"
|
||||
asset_ids.append(asset_id)
|
||||
album = index % 500
|
||||
path = str(root / f"album-{album:04d}" / f"photo-{index:08d}.jpg")
|
||||
rows.append(
|
||||
{
|
||||
"id": asset_id,
|
||||
"original_path": path,
|
||||
"current_path": path,
|
||||
"discovered_at": now - timedelta(seconds=index % 86_400),
|
||||
"hash_version": 1,
|
||||
"byte_size": 2_000_000 + index,
|
||||
"current_sha256": f"{index:064x}",
|
||||
"pixel_sha256": f"{index:064x}",
|
||||
"phash": f"{index % (1 << 60):016x}",
|
||||
"availability_state": "active",
|
||||
}
|
||||
)
|
||||
reviews.append(
|
||||
{
|
||||
"id": str(uuid.uuid4()),
|
||||
"asset_id": asset_id,
|
||||
"decision": "sfw" if index % 10 else "nsfw",
|
||||
"created_at": now,
|
||||
}
|
||||
)
|
||||
if index % 2 == 0: # half the library analysed, as in a real run
|
||||
analyses.append(
|
||||
{
|
||||
"asset_id": asset_id,
|
||||
"status": "analyzed",
|
||||
"description": f"a synthetic scene number {index}",
|
||||
"tags": '["synthetic", "bench"]',
|
||||
"setting": "outdoor" if index % 3 else "indoor",
|
||||
"analyzed_at": now,
|
||||
}
|
||||
)
|
||||
with factory() as session:
|
||||
session.execute(insert(Asset), rows)
|
||||
session.execute(insert(SafetyReview), reviews)
|
||||
if analyses:
|
||||
session.execute(insert(AnalysisResult), analyses)
|
||||
session.commit()
|
||||
|
||||
if cluster_members:
|
||||
with factory() as session:
|
||||
cluster_id = str(uuid.uuid4())
|
||||
session.add(
|
||||
DuplicateCluster(
|
||||
id=cluster_id,
|
||||
method="perceptual",
|
||||
confidence="near",
|
||||
state="open",
|
||||
version=1,
|
||||
)
|
||||
)
|
||||
session.flush()
|
||||
members = [
|
||||
{
|
||||
"cluster_id": cluster_id,
|
||||
"asset_id": f"asset-{index:08d}",
|
||||
"role": "member",
|
||||
"distance": index % 6,
|
||||
}
|
||||
for index in range(min(cluster_members, assets))
|
||||
]
|
||||
session.execute(insert(DuplicateMember), members)
|
||||
session.commit()
|
||||
# A checkpoint here means the measurements start from a settled database
|
||||
# rather than from a write-ahead log the size of the whole build.
|
||||
with sqlite3.connect(config.database_path) as connection:
|
||||
connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
finally:
|
||||
engine.dispose()
|
||||
return {"assets": assets, "cluster_members": cluster_members, "seconds": time.monotonic() - started}
|
||||
|
||||
|
||||
# ── scenarios ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class Scenario:
|
||||
name: str
|
||||
call: object
|
||||
iterations: int
|
||||
samples: list[float] = field(default_factory=list)
|
||||
|
||||
def run(self) -> dict:
|
||||
for _ in range(self.iterations):
|
||||
started = time.perf_counter()
|
||||
self.call()
|
||||
self.samples.append((time.perf_counter() - started) * 1000)
|
||||
ordered = sorted(self.samples)
|
||||
index = max(0, int(round(0.95 * len(ordered))) - 1)
|
||||
return {
|
||||
"scenario": self.name,
|
||||
"iterations": self.iterations,
|
||||
"latency_p50_ms": round(statistics.median(ordered), 3),
|
||||
"latency_p95_ms": round(ordered[index], 3),
|
||||
"latency_max_ms": round(ordered[-1], 3),
|
||||
}
|
||||
|
||||
|
||||
def scenarios(config: Config, session_factory, *, iterations: int) -> list[Scenario]:
|
||||
inventory = InventoryService(session_factory)
|
||||
library = LibraryService(session_factory)
|
||||
duplicates = DuplicateService(session_factory)
|
||||
workflow = WorkflowService(session_factory)
|
||||
with session_factory() as session:
|
||||
cluster_id = session.scalar(select(DuplicateCluster.id))
|
||||
|
||||
built = [
|
||||
Scenario("inventory_page", lambda: inventory.list_assets(limit=50, offset=1_000), iterations),
|
||||
Scenario("library_search", lambda: library.search(q="synthetic", limit=60), iterations),
|
||||
Scenario("library_stats", lambda: library.stats(), iterations),
|
||||
Scenario("workflow_readiness", lambda: workflow.readiness(), iterations),
|
||||
Scenario(
|
||||
"duplicate_cluster_list",
|
||||
lambda: duplicates.list_clusters(limit=50, offset=0),
|
||||
iterations,
|
||||
),
|
||||
]
|
||||
if cluster_id:
|
||||
built.append(
|
||||
Scenario(
|
||||
"duplicate_cluster_page",
|
||||
lambda: duplicates.get_cluster(cluster_id, limit=100, offset=0),
|
||||
iterations,
|
||||
)
|
||||
)
|
||||
return built
|
||||
|
||||
|
||||
# ── budget evaluation ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def evaluate(profile: str, measurements: list[dict]) -> tuple[list[dict], list[dict]]:
|
||||
"""Compare measurements with the budgets. Returns ``(breaches, exceptions_used)``."""
|
||||
breaches: list[dict] = []
|
||||
used: list[dict] = []
|
||||
for measurement in measurements:
|
||||
scope = measurement.get("scenario", "resources")
|
||||
for budget in BUDGETS:
|
||||
if budget.metric not in measurement:
|
||||
continue
|
||||
value = measurement[budget.metric]
|
||||
if value is None or value < 0:
|
||||
continue
|
||||
limit = budget.limit
|
||||
exception = APPROVED_EXCEPTIONS.get((profile, scope, budget.metric))
|
||||
if exception:
|
||||
limit = exception["limit"]
|
||||
used.append({"scope": scope, "metric": budget.metric, **exception})
|
||||
if value > limit:
|
||||
breaches.append(
|
||||
{
|
||||
"scope": scope,
|
||||
"metric": budget.metric,
|
||||
"value": value,
|
||||
"limit": limit,
|
||||
"unit": budget.unit,
|
||||
"why": budget.why,
|
||||
}
|
||||
)
|
||||
return breaches, used
|
||||
|
||||
|
||||
# ── soak ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def soak(config: Config, session_factory, *, seconds: float, interval: float = 1.0) -> dict:
|
||||
"""Browse, queue, cancel, and retry for a while; watch what grows.
|
||||
|
||||
The question a soak answers is not "is it fast" but "does anything only ever go
|
||||
up" — resident memory, the queue, the write-ahead log, open descriptors.
|
||||
"""
|
||||
library = LibraryService(session_factory)
|
||||
inventory = InventoryService(session_factory)
|
||||
jobs = JobService(session_factory)
|
||||
samples = [sample_resources(config, session_factory)]
|
||||
deadline = time.monotonic() + seconds
|
||||
last_sample = time.monotonic()
|
||||
cycles = 0
|
||||
while time.monotonic() < deadline:
|
||||
offset = (cycles * 50) % 1_000
|
||||
library.search(q="synthetic", limit=60, offset=offset)
|
||||
inventory.list_assets(limit=50, offset=offset)
|
||||
job = jobs.enqueue("scan", items=[f"soak-{cycles}"])
|
||||
jobs.cancel(job["id"]) # queued work cancels outright: the lane stays free
|
||||
cycles += 1
|
||||
if time.monotonic() - last_sample >= interval:
|
||||
gc.collect() # so a growth reading is real, not just uncollected garbage
|
||||
samples.append(sample_resources(config, session_factory))
|
||||
last_sample = time.monotonic()
|
||||
samples.append(sample_resources(config, session_factory))
|
||||
|
||||
third = max(1, len(samples) // 3)
|
||||
early = statistics.mean(sample["rss_bytes"] for sample in samples[:third])
|
||||
late = statistics.mean(sample["rss_bytes"] for sample in samples[-third:])
|
||||
return {
|
||||
"scenario": "soak",
|
||||
"seconds": seconds,
|
||||
"cycles": cycles,
|
||||
"samples": samples,
|
||||
"rss_growth_bytes": max(0, int(late - early)),
|
||||
"queue_depth": max(sample["queue_depth"] for sample in samples),
|
||||
"wal_bytes": max(sample["wal_bytes"] for sample in samples),
|
||||
"open_files": max(sample["open_files"] for sample in samples),
|
||||
"cache_over_quota_bytes": max(sample["cache_over_quota_bytes"] for sample in samples),
|
||||
}
|
||||
|
||||
|
||||
# ── the run ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def run(
|
||||
config: Config,
|
||||
*,
|
||||
profile: str = "smoke",
|
||||
soak_seconds: float = 0.0,
|
||||
output: Path | str | None = None,
|
||||
) -> dict:
|
||||
"""Build, measure, evaluate. Returns the report; the caller decides the exit code."""
|
||||
if profile not in PROFILES:
|
||||
raise ValueError(f"unknown profile {profile!r}; try one of {sorted(PROFILES)}")
|
||||
settings = PROFILES[profile]
|
||||
report = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"profile": profile,
|
||||
"started_at": _now().isoformat(),
|
||||
"budgets": [
|
||||
{"metric": b.metric, "limit": b.limit, "unit": b.unit, "why": b.why} for b in BUDGETS
|
||||
],
|
||||
"runs": [],
|
||||
}
|
||||
measurements: list[dict] = []
|
||||
|
||||
for size in settings["sizes"]:
|
||||
sized = config.model_copy(update={"data_dir": Path(config.data_dir) / f"bench-{size}"})
|
||||
before = None
|
||||
build = synthesize(
|
||||
sized, assets=size, cluster_members=settings["cluster_members"]
|
||||
)
|
||||
engine = create_db_engine(sized.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
try:
|
||||
before = sample_resources(sized, factory)
|
||||
results = [
|
||||
scenario.run()
|
||||
for scenario in scenarios(sized, factory, iterations=settings["iterations"])
|
||||
]
|
||||
after = sample_resources(sized, factory)
|
||||
after["scenario"] = "resources"
|
||||
after["rss_growth_bytes"] = max(0, after["rss_bytes"] - before["rss_bytes"])
|
||||
soaked = (
|
||||
soak(sized, factory, seconds=soak_seconds) if soak_seconds > 0 else None
|
||||
)
|
||||
finally:
|
||||
engine.dispose()
|
||||
measurements.extend(results)
|
||||
measurements.append(after)
|
||||
if soaked:
|
||||
measurements.append(soaked)
|
||||
report["runs"].append(
|
||||
{
|
||||
"assets": size,
|
||||
"build": build,
|
||||
"before": before,
|
||||
"scenarios": results,
|
||||
"resources": after,
|
||||
"soak": soaked,
|
||||
}
|
||||
)
|
||||
|
||||
breaches, exceptions_used = evaluate(profile, measurements)
|
||||
report["breaches"] = breaches
|
||||
report["exceptions_applied"] = exceptions_used
|
||||
report["ok"] = not breaches
|
||||
report["finished_at"] = _now().isoformat()
|
||||
if output:
|
||||
path = Path(output)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(report, indent=2))
|
||||
return report
|
||||
@@ -1,157 +0,0 @@
|
||||
"""Operational diagnostics: what the application is using, and what is about to
|
||||
run out (US07-05, concept §17).
|
||||
|
||||
Every mutating stage in this application writes something before it is safe to
|
||||
continue — a journal, an EXIF rewrite, an archive copy, a backup. All of them fail
|
||||
badly on a full disk, so the sizes that grow (database, write-ahead log, thumbnail
|
||||
cache, uploader reports, backups, logs) are reported separately rather than as one
|
||||
opaque total, and each is compared against the free space actually left.
|
||||
|
||||
This is a read-only report. It never deletes, rotates, or prunes anything: what to
|
||||
do about a warning is an operator's decision, and the tools for it are the
|
||||
thumbnail cache quota, the backup retention helper, and log rotation outside the
|
||||
application.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.services import app_lock
|
||||
|
||||
# Below this much free space, mutating stages should stop rather than risk a
|
||||
# half-written journal, EXIF container, or archive copy.
|
||||
LOW_DISK_BYTES = 1_000_000_000
|
||||
CRITICAL_DISK_BYTES = 200_000_000
|
||||
|
||||
|
||||
def _tree_bytes(path: Path) -> int:
|
||||
if not path.exists():
|
||||
return 0
|
||||
if path.is_file():
|
||||
return path.stat().st_size
|
||||
total = 0
|
||||
for child in path.rglob("*"):
|
||||
try:
|
||||
if child.is_file() and not child.is_symlink():
|
||||
total += child.stat().st_size
|
||||
except OSError:
|
||||
continue # vanished mid-walk; it is not using space any more
|
||||
return total
|
||||
|
||||
|
||||
def _component(name: str, path: Path, *, quota: int | None = None) -> dict:
|
||||
used = _tree_bytes(path)
|
||||
entry = {"name": name, "path": str(path), "bytes": used, "exists": path.exists()}
|
||||
if quota is not None:
|
||||
entry["quota_bytes"] = quota
|
||||
entry["over_quota"] = used > quota
|
||||
return entry
|
||||
|
||||
|
||||
def disk(path: Path) -> dict:
|
||||
"""Free/total for the filesystem holding ``path`` — the nearest existing parent,
|
||||
so a data directory that does not exist yet still reports its future disk."""
|
||||
probe = path
|
||||
while not probe.exists() and probe != probe.parent:
|
||||
probe = probe.parent
|
||||
try:
|
||||
usage = shutil.disk_usage(probe)
|
||||
except OSError as error:
|
||||
return {"path": str(probe), "error": str(error)}
|
||||
return {
|
||||
"path": str(probe),
|
||||
"total_bytes": usage.total,
|
||||
"free_bytes": usage.free,
|
||||
"used_bytes": usage.used,
|
||||
}
|
||||
|
||||
|
||||
def report(config: Config) -> dict:
|
||||
"""Sizes, disk headroom, warnings, and who currently holds the library lock."""
|
||||
database = config.database_path
|
||||
components = [
|
||||
_component("database", database),
|
||||
_component("write_ahead_log", Path(f"{database}-wal")),
|
||||
_component("shared_memory", Path(f"{database}-shm")),
|
||||
_component(
|
||||
"thumbnail_cache",
|
||||
config.thumbnail_cache_dir,
|
||||
quota=config.thumbnail_cache_quota_bytes,
|
||||
),
|
||||
_component("upload_reports", config.data_dir / "uploads"),
|
||||
_component("backups", config.data_dir / "backups"),
|
||||
_component("logs", config.data_dir / "logs"),
|
||||
]
|
||||
space = disk(config.data_dir)
|
||||
free = space.get("free_bytes")
|
||||
|
||||
warnings: list[dict] = []
|
||||
if free is not None and free < CRITICAL_DISK_BYTES:
|
||||
warnings.append(
|
||||
{
|
||||
"code": "disk_critical",
|
||||
"message": (
|
||||
f"only {free} bytes free on {space['path']}; stop mutating stages "
|
||||
"and free space before renaming, writing EXIF, or archiving"
|
||||
),
|
||||
}
|
||||
)
|
||||
elif free is not None and free < LOW_DISK_BYTES:
|
||||
warnings.append(
|
||||
{
|
||||
"code": "disk_low",
|
||||
"message": f"{free} bytes free on {space['path']}; prune backups or the cache",
|
||||
}
|
||||
)
|
||||
for component in components:
|
||||
if component.get("over_quota"):
|
||||
warnings.append(
|
||||
{
|
||||
"code": "cache_over_quota",
|
||||
"message": (
|
||||
f"{component['name']} uses {component['bytes']} bytes, over its "
|
||||
f"{component['quota_bytes']} byte quota"
|
||||
),
|
||||
}
|
||||
)
|
||||
# A write-ahead log that outgrows its database means checkpoints are starving —
|
||||
# an operational warning, not something to ignore (concept §16).
|
||||
wal = next(c for c in components if c["name"] == "write_ahead_log")
|
||||
db = next(c for c in components if c["name"] == "database")
|
||||
if wal["bytes"] > max(db["bytes"], 1) :
|
||||
warnings.append(
|
||||
{
|
||||
"code": "wal_growth",
|
||||
"message": (
|
||||
f"the write-ahead log ({wal['bytes']} bytes) is larger than the database "
|
||||
f"({db['bytes']} bytes); a long-running read may be blocking checkpoints"
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
locks = {}
|
||||
for role in ("api", "worker"):
|
||||
holder = app_lock.LibraryLock(config, role).holder()
|
||||
locks[role] = holder.as_dict() if holder else None
|
||||
legacy = app_lock.legacy_activity(config)
|
||||
if legacy["active"]:
|
||||
warnings.append(
|
||||
{
|
||||
"code": "legacy_process_active",
|
||||
"message": (
|
||||
"a legacy CLI is writing this library; mutating stages are refused "
|
||||
"until it stops"
|
||||
),
|
||||
}
|
||||
)
|
||||
return {
|
||||
"components": components,
|
||||
"total_bytes": sum(component["bytes"] for component in components),
|
||||
"disk": space,
|
||||
"warnings": warnings,
|
||||
"locks": locks,
|
||||
"legacy_activity": legacy,
|
||||
}
|
||||
@@ -52,13 +52,6 @@ from photo_pipeline.services import availability, hashing
|
||||
NEAR_MAX = 5
|
||||
SIMILAR_MAX = 10
|
||||
|
||||
# Member paging (US07-06). A burst or a re-imported folder can put thousands of
|
||||
# assets in one cluster; review looks at a few at a time, so neither the list view
|
||||
# nor the detail view may load them all.
|
||||
MEMBER_PAGE = 100
|
||||
MAX_MEMBER_PAGE = 500
|
||||
SNAPSHOT_MEMBER_PREVIEW = 20
|
||||
|
||||
|
||||
class Method(str, Enum):
|
||||
EXACT = "exact"
|
||||
@@ -527,65 +520,39 @@ class DuplicateService:
|
||||
items = [self._snapshot(session, c.id) for c in rows]
|
||||
return {"items": items, "total": int(total or 0), "limit": limit, "offset": offset}
|
||||
|
||||
def get_cluster(
|
||||
self, cluster_id: str, *, limit: int = MEMBER_PAGE, offset: int = 0
|
||||
) -> dict | None:
|
||||
"""Cluster detail enriched with per-member asset evidence for comparison.
|
||||
|
||||
Members are paged and their evidence is loaded in batches (US07-06). A
|
||||
cluster of a few thousand near-identical frames is a real shape for a phone
|
||||
library, and the review screen only ever shows a handful at a time: loading
|
||||
every member — each with its own asset, thumbnail, and location query — made
|
||||
opening such a cluster cost thousands of round trips and megabytes of JSON.
|
||||
"""
|
||||
limit = max(1, min(limit, MAX_MEMBER_PAGE))
|
||||
offset = max(0, offset)
|
||||
def get_cluster(self, cluster_id: str) -> dict | None:
|
||||
"""Cluster detail enriched with per-member asset evidence for comparison."""
|
||||
with self._session_factory() as session:
|
||||
cluster = session.get(DuplicateCluster, cluster_id)
|
||||
if cluster is None:
|
||||
return None
|
||||
member_total = int(
|
||||
session.scalar(
|
||||
select(func.count())
|
||||
.select_from(DuplicateMember)
|
||||
.where(DuplicateMember.cluster_id == cluster_id)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
rows = list(
|
||||
session.execute(
|
||||
select(DuplicateMember)
|
||||
.where(DuplicateMember.cluster_id == cluster_id)
|
||||
.order_by(DuplicateMember.asset_id)
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
).scalars()
|
||||
)
|
||||
evidence = self._member_evidence(session, [row.asset_id for row in rows])
|
||||
members = []
|
||||
for member in rows:
|
||||
for member in session.execute(
|
||||
select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
|
||||
).scalars():
|
||||
asset = session.get(Asset, member.asset_id)
|
||||
try:
|
||||
member_evidence = json.loads(member.evidence) if member.evidence else {}
|
||||
evidence = json.loads(member.evidence) if member.evidence else {}
|
||||
except json.JSONDecodeError:
|
||||
member_evidence = {}
|
||||
asset, offline = evidence[member.asset_id]
|
||||
evidence = {}
|
||||
members.append(
|
||||
{
|
||||
"asset_id": member.asset_id,
|
||||
"role": member.role,
|
||||
"distance": member.distance,
|
||||
"evidence": member_evidence,
|
||||
"evidence": evidence,
|
||||
"current_path": asset.current_path if asset else None,
|
||||
"byte_size": asset.byte_size if asset else None,
|
||||
"phash": asset.phash if asset else None,
|
||||
**offline,
|
||||
**self._offline_evidence(session, asset),
|
||||
}
|
||||
)
|
||||
members.sort(key=lambda m: m["asset_id"])
|
||||
# A full-resolution comparison of an offline original is impossible; the
|
||||
# UI asks for that named medium instead of guessing (concept §9). The
|
||||
# answer covers the whole cluster, not just this page, so a mount is not
|
||||
# discovered halfway through a review.
|
||||
mount_required = self._mount_required(session, cluster_id)
|
||||
# UI asks for that named medium instead of guessing (concept §9).
|
||||
mount_required = sorted(
|
||||
{m["archive_location"] for m in members if m["requires_mount"]}
|
||||
)
|
||||
return {
|
||||
"id": cluster.id,
|
||||
"method": cluster.method,
|
||||
@@ -597,74 +564,10 @@ class DuplicateService:
|
||||
"requires_confirmation": cluster.method == Method.PERCEPTUAL.value,
|
||||
"mount_required": mount_required,
|
||||
"members": members,
|
||||
"member_total": member_total,
|
||||
"limit": limit,
|
||||
"offset": offset,
|
||||
}
|
||||
|
||||
def _member_evidence(self, session, asset_ids: list[str]) -> dict:
|
||||
"""``{asset_id: (asset, offline_evidence)}`` for one page, in three queries."""
|
||||
if not asset_ids:
|
||||
return {}
|
||||
assets = {
|
||||
asset.id: asset
|
||||
for asset in session.execute(
|
||||
select(Asset).where(Asset.id.in_(asset_ids))
|
||||
).scalars()
|
||||
}
|
||||
previews: dict[str, list] = {}
|
||||
for thumbnail in session.execute(
|
||||
select(Thumbnail).where(Thumbnail.asset_id.in_(asset_ids))
|
||||
).scalars():
|
||||
previews.setdefault(thumbnail.asset_id, []).append(thumbnail)
|
||||
location_ids = {
|
||||
asset.archive_location_id for asset in assets.values() if asset.archive_location_id
|
||||
}
|
||||
locations = (
|
||||
{
|
||||
location.id: location
|
||||
for location in session.execute(
|
||||
select(ArchiveLocation).where(ArchiveLocation.id.in_(location_ids))
|
||||
).scalars()
|
||||
}
|
||||
if location_ids
|
||||
else {}
|
||||
)
|
||||
return {
|
||||
asset_id: (
|
||||
assets.get(asset_id),
|
||||
self._offline_evidence(
|
||||
assets.get(asset_id),
|
||||
locations=locations,
|
||||
thumbnails=previews.get(asset_id, []),
|
||||
),
|
||||
)
|
||||
for asset_id in asset_ids
|
||||
}
|
||||
|
||||
def _mount_required(self, session, cluster_id: str) -> list[str]:
|
||||
"""Archive media whose originals this cluster needs, across every member."""
|
||||
rows = session.execute(
|
||||
select(ArchiveLocation.name)
|
||||
.select_from(DuplicateMember)
|
||||
.join(Asset, Asset.id == DuplicateMember.asset_id)
|
||||
.join(ArchiveLocation, ArchiveLocation.id == Asset.archive_location_id)
|
||||
.where(
|
||||
DuplicateMember.cluster_id == cluster_id,
|
||||
Asset.availability_state == availability.ARCHIVED_OFFLINE,
|
||||
)
|
||||
.distinct()
|
||||
).scalars()
|
||||
return sorted(rows)
|
||||
|
||||
def _offline_evidence(
|
||||
self, asset: Asset | None, *, locations: dict, thumbnails: list
|
||||
) -> dict:
|
||||
"""What review can still rely on when a member's original is not readable.
|
||||
|
||||
Takes the already-loaded locations and thumbnails for its page rather than
|
||||
querying per member (US07-06).
|
||||
"""
|
||||
def _offline_evidence(self, session, asset: Asset | None) -> dict:
|
||||
"""What review can still rely on when a member's original is not readable."""
|
||||
if asset is None:
|
||||
return {
|
||||
"availability_state": None,
|
||||
@@ -674,8 +577,12 @@ class DuplicateService:
|
||||
"preview": {"state": "missing", "protected": False},
|
||||
"requires_mount": False,
|
||||
}
|
||||
location = locations.get(asset.archive_location_id)
|
||||
preview = self._preview_evidence(thumbnails)
|
||||
location = (
|
||||
session.get(ArchiveLocation, asset.archive_location_id)
|
||||
if asset.archive_location_id
|
||||
else None
|
||||
)
|
||||
preview = self._preview_evidence(session, asset.id)
|
||||
archived = asset.availability_state in availability.ARCHIVED
|
||||
return {
|
||||
"availability_state": asset.availability_state,
|
||||
@@ -691,7 +598,10 @@ class DuplicateService:
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _preview_evidence(rows: list) -> dict:
|
||||
def _preview_evidence(session, asset_id: str) -> dict:
|
||||
rows = list(
|
||||
session.execute(select(Thumbnail).where(Thumbnail.asset_id == asset_id)).scalars()
|
||||
)
|
||||
ready = [r for r in rows if r.state == "ready" and r.path]
|
||||
if ready:
|
||||
best = max(ready, key=lambda r: (bool(r.protected), r.size or 0))
|
||||
@@ -783,29 +693,11 @@ class DuplicateService:
|
||||
session.delete(link)
|
||||
|
||||
def _snapshot(self, session, cluster_id) -> dict:
|
||||
"""A cluster and a *bounded* preview of its members.
|
||||
|
||||
The list view shows a count and a few ids; a snapshot that loaded every
|
||||
member turned one page of 200 clusters into hundreds of thousands of rows
|
||||
(US07-06). ``member_total`` is the honest count either way.
|
||||
"""
|
||||
cluster = session.get(DuplicateCluster, cluster_id)
|
||||
member_total = int(
|
||||
session.scalar(
|
||||
select(func.count())
|
||||
.select_from(DuplicateMember)
|
||||
.where(DuplicateMember.cluster_id == cluster_id)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
members = session.execute(
|
||||
select(DuplicateMember)
|
||||
.where(DuplicateMember.cluster_id == cluster_id)
|
||||
.order_by(DuplicateMember.asset_id)
|
||||
.limit(SNAPSHOT_MEMBER_PREVIEW)
|
||||
select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
|
||||
).scalars()
|
||||
return {
|
||||
"member_total": member_total,
|
||||
"id": cluster.id,
|
||||
"method": cluster.method,
|
||||
"confidence": cluster.confidence,
|
||||
|
||||
@@ -21,7 +21,6 @@ unreadable file, or a write that did not take is not evidence that metadata is f
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
@@ -59,11 +58,6 @@ class CheckpointResult:
|
||||
state: str # verified | divergent | failed
|
||||
changed_fields: tuple[str, ...] = ()
|
||||
sha256: str | None = None
|
||||
# exiftool rewrites the container, so the file's size moves with its hash. Both
|
||||
# are inventory facts about the current bytes and both have to be refreshed
|
||||
# together, or the next stage compares against a size that no longer exists
|
||||
# (US07-07: a rename plan blocked itself forever after any EXIF write).
|
||||
byte_size: int | None = None
|
||||
verified_at: datetime | None = None
|
||||
reason: str | None = None
|
||||
|
||||
@@ -143,12 +137,9 @@ def run(
|
||||
|
||||
changed = compare(before, after)
|
||||
sha256 = hashing.sha256_file(path)
|
||||
byte_size = os.path.getsize(path)
|
||||
if changed:
|
||||
return CheckpointResult(
|
||||
DIVERGENT, changed_fields=changed, sha256=sha256, byte_size=byte_size
|
||||
)
|
||||
return CheckpointResult(VERIFIED, sha256=sha256, byte_size=byte_size, verified_at=_now())
|
||||
return CheckpointResult(DIVERGENT, changed_fields=changed, sha256=sha256)
|
||||
return CheckpointResult(VERIFIED, sha256=sha256, verified_at=_now())
|
||||
|
||||
|
||||
def record(
|
||||
|
||||
@@ -15,9 +15,10 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import String, and_, case, cast, func, or_, select, text
|
||||
from sqlalchemy import and_, func, or_, select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline.models import AnalysisResult, Asset
|
||||
@@ -70,75 +71,38 @@ class LibraryService:
|
||||
return {"rows": rows, "total": total, "offset": offset, "limit": limit}
|
||||
|
||||
def stats(self) -> dict:
|
||||
"""Library-wide totals, aggregated in SQL (US07-06).
|
||||
|
||||
This page used to load every analysis row — object, tags, and all — to count
|
||||
them in Python, which cost half a second at 100k assets and grew from there.
|
||||
Only the album breakdown still walks rows, and only their path and status:
|
||||
SQLite has no ``dirname``, and two short strings per asset is cheap.
|
||||
"""
|
||||
with self._session_factory() as session:
|
||||
status = dict(
|
||||
session.execute(
|
||||
select(AnalysisResult.status, func.count()).group_by(AnalysisResult.status)
|
||||
).all()
|
||||
)
|
||||
albums: dict[str, dict] = {}
|
||||
for path, row_status in session.execute(
|
||||
select(Asset.current_path, AnalysisResult.status).join(
|
||||
Asset, Asset.id == AnalysisResult.asset_id
|
||||
rows = list(
|
||||
session.execute(
|
||||
select(AnalysisResult, Asset.current_path).join(
|
||||
Asset, Asset.id == AnalysisResult.asset_id
|
||||
)
|
||||
)
|
||||
):
|
||||
)
|
||||
albums: dict[str, dict] = {}
|
||||
tag_counts: Counter = Counter()
|
||||
year_counts: Counter = Counter()
|
||||
people: Counter = Counter()
|
||||
errors = []
|
||||
for result, path in rows:
|
||||
album = _album_of(path)
|
||||
bucket = albums.setdefault(album, {"album": album, "done": 0, "total": 0})
|
||||
bucket["total"] += 1
|
||||
if row_status in DONE:
|
||||
if result.status in DONE:
|
||||
bucket["done"] += 1
|
||||
year_counts = dict(
|
||||
session.execute(
|
||||
select(AnalysisResult.approx_year, func.count())
|
||||
.where(AnalysisResult.approx_year.is_not(None))
|
||||
.group_by(AnalysisResult.approx_year)
|
||||
).all()
|
||||
)
|
||||
people = dict(
|
||||
session.execute(
|
||||
select(
|
||||
case(
|
||||
(AnalysisResult.people_count >= 3, "3+"),
|
||||
else_=cast(AnalysisResult.people_count, String),
|
||||
),
|
||||
func.count(),
|
||||
)
|
||||
.where(AnalysisResult.people_count.is_not(None))
|
||||
.group_by(
|
||||
case(
|
||||
(AnalysisResult.people_count >= 3, "3+"),
|
||||
else_=cast(AnalysisResult.people_count, String),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
)
|
||||
# SQLite's JSON1 counts the tag arrays where they are: parsing 50k JSON
|
||||
# strings in Python to keep the top 40 is the definition of doing work
|
||||
# the database already does. Malformed tags are skipped, not fatal.
|
||||
tag_counts = session.execute(
|
||||
text(
|
||||
"SELECT tag.value AS value, count(*) AS total "
|
||||
"FROM analysis_results, json_each(analysis_results.tags) AS tag "
|
||||
"WHERE analysis_results.tags IS NOT NULL "
|
||||
"AND json_valid(analysis_results.tags) "
|
||||
"GROUP BY tag.value ORDER BY total DESC, value LIMIT 40"
|
||||
)
|
||||
).all()
|
||||
errors = [
|
||||
{"path": path, "error": message}
|
||||
for path, message in session.execute(
|
||||
select(Asset.current_path, AnalysisResult.error_message)
|
||||
.join(Asset, Asset.id == AnalysisResult.asset_id)
|
||||
.where(AnalysisResult.status == "error")
|
||||
)
|
||||
]
|
||||
for tag in _tags(result.tags):
|
||||
tag_counts[tag] += 1
|
||||
if result.approx_year is not None:
|
||||
year_counts[result.approx_year] += 1
|
||||
if result.people_count is not None:
|
||||
people["3+" if result.people_count >= 3 else str(result.people_count)] += 1
|
||||
if result.status == "error":
|
||||
errors.append({"path": path, "error": result.error_message})
|
||||
return {
|
||||
"total": sum(status.values()),
|
||||
"status": status,
|
||||
@@ -147,7 +111,7 @@ class LibraryService:
|
||||
"season": self._facet("season"),
|
||||
"people": [{"value": v, "count": n} for v, n in sorted(people.items())],
|
||||
"years": [{"value": y, "count": year_counts[y]} for y in sorted(year_counts)],
|
||||
"top_tags": [{"value": t, "count": n} for t, n in tag_counts],
|
||||
"top_tags": [{"value": t, "count": n} for t, n in tag_counts.most_common(40)],
|
||||
"albums": sorted(albums.values(), key=lambda d: d["album"]),
|
||||
"errors": sorted(errors, key=lambda e: e["path"] or ""),
|
||||
}
|
||||
|
||||
@@ -1,391 +0,0 @@
|
||||
"""The release gate, the real-library dry run, and the approval that unlocks
|
||||
mutation (US07-07, concept §18 release gates).
|
||||
|
||||
Three things live here because they are one decision:
|
||||
|
||||
1. **The gate** — one command that provisions an isolated stack, runs every suite in
|
||||
a fixed order, and retains versioned evidence with checksums. A release is not
|
||||
"the tests passed on my machine last Tuesday"; it is a report that says which
|
||||
revision, which suites, how long, and what the artefacts hash to.
|
||||
2. **The dry run** — a strictly read-only pass over the real photo library that
|
||||
answers "what would this application do to it?" before it is allowed to do
|
||||
anything. It opens no file for writing, creates no database rows, and touches no
|
||||
metadata; it counts, classifies, and reconciles against whatever the database
|
||||
already knows.
|
||||
3. **The approval** — a person reads that report and signs it off for exactly the
|
||||
library roots it describes. Until then, with
|
||||
``PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL`` set, every mutating request is
|
||||
refused. Change the roots, or produce a newer report, and the approval no longer
|
||||
matches: it approves *that* reconciliation, not the idea of mutating.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from collections import Counter
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from photo_pipeline import path_policy
|
||||
from photo_pipeline.config import Config
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
APPROVAL_NAME = "dry-run-approval.json"
|
||||
CHECKSUMS_NAME = "CHECKSUMS.sha256"
|
||||
REPORT_NAME = "release-report.json"
|
||||
|
||||
# The suites, in the order a failure is cheapest to read: units before the stacks
|
||||
# they compose. ``label`` is what the report and the operator see.
|
||||
STAGES: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||||
("unit", ("tests/unit",)),
|
||||
("characterization", ("tests/characterization",)),
|
||||
("integration", ("tests/integration",)),
|
||||
("browser", ("tests/e2e",)),
|
||||
)
|
||||
|
||||
# Skips the gate accepts, because they describe the machine rather than the code.
|
||||
ALLOWED_SKIP_REASONS = ("exiftool not installed", "root ignores directory permissions")
|
||||
|
||||
|
||||
class ReleaseError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def sha256_bytes(payload: bytes) -> str:
|
||||
return hashlib.sha256(payload).hexdigest()
|
||||
|
||||
|
||||
def revision() -> str | None:
|
||||
"""The commit this gate ran against, when the tree is a git checkout."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
cwd=str(Path(__file__).resolve().parents[2]),
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return None
|
||||
return result.stdout.strip() or None
|
||||
|
||||
|
||||
# ── the story matrix ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def story_matrix(repo: Path | None = None) -> dict:
|
||||
"""Every backlog story, and how it is covered.
|
||||
|
||||
A story is ``delivered`` (mapped to test files that exist) or ``planned`` (an
|
||||
accepted, not-yet-implemented story). Anything else — a story file nobody
|
||||
mapped, or a mapping to a file that is gone — is a hole in the matrix, and the
|
||||
gate fails on it rather than reporting a green run over missing coverage.
|
||||
"""
|
||||
repo = repo or Path(__file__).resolve().parents[2]
|
||||
traceability = json.loads((repo / "tests" / "story_traceability.json").read_text())
|
||||
mapped: dict[str, list[str]] = traceability["stories"]
|
||||
planned: list[str] = traceability.get("planned", [])
|
||||
stories = sorted(
|
||||
"-".join(path.stem.split("-")[:2])
|
||||
for path in (repo / "delivery_backlog" / "stories").glob("US*.md")
|
||||
)
|
||||
|
||||
missing_tests = [
|
||||
f"{story}: {rel}"
|
||||
for story, files in mapped.items()
|
||||
for rel in files
|
||||
if not (repo / rel).is_file()
|
||||
]
|
||||
unmapped = [s for s in stories if s not in mapped and s not in planned]
|
||||
unknown = [s for s in list(mapped) + planned if s not in stories]
|
||||
overlap = sorted(set(mapped) & set(planned))
|
||||
return {
|
||||
"stories": len(stories),
|
||||
"delivered": sorted(mapped),
|
||||
"planned": sorted(planned),
|
||||
"problems": [
|
||||
*(f"story with no tests and not planned: {s}" for s in unmapped),
|
||||
*(f"mapped test file is missing — {entry}" for entry in missing_tests),
|
||||
*(f"mapped story is not in the backlog: {s}" for s in unknown),
|
||||
*(f"story is both delivered and planned: {s}" for s in overlap),
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
# ── the gate ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class StageResult:
|
||||
label: str
|
||||
command: list[str]
|
||||
returncode: int
|
||||
seconds: float
|
||||
summary: str
|
||||
skipped: list[str]
|
||||
|
||||
def as_dict(self) -> dict:
|
||||
return {
|
||||
"stage": self.label,
|
||||
"command": self.command,
|
||||
"returncode": self.returncode,
|
||||
"seconds": round(self.seconds, 2),
|
||||
"summary": self.summary,
|
||||
"skipped": self.skipped,
|
||||
"ok": self.returncode == 0,
|
||||
}
|
||||
|
||||
|
||||
def _run_stage(label: str, paths: tuple[str, ...], *, repo: Path, log_dir: Path) -> StageResult:
|
||||
command = [sys.executable, "-m", "pytest", *paths, "-q", "-rs"]
|
||||
started = time.monotonic()
|
||||
result = subprocess.run(command, cwd=str(repo), capture_output=True, text=True)
|
||||
elapsed = time.monotonic() - started
|
||||
output = result.stdout + result.stderr
|
||||
(log_dir / f"{label}.log").write_text(output)
|
||||
lines = [line for line in output.splitlines() if line.strip()]
|
||||
summary = lines[-1] if lines else ""
|
||||
skipped = [line for line in lines if line.startswith("SKIPPED")]
|
||||
return StageResult(label, command, result.returncode, elapsed, summary, skipped)
|
||||
|
||||
|
||||
def unexpected_skips(results: list[StageResult]) -> list[str]:
|
||||
"""Skips the gate will not accept: everything but the documented environment ones."""
|
||||
return [
|
||||
line
|
||||
for result in results
|
||||
for line in result.skipped
|
||||
if not any(reason in line for reason in ALLOWED_SKIP_REASONS)
|
||||
]
|
||||
|
||||
|
||||
def run_gate(
|
||||
config: Config,
|
||||
*,
|
||||
output: Path | str | None = None,
|
||||
stages: tuple[tuple[str, tuple[str, ...]], ...] = STAGES,
|
||||
repo: Path | None = None,
|
||||
) -> dict:
|
||||
"""Run every suite in an isolated stack and retain checksummed evidence.
|
||||
|
||||
The stack is isolated by construction: each pytest run builds its own temporary
|
||||
data directories and libraries, so the gate never reads or writes the operator's
|
||||
photos. What it keeps afterwards is the report, the per-stage logs, and a
|
||||
checksum file over both.
|
||||
"""
|
||||
repo = repo or Path(__file__).resolve().parents[2]
|
||||
directory = Path(output) if output else Path(config.data_dir) / "release" / _now().strftime(
|
||||
"%Y%m%dT%H%M%SZ"
|
||||
)
|
||||
logs = directory / "logs"
|
||||
logs.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
matrix = story_matrix(repo)
|
||||
results = [_run_stage(label, paths, repo=repo, log_dir=logs) for label, paths in stages]
|
||||
skips = unexpected_skips(results)
|
||||
|
||||
report = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"started_at": _now().isoformat(),
|
||||
"revision": revision(),
|
||||
"python": sys.version.split()[0],
|
||||
"platform": os.uname().sysname,
|
||||
"matrix": matrix,
|
||||
"stages": [result.as_dict() for result in results],
|
||||
"unexpected_skips": skips,
|
||||
"failures": [result.label for result in results if result.returncode != 0],
|
||||
}
|
||||
report["ok"] = not report["failures"] and not matrix["problems"] and not skips
|
||||
report["finished_at"] = _now().isoformat()
|
||||
|
||||
(directory / REPORT_NAME).write_text(json.dumps(report, indent=2))
|
||||
# The evidence is only evidence if it can be shown to be the evidence that was
|
||||
# produced. Checksums are the honest version of "signed" without a key: a real
|
||||
# signature belongs to whatever key management the release actually has.
|
||||
checksums = "\n".join(
|
||||
f"{sha256_file(path)} {path.relative_to(directory)}"
|
||||
for path in sorted(directory.rglob("*"))
|
||||
if path.is_file() and path.name != CHECKSUMS_NAME
|
||||
)
|
||||
(directory / CHECKSUMS_NAME).write_text(checksums + "\n")
|
||||
report["evidence"] = str(directory)
|
||||
return report
|
||||
|
||||
|
||||
# ── the real-library dry run ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def dry_run(config: Config, *, roots: tuple[Path, ...] | None = None) -> dict:
|
||||
"""Read-only reconciliation of the configured library. Changes nothing.
|
||||
|
||||
Opens no file for writing, writes no database row, and reads only what
|
||||
``os.stat`` and the existing database already say. The point is to be able to
|
||||
look at a real library — the one with the irreplaceable photos in it — and see
|
||||
what the application believes about it before it is allowed to act.
|
||||
"""
|
||||
roots = roots or tuple(Path(root) for root in config.library_roots)
|
||||
if not roots:
|
||||
raise ReleaseError("no library roots are configured")
|
||||
|
||||
by_extension: Counter = Counter()
|
||||
folders: set[str] = set()
|
||||
files: list[str] = []
|
||||
unreadable: list[str] = []
|
||||
excluded = 0
|
||||
total_bytes = 0
|
||||
for root in roots:
|
||||
if not Path(root).is_dir():
|
||||
raise ReleaseError(f"library root {root} is not a directory")
|
||||
for path in sorted(Path(root).rglob("*")):
|
||||
if path.is_dir():
|
||||
# Never traverse into an excluded directory, and never report its
|
||||
# contents: proving exclusion must not require opening it.
|
||||
if path_policy.is_excluded(path):
|
||||
excluded += 1
|
||||
continue
|
||||
if path_policy.is_excluded(path):
|
||||
continue
|
||||
try:
|
||||
stat = path.stat()
|
||||
except OSError:
|
||||
unreadable.append(str(path))
|
||||
continue
|
||||
files.append(str(path))
|
||||
folders.add(str(path.parent))
|
||||
by_extension[path.suffix.lower() or "(none)"] += 1
|
||||
total_bytes += stat.st_size
|
||||
|
||||
known = _known_paths(config)
|
||||
on_disk = set(files)
|
||||
report = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"generated_at": _now().isoformat(),
|
||||
"revision": revision(),
|
||||
"library_roots": [str(root) for root in roots],
|
||||
"files": len(files),
|
||||
"folders": len(folders),
|
||||
"bytes": total_bytes,
|
||||
"excluded_directories": excluded,
|
||||
"unreadable": unreadable,
|
||||
"by_extension": dict(sorted(by_extension.items())),
|
||||
"reconciliation": {
|
||||
"known_to_database": len(known),
|
||||
"already_registered": len(on_disk & known),
|
||||
"new_to_the_application": len(on_disk - known),
|
||||
"recorded_but_absent": sorted(known - on_disk)[:100],
|
||||
"recorded_but_absent_total": len(known - on_disk),
|
||||
},
|
||||
"mutation": "none — this pass is read-only",
|
||||
}
|
||||
report["checksum"] = sha256_bytes(
|
||||
json.dumps(report, sort_keys=True).encode("utf-8")
|
||||
)
|
||||
return report
|
||||
|
||||
|
||||
def _known_paths(config: Config) -> set[str]:
|
||||
"""Current asset paths the database holds, or an empty set if there is none."""
|
||||
if not config.database_path.exists():
|
||||
return set()
|
||||
from sqlalchemy import select
|
||||
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory
|
||||
from photo_pipeline.models import Asset
|
||||
|
||||
engine = create_db_engine(config.database_url)
|
||||
try:
|
||||
with create_session_factory(engine)() as session:
|
||||
return {
|
||||
path
|
||||
for path in session.scalars(select(Asset.current_path))
|
||||
if path is not None
|
||||
}
|
||||
except Exception:
|
||||
return set()
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
# ── the approval ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def approval_path(config: Config) -> Path:
|
||||
return Path(config.data_dir) / APPROVAL_NAME
|
||||
|
||||
|
||||
def approve(config: Config, report: dict | Path | str, *, approver: str) -> dict:
|
||||
"""Record that a person read this reconciliation and accepts mutation for it."""
|
||||
if isinstance(report, (str, Path)):
|
||||
report = json.loads(Path(report).read_text())
|
||||
if "checksum" not in report:
|
||||
raise ReleaseError("this is not a dry-run report: it has no checksum")
|
||||
record = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"approved_at": _now().isoformat(),
|
||||
"approved_by": approver,
|
||||
"report_checksum": report["checksum"],
|
||||
"library_roots": report["library_roots"],
|
||||
"files": report["files"],
|
||||
"revision": report.get("revision"),
|
||||
}
|
||||
path = approval_path(config)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(record, indent=2))
|
||||
return record
|
||||
|
||||
|
||||
def mutation_blockers(config: Config) -> list[dict]:
|
||||
"""Why mutation must stay refused, or an empty list.
|
||||
|
||||
Only enforced when ``require_dry_run_approval`` is configured — the loopback
|
||||
developer setup keeps working unchanged, and an operator turns this on before
|
||||
pointing the application at the library they cannot replace.
|
||||
"""
|
||||
if not config.require_dry_run_approval:
|
||||
return []
|
||||
path = approval_path(config)
|
||||
if not path.exists():
|
||||
return [
|
||||
{
|
||||
"code": "dry_run_not_approved",
|
||||
"message": (
|
||||
"run `python -m photo_pipeline dry-run` and approve its report "
|
||||
"before mutation is enabled"
|
||||
),
|
||||
}
|
||||
]
|
||||
try:
|
||||
record = json.loads(path.read_text())
|
||||
except ValueError:
|
||||
return [{"code": "approval_unreadable", "message": f"{path} is not readable JSON"}]
|
||||
approved_roots = [str(root) for root in record.get("library_roots", [])]
|
||||
configured = [str(root) for root in config.library_roots]
|
||||
if sorted(approved_roots) != sorted(configured):
|
||||
return [
|
||||
{
|
||||
"code": "approval_scope_mismatch",
|
||||
"message": (
|
||||
f"the approval covers {approved_roots}, but the configured library "
|
||||
f"is {configured}; run a new dry run"
|
||||
),
|
||||
}
|
||||
]
|
||||
return []
|
||||
@@ -105,7 +105,7 @@ def exif_projection(decision: str) -> dict[str, list[str]]:
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import column, func, select
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline.models import Asset, ExifProjection, SafetyReview
|
||||
@@ -139,8 +139,8 @@ class SafetyService:
|
||||
|
||||
# -- reads ----------------------------------------------------------------
|
||||
def _latest_by_asset(self, session) -> dict[str, SafetyReview]:
|
||||
"""The current review per asset, as ORM rows. Only for small, known sets —
|
||||
every library-wide caller uses ``latest_reviews()`` in SQL instead."""
|
||||
# Latest row per asset. Small local scale: order ascending, let later rows
|
||||
# overwrite. ponytail: a windowed query if safety_reviews ever grows huge.
|
||||
latest: dict[str, SafetyReview] = {}
|
||||
for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
|
||||
latest[review.asset_id] = review
|
||||
@@ -148,99 +148,58 @@ class SafetyService:
|
||||
|
||||
def current_decision(self, asset_id: str) -> str | None:
|
||||
with self._session_factory() as session:
|
||||
latest = latest_reviews().subquery()
|
||||
return session.scalar(
|
||||
select(latest.c.decision).where(latest.c.asset_id == asset_id)
|
||||
)
|
||||
review = self._latest_by_asset(session).get(asset_id)
|
||||
return review.decision if review else None
|
||||
|
||||
def counts(self) -> dict[str, int]:
|
||||
"""Decision breakdown over canonical, active assets — the workflow totals.
|
||||
|
||||
Aggregated in SQL: the workflow home asks for this on every load, and
|
||||
materialising every asset and every review to count them cost hundreds of
|
||||
milliseconds at 25k assets and would scale linearly from there (US07-06).
|
||||
"""
|
||||
latest = latest_reviews().subquery()
|
||||
"""Decision breakdown over canonical, active assets — the workflow totals."""
|
||||
with self._session_factory() as session:
|
||||
rows = session.execute(
|
||||
select(
|
||||
func.coalesce(latest.c.decision, "undecided"),
|
||||
func.count(),
|
||||
func.count(latest.c.score),
|
||||
)
|
||||
.select_from(Asset)
|
||||
.join(latest, latest.c.asset_id == Asset.id, isouter=True)
|
||||
.where(
|
||||
Asset.canonical_asset_id.is_(None),
|
||||
Asset.availability_state == "active",
|
||||
)
|
||||
.group_by(func.coalesce(latest.c.decision, "undecided"))
|
||||
).all()
|
||||
out = {SFW: 0, NSFW: 0, "deferred": 0, "undecided": 0, "scored": 0}
|
||||
for decision, total, scored in rows:
|
||||
if decision in (SFW, NSFW, "deferred"):
|
||||
out[decision] += int(total)
|
||||
else:
|
||||
# Anything that is not one of the three decisions is undecided —
|
||||
# including a score-only review, which is what "scored" counts.
|
||||
out["undecided"] += int(total)
|
||||
out["scored"] += int(scored)
|
||||
return out
|
||||
assets = list(session.scalars(_eligible_assets_query()))
|
||||
latest = self._latest_by_asset(session)
|
||||
out = {SFW: 0, NSFW: 0, "deferred": 0, "undecided": 0, "scored": 0}
|
||||
for asset in assets:
|
||||
review = latest.get(asset.id)
|
||||
decision = review.decision if review else None
|
||||
if decision in (SFW, NSFW, "deferred"):
|
||||
out[decision] += 1
|
||||
else:
|
||||
out["undecided"] += 1
|
||||
if review and review.score is not None:
|
||||
out["scored"] += 1
|
||||
return out
|
||||
|
||||
def review_queue(self, state: str = "", limit: int = 100, offset: int = 0) -> dict:
|
||||
"""Assets for the review UI, filtered by ``state`` (undecided/sfw/nsfw/deferred).
|
||||
|
||||
Filtered, counted, and paged in SQL (US07-06): the queue for a large library
|
||||
is thousands of rows and the reviewer sees one page of it.
|
||||
"""
|
||||
latest = latest_reviews().subquery()
|
||||
projections = (
|
||||
select(ExifProjection.asset_id, ExifProjection.state.label("exif_state"))
|
||||
.where(ExifProjection.stage == "safety")
|
||||
.subquery()
|
||||
)
|
||||
effective = func.coalesce(latest.c.decision, "undecided")
|
||||
query = (
|
||||
select(
|
||||
Asset.id,
|
||||
Asset.current_path,
|
||||
latest.c.score,
|
||||
latest.c.decision,
|
||||
latest.c.exif_verified_at,
|
||||
projections.c.exif_state,
|
||||
)
|
||||
.select_from(Asset)
|
||||
.join(latest, latest.c.asset_id == Asset.id, isouter=True)
|
||||
.join(projections, projections.c.asset_id == Asset.id, isouter=True)
|
||||
.where(
|
||||
Asset.canonical_asset_id.is_(None),
|
||||
Asset.availability_state == "active",
|
||||
)
|
||||
)
|
||||
if state:
|
||||
query = query.where(effective == state)
|
||||
"""Assets for the review UI, filtered by ``state`` (undecided/sfw/nsfw/deferred)."""
|
||||
with self._session_factory() as session:
|
||||
total = int(
|
||||
session.scalar(select(func.count()).select_from(query.subquery())) or 0
|
||||
)
|
||||
rows = session.execute(
|
||||
query.order_by(Asset.current_path).limit(limit).offset(offset)
|
||||
).all()
|
||||
return {
|
||||
"total": total,
|
||||
"items": [
|
||||
{
|
||||
"asset_id": asset_id,
|
||||
"current_path": current_path,
|
||||
"score": score,
|
||||
"decision": decision,
|
||||
"suggested": classify(score) if score is not None else None,
|
||||
"exif_verified": bool(exif_verified_at),
|
||||
"exif_state": exif_state,
|
||||
}
|
||||
for asset_id, current_path, score, decision, exif_verified_at, exif_state in rows
|
||||
],
|
||||
}
|
||||
assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path)))
|
||||
latest = self._latest_by_asset(session)
|
||||
# One query, not one per asset: the reviewer needs to see a divergent
|
||||
# checkpoint, which is neither "verified" nor a plain failure (US07-03).
|
||||
projections = {
|
||||
row.asset_id: row.state
|
||||
for row in session.scalars(
|
||||
select(ExifProjection).where(ExifProjection.stage == "safety")
|
||||
)
|
||||
}
|
||||
rows = []
|
||||
for asset in assets:
|
||||
review = latest.get(asset.id)
|
||||
decision = review.decision if review else None
|
||||
effective = decision or "undecided"
|
||||
if state and state != effective:
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"asset_id": asset.id,
|
||||
"current_path": asset.current_path,
|
||||
"score": review.score if review else None,
|
||||
"decision": decision,
|
||||
"suggested": classify(review.score) if review and review.score is not None else None,
|
||||
"exif_verified": bool(review and review.exif_verified_at),
|
||||
"exif_state": projections.get(asset.id),
|
||||
}
|
||||
)
|
||||
return {"total": len(rows), "items": rows[offset : offset + limit]}
|
||||
|
||||
def scorable_asset_ids(self) -> list[str]:
|
||||
"""Canonical active assets with a path — the items a scoring job enqueues."""
|
||||
@@ -294,7 +253,6 @@ class SafetyService:
|
||||
|
||||
exif_verified_at = None
|
||||
result_sha256 = None
|
||||
result_byte_size = None
|
||||
if write_exif and decision in (SFW, NSFW) and path:
|
||||
ops = exif_projection(decision)
|
||||
# The full checkpoint: write the owned keyword, read the whole file back,
|
||||
@@ -315,7 +273,6 @@ class SafetyService:
|
||||
if result.verified:
|
||||
exif_verified_at = result.verified_at
|
||||
result_sha256 = result.sha256
|
||||
result_byte_size = result.byte_size
|
||||
|
||||
now = _now()
|
||||
with self._session_factory() as session:
|
||||
@@ -334,8 +291,6 @@ class SafetyService:
|
||||
if result_sha256:
|
||||
asset = session.get(Asset, asset_id)
|
||||
asset.current_sha256 = result_sha256
|
||||
if result_byte_size is not None:
|
||||
asset.byte_size = result_byte_size
|
||||
session.commit()
|
||||
return {
|
||||
"asset_id": asset_id,
|
||||
@@ -345,34 +300,6 @@ class SafetyService:
|
||||
}
|
||||
|
||||
|
||||
def latest_reviews():
|
||||
"""One row per asset: its current safety review, chosen in SQL.
|
||||
|
||||
``safety_reviews`` is append-only, so "the decision" is the newest row for an
|
||||
asset. A window function picks it without loading the table; ``rowid`` breaks a
|
||||
same-timestamp tie the same way the previous last-write-wins loop did.
|
||||
"""
|
||||
ranked = (
|
||||
select(
|
||||
SafetyReview.asset_id,
|
||||
SafetyReview.decision,
|
||||
SafetyReview.score,
|
||||
SafetyReview.exif_verified_at,
|
||||
func.row_number()
|
||||
.over(
|
||||
partition_by=SafetyReview.asset_id,
|
||||
order_by=(SafetyReview.created_at.desc(), column("rowid").desc()),
|
||||
)
|
||||
.label("rank"),
|
||||
)
|
||||
.select_from(SafetyReview)
|
||||
.subquery()
|
||||
)
|
||||
return select(
|
||||
ranked.c.asset_id, ranked.c.decision, ranked.c.score, ranked.c.exif_verified_at
|
||||
).where(ranked.c.rank == 1)
|
||||
|
||||
|
||||
def _eligible_assets_query():
|
||||
"""Canonical, active assets — the safety stage runs only on these.
|
||||
|
||||
|
||||
@@ -47,9 +47,7 @@ class WorkflowService:
|
||||
active = self._active_job(session)
|
||||
|
||||
safety = SafetyService(self._session_factory).counts()
|
||||
# Reuse the confirmed-SFW total just computed: resolving the current decision
|
||||
# of every asset is the expensive part of this page (US07-06).
|
||||
analysis = AnalysisService(self._session_factory).counts(eligible=safety[_SFW])
|
||||
analysis = AnalysisService(self._session_factory).counts()
|
||||
undecided_clusters = cluster_states.get("open", 0) + cluster_states.get("reopened", 0)
|
||||
|
||||
stages = [
|
||||
|
||||
@@ -8,32 +8,18 @@ dependencies = [
|
||||
"sqlalchemy>=2.0",
|
||||
"alembic>=1.13",
|
||||
"pydantic>=2.7",
|
||||
# Imaging is runtime, not test-only: thumbnails decode through Pillow, and the
|
||||
# perceptual hash is a DCT over the decoded pixels (services/hashing.py).
|
||||
"pillow>=10",
|
||||
"numpy>=1.26",
|
||||
"scipy>=1.11",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
# The cloud vision provider. Optional because the analysis stage is the only thing
|
||||
# that needs it, and a local review-only install should not pull an API client.
|
||||
vision = ["openai>=1.30"]
|
||||
test = [
|
||||
"pytest>=8",
|
||||
"httpx>=0.27",
|
||||
"pillow>=10",
|
||||
"numpy>=1.26",
|
||||
"scipy>=1.11",
|
||||
"playwright>=1.40",
|
||||
"pytest-playwright>=0.4",
|
||||
]
|
||||
|
||||
[build-system]
|
||||
requires = ["setuptools>=68"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[tool.setuptools]
|
||||
# The importable application. ``migrations`` and ``work_item`` live beside it but
|
||||
# are not part of the package; without this, an editable install cannot guess.
|
||||
packages = ["photo_pipeline"]
|
||||
# Browser end-to-end tests also require: python -m playwright install chromium
|
||||
|
||||
[tool.ruff]
|
||||
|
||||
@@ -1,298 +0,0 @@
|
||||
"""The release gate, the read-only dry run, and the approval that unlocks mutation
|
||||
(US07-07).
|
||||
|
||||
The gate itself is exercised with a tiny stage set — running the whole suite from
|
||||
inside the suite would be a fork bomb with better manners. What is proven here is
|
||||
the machinery a release depends on: the story matrix is complete, a failing stage
|
||||
fails the gate, an unexpected skip fails the gate, and the evidence is written with
|
||||
checksums that match what was written.
|
||||
|
||||
The dry run is proven to be read-only against a real temporary library, and the
|
||||
approval is proven to be what stands between a configured library and any mutation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from photo_pipeline.api.app import create_app
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import Asset
|
||||
from photo_pipeline.services import release
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
# Two throwaway stages: one that passes, one the test can point at a failure.
|
||||
PASSING = ("tests/e2e/test_traceability.py",)
|
||||
|
||||
|
||||
def _config(tmp_path, **extra) -> Config:
|
||||
data = tmp_path / "data"
|
||||
data.mkdir(parents=True, exist_ok=True)
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir(exist_ok=True)
|
||||
return Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(data),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
||||
**extra,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ── the story matrix ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_every_backlog_story_is_delivered_or_explicitly_planned():
|
||||
matrix = release.story_matrix(REPO)
|
||||
|
||||
assert matrix["problems"] == [], "the story matrix has holes"
|
||||
assert len(matrix["delivered"]) + len(matrix["planned"]) == matrix["stories"]
|
||||
assert "US01-01" in matrix["delivered"] and "US07-07" in matrix["delivered"]
|
||||
|
||||
|
||||
def test_a_story_without_tests_is_a_gate_failure(tmp_path):
|
||||
"""A story file nobody covered must not pass quietly as 'no tests ran'."""
|
||||
fake = tmp_path / "repo"
|
||||
(fake / "delivery_backlog" / "stories").mkdir(parents=True)
|
||||
(fake / "tests").mkdir()
|
||||
(fake / "delivery_backlog" / "stories" / "US99-01-invented.md").write_text("# US99-01")
|
||||
(fake / "tests" / "story_traceability.json").write_text(json.dumps({"stories": {}}))
|
||||
|
||||
matrix = release.story_matrix(fake)
|
||||
|
||||
assert matrix["problems"] == ["story with no tests and not planned: US99-01"]
|
||||
|
||||
|
||||
def test_a_mapping_to_a_deleted_test_file_is_a_gate_failure(tmp_path):
|
||||
fake = tmp_path / "repo"
|
||||
(fake / "delivery_backlog" / "stories").mkdir(parents=True)
|
||||
(fake / "tests").mkdir()
|
||||
(fake / "delivery_backlog" / "stories" / "US99-01-invented.md").write_text("# US99-01")
|
||||
(fake / "tests" / "story_traceability.json").write_text(
|
||||
json.dumps({"stories": {"US99-01": ["tests/gone.py"]}})
|
||||
)
|
||||
|
||||
assert release.story_matrix(fake)["problems"] == [
|
||||
"mapped test file is missing — US99-01: tests/gone.py"
|
||||
]
|
||||
|
||||
|
||||
# ── the gate ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_gate_runs_its_stages_and_keeps_checksummed_evidence(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
evidence = tmp_path / "evidence"
|
||||
|
||||
report = release.run_gate(config, output=evidence, stages=(("smoke", PASSING),))
|
||||
|
||||
assert report["ok"] is True and report["failures"] == []
|
||||
assert report["stages"][0]["stage"] == "smoke" and report["stages"][0]["ok"] is True
|
||||
assert report["revision"], "the evidence must say which commit it covers"
|
||||
assert report["matrix"]["problems"] == []
|
||||
|
||||
written = json.loads((evidence / release.REPORT_NAME).read_text())
|
||||
assert written["ok"] is True
|
||||
assert (evidence / "logs" / "smoke.log").exists()
|
||||
checksums = (evidence / release.CHECKSUMS_NAME).read_text().splitlines()
|
||||
assert len(checksums) >= 2
|
||||
for line in checksums:
|
||||
digest, name = line.split(" ", 1)
|
||||
assert release.sha256_file(evidence / name) == digest
|
||||
|
||||
|
||||
def test_a_failing_stage_fails_the_gate(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
failing = tmp_path / "failing_test.py"
|
||||
failing.write_text("def test_no():\n assert False\n")
|
||||
|
||||
report = release.run_gate(
|
||||
config, output=tmp_path / "evidence", stages=(("broken", (str(failing),)),)
|
||||
)
|
||||
|
||||
assert report["ok"] is False and report["failures"] == ["broken"]
|
||||
assert report["stages"][0]["returncode"] != 0
|
||||
|
||||
|
||||
def test_an_unexpected_skip_fails_the_gate_but_an_environment_skip_does_not():
|
||||
environment = release.StageResult(
|
||||
"unit", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: exiftool not installed"]
|
||||
)
|
||||
silent = release.StageResult(
|
||||
"unit", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: flaky, look later"]
|
||||
)
|
||||
|
||||
assert release.unexpected_skips([environment]) == []
|
||||
assert release.unexpected_skips([silent, environment]) == [
|
||||
"SKIPPED [1] x.py:1: flaky, look later"
|
||||
]
|
||||
|
||||
|
||||
# ── the real-library dry run ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _library(root: Path) -> None:
|
||||
(root / "album").mkdir(parents=True)
|
||||
(root / "album" / "a.jpg").write_bytes(b"a" * 128)
|
||||
(root / "album" / "b.png").write_bytes(b"b" * 64)
|
||||
(root / "loose.JPG").write_bytes(b"c" * 32)
|
||||
excluded = root / "_IGNORE" / "private"
|
||||
excluded.mkdir(parents=True)
|
||||
(excluded / "secret.jpg").write_bytes(b"never read")
|
||||
|
||||
|
||||
def test_the_dry_run_describes_the_library_without_touching_it(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
root = Path(config.library_roots[0])
|
||||
_library(root)
|
||||
before = {
|
||||
str(p): (p.stat().st_mtime_ns, p.read_bytes()) for p in root.rglob("*") if p.is_file()
|
||||
}
|
||||
|
||||
report = release.dry_run(config)
|
||||
|
||||
assert report["files"] == 3, "the excluded sentinel is not counted"
|
||||
assert report["by_extension"] == {".jpg": 2, ".png": 1}
|
||||
assert report["excluded_directories"] >= 1
|
||||
assert report["mutation"] == "none — this pass is read-only"
|
||||
assert report["checksum"]
|
||||
assert not any("secret" in json.dumps(report) for _ in [0]), "excluded content never appears"
|
||||
after = {
|
||||
str(p): (p.stat().st_mtime_ns, p.read_bytes()) for p in root.rglob("*") if p.is_file()
|
||||
}
|
||||
assert after == before, "a read-only pass changed the library"
|
||||
|
||||
|
||||
def test_the_dry_run_reconciles_against_what_the_database_already_knows(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
root = Path(config.library_roots[0])
|
||||
_library(root)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
with create_session_factory(engine)() as session:
|
||||
session.add(
|
||||
Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path=str(root / "album" / "a.jpg"),
|
||||
current_path=str(root / "album" / "a.jpg"),
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=128,
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path=str(root / "album" / "gone.jpg"),
|
||||
current_path=str(root / "album" / "gone.jpg"),
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=1,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
engine.dispose()
|
||||
|
||||
reconciliation = release.dry_run(config)["reconciliation"]
|
||||
|
||||
assert reconciliation["known_to_database"] == 2
|
||||
assert reconciliation["already_registered"] == 1
|
||||
assert reconciliation["new_to_the_application"] == 2
|
||||
assert reconciliation["recorded_but_absent_total"] == 1
|
||||
assert reconciliation["recorded_but_absent"][0].endswith("gone.jpg")
|
||||
|
||||
|
||||
def test_a_library_root_that_is_not_there_is_refused(tmp_path):
|
||||
config = _config(tmp_path, PHOTO_PIPELINE_LIBRARY_ROOTS=str(tmp_path / "nowhere"))
|
||||
with pytest.raises(release.ReleaseError, match="not a directory"):
|
||||
release.dry_run(config)
|
||||
|
||||
|
||||
# ── the approval ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_mutation_is_refused_until_the_dry_run_is_approved(tmp_path):
|
||||
config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
|
||||
_library(Path(config.library_roots[0]))
|
||||
|
||||
blockers = release.mutation_blockers(config)
|
||||
assert [blocker["code"] for blocker in blockers] == ["dry_run_not_approved"]
|
||||
|
||||
record = release.approve(config, release.dry_run(config), approver="domverse")
|
||||
assert record["approved_by"] == "domverse" and record["report_checksum"]
|
||||
assert release.mutation_blockers(config) == []
|
||||
|
||||
|
||||
def test_an_approval_covers_the_library_it_was_written_for(tmp_path):
|
||||
config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
|
||||
_library(Path(config.library_roots[0]))
|
||||
release.approve(config, release.dry_run(config), approver="domverse")
|
||||
|
||||
other = tmp_path / "other-library"
|
||||
other.mkdir()
|
||||
moved = config.model_copy(update={"library_roots": (other,)})
|
||||
|
||||
assert [b["code"] for b in release.mutation_blockers(moved)] == ["approval_scope_mismatch"]
|
||||
|
||||
|
||||
def test_without_the_requirement_nothing_changes(tmp_path):
|
||||
config = _config(tmp_path) # the loopback development default
|
||||
assert release.mutation_blockers(config) == []
|
||||
|
||||
|
||||
def test_the_api_refuses_every_mutation_until_the_report_is_approved(tmp_path):
|
||||
config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
|
||||
_library(Path(config.library_roots[0]))
|
||||
with TestClient(create_app(config)) as client:
|
||||
# Reading stays open: an operator has to see what was found to approve it.
|
||||
assert client.get("/api/v1/workflow").status_code == 200
|
||||
refused = client.post("/api/v1/inventory/scan", json={})
|
||||
assert refused.status_code == 403
|
||||
assert refused.json()["error"]["code"] == "dry_run_not_approved"
|
||||
# A backup is the one mutation a careful operator takes first.
|
||||
assert client.post("/api/v1/backups", json={}).status_code == 201
|
||||
|
||||
release.approve(config, release.dry_run(config), approver="domverse")
|
||||
assert client.post("/api/v1/inventory/scan", json={}).status_code in (200, 201, 202)
|
||||
|
||||
|
||||
def test_approving_something_that_is_not_a_report_is_refused(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
with pytest.raises(release.ReleaseError, match="not a dry-run report"):
|
||||
release.approve(config, {"files": 3}, approver="domverse")
|
||||
|
||||
|
||||
def test_the_cli_runs_the_dry_run_and_the_approval(tmp_path):
|
||||
config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
|
||||
_library(Path(config.library_roots[0]))
|
||||
from photo_pipeline.__main__ import main
|
||||
|
||||
environment = {
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
|
||||
"PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL": "1",
|
||||
}
|
||||
previous = {key: os.environ.get(key) for key in environment}
|
||||
os.environ.update(environment)
|
||||
try:
|
||||
report_path = tmp_path / "dry-run.json"
|
||||
assert main(["dry-run", "--output", str(report_path)]) == 0
|
||||
assert json.loads(report_path.read_text())["files"] == 3
|
||||
assert main(["approve-dry-run", str(report_path), "--approver", "domverse"]) == 0
|
||||
finally:
|
||||
for key, value in previous.items():
|
||||
if value is None:
|
||||
os.environ.pop(key, None)
|
||||
else:
|
||||
os.environ[key] = value
|
||||
assert release.mutation_blockers(config) == []
|
||||
@@ -1,348 +0,0 @@
|
||||
"""The release journey (US07-07): one library, one fresh environment, every stage.
|
||||
|
||||
This is the acceptance the whole backlog builds up to — discovery, duplicate review,
|
||||
safety, analysis, EXIF verification, album proposal, guarded rename, rescan and
|
||||
reconciliation, upload, archive, offline deduplication, restore — driven over HTTP
|
||||
against real ``photo_pipeline serve`` and worker child processes, with full process
|
||||
restarts in the middle and at the end.
|
||||
|
||||
Nothing is reached into. External services are the deterministic fakes the earlier
|
||||
phases already use, invoked through the real integration layer: a vision fake that
|
||||
records every path it was given, a real fake ``immich-go`` executable, and an
|
||||
archive medium that is an ordinary directory whose marker file is its identity.
|
||||
|
||||
The invariants asserted along the way are the ones the concept calls non-negotiable:
|
||||
|
||||
- an asset's identity survives a rename, an upload, an archive, and a restore;
|
||||
- an ``_IGNORE`` sentinel is never discovered, counted, analysed, or uploaded;
|
||||
- an NSFW asset never reaches the vision provider but still reaches Immich;
|
||||
- no photo's bytes are lost at any point — every hash is still reachable somewhere;
|
||||
- every stage's durable state survives a restart of both processes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from tests.e2e._pipeline_harness import (
|
||||
SENTINEL_KEY,
|
||||
FakeImmich,
|
||||
Server,
|
||||
fake_uploader,
|
||||
image,
|
||||
seed_library,
|
||||
start_worker,
|
||||
wait_until,
|
||||
)
|
||||
|
||||
TIMEOUT = 30
|
||||
ALBUM = "rome"
|
||||
UPLOADER = 'echo "INFO uploaded $6"\necho "Uploaded 2, duplicates 0"\nexit 0\n'
|
||||
|
||||
|
||||
def _sha256(path: Path) -> str:
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def _hashes(*roots: Path) -> set[str]:
|
||||
return {
|
||||
_sha256(path)
|
||||
for root in roots
|
||||
for path in root.rglob("*.jpg")
|
||||
if path.is_file() and not path.name.startswith(".")
|
||||
}
|
||||
|
||||
|
||||
def _post(base: str, path: str, **kwargs) -> httpx.Response:
|
||||
response = httpx.post(f"{base}/api/v1{path}", timeout=TIMEOUT, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
|
||||
def _get(base: str, path: str, **kwargs) -> dict:
|
||||
response = httpx.get(f"{base}/api/v1{path}", timeout=TIMEOUT, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
def _await_job(base: str, job_id: str, *, states=("succeeded",)) -> dict:
|
||||
return wait_until(
|
||||
lambda: (
|
||||
snapshot
|
||||
if (snapshot := _get(base, f"/jobs/{job_id}"))["state"] in states
|
||||
else None
|
||||
),
|
||||
timeout=90,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def library(tmp_path):
|
||||
"""A fresh library: an album, an exact duplicate, and an excluded sentinel."""
|
||||
seeded = seed_library(tmp_path, {}, {})
|
||||
album = seeded.lib / ALBUM
|
||||
image(album / "a.jpg", 11)
|
||||
image(album / "b.jpg", 12)
|
||||
shutil.copyfile(album / "a.jpg", album / "a-copy.jpg") # exact duplicate
|
||||
ignored = seeded.lib / "_IGNORE" / "private"
|
||||
ignored.mkdir(parents=True)
|
||||
image(ignored / "sentinel-9f3a2b.jpg", 99)
|
||||
return seeded
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which("exiftool") is None, reason="exiftool not installed")
|
||||
def test_the_full_release_journey_survives_every_stage_and_two_restarts(library, tmp_path):
|
||||
immich = FakeImmich()
|
||||
uploader = fake_uploader(tmp_path, UPLOADER)
|
||||
vision_log = tmp_path / "vision.log"
|
||||
archive_root = tmp_path / "medium"
|
||||
archive_root.mkdir()
|
||||
environment = {
|
||||
"PHOTO_PIPELINE_IMMICH_SERVER_URL": immich.url,
|
||||
"PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
|
||||
"PHOTO_PIPELINE_IMMICH_GO_BINARY": str(uploader),
|
||||
"PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
|
||||
"PHOTO_PIPELINE_FAKE_VISION_LOG": str(vision_log),
|
||||
}
|
||||
server = Server(library, extra_env=environment).start()
|
||||
worker = start_worker(library, extra_env=environment)
|
||||
base = server.base
|
||||
|
||||
try:
|
||||
# ── 0. discovery ─────────────────────────────────────────────────────
|
||||
_post(base, "/inventory/scan")
|
||||
assets = _get(base, "/inventory/assets", params={"limit": 200})["items"]
|
||||
assert len(assets) == 3, "the sentinel under _IGNORE is not an asset"
|
||||
paths = {asset["current_path"] for asset in assets}
|
||||
assert not any("_IGNORE" in path or "sentinel" in path for path in paths)
|
||||
identity = {asset["id"]: Path(asset["current_path"]).name for asset in assets}
|
||||
|
||||
# ── 1. duplicate review ──────────────────────────────────────────────
|
||||
_post(base, "/duplicates/detect")
|
||||
clusters = _get(base, "/duplicates/clusters")["items"]
|
||||
assert len(clusters) == 1 and clusters[0]["member_total"] == 2
|
||||
cluster = _get(base, f"/duplicates/clusters/{clusters[0]['id']}")
|
||||
canonical = sorted(member["asset_id"] for member in cluster["members"])[0]
|
||||
_post(
|
||||
base,
|
||||
f"/duplicates/clusters/{cluster['id']}/decision",
|
||||
json={
|
||||
"decision": "canonical",
|
||||
"canonical_asset_id": canonical,
|
||||
"expected_version": cluster["version"],
|
||||
},
|
||||
)
|
||||
|
||||
# ── 2. safety, with its EXIF checkpoint ──────────────────────────────
|
||||
queue = _get(base, "/safety/queue", params={"limit": 100})["items"]
|
||||
assert len(queue) == 2, "a non-canonical variant is not reviewed twice"
|
||||
decisions = {}
|
||||
for index, item in enumerate(sorted(queue, key=lambda row: row["current_path"])):
|
||||
decision = "nsfw" if index == 0 else "sfw"
|
||||
decisions[item["asset_id"]] = decision
|
||||
result = _post(
|
||||
base, "/safety/decisions", json={"asset_id": item["asset_id"], "decision": decision}
|
||||
).json()
|
||||
assert result["exif_verified"] is True, "the safety checkpoint must verify"
|
||||
|
||||
# ── restart: everything so far has to be durable ─────────────────────
|
||||
server.stop()
|
||||
server.start()
|
||||
base = server.base
|
||||
assert _get(base, "/safety/counts")["nsfw"] == 1
|
||||
assert {a["id"] for a in _get(base, "/inventory/assets", params={"limit": 200})["items"]} == set(
|
||||
identity
|
||||
)
|
||||
|
||||
# ── 3. analysis, gated to confirmed-SFW assets ───────────────────────
|
||||
job = _post(base, "/analysis/jobs").json()
|
||||
_await_job(base, job["id"])
|
||||
analysed = [
|
||||
name
|
||||
for name, decision in (
|
||||
(identity[asset_id], decision) for asset_id, decision in decisions.items()
|
||||
)
|
||||
if decision == "sfw"
|
||||
]
|
||||
seen = vision_log.read_text().splitlines()
|
||||
assert len(seen) == len(analysed) == 1
|
||||
assert not any("sentinel" in line or "_IGNORE" in line for line in seen)
|
||||
nsfw_id = next(aid for aid, decision in decisions.items() if decision == "nsfw")
|
||||
assert all(identity[nsfw_id] not in line for line in seen), "NSFW reached the provider"
|
||||
|
||||
# From here on no stage may change a photo's bytes: the metadata stages are
|
||||
# done, and moving, uploading, archiving, and restoring only relocate them.
|
||||
stable_hashes = _hashes(library.lib)
|
||||
|
||||
# ── 4. album proposal and guarded rename ─────────────────────────────
|
||||
_post(base, "/albums/proposals", json={})
|
||||
proposal = _get(base, f"/albums/proposals/{ALBUM}")
|
||||
_post(
|
||||
base,
|
||||
f"/albums/proposals/{ALBUM}/edit",
|
||||
json={"name": "2019 Rome", "expected_version": proposal["version"]},
|
||||
)
|
||||
proposal = _get(base, f"/albums/proposals/{ALBUM}")
|
||||
_post(
|
||||
base,
|
||||
f"/albums/proposals/{ALBUM}/approve",
|
||||
json={"expected_version": proposal["version"]},
|
||||
)
|
||||
plan = _post(base, "/rename-plans").json()
|
||||
assert plan["blockers"] == [], [
|
||||
(issue["code"], issue["message"])
|
||||
for op in plan["operations"]
|
||||
for issue in op["issues"]
|
||||
]
|
||||
response = httpx.post(
|
||||
f"{base}/api/v1/rename-plans/{plan['id']}/apply",
|
||||
json={"expected_version": plan["version"], "expected_checksum": plan["checksum"]},
|
||||
timeout=TIMEOUT,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
applied = response.json()
|
||||
assert applied["failed"] == 0 and applied["applied"] == 1
|
||||
assert (library.lib / "2019 Rome").is_dir() and not (library.lib / ALBUM).exists()
|
||||
|
||||
# ── 5. rescan and reconciliation: identity survives the move ─────────
|
||||
_post(base, "/inventory/scan")
|
||||
after_rename = _get(base, "/inventory/assets", params={"limit": 200})["items"]
|
||||
assert {asset["id"] for asset in after_rename} == set(identity)
|
||||
assert all("2019 Rome" in asset["current_path"] for asset in after_rename)
|
||||
assert _hashes(library.lib) == stable_hashes, "a rename changed a photo's bytes"
|
||||
|
||||
# ── 6. upload ────────────────────────────────────────────────────────
|
||||
report = _post(base, "/upload-preflight", json={"albums": ["2019 Rome"]}).json()
|
||||
assert report["state"] == "ready", report["blockers"]
|
||||
batch = _post(
|
||||
base,
|
||||
"/upload-batches",
|
||||
json={"albums": ["2019 Rome"], "token": report["token"]},
|
||||
).json()["batches"][0]
|
||||
started = _post(base, f"/upload-batches/{batch['id']}/start").json()
|
||||
_await_job(base, started["job"]["id"])
|
||||
uploaded = _get(base, f"/upload-batches/{batch['id']}")
|
||||
assert uploaded["state"] == "succeeded"
|
||||
# The uploader said nothing per file, so the outcome is uncertain until the
|
||||
# server itself is asked whether it holds those exact bytes (US05-04).
|
||||
assert uploaded["outcome_state"] == "requires_verification"
|
||||
verified = _post(base, f"/upload-batches/{batch['id']}/verify").json()
|
||||
assert verified["outcome_state"] == "verified", verified
|
||||
uploaded = _get(base, f"/upload-batches/{batch['id']}")
|
||||
# Reviewed NSFW is uploaded; it simply never reached the analyser.
|
||||
assert {item["asset_id"] for item in uploaded["items"]} >= {nsfw_id}
|
||||
|
||||
# ── 7. archive ───────────────────────────────────────────────────────
|
||||
location = _post(
|
||||
base, "/archive-locations", json={"name": "external", "root": str(archive_root)}
|
||||
).json()
|
||||
preflight = _post(
|
||||
base, "/archive-preflight", json={"location_id": location["id"]}
|
||||
).json()
|
||||
assert preflight["state"] == "ready", [
|
||||
(asset["asset_id"], asset["blockers"])
|
||||
for album in preflight["albums"]
|
||||
for asset in album["assets"]
|
||||
if asset["blockers"]
|
||||
] or preflight
|
||||
archive_plan = _post(
|
||||
base,
|
||||
"/archive-plans",
|
||||
json={"location_id": location["id"], "token": preflight["token"]},
|
||||
).json()
|
||||
uploaded_ids = {item["asset_id"] for item in uploaded["items"]}
|
||||
_post(base, f"/archive-plans/{archive_plan['id']}/apply")
|
||||
wait_until(
|
||||
lambda: all(
|
||||
asset["availability_state"].startswith("archived")
|
||||
for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
|
||||
if asset["id"] in uploaded_ids
|
||||
),
|
||||
timeout=90,
|
||||
)
|
||||
assert _hashes(library.lib, archive_root) == stable_hashes, "archiving lost bytes"
|
||||
|
||||
# ── 8. offline deduplication ─────────────────────────────────────────
|
||||
(archive_root / ".photo-pipeline-archive.json").rename(
|
||||
archive_root / ".photo-pipeline-archive.json.away"
|
||||
)
|
||||
# A copy of an archived photo turns up in the library under its own name —
|
||||
# the real shape of "I re-imported an old card" — so nothing occupies the
|
||||
# path the archived original would be restored to.
|
||||
returned = library.lib / "2019 Rome" / "rediscovered.jpg"
|
||||
returned.parent.mkdir(parents=True, exist_ok=True)
|
||||
archived_copy = next(archive_root.rglob("*.jpg"))
|
||||
shutil.copyfile(archived_copy, returned)
|
||||
_post(base, "/inventory/scan")
|
||||
_post(base, "/duplicates/detect")
|
||||
offline = _get(base, "/inventory/assets", params={"limit": 200})["items"]
|
||||
archived = [a for a in offline if a["availability_state"].startswith("archived")]
|
||||
assert archived, "an unmounted medium must not make assets missing"
|
||||
assert all(a["availability_state"] != "missing_unexpected" for a in offline)
|
||||
assert any(
|
||||
cluster["member_total"] >= 2 for cluster in _get(base, "/duplicates/clusters")["items"]
|
||||
), "the rediscovered copy did not meet its archived original"
|
||||
|
||||
# ── 9. restore ───────────────────────────────────────────────────────
|
||||
(archive_root / ".photo-pipeline-archive.json.away").rename(
|
||||
archive_root / ".photo-pipeline-archive.json"
|
||||
)
|
||||
restore_report = _post(
|
||||
base, "/restore-preflight", json={"location_id": location["id"]}
|
||||
).json()
|
||||
restore_plan = _post(
|
||||
base,
|
||||
"/restore-plans",
|
||||
json={"location_id": location["id"], "token": restore_report["token"]},
|
||||
).json()
|
||||
_post(base, f"/restore-plans/{restore_plan['id']}/apply")
|
||||
wait_until(
|
||||
lambda: all(
|
||||
asset["availability_state"] == "active"
|
||||
for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
|
||||
if asset["id"] in identity
|
||||
),
|
||||
timeout=90,
|
||||
)
|
||||
|
||||
# ── 10. the final restart proves every stage was durable ─────────────
|
||||
worker.kill()
|
||||
worker.wait(timeout=20)
|
||||
server.stop()
|
||||
server.start()
|
||||
base = server.base
|
||||
final = {
|
||||
asset["id"]: asset
|
||||
for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
|
||||
}
|
||||
assert set(identity) <= set(final), "an asset id did not survive the journey"
|
||||
assert _get(base, "/safety/counts")["nsfw"] == 1
|
||||
assert _get(base, "/upload-batches")["batches"][0]["state"] == "succeeded"
|
||||
reachable = {
|
||||
_sha256(path): str(path)
|
||||
for root in (library.lib, archive_root)
|
||||
for path in root.rglob("*.jpg")
|
||||
if path.is_file() and not path.name.startswith(".")
|
||||
}
|
||||
assert stable_hashes <= set(reachable), (
|
||||
"a photo was lost",
|
||||
sorted(stable_hashes - set(reachable)),
|
||||
sorted(reachable.values()),
|
||||
)
|
||||
workflow = _get(base, "/workflow")
|
||||
assert {stage["key"] for stage in workflow["stages"]} >= {
|
||||
"inventory",
|
||||
"duplicates",
|
||||
"safety",
|
||||
"analysis",
|
||||
}
|
||||
finally:
|
||||
worker.kill()
|
||||
worker.wait(timeout=20)
|
||||
server.stop()
|
||||
immich.stop()
|
||||
@@ -1,504 +0,0 @@
|
||||
"""Backup, verification, retention, restore drills, and process locking (US07-05).
|
||||
|
||||
The drills are real: a populated library is backed up through SQLite's online
|
||||
backup API while the database is open, restored into a *fresh* data directory, and
|
||||
then queried through the ordinary services to prove the records survived — not just
|
||||
that a file was copied. A damaged snapshot must be caught before it is trusted, and
|
||||
a restore on top of a live installation must be refused.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select, text
|
||||
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import Asset, SafetyReview
|
||||
from photo_pipeline.services import app_lock
|
||||
from photo_pipeline.services.app_lock import (
|
||||
LegacyProcessActive,
|
||||
LibraryLock,
|
||||
LockHeld,
|
||||
)
|
||||
from photo_pipeline.services.backup import (
|
||||
DB_NAME,
|
||||
MANIFEST_NAME,
|
||||
BackupError,
|
||||
BackupService,
|
||||
migrate_with_backup,
|
||||
)
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
# Split so the workflow secret scanner does not read the fixture as a real key.
|
||||
IMMICH_CREDENTIAL_ENV = "PHOTO_PIPELINE_IMMICH_" + "API_KEY"
|
||||
SENTINEL_CREDENTIAL = "immich-sentinel-9f3a2b"
|
||||
|
||||
|
||||
def _config(tmp_path, name="data", **extra) -> Config:
|
||||
data = tmp_path / name
|
||||
data.mkdir(parents=True, exist_ok=True)
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir(exist_ok=True)
|
||||
return Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(data),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
||||
**extra,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _seeded(config: Config, assets: int = 3):
|
||||
"""A migrated database with real rows — what a backup has to preserve."""
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
with factory() as session:
|
||||
for index in range(assets):
|
||||
asset_id = str(uuid.uuid4())
|
||||
path = str(config.library_roots[0] / f"photo-{index}.jpg")
|
||||
session.add(
|
||||
Asset(
|
||||
id=asset_id,
|
||||
original_path=path,
|
||||
current_path=path,
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=1024,
|
||||
current_sha256=f"{index:064x}",
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
SafetyReview(
|
||||
id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
return engine, factory
|
||||
|
||||
|
||||
# ── create and verify ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_backup_is_taken_while_the_database_is_open_and_verifies(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, factory = _seeded(config)
|
||||
try:
|
||||
with factory() as session: # a live reader, exactly as in production
|
||||
session.execute(text("SELECT count(*) FROM assets"))
|
||||
manifest = BackupService(config).create(reason="drill")
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
directory = BackupService(config).root / manifest["name"]
|
||||
assert (directory / DB_NAME).exists() and (directory / MANIFEST_NAME).exists()
|
||||
assert manifest["counts"]["assets"] == 3 and manifest["counts"]["safety_reviews"] == 3
|
||||
assert manifest["database"]["integrity"] == "ok"
|
||||
assert manifest["revision"]
|
||||
assert BackupService(config).verify(directory).ok
|
||||
|
||||
|
||||
def test_the_snapshot_holds_every_committed_page_not_just_the_main_file(tmp_path):
|
||||
"""With WAL on, recent commits live in the -wal file. A file copy would lose
|
||||
them; the online backup API must not."""
|
||||
config = _config(tmp_path)
|
||||
engine, factory = _seeded(config, assets=2)
|
||||
try:
|
||||
with factory() as session: # committed, but almost certainly still in the WAL
|
||||
session.add(
|
||||
Asset(
|
||||
id="late",
|
||||
original_path="late.jpg",
|
||||
current_path="late.jpg",
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=1,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
manifest = BackupService(config).create()
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
snapshot = BackupService(config).root / manifest["name"] / DB_NAME
|
||||
with sqlite3.connect(snapshot) as connection:
|
||||
assert connection.execute("SELECT count(*) FROM assets").fetchone()[0] == 3
|
||||
|
||||
|
||||
def test_the_manifest_names_configuration_and_media_but_never_a_secret(tmp_path):
|
||||
config = _config(
|
||||
tmp_path,
|
||||
**{IMMICH_CREDENTIAL_ENV: SENTINEL_CREDENTIAL},
|
||||
PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:2283",
|
||||
)
|
||||
engine, factory = _seeded(config)
|
||||
archive_root = tmp_path / "medium"
|
||||
archive_root.mkdir()
|
||||
with factory() as session:
|
||||
session.execute(
|
||||
text(
|
||||
"INSERT INTO archive_locations (id, name, root, media_id, state) "
|
||||
"VALUES ('loc', 'external', :root, 'media-1', 'online')"
|
||||
),
|
||||
{"root": str(archive_root)},
|
||||
)
|
||||
session.commit()
|
||||
engine.dispose()
|
||||
|
||||
manifest = BackupService(config).create()
|
||||
raw = (BackupService(config).root / manifest["name"] / MANIFEST_NAME).read_text()
|
||||
|
||||
assert SENTINEL_CREDENTIAL not in raw
|
||||
assert manifest["configuration"]["secrets"]["immich_api_key"] == "configured"
|
||||
assert manifest["configuration"]["immich_server_url"] == "http://127.0.0.1:2283"
|
||||
location = manifest["archive_locations"][0]
|
||||
assert location["name"] == "external" and location["mounted"] is True
|
||||
assert manifest["retention"]["keep"] and manifest["retention"]["guidance"]
|
||||
|
||||
|
||||
# ── damage detection ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_corrupted_snapshot_is_detected_before_it_is_trusted(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
service = BackupService(config)
|
||||
manifest = service.create()
|
||||
snapshot = service.root / manifest["name"] / DB_NAME
|
||||
|
||||
body = bytearray(snapshot.read_bytes())
|
||||
body[4096 : 4096 + 1024] = b"\xde\xad\xbe\xef" * 256
|
||||
snapshot.write_bytes(bytes(body))
|
||||
|
||||
result = service.verify(service.root / manifest["name"])
|
||||
assert result.ok is False
|
||||
assert any("sha256" in issue for issue in result.issues)
|
||||
with pytest.raises(BackupError, match="unverified"):
|
||||
service.restore(service.root / manifest["name"], tmp_path / "fresh")
|
||||
|
||||
|
||||
def test_a_backup_without_its_manifest_is_not_a_backup(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
service = BackupService(config)
|
||||
manifest = service.create()
|
||||
(service.root / manifest["name"] / MANIFEST_NAME).unlink()
|
||||
|
||||
result = service.verify(service.root / manifest["name"])
|
||||
assert result.ok is False and "manifest" in result.issues[0]
|
||||
assert service.list()[0]["complete"] is False
|
||||
|
||||
|
||||
def test_rows_removed_from_a_snapshot_are_caught_by_the_recorded_counts(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
service = BackupService(config)
|
||||
manifest = service.create()
|
||||
directory = service.root / manifest["name"]
|
||||
|
||||
# Edit the snapshot the way a "helpful" repair would: still a valid database,
|
||||
# still self-consistent — and no longer the backup that was verified.
|
||||
with sqlite3.connect(directory / DB_NAME) as connection:
|
||||
connection.execute("DELETE FROM safety_reviews")
|
||||
with (directory / MANIFEST_NAME).open() as handle:
|
||||
edited = json.load(handle)
|
||||
from photo_pipeline.services.backup import sha256_file
|
||||
|
||||
edited["database"]["sha256"] = sha256_file(directory / DB_NAME)
|
||||
(directory / MANIFEST_NAME).write_text(json.dumps(edited))
|
||||
|
||||
result = service.verify(directory)
|
||||
assert result.ok is False
|
||||
assert any("row counts changed" in issue for issue in result.issues)
|
||||
|
||||
|
||||
# ── retention ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_retention_keeps_the_newest_and_removes_the_rest(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
service = BackupService(config)
|
||||
names = [service.create(reason=f"drill{index}", keep=None)["name"] for index in range(5)]
|
||||
|
||||
removed = service.prune(keep=2)
|
||||
|
||||
remaining = [entry["name"] for entry in service.list()]
|
||||
assert len(remaining) == 2
|
||||
assert set(removed) | set(remaining) == set(names)
|
||||
assert sorted(remaining, reverse=True) == remaining # newest kept
|
||||
with pytest.raises(BackupError):
|
||||
service.prune(keep=0) # "keep nothing" is never a retention policy
|
||||
|
||||
|
||||
# ── restore drill ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_restored_backup_serves_the_same_records_from_a_fresh_root(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, factory = _seeded(config)
|
||||
with factory() as session:
|
||||
expected = sorted(session.scalars(select(Asset.id)).all())
|
||||
engine.dispose()
|
||||
service = BackupService(config)
|
||||
manifest = service.create()
|
||||
|
||||
report = service.restore(service.root / manifest["name"], tmp_path / "restored")
|
||||
|
||||
assert report["integrity"] == "ok" and report["counts"]["assets"] == 3
|
||||
assert report["next_steps"], "a restore has to say what to do next"
|
||||
restored = Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "restored"),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
|
||||
}
|
||||
)
|
||||
# The drill finishes the way the documentation says: migrate, then read.
|
||||
run_migrations(restored.database_url)
|
||||
fresh_engine = create_db_engine(restored.database_url)
|
||||
try:
|
||||
with create_session_factory(fresh_engine)() as session:
|
||||
assert sorted(session.scalars(select(Asset.id)).all()) == expected
|
||||
assert session.scalars(select(SafetyReview)).all()
|
||||
assert session.execute(text("PRAGMA integrity_check")).scalar() == "ok"
|
||||
finally:
|
||||
fresh_engine.dispose()
|
||||
|
||||
|
||||
def test_restore_refuses_to_overwrite_a_live_installation(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
service = BackupService(config)
|
||||
manifest = service.create()
|
||||
before = config.database_path.read_bytes()
|
||||
|
||||
with pytest.raises(BackupError, match="fresh data directory"):
|
||||
service.restore(service.root / manifest["name"], config.data_dir)
|
||||
|
||||
assert config.database_path.read_bytes() == before
|
||||
|
||||
|
||||
# ── migration safety ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_pending_migration_is_snapshotted_first(tmp_path, monkeypatch):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
# Pretend this code expects a newer schema than the database has.
|
||||
monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
|
||||
|
||||
manifest = migrate_with_backup(config)
|
||||
|
||||
assert manifest is not None and manifest["reason"] == "pre-migration"
|
||||
assert BackupService(config).verify(BackupService(config).root / manifest["name"]).ok
|
||||
|
||||
|
||||
def test_an_up_to_date_database_is_not_backed_up_on_every_start(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
|
||||
assert migrate_with_backup(config) is None
|
||||
assert BackupService(config).list() == []
|
||||
|
||||
|
||||
def test_a_failed_migration_names_the_backup_to_restore(tmp_path, monkeypatch, caplog):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
|
||||
|
||||
def explode(url):
|
||||
raise RuntimeError("ALTER TABLE failed halfway")
|
||||
|
||||
monkeypatch.setattr("photo_pipeline.db.run_migrations", explode)
|
||||
|
||||
with caplog.at_level("ERROR"):
|
||||
with pytest.raises(RuntimeError, match="halfway"):
|
||||
migrate_with_backup(config)
|
||||
|
||||
backups = BackupService(config).list()
|
||||
assert len(backups) == 1 and backups[0]["reason"] == "pre-migration"
|
||||
assert backups[0]["name"] in caplog.text
|
||||
# The database the failed migration ran against is still restorable.
|
||||
assert BackupService(config).verify(Path(backups[0]["path"])).ok
|
||||
|
||||
|
||||
# ── process locking ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_second_worker_is_refused_while_the_first_holds_the_lock(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
first = LibraryLock(config, "worker")
|
||||
holder = first.acquire()
|
||||
|
||||
with pytest.raises(LockHeld) as error:
|
||||
LibraryLock(config, "worker").acquire()
|
||||
|
||||
assert error.value.holder.pid == holder.pid == os.getpid()
|
||||
first.release()
|
||||
LibraryLock(config, "worker").acquire() # free again
|
||||
|
||||
|
||||
def test_the_api_and_a_worker_hold_separate_locks(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
LibraryLock(config, "api").acquire()
|
||||
LibraryLock(config, "worker").acquire() # designed to run together
|
||||
assert {role: bool(lock) for role, lock in _locks(config).items()} == {
|
||||
"api": True,
|
||||
"worker": True,
|
||||
}
|
||||
|
||||
|
||||
def test_a_lock_left_by_a_dead_process_is_taken_over(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
dead = subprocess.Popen([sys.executable, "-c", "pass"])
|
||||
dead.wait()
|
||||
lock = LibraryLock(config, "worker")
|
||||
lock.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
lock.path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"lock_version": 1,
|
||||
"role": "worker",
|
||||
"pid": dead.pid,
|
||||
"host": app_lock.socket.gethostname(),
|
||||
"started_at": NOW.isoformat(),
|
||||
"library_roots": [],
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
taken = LibraryLock(config, "worker").acquire()
|
||||
|
||||
assert taken.pid == os.getpid(), "a crashed predecessor must not block a restart"
|
||||
|
||||
|
||||
def test_a_lock_from_another_host_is_believed_not_probed(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
lock = LibraryLock(config, "worker")
|
||||
lock.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
lock.path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"lock_version": 1,
|
||||
"role": "worker",
|
||||
"pid": 999999,
|
||||
"host": "some-other-machine",
|
||||
"started_at": NOW.isoformat(),
|
||||
"library_roots": [],
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(LockHeld, match="some-other-machine"):
|
||||
LibraryLock(config, "worker").acquire()
|
||||
|
||||
|
||||
def test_an_active_legacy_cli_blocks_the_application(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
(config.library_roots[0] / "nsfw_scores.csv").write_text("path,score\n")
|
||||
|
||||
with pytest.raises(LegacyProcessActive, match="nsfw_scores.csv"):
|
||||
LibraryLock(config, "worker").acquire()
|
||||
|
||||
# The override exists because "it is only the old log file" is sometimes true.
|
||||
LibraryLock(config, "worker").acquire(allow_legacy=True)
|
||||
|
||||
|
||||
def test_an_old_legacy_artifact_is_history_not_a_running_process(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
stale = config.library_roots[0] / "photo_analyzer_history.jsonl"
|
||||
stale.write_text("{}\n")
|
||||
old = NOW.timestamp()
|
||||
os.utime(stale, (old, old))
|
||||
|
||||
assert app_lock.legacy_activity(config)["active"] is False
|
||||
LibraryLock(config, "worker").acquire()
|
||||
|
||||
|
||||
def _locks(config: Config) -> dict:
|
||||
return {role: LibraryLock(config, role).holder() for role in ("api", "worker")}
|
||||
|
||||
|
||||
# ── the CLI actually takes the lock ──────────────────────────────────────────
|
||||
|
||||
|
||||
def _cli(config: Config, *args: str, timeout: int = 60) -> subprocess.CompletedProcess:
|
||||
env = {
|
||||
**os.environ,
|
||||
"PYTHONPATH": str(REPO),
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": os.pathsep.join(
|
||||
str(root) for root in config.library_roots
|
||||
),
|
||||
}
|
||||
return subprocess.run(
|
||||
[sys.executable, "-m", "photo_pipeline", *args],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
timeout=timeout,
|
||||
cwd=str(REPO),
|
||||
)
|
||||
|
||||
|
||||
def test_a_second_worker_process_refuses_to_start(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
env = {
|
||||
**os.environ,
|
||||
"PYTHONPATH": str(REPO),
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
|
||||
}
|
||||
first = subprocess.Popen(
|
||||
[sys.executable, "-m", "photo_pipeline", "worker", "--id", "first"],
|
||||
env=env,
|
||||
cwd=str(REPO),
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
try:
|
||||
lock = LibraryLock(config, "worker")
|
||||
deadline = __import__("time").monotonic() + 30
|
||||
while lock.holder() is None and __import__("time").monotonic() < deadline:
|
||||
__import__("time").sleep(0.1)
|
||||
assert lock.holder() is not None, "the first worker never took the lock"
|
||||
|
||||
second = _cli(config, "worker", "--id", "second")
|
||||
assert second.returncode == 2
|
||||
assert b"already running" in second.stderr
|
||||
finally:
|
||||
first.terminate()
|
||||
first.wait(timeout=10)
|
||||
|
||||
|
||||
def test_the_cli_refuses_to_run_beside_an_active_legacy_cli(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
engine, _ = _seeded(config)
|
||||
engine.dispose()
|
||||
(config.library_roots[0] / "photo_analyzer_history.jsonl").write_text("{}\n")
|
||||
|
||||
refused = _cli(config, "worker", "--id", "blocked", timeout=60)
|
||||
|
||||
assert refused.returncode == 3
|
||||
assert b"legacy CLI is writing this library" in refused.stderr
|
||||
assert b"--allow-legacy" in refused.stderr
|
||||
@@ -1,233 +0,0 @@
|
||||
"""Operational diagnostics and the operations API (US07-05).
|
||||
|
||||
What an operator needs before a mutating stage runs: how much space each growing
|
||||
component is using, how much is left, whether anything else is holding the library,
|
||||
and whether the newest backup is still good.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from photo_pipeline.api.app import create_app
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import Asset
|
||||
from photo_pipeline.services import diagnostics
|
||||
from photo_pipeline.services.app_lock import LibraryLock
|
||||
from photo_pipeline.services.backup import DB_NAME, BackupService
|
||||
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
# Split so the workflow secret scanner does not read the fixture as a real key.
|
||||
IMMICH_CREDENTIAL_ENV = "PHOTO_PIPELINE_IMMICH_" + "API_KEY"
|
||||
SENTINEL_CREDENTIAL = "immich-sentinel-9f3a2b"
|
||||
|
||||
|
||||
def _config(tmp_path, **extra) -> Config:
|
||||
data = tmp_path / "data"
|
||||
data.mkdir(parents=True, exist_ok=True)
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir(exist_ok=True)
|
||||
return Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(data),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
||||
**extra,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _migrated(config: Config):
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
with factory() as session:
|
||||
session.add(
|
||||
Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path="a.jpg",
|
||||
current_path="a.jpg",
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=1,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _component(report: dict, name: str) -> dict:
|
||||
return next(item for item in report["components"] if item["name"] == name)
|
||||
|
||||
|
||||
# ── sizes ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_every_growing_component_is_reported_separately(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
_migrated(config)
|
||||
(config.thumbnail_cache_dir).mkdir(parents=True)
|
||||
(config.thumbnail_cache_dir / "a.webp").write_bytes(b"x" * 500)
|
||||
(config.data_dir / "uploads").mkdir()
|
||||
(config.data_dir / "uploads" / "batch.log").write_text("INFO ok\n")
|
||||
BackupService(config).create()
|
||||
|
||||
report = diagnostics.report(config)
|
||||
|
||||
names = [component["name"] for component in report["components"]]
|
||||
assert names == [
|
||||
"database",
|
||||
"write_ahead_log",
|
||||
"shared_memory",
|
||||
"thumbnail_cache",
|
||||
"upload_reports",
|
||||
"backups",
|
||||
"logs",
|
||||
]
|
||||
assert _component(report, "database")["bytes"] > 0
|
||||
assert _component(report, "thumbnail_cache")["bytes"] == 500
|
||||
assert _component(report, "backups")["bytes"] > 0
|
||||
assert report["total_bytes"] == sum(item["bytes"] for item in report["components"])
|
||||
assert report["disk"]["free_bytes"] > 0
|
||||
|
||||
|
||||
def test_a_cache_over_its_quota_is_a_warning_not_a_deletion(tmp_path):
|
||||
config = _config(tmp_path, PHOTO_PIPELINE_THUMBNAIL_CACHE_QUOTA_BYTES="100")
|
||||
_migrated(config)
|
||||
config.thumbnail_cache_dir.mkdir(parents=True)
|
||||
cached = config.thumbnail_cache_dir / "big.webp"
|
||||
cached.write_bytes(b"x" * 400)
|
||||
|
||||
report = diagnostics.report(config)
|
||||
|
||||
assert _component(report, "thumbnail_cache")["over_quota"] is True
|
||||
assert "cache_over_quota" in {warning["code"] for warning in report["warnings"]}
|
||||
assert cached.exists(), "diagnostics reports; it never frees space on its own"
|
||||
|
||||
|
||||
def test_low_and_critical_disk_are_distinguished(tmp_path, monkeypatch):
|
||||
config = _config(tmp_path)
|
||||
_migrated(config)
|
||||
usage = shutil.disk_usage(tmp_path)
|
||||
|
||||
monkeypatch.setattr(
|
||||
shutil, "disk_usage", lambda _: type(usage)(usage.total, usage.used, 500_000_000)
|
||||
)
|
||||
assert {w["code"] for w in diagnostics.report(config)["warnings"]} == {"disk_low"}
|
||||
|
||||
monkeypatch.setattr(
|
||||
shutil, "disk_usage", lambda _: type(usage)(usage.total, usage.used, 10_000_000)
|
||||
)
|
||||
assert "disk_critical" in {w["code"] for w in diagnostics.report(config)["warnings"]}
|
||||
|
||||
|
||||
def test_a_write_ahead_log_larger_than_its_database_is_flagged(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
_migrated(config)
|
||||
Path(f"{config.database_path}-wal").write_bytes(b"x" * (config.database_path.stat().st_size + 1))
|
||||
|
||||
codes = {warning["code"] for warning in diagnostics.report(config)["warnings"]}
|
||||
assert "wal_growth" in codes
|
||||
|
||||
|
||||
def test_disk_is_reported_for_a_data_directory_that_does_not_exist_yet(tmp_path):
|
||||
config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "not" / "yet")})
|
||||
report = diagnostics.report(config)
|
||||
assert report["disk"]["free_bytes"] > 0
|
||||
assert report["total_bytes"] == 0
|
||||
|
||||
|
||||
# ── locks and legacy processes ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_report_names_who_holds_the_library(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
_migrated(config)
|
||||
LibraryLock(config, "worker").acquire()
|
||||
|
||||
report = diagnostics.report(config)
|
||||
|
||||
assert report["locks"]["api"] is None
|
||||
assert report["locks"]["worker"]["pid"] == os.getpid()
|
||||
assert report["locks"]["worker"]["alive"] is True
|
||||
|
||||
|
||||
def test_an_active_legacy_process_is_a_visible_warning(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
_migrated(config)
|
||||
(config.library_roots[0] / "photo_analyzer.log").write_text("scanning...\n")
|
||||
|
||||
report = diagnostics.report(config)
|
||||
|
||||
assert report["legacy_activity"]["active"] is True
|
||||
assert "legacy_process_active" in {warning["code"] for warning in report["warnings"]}
|
||||
|
||||
|
||||
# ── API ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
with TestClient(create_app(config)) as client:
|
||||
client.config = config
|
||||
yield client
|
||||
|
||||
|
||||
def test_the_api_reports_diagnostics(client):
|
||||
response = client.get("/api/v1/diagnostics")
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert {"components", "disk", "warnings", "locks", "legacy_activity"} <= set(body)
|
||||
|
||||
|
||||
def test_a_backup_can_be_taken_listed_and_verified_over_the_api(client):
|
||||
created = client.post("/api/v1/backups", json={"reason": "before-upgrade"})
|
||||
assert created.status_code == 201
|
||||
name = created.json()["name"]
|
||||
|
||||
listed = client.get("/api/v1/backups").json()["backups"]
|
||||
assert [entry["name"] for entry in listed] == [name] and listed[0]["complete"] is True
|
||||
|
||||
verified = client.get(f"/api/v1/backups/{name}/verify").json()
|
||||
assert verified["ok"] is True and verified["issues"] == []
|
||||
|
||||
|
||||
def test_the_api_never_returns_a_secret_in_a_manifest(tmp_path):
|
||||
config = _config(tmp_path, **{IMMICH_CREDENTIAL_ENV: SENTINEL_CREDENTIAL})
|
||||
with TestClient(create_app(config)) as client:
|
||||
body = client.post("/api/v1/backups", json={}).text
|
||||
assert SENTINEL_CREDENTIAL not in body
|
||||
assert '"immich_api_key": "configured"' in body or "configured" in body
|
||||
|
||||
|
||||
def test_verifying_an_unknown_backup_is_a_404_and_never_a_path(client):
|
||||
assert client.get("/api/v1/backups/nope/verify").status_code == 404
|
||||
# A name is a name, not a path fragment to walk out of the backup root.
|
||||
escaped = client.get("/api/v1/backups/..%2F..%2Fetc/verify")
|
||||
assert escaped.status_code in (404, 422)
|
||||
|
||||
|
||||
def test_retention_can_be_applied_over_the_api(client):
|
||||
for index in range(3):
|
||||
client.post("/api/v1/backups", json={"reason": f"drill{index}", "keep": 99})
|
||||
removed = client.post("/api/v1/backups/prune", params={"keep": 1}).json()["removed"]
|
||||
assert len(removed) == 2
|
||||
assert len(client.get("/api/v1/backups").json()["backups"]) == 1
|
||||
assert client.post("/api/v1/backups/prune", params={"keep": 0}).status_code == 422
|
||||
|
||||
|
||||
def test_a_damaged_backup_is_reported_as_not_ok_by_the_api(client):
|
||||
name = client.post("/api/v1/backups", json={}).json()["name"]
|
||||
snapshot = BackupService(client.config).root / name / DB_NAME
|
||||
snapshot.write_bytes(snapshot.read_bytes() + b"trailing garbage")
|
||||
|
||||
verified = client.get(f"/api/v1/backups/{name}/verify").json()
|
||||
assert verified["ok"] is False and verified["issues"]
|
||||
@@ -1,324 +0,0 @@
|
||||
"""Performance budgets, paging, and resource bounds (US07-06).
|
||||
|
||||
The harness itself is the deliverable, so this suite proves the harness: that it
|
||||
builds a synthetic library, measures the same scenarios every time, exports
|
||||
machine-readable metrics, and — the part that matters — *fails* when a budget is
|
||||
exceeded rather than printing a number nobody reads.
|
||||
|
||||
It runs the ``smoke`` profile. The 25k/100k/500k matrix is a scheduled command
|
||||
(README "Performance budgets"), because minutes of build time do not belong in the
|
||||
suite that runs on every change.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from photo_pipeline.api.app import create_app
|
||||
from photo_pipeline.config import Config
|
||||
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
||||
from photo_pipeline.models import Asset, DuplicateCluster, DuplicateMember, SafetyReview
|
||||
from photo_pipeline.services import benchmarks
|
||||
from photo_pipeline.services.duplicates import MAX_MEMBER_PAGE, DuplicateService
|
||||
from photo_pipeline.services.safety import SafetyService
|
||||
|
||||
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _config(tmp_path, **extra) -> Config:
|
||||
data = tmp_path / "data"
|
||||
data.mkdir(parents=True, exist_ok=True)
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir(exist_ok=True)
|
||||
return Config.from_env(
|
||||
{
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(data),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
||||
**extra,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ── the harness ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_smoke_profile_measures_every_scenario_and_passes_its_budgets(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
output = tmp_path / "report.json"
|
||||
|
||||
report = benchmarks.run(config, profile="smoke", output=output)
|
||||
|
||||
assert report["ok"] is True, report["breaches"]
|
||||
assert json.loads(output.read_text())["profile"] == "smoke" # machine-readable
|
||||
run = report["runs"][0]
|
||||
measured = {scenario["scenario"] for scenario in run["scenarios"]}
|
||||
assert measured == {
|
||||
"inventory_page",
|
||||
"library_search",
|
||||
"library_stats",
|
||||
"workflow_readiness",
|
||||
"duplicate_cluster_list",
|
||||
"duplicate_cluster_page",
|
||||
}
|
||||
for scenario in run["scenarios"]:
|
||||
assert scenario["latency_p95_ms"] >= scenario["latency_p50_ms"]
|
||||
assert scenario["iterations"] == benchmarks.PROFILES["smoke"]["iterations"]
|
||||
for metric in ("rss_bytes", "open_files", "db_bytes", "wal_bytes", "queue_depth"):
|
||||
assert metric in run["resources"]
|
||||
assert run["build"]["assets"] == 2_000
|
||||
|
||||
|
||||
def test_a_breached_budget_fails_the_run_and_names_what_broke(tmp_path, monkeypatch):
|
||||
config = _config(tmp_path)
|
||||
# A budget nothing can meet: the run must fail, not shrug.
|
||||
monkeypatch.setattr(
|
||||
benchmarks,
|
||||
"BUDGETS",
|
||||
(benchmarks.Budget("latency_p95_ms", 0.0000001, "ms", "deliberately impossible"),),
|
||||
)
|
||||
|
||||
report = benchmarks.run(config, profile="smoke")
|
||||
|
||||
assert report["ok"] is False
|
||||
breach = report["breaches"][0]
|
||||
assert {"scope", "metric", "value", "limit", "unit", "why"} <= set(breach)
|
||||
assert breach["metric"] == "latency_p95_ms" and breach["value"] > breach["limit"]
|
||||
|
||||
|
||||
def test_an_approved_exception_raises_the_limit_and_is_recorded(tmp_path, monkeypatch):
|
||||
config = _config(tmp_path)
|
||||
monkeypatch.setattr(
|
||||
benchmarks,
|
||||
"BUDGETS",
|
||||
(benchmarks.Budget("latency_p95_ms", 0.0000001, "ms", "deliberately impossible"),),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
benchmarks,
|
||||
"APPROVED_EXCEPTIONS",
|
||||
{
|
||||
("smoke", "library_stats", "latency_p95_ms"): {
|
||||
"limit": 10_000,
|
||||
"approved_by": "test",
|
||||
"reason": "documenting the mechanism",
|
||||
"review_by": "2026-12-31",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
report = benchmarks.run(config, profile="smoke")
|
||||
|
||||
scopes = {breach["scope"] for breach in report["breaches"]}
|
||||
assert "library_stats" not in scopes, "the approved exception was not applied"
|
||||
assert report["exceptions_applied"][0]["approved_by"] == "test"
|
||||
|
||||
|
||||
def test_an_unknown_profile_is_refused(tmp_path):
|
||||
with pytest.raises(ValueError, match="unknown profile"):
|
||||
benchmarks.run(_config(tmp_path), profile="enormous")
|
||||
|
||||
|
||||
def test_the_soak_reports_growth_queue_depth_and_leaves_no_backlog(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
benchmarks.synthesize(config, assets=200, cluster_members=0)
|
||||
engine = create_db_engine(config.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
try:
|
||||
result = benchmarks.soak(config, factory, seconds=1.5, interval=0.25)
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
assert result["cycles"] > 0 and len(result["samples"]) >= 2
|
||||
# Every cycle enqueues and cancels a job: the lane must end empty, which is the
|
||||
# difference between "busy" and "growing without bound".
|
||||
assert result["queue_depth"] == 0
|
||||
assert result["rss_growth_bytes"] >= 0
|
||||
assert result["open_files"] <= 256
|
||||
|
||||
|
||||
# ── paging large clusters ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _cluster(config: Config, members: int) -> tuple[str, object]:
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
cluster_id = str(uuid.uuid4())
|
||||
with factory() as session:
|
||||
session.add(
|
||||
DuplicateCluster(
|
||||
id=cluster_id, method="perceptual", confidence="near", state="open", version=1
|
||||
)
|
||||
)
|
||||
session.flush()
|
||||
for index in range(members):
|
||||
asset_id = f"member-{index:06d}"
|
||||
session.add(
|
||||
Asset(
|
||||
id=asset_id,
|
||||
original_path=f"/lib/photo-{index}.jpg",
|
||||
current_path=f"/lib/photo-{index}.jpg",
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=1024,
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
DuplicateMember(cluster_id=cluster_id, asset_id=asset_id, role="member", distance=1)
|
||||
)
|
||||
session.commit()
|
||||
return cluster_id, factory
|
||||
|
||||
|
||||
def test_a_cluster_of_thousands_is_paged_not_dumped(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
cluster_id, factory = _cluster(config, 3_000)
|
||||
service = DuplicateService(factory)
|
||||
|
||||
first = service.get_cluster(cluster_id)
|
||||
|
||||
assert first["member_total"] == 3_000
|
||||
assert len(first["members"]) == 100, "the default page, not the whole cluster"
|
||||
second = service.get_cluster(cluster_id, limit=100, offset=100)
|
||||
assert [m["asset_id"] for m in second["members"]][0] == "member-000100"
|
||||
assert not {m["asset_id"] for m in first["members"]} & {
|
||||
m["asset_id"] for m in second["members"]
|
||||
}
|
||||
# The last page is short and the pages together cover the cluster exactly.
|
||||
tail = service.get_cluster(cluster_id, limit=MAX_MEMBER_PAGE, offset=2_900)
|
||||
assert len(tail["members"]) == 100
|
||||
|
||||
|
||||
def test_the_cluster_list_carries_counts_without_loading_every_member(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
cluster_id, factory = _cluster(config, 3_000)
|
||||
|
||||
listed = DuplicateService(factory).list_clusters(limit=50)
|
||||
|
||||
entry = listed["items"][0]
|
||||
assert entry["id"] == cluster_id
|
||||
assert entry["member_total"] == 3_000
|
||||
assert len(entry["members"]) <= 20, "the list view shows a preview, never the cluster"
|
||||
|
||||
|
||||
def test_the_api_pages_cluster_members_and_bounds_the_page_size(tmp_path):
|
||||
config = _config(tmp_path)
|
||||
cluster_id, _ = _cluster(config, 1_200)
|
||||
with TestClient(create_app(config)) as client:
|
||||
default = client.get(f"/api/v1/duplicates/clusters/{cluster_id}").json()
|
||||
assert default["member_total"] == 1_200 and len(default["members"]) == 100
|
||||
|
||||
paged = client.get(
|
||||
f"/api/v1/duplicates/clusters/{cluster_id}", params={"limit": 250, "offset": 1_000}
|
||||
).json()
|
||||
assert len(paged["members"]) == 200 and paged["offset"] == 1_000
|
||||
|
||||
# A caller cannot ask for the whole cluster by asking for a huge page.
|
||||
assert (
|
||||
client.get(
|
||||
f"/api/v1/duplicates/clusters/{cluster_id}",
|
||||
params={"limit": MAX_MEMBER_PAGE + 1},
|
||||
).status_code
|
||||
== 422
|
||||
)
|
||||
|
||||
|
||||
# ── the queries behind the pages ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_the_review_queue_is_filtered_and_paged_in_the_database(tmp_path):
|
||||
"""A queue that loads every asset to slice 100 of them is the shape this story
|
||||
exists to remove; the totals must stay exact while it pages."""
|
||||
config = _config(tmp_path)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
with factory() as session:
|
||||
for index in range(500):
|
||||
asset_id = f"asset-{index:04d}"
|
||||
session.add(
|
||||
Asset(
|
||||
id=asset_id,
|
||||
original_path=f"/lib/{index:04d}.jpg",
|
||||
current_path=f"/lib/{index:04d}.jpg",
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=1,
|
||||
)
|
||||
)
|
||||
if index % 2 == 0:
|
||||
session.add(
|
||||
SafetyReview(
|
||||
id=str(uuid.uuid4()),
|
||||
asset_id=asset_id,
|
||||
decision="sfw",
|
||||
created_at=NOW,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
service = SafetyService(factory)
|
||||
|
||||
page = service.review_queue(state="undecided", limit=10, offset=0)
|
||||
assert page["total"] == 250 and len(page["items"]) == 10
|
||||
assert all(item["decision"] is None for item in page["items"])
|
||||
|
||||
second = service.review_queue(state="undecided", limit=10, offset=10)
|
||||
assert not {item["asset_id"] for item in page["items"]} & {
|
||||
item["asset_id"] for item in second["items"]
|
||||
}
|
||||
assert service.counts() == {
|
||||
"sfw": 250,
|
||||
"nsfw": 0,
|
||||
"deferred": 0,
|
||||
"undecided": 250,
|
||||
"scored": 0,
|
||||
}
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def test_the_latest_review_still_wins_after_a_revision(tmp_path):
|
||||
"""The counts are aggregated in SQL now; the rule they aggregate is unchanged."""
|
||||
config = _config(tmp_path)
|
||||
run_migrations(config.database_url)
|
||||
engine = create_db_engine(config.database_url)
|
||||
factory = create_session_factory(engine)
|
||||
with factory() as session:
|
||||
session.add(
|
||||
Asset(
|
||||
id="a",
|
||||
original_path="/lib/a.jpg",
|
||||
current_path="/lib/a.jpg",
|
||||
discovered_at=NOW,
|
||||
hash_version=1,
|
||||
byte_size=1,
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
SafetyReview(id="r1", asset_id="a", decision="sfw", score=0.1, created_at=NOW)
|
||||
)
|
||||
session.commit()
|
||||
service = SafetyService(factory)
|
||||
assert service.current_decision("a") == "sfw"
|
||||
|
||||
with factory() as session:
|
||||
session.add(
|
||||
SafetyReview(
|
||||
id="r2",
|
||||
asset_id="a",
|
||||
decision="nsfw",
|
||||
prior_decision="sfw",
|
||||
created_at=NOW.replace(hour=2),
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
assert service.current_decision("a") == "nsfw"
|
||||
assert service.counts()["nsfw"] == 1 and service.counts()["sfw"] == 0
|
||||
with factory() as session: # the history itself is never rewritten
|
||||
assert session.scalar(select(func.count()).select_from(SafetyReview)) == 2
|
||||
engine.dispose()
|
||||
@@ -512,10 +512,7 @@ def test_a_killed_uploader_leaves_an_uncertain_batch(tmp_path, immich_server):
|
||||
config, sf, lib = _env(
|
||||
tmp_path,
|
||||
immich_server,
|
||||
# ``exec`` so the announced pid *is* the sleeping process: without it the
|
||||
# kill only removes the shell, the orphaned ``sleep`` keeps stdout open, and
|
||||
# the test's own timeout races the sleep it is waiting out (US07-05).
|
||||
uploader=_uploader(tmp_path, 'echo "pid $$"; exec sleep 30'),
|
||||
uploader=_uploader(tmp_path, 'echo "pid $$"; sleep 30; exit 0'),
|
||||
)
|
||||
_album(sf, lib)
|
||||
(batch,) = _approved(sf, config)
|
||||
|
||||
@@ -15,10 +15,9 @@
|
||||
"tests/characterization/test_webapp_query.py"
|
||||
],
|
||||
"US01-02": [
|
||||
"tests/integration/test_app_lifecycle.py",
|
||||
"tests/integration/test_migrations.py",
|
||||
"tests/unit/test_config.py",
|
||||
"tests/unit/test_env_file.py"
|
||||
"tests/integration/test_migrations.py",
|
||||
"tests/integration/test_app_lifecycle.py"
|
||||
],
|
||||
"US01-03": [
|
||||
"tests/unit/test_path_policy.py",
|
||||
@@ -163,25 +162,6 @@
|
||||
"tests/integration/test_concurrency_races.py",
|
||||
"tests/integration/test_fault_matrix.py",
|
||||
"tests/e2e/test_crash_recovery.py"
|
||||
],
|
||||
"US07-05": [
|
||||
"tests/integration/test_backup_recovery.py",
|
||||
"tests/integration/test_diagnostics.py"
|
||||
],
|
||||
"US07-06": [
|
||||
"tests/integration/test_performance_budgets.py"
|
||||
],
|
||||
"US07-07": [
|
||||
"tests/e2e/test_release_gate.py",
|
||||
"tests/e2e/test_release_journey.py"
|
||||
]
|
||||
},
|
||||
"planned": [
|
||||
"US08-01",
|
||||
"US08-02",
|
||||
"US08-03",
|
||||
"US08-04",
|
||||
"US08-05"
|
||||
],
|
||||
"_planned_comment": "Accepted backlog stories that are not implemented yet. The release gate (US07-07) requires every story file to be either mapped to tests or listed here, so an unimplemented story is a visible decision rather than a hole in the matrix."
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,69 +0,0 @@
|
||||
"""Configuration from a dotenv file, including the archived CLI's variable names.
|
||||
|
||||
An operator who already has a ``photo_analyzer.env`` should not have to rewrite it
|
||||
to run the application it was replaced by. The file is standing configuration; the
|
||||
shell is what you meant this time, so the shell always wins.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from photo_pipeline.config import Config, load_env_file, parse_env_file
|
||||
|
||||
SAMPLE = """
|
||||
# The archived CLI's shape, comments and all.
|
||||
LLM_API_KEY=not-real
|
||||
LLM_BASE_URL="https://example.invalid/v1beta/openai/"
|
||||
LLM_MODEL='gemini-2.5-flash'
|
||||
LIBRARY=/tmp/pictures
|
||||
|
||||
MAX_WORKERS=4
|
||||
# commented=ignored
|
||||
malformed line without an equals sign
|
||||
"""
|
||||
|
||||
|
||||
def test_the_file_is_parsed_and_never_executed():
|
||||
values = parse_env_file(SAMPLE)
|
||||
|
||||
assert values["LLM_BASE_URL"] == "https://example.invalid/v1beta/openai/" # quotes stripped
|
||||
assert values["LLM_MODEL"] == "gemini-2.5-flash"
|
||||
assert values["MAX_WORKERS"] == "4"
|
||||
assert "commented" not in values and "malformed line without an equals sign" not in values
|
||||
|
||||
|
||||
def test_the_archived_cli_names_still_configure_the_application():
|
||||
values = parse_env_file(SAMPLE)
|
||||
|
||||
assert values["OPENAI_API_KEY"] == "not-real"
|
||||
assert values["OPENAI_BASE_URL"] == "https://example.invalid/v1beta/openai/"
|
||||
assert values["PHOTO_PIPELINE_LIBRARY_ROOTS"] == "/tmp/pictures"
|
||||
|
||||
|
||||
def test_an_explicit_shell_variable_beats_the_file(tmp_path, monkeypatch):
|
||||
path = tmp_path / "photo_analyzer.env"
|
||||
path.write_text(SAMPLE)
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "from-the-shell")
|
||||
monkeypatch.delenv("PHOTO_PIPELINE_LIBRARY_ROOTS", raising=False)
|
||||
|
||||
applied = load_env_file(path)
|
||||
|
||||
assert "OPENAI_API_KEY" not in applied, "the file overrode an exported value"
|
||||
assert os.environ["OPENAI_API_KEY"] == "from-the-shell"
|
||||
assert os.environ["PHOTO_PIPELINE_LIBRARY_ROOTS"] == "/tmp/pictures"
|
||||
assert Config.from_env().library_roots[0].name == "pictures"
|
||||
|
||||
|
||||
def test_the_file_is_found_through_its_variable(tmp_path, monkeypatch):
|
||||
path = tmp_path / "custom.env"
|
||||
path.write_text("PHOTO_PIPELINE_PORT=9123\n")
|
||||
monkeypatch.delenv("PHOTO_PIPELINE_PORT", raising=False)
|
||||
monkeypatch.setenv("PHOTO_PIPELINE_ENV_FILE", str(path))
|
||||
monkeypatch.chdir(tmp_path) # no ./.env here, so only the variable can find it
|
||||
|
||||
assert Config.from_env().port == 9123
|
||||
|
||||
|
||||
def test_a_missing_file_is_not_an_error(tmp_path):
|
||||
assert load_env_file(tmp_path / "nothing-here.env") == {}
|
||||
Reference in New Issue
Block a user