From 80ca38af0a84163ec59f33be2c98ec20981effef Mon Sep 17 00:00:00 2001 From: Antoine Vandeplanque Date: Wed, 9 Sep 2026 22:14:22 +0200 Subject: [PATCH] add dockerfile, docker config, pipeline and container entrypoint --- .dockerignore | 12 ++ Dockerfile | 30 ++++ PROJECT_PLAN.md | 7 + README.md | 36 +++- docs/DEPLOYMENT.md | 234 +++++++++++++++++++++++++ ops/app/.env.example | 17 ++ ops/app/compose.yaml | 58 ++++++ pyproject.toml | 1 + src/car_traker/collect.py | 61 +++++-- src/car_traker/container_entrypoint.py | 69 ++++++++ src/car_traker/pipeline.py | 106 +++++++++++ tests/test_browser_pipeline.py | 202 +++++++++++++++++++++ tests/test_collect_browser.py | 43 +++++ tests/test_container_entrypoint.py | 57 ++++++ tests/test_pipeline.py | 222 +++++++++++++++++++++++ 15 files changed, 1136 insertions(+), 19 deletions(-) create mode 100644 .dockerignore create mode 100644 Dockerfile create mode 100644 docs/DEPLOYMENT.md create mode 100644 ops/app/.env.example create mode 100644 ops/app/compose.yaml create mode 100644 src/car_traker/container_entrypoint.py create mode 100644 src/car_traker/pipeline.py create mode 100644 tests/test_browser_pipeline.py create mode 100644 tests/test_collect_browser.py create mode 100644 tests/test_container_entrypoint.py create mode 100644 tests/test_pipeline.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..b43cc58 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,12 @@ +# Allowlist the build context: credentials, data, Git history and tests stay out. +** +!Dockerfile +!pyproject.toml +!uv.lock +!README.md +!src/ +!src/car_traker/ +!src/car_traker/** +src/**/__pycache__/ +src/**/*.pyc +!searches.toml diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..102d631 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,30 @@ +FROM ghcr.io/astral-sh/uv:0.9.3 AS uv +FROM python:3.13-slim-bookworm + +COPY --from=uv /uv /usr/local/bin/uv +ENV UV_PROJECT_ENVIRONMENT=/opt/venv \ + UV_PYTHON_DOWNLOADS=never \ + UV_LINK_MODE=copy \ + PLAYWRIGHT_BROWSERS_PATH=/opt/playwright \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + HOME=/tmp \ + PATH=/opt/venv/bin:$PATH +WORKDIR /app + +COPY pyproject.toml uv.lock README.md ./ +RUN uv sync --locked --no-dev --no-install-project \ + && python -m playwright install --with-deps chromium \ + && apt-get update \ + && apt-get install -y --no-install-recommends xvfb \ + && rm -rf /var/lib/apt/lists/* /tmp/uv-cache /tmp/.cache \ + && useradd --uid 1000 --no-create-home app \ + && mkdir /data \ + && chown app:app /data +COPY src/ ./src/ +RUN uv sync --locked --no-dev --no-editable && rm -rf /tmp/.cache +COPY searches.toml ./searches.toml + +USER 1000:1000 +ENTRYPOINT ["python", "-m", "car_traker.container_entrypoint"] +CMD ["car-traker-pipeline", "--headed", "--config", "/app/searches.toml", "--output", "/data", "--family", "saab-900", "--max-pages", "1", "--max-details", "1"] diff --git a/PROJECT_PLAN.md b/PROJECT_PLAN.md index 7642be3..79a4b9e 100644 --- a/PROJECT_PLAN.md +++ b/PROJECT_PLAN.md @@ -24,6 +24,13 @@ PostgreSQL containers are disposable tests only. Homelab provisioning and remote connection setup remain user actions documented in `docs/POSTGRESQL.md`; no homelab database has been accessed or modified during implementation. +The manual homelab job is now packaged in `Dockerfile` and `ops/app/compose.yaml`. +`car-traker-pipeline` performs a database preflight, bounded collection, and an +idempotent import of the exact returned manifest, retaining partial coverage and +separate operational outcomes. There is no scheduler or automatic deployment. +See `docs/DEPLOYMENT.md`; container tests use isolated PostgreSQL 17 and synthetic +browser responses. Actual live access from the homelab still needs verification. + ## Purpose Build a French used-car asking-price observatory focused initially on youngtimers. diff --git a/README.md b/README.md index eca48ff..63583af 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,8 @@ First implementation: collect unnormalized listing observations from Leboncoin search pages and individual ads, including descriptions and source-provided locations. Finalized raw runs can now be cleaned and imported idempotently into PostgreSQL. -Scheduling and deployment are not implemented yet. +Manual Docker Compose execution is available; scheduling and automated deployment +remain deferred. Asking prices are not sale prices; disappearing listings do not prove a sale. ## Run @@ -51,12 +52,16 @@ src/car_traker/ # Installed Python package __main__.py # python -m car_traker entry point collect.py # Collector and CLI database.py # PostgreSQL migrations and import CLI + pipeline.py # Manual collect -> raw -> import job + container_entrypoint.py # Virtual display and graceful shutdown transform.py # Pure, conservative silver field cleaning migrations/ # Versioned SQL schema and analytical views tests/ # Offline tests docs/POSTGRESQL.md # Homelab provisioning, import, queries, database tests docs/CI.md # Gitea 1.23 CI and repository-scoped runner registration ops/ci-runner/ # Optional CI runner infrastructure, not app deployment +ops/app/ # Manual homelab job and runtime configuration example +docs/DEPLOYMENT.md # Storage, credentials, network access, and recovery searches.toml # Search configuration pyproject.toml # Dependencies, build configuration, and CLI entry point data/ # Local snapshots (ignored by Git) @@ -85,6 +90,35 @@ separate timestamps; source coverage and data-quality issues remain explicit. SQL views expose typed listing history, latest observations, price history, coverage, and validation issues. Generation/trim inference, redaction, and scheduling are deferred. +## Manual Homelab Job + +`car-traker-pipeline` checks database connectivity/schema, collects raw observations, +and imports that exact finalized run. Partial/failed collections remain marked as +such even when their import succeeds. A lock prevents overlapping pipeline jobs on +the same output directory, and `pipeline.json` records the import outcome separately +from the immutable raw manifest. + +For a laptop smoke run with the existing database connection configured: + +```sh +uv run car-traker-pipeline --headed --family saab-900 --max-pages 1 --max-details 1 +``` + +For Docker on the homelab, follow [manual deployment](docs/DEPLOYMENT.md) first to +configure private raw storage, the password file, and container network access. +Then, from the homelab checkout: + +```sh +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml build app +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app car-traker-db check +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app +``` + +The default image command is bounded to Saab 900, one search page and one detail +request. It uses headed Chromium through a virtual display, not a desktop/VNC +service. No PostgreSQL server, scheduler, or automatic migration is added. +Live Leboncoin access from the homelab still needs a manual smoke test. + ## Raw Format **Gzip-compressed JSON Lines (`.jsonl.gz`)**, with one observation per line and one diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md new file mode 100644 index 0000000..5e849dd --- /dev/null +++ b/docs/DEPLOYMENT.md @@ -0,0 +1,234 @@ +# Manual Homelab Deployment + +## Scope +The next milestone is one manual **collection -> raw files -> PostgreSQL import** +job on the homelab. Development stays on the laptop; build and run from a repository +checkout on the existing Debian VM, alongside Gitea, its runner, and PostgreSQL 17. +Use the existing database and role, currently both `car_tracker`. +Do not provision another production PostgreSQL server or recreate those objects. + +`ops/app/compose.yaml` is the one-shot application stack, separate from +`ops/ci-runner`. There is no scheduling, restart loop, or automatic CI deployment. +Run commands from the homelab checkout's root with Docker Engine and Compose +available. Review the revision and `searches.toml`; do not run these in CI. +After these changes are published, update the homelab checkout before building. + +## Private Runtime Configuration +Use an external, nonsecret Compose env file, for example +`/srv/car-tracker/app.env`. These absolute paths are examples, not secrets. +Inspect existing files first: never overwrite an existing env file or `.pgpass`. +Use `ops/app/.env.example` as the reference when creating a new file locally: + +```dotenv +APP_UID=1000 +APP_GID=1000 +RAW_DATA_DIR=/srv/car-tracker/raw +DB_CONFIG_DIR=/srv/car-tracker/db +PGHOST=postgres.your-lan.example +PGPORT=5432 +PGDATABASE=car_tracker +PGUSER=car_tracker +PGSSLMODE=require +APP_SUBNET=172.30.17.0/28 +``` + +Replace the hostname with the existing VM's reachable LAN/VPN hostname and choose +the numeric `APP_UID`/`APP_GID` that will own the mounted files. Keep the application +user non-root. Both bind directories must already exist; Compose refuses to create +missing source directories. After inspecting `/srv` and any existing application +directories, create only missing directories and apply permissions deliberately: + +```sh +sudo install -d -m 0700 -o 1000 -g 1000 /srv/car-tracker/raw /srv/car-tracker/db +``` + +Use your chosen UID/GID instead of `1000` in all ownership commands. Both +`RAW_DATA_DIR` and `DB_CONFIG_DIR` must have mode `0700` and the correct ownership. +The raw directory must be writable by the container user; the DB directory must be +traversable by that user even though its container mount is read-only. + +Create or carefully edit `/srv/car-tracker/db/pgpass` locally, using a private +editor/session. Its matching entry has this form (do not use the placeholder): + +```text +postgres.your-lan.example:5432:car_tracker:car_tracker: +``` + +Escape `:` and `\` in the password according to libpq password-file rules. +Do not put passwords in commands, the env file, Git, logs, or chat. Do not send +tokens or credentials to anyone to complete this setup. Preserve existing password +files and entries rather than replacing them blindly. Protect the dedicated file: + +```sh +sudo chown 1000:1000 /srv/car-tracker/db/pgpass +sudo chmod 0600 /srv/car-tracker/db/pgpass +``` + +The password is supplied only through `/run/db/pgpass`, mounted from +`DB_CONFIG_DIR`. Incorrect ownership prevents reads; libpq ignores a password file +with overly permissive permissions. Keep this directory dedicated to app DB files, +not unrelated credentials. + +Compose interpolates the external env file but passes only the runtime whitelist: +`PGHOST`, `PGPORT`, `PGDATABASE`, `PGUSER`, and `PGSSLMODE`, plus fixed +`PGPASSFILE=/run/db/pgpass` and `PGSSLROOTCERT=/run/db/root.crt`. +It does not pass through `DATABASE_URL`, `PGPASSWORD`, or arbitrary host variables. +Shell variables can override Compose env-file interpolation; unset stale matching +variables before using this configuration. Always specify `--env-file` explicitly. + +## TLS And Network Access +`PGSSLMODE` defaults to `require`: require encryption, but do not treat this as +server identity verification. Prefer `verify-full` with the trusted CA certificate +at `/srv/car-tracker/db/root.crt`, readable by the application user. The existing +directory mount exposes it as `/run/db/root.crt`; `PGHOST` must match the server +certificate. Configure PostgreSQL server TLS before running. Never fix production +connectivity by disabling TLS or adding `trust` authentication. + +`localhost` inside the container is not the VM's PostgreSQL server. Prefer the VM's +LAN/VPN hostname. Compose also provides `host.docker.internal` through +`extra_hosts: host-gateway`, but PostgreSQL must listen on that gateway address if +you use it. A listener only on loopback or the LAN address does not necessarily +accept gateway connections. Certificate hostname matching still applies. + +The app has a dedicated bridge subnet, default `172.30.17.0/28`. Check existing +Docker, LAN, and VPN routes for conflicts before starting; configure `APP_SUBNET` +to a non-overlapping range if needed. Coordinate any network changes with the +existing runner and services. Do not expose PostgreSQL publicly. + +PostgreSQL may see a container client address different from the laptop's address. +The laptop's working HBA rule is therefore not proof this job can connect. Inspect +the actual client address in PostgreSQL logs and existing firewall/HBA rules. A new +narrow rule may be needed for the default app subnet: + +```text +hostssl car_tracker car_tracker 172.30.17.0/28 scram-sha-256 +``` + +Use the configured subnet and actual observed addressing, not a blindly copied +rule. HBA rules are first-match; preserve unrelated rules and permit only the +necessary source/port through the firewall. Check `pg_hba_file_rules` for errors +and reload authentication changes with `SELECT pg_reload_conf();` as an admin. +Changing `listen_addresses` requires a PostgreSQL restart: coordinate downtime for +the shared VM's clients rather than restarting casually. See +[PostgreSQL troubleshooting](POSTGRESQL.md#connection-troubleshooting) for details; +that document's `car_traker` names are examples, not this VM's `car_tracker` names. + +## Build And Smoke Run +Validate without printing interpolated configuration, then build on the homelab: + +```sh +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml config --quiet +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml build app +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app car-traker-db check +``` + +`check` is read-only connectivity testing, not a schema or migration-privilege +check. The pipeline additionally checks access to `silver.collection_runs` before +opening the browser. No new migrations are required for this milestone. If the +existing packaged migrations have not yet been applied, apply them explicitly: + +```sh +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app car-traker-db migrate +``` + +Migrations are not automatic. Once DB access/schema are ready, run the image's +default bounded headed pipeline: Saab 900, one page per selected search, at most +one unique detail request for the run, with the default five-second request delay. + +```sh +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app +``` + +Inspect the reported manifest and import outcome before expanding. After a usable +smoke run, this is the explicit larger manual command: + +```sh +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app \ + car-traker-pipeline --headed --config /app/searches.toml --output /data \ + --family saab-900 --max-pages 3 --max-details 5 +``` + +Limits can intentionally produce partial coverage. Do not loop or retry a blocked +source automatically; stop and investigate access failures before another scrape. + +## Outcomes And Recovery +The pipeline prints `Manifest: /data///manifest.json`, reports a JSON +summary after successful import, and imports +that exact finalized run, including partial or failed collection runs. Host files +are under `/srv/car-tracker/raw///`. Import success does not imply +complete source coverage. Keep finalized raw files and manifests immutable. + +| Pipeline exit | Meaning | +| --- | --- | +| 0 | Complete collection imported, without rejected records | +| 1 | Incomplete collection imported, without rejected records | +| 2 | Import or replay has rejected records; inspect validation issues | +| 3 | Operational, configuration, locking, or database failure | +| 130 | Cancellation/interruption | + +`pipeline.json` is a separate operational sidecar, not a rewrite of raw coverage. +It records `collection_status`, `import_status`, and an import summary on success. +An interrupted import or lost DB response can leave `import_status: unknown`: +the transaction may already have committed. Recover by idempotently replaying the +exact manifest with the DB CLI, not by rerunning the scrape. Substitute the exact +date/run ID printed by the pipeline: + +```sh +docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app \ + car-traker-db import '/data///manifest.json' +``` + +An identical replay reports `already_imported` without duplicating observations. +The standalone DB CLI does not update `pipeline.json`; retain its replay result +separately. Its exit codes differ; see [import documentation](POSTGRESQL.md#migrate-and-import). +Do not alter raw files to force a replay past integrity checks. + +`browser_cleanup.json` may record a nonfatal browser teardown error after raw +finalization; the pipeline can still import that run. Allow graceful cancellation +to finish. A hard kill, OOM, or expired stop grace can leave a `running` manifest, +no sidecar, or stale `running` import state. A `running` manifest is not imported; +investigate rather than marking it complete or assuming a DB rollback occurred. + +The output-root `.pipeline.lock` covers the whole manual pipeline job, including +collection and import. Do not delete an active lock file; the file can remain after +exit and its presence alone does not mean a job is active. Standalone collection +and DB CLIs do not honor this lock. Never overlap them with a pipeline or other +manual job using the same output root, including during recovery. + +## Runtime Boundaries + +Compose runs non-root with a read-only root filesystem, all capabilities dropped, +and `no-new-privileges`. There is no Docker socket mount or published port. +Runtime limits are 1.5 CPUs, 2 GiB memory, 256 PIDs, 512 MiB shared memory, +512 MiB `/tmp` tmpfs, a 1 MiB X11 socket tmpfs, and a 30-second stop grace period. These do not constrain +image builds. Keep load low on the VM shared with Gitea, its runner, and PostgreSQL. + +The image installs Chromium matching the locked Playwright version. Its custom +Python entrypoint starts Xvfb for headed operation, forwards SIGTERM to the child, +and waits for child cleanup before removing remaining processes and the display. +No external GUI or noVNC service is needed. + +Chromium's sandbox remains disabled by Playwright's default launch behavior; an +explicit sandbox trial failed under host security restrictions. These container +settings are not a claim of secure browser isolation. Do not add privileged mode, +`SYS_ADMIN`, or `seccomp=unconfined` to work around this. The browser container can +read mounted DB credentials and shares the VM kernel. Prefer a separate VM for +future browser isolation rather than weakening the shared host's protections. + +## Verification And Retention + +Live source access from this homelab has not been verified here. All 109 tests +passed inside the built non-root image with its read-only filesystem and resource +limits, using isolated PostgreSQL 17 and synthetic browser responses. An actual +Docker stop also allowed the child to finish cleanup and exit 130 before the +display was removed. These tests do not guarantee live source access. +For optional testing, see [PostgreSQL tests](POSTGRESQL.md#tests) and +`tests/test_browser_pipeline.py`. Browser tests require `RUN_BROWSER_TESTS=1` and +`TEST_DATABASE_URL`; use only a disposable database explicitly ending in `_test`, +never homelab production. Tests truncate data. Default CI skips browser tests. + +Retention is still manual: start with 30 days for raw files and private database +copies, including descriptions and raw payloads, and account for relevant backups. +Deleting raw files does not delete DB copies. No retention or backup automation is +provided here; arrange and verify private backups manually before relying on the +data. Keep raw files, DB copies, and backups private and outside Git/CI artifacts. diff --git a/ops/app/.env.example b/ops/app/.env.example new file mode 100644 index 0000000..16b7d19 --- /dev/null +++ b/ops/app/.env.example @@ -0,0 +1,17 @@ +# Nonsecret Compose settings. Use a separate runtime env file on the homelab. +# Do not put passwords or DATABASE_URL in this file. +APP_UID=1000 +APP_GID=1000 +RAW_DATA_DIR=/srv/car-tracker/raw +DB_CONFIG_DIR=/srv/car-tracker/db + +# Use the existing VM's reachable PostgreSQL hostname/IP, not container localhost. +PGHOST=postgres.your-lan.example +PGPORT=5432 +PGDATABASE=car_tracker +PGUSER=car_tracker + +# Encryption required. Prefer verify-full once root.crt and hostname are configured. +PGSSLMODE=require +# Check this subnet does not overlap existing Docker/LAN/VPN networks. +APP_SUBNET=172.30.17.0/28 diff --git a/ops/app/compose.yaml b/ops/app/compose.yaml new file mode 100644 index 0000000..783f734 --- /dev/null +++ b/ops/app/compose.yaml @@ -0,0 +1,58 @@ +name: car-traker-app + +services: + app: + build: + context: ../.. + dockerfile: Dockerfile + image: car-traker:local + # One-shot manual job only. Never restart a blocked scraper automatically. + restart: "no" + init: true + stop_grace_period: 30s + user: "${APP_UID:-1000}:${APP_GID:-1000}" + read_only: true + cap_drop: [ALL] + security_opt: + - no-new-privileges:true + cpus: 1.5 + mem_limit: 2g + pids_limit: 256 + shm_size: 512m + tmpfs: + - /tmp:rw,nosuid,nodev,size=512m,mode=1777 + - /tmp/.X11-unix:rw,nosuid,nodev,size=1m,mode=1777 + environment: + PGHOST: "${PGHOST:?Set the existing PostgreSQL hostname or IP}" + PGPORT: "${PGPORT:-5432}" + PGDATABASE: "${PGDATABASE:?Set the application database name}" + PGUSER: "${PGUSER:?Set the application database role}" + PGSSLMODE: "${PGSSLMODE:-require}" + PGPASSFILE: /run/db/pgpass + PGSSLROOTCERT: /run/db/root.crt + extra_hosts: + - "host.docker.internal:host-gateway" + volumes: + - type: bind + source: "${RAW_DATA_DIR:?Set an existing private raw storage directory}" + target: /data + bind: + create_host_path: false + - type: bind + source: "${DB_CONFIG_DIR:?Set an existing private directory containing pgpass}" + target: /run/db + read_only: true + bind: + create_host_path: false + - type: bind + source: ../../searches.toml + target: /app/searches.toml + read_only: true + networks: + app: + +networks: + app: + ipam: + config: + - subnet: "${APP_SUBNET:-172.30.17.0/28}" diff --git a/pyproject.toml b/pyproject.toml index 205e4cc..7126cbd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,7 @@ select = ["E4", "E7", "E9", "F"] [project.scripts] car-traker = "car_traker.collect:main" car-traker-db = "car_traker.database:main" +car-traker-pipeline = "car_traker.pipeline:main" [build-system] requires = ["setuptools>=77"] diff --git a/src/car_traker/collect.py b/src/car_traker/collect.py index 54c5083..3f697a9 100644 --- a/src/car_traker/collect.py +++ b/src/car_traker/collect.py @@ -4,6 +4,7 @@ import argparse import gzip import json import math +import sys import time import tomllib import uuid @@ -257,8 +258,8 @@ def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, m return path, manifest["status"] -def main(): - parser = argparse.ArgumentParser(description=__doc__) +def parse_collection_args(argv=None, description=None): + parser = argparse.ArgumentParser(description=description or __doc__) parser.add_argument("--config", type=Path, default=Path("searches.toml")) parser.add_argument("--output", type=Path, default=Path("data/raw/leboncoin")) parser.add_argument("--family", help="Collect only this configured family") @@ -267,7 +268,7 @@ def main(): parser.add_argument("--headed", action="store_true") parser.add_argument("--search-only", action="store_true", help="Skip individual listing pages") parser.add_argument("--max-details", type=int, help="Maximum unique detail requests per run") - args = parser.parse_args() + args = parser.parse_args(argv) if args.max_pages < 1 or not math.isfinite(args.delay) or args.delay < 1: parser.error("max-pages must be positive and delay must be finite and >= 1 second") if args.max_details is not None and args.max_details < 1: @@ -284,26 +285,50 @@ def main(): searches = [s for s in searches if s["family"] == args.family] if not searches: parser.error("no matching searches") + args.searches = searches + return args + +def collect_browser(args): from playwright.sync_api import sync_playwright - with sync_playwright() as playwright: - browser = playwright.chromium.launch(headless=not args.headed) + result = None + try: + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=not args.headed) + try: + context = browser.new_context(locale="fr-FR") + page = context.new_page() + + def fetch(url): + response = page.goto(url, wait_until="domcontentloaded", timeout=60_000) + if response is None or response.status != 200: + raise SourceAccessError(response.status if response else None) + return page.locator("script#__NEXT_DATA__").text_content(timeout=15_000) + + result = collect( + fetch, args.searches, args.output, args.max_pages, args.delay, + args.search_only, args.max_details, + ) + finally: + browser.close() + except KeyboardInterrupt: + if result is not None: + print(f"Finalized manifest before interruption: {result[0]}", file=sys.stderr) + raise + except Exception as exc: + if result is None: + raise + # Browser teardown must not discard an already-finalized raw run. + print(f"Browser cleanup failed ({type(exc).__name__}); raw run is finalized.", file=sys.stderr) try: - context = browser.new_context(locale="fr-FR") - page = context.new_page() + write_json(result[0].parent / "browser_cleanup.json", {"error_type": type(exc).__name__}) + except OSError: + print("Could not record browser cleanup failure.", file=sys.stderr) + return result - def fetch(url): - response = page.goto(url, wait_until="domcontentloaded", timeout=60_000) - if response is None or response.status != 200: - raise SourceAccessError(response.status if response else None) - return page.locator("script#__NEXT_DATA__").text_content(timeout=15_000) - path, status = collect( - fetch, searches, args.output, args.max_pages, args.delay, - args.search_only, args.max_details, - ) - finally: - browser.close() +def main(): + path, status = collect_browser(parse_collection_args()) print(f"{status}: {path}") return 0 if status == "complete" else 1 diff --git a/src/car_traker/container_entrypoint.py b/src/car_traker/container_entrypoint.py new file mode 100644 index 0000000..247b751 --- /dev/null +++ b/src/car_traker/container_entrypoint.py @@ -0,0 +1,69 @@ +"""Keep the virtual display alive until the command finishes handling shutdown.""" + +import os +import select +import signal +import subprocess +import sys + + +def main(): + if len(sys.argv) < 2: + print("A container command is required.", file=sys.stderr) + return 3 + display = child = None + stopped = None + + def forward(signum, frame): + nonlocal stopped + stopped = signum + if child is not None and child.poll() is None: + child.send_signal(signum) + + previous = {sig: signal.signal(sig, forward) for sig in (signal.SIGINT, signal.SIGTERM)} + reader, writer = os.pipe() + try: + display = subprocess.Popen([ + "Xvfb", "-displayfd", str(writer), "-screen", "0", "1280x1024x24", "-nolisten", "tcp", + ], pass_fds=(writer,), start_new_session=True) + os.close(writer) + writer = None + ready, _, _ = select.select([reader], [], [], 10) + if stopped: + return 128 + stopped + number = os.read(reader, 32).decode("ascii").strip() if ready else "" + if not number.isdigit(): + print("Virtual display did not start.", file=sys.stderr) + return 3 + child = subprocess.Popen(sys.argv[1:], env={**os.environ, "DISPLAY": f":{number}"}, + start_new_session=True) + if stopped: + child.send_signal(stopped) + code = child.wait() + return code if code >= 0 else 128 - code + except (OSError, ValueError): + print("Could not start the virtual display or container command.", file=sys.stderr) + return 3 + finally: + # Only clean up descendants after the command has finalized its files/transaction. + if child is not None: + try: + os.killpg(child.pid, signal.SIGKILL) + except ProcessLookupError: + pass + if display is not None: + display.terminate() + try: + display.wait(timeout=5) + except subprocess.TimeoutExpired: + display.kill() + display.wait() + os.close(reader) + if writer is not None: + os.close(writer) + for sig, handler in previous.items(): + signal.signal(sig, handler) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/car_traker/pipeline.py b/src/car_traker/pipeline.py new file mode 100644 index 0000000..06bef08 --- /dev/null +++ b/src/car_traker/pipeline.py @@ -0,0 +1,106 @@ +"""Collect raw observations and import that exact run into PostgreSQL.""" + +from contextlib import redirect_stderr +import fcntl +import io +import json +import os +import signal +import sys + +import psycopg + +from .collect import collect_browser, parse_collection_args, write_json +from .database import InputError, database_error_hint, import_run + + +def run_pipeline(args): + """Run one locked collection/import job and return its CLI exit code.""" + try: + if not any(os.environ.get(key) for key in ( + "DATABASE_URL", "PGSERVICE", "PGHOST", "PGHOSTADDR", + )): + raise InputError("database_configuration_required") + conninfo = os.environ.get("DATABASE_URL", "") + args.output.mkdir(mode=0o700, parents=True, exist_ok=True) + # Keep the inode: unlinking would let another process lock a different file. + with (args.output / ".pipeline.lock").open("a") as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + print("Another pipeline job holds the output lock.", file=sys.stderr) + return 3 + sidecar = None + state = None + try: + with psycopg.connect(conninfo, connect_timeout=10, autocommit=True) as conn: + conn.execute("SELECT 1 FROM silver.collection_runs LIMIT 0") + path, status = collect_browser(args) + print(f"Manifest: {path}", flush=True) + sidecar = path.parent / "pipeline.json" + state = {"schema_version": 1, "collection_status": status, "import_status": "not_attempted"} + manifest = json.loads(path.read_text(encoding="utf-8")) + if manifest.get("error_type") == "KeyboardInterrupt": + raise KeyboardInterrupt + write_json(sidecar, state) + state["import_status"] = "running" + write_json(sidecar, state) + with psycopg.connect(conninfo, connect_timeout=10, autocommit=True) as conn: + summary = import_run(conn, path) + state.update(import_status="succeeded", import_summary=summary) + write_json(sidecar, state) + print(json.dumps(state), flush=True) + if summary.get("rejected", 0): + return 2 + return 0 if status == "complete" else 1 + except (Exception, KeyboardInterrupt) as exc: + if state is not None: + if state["import_status"] == "running": + # A lost response or interruption can occur after PostgreSQL commits. + state["import_status"] = "unknown" + state["error_type"] = type(exc).__name__ + try: + write_json(sidecar, state) + except Exception: + print("Could not write pipeline outcome; raw files are retained.", file=sys.stderr) + raise + except KeyboardInterrupt: + print("Pipeline interrupted; raw files are retained.", file=sys.stderr) + return 130 + except InputError: + print("Pipeline input/configuration error; check configuration and raw files.", file=sys.stderr) + except psycopg.Error as exc: + print(f"Database error: {database_error_hint(exc)}", file=sys.stderr) + except Exception: + print("Pipeline failed; check browser, configuration and storage. Raw files are retained.", file=sys.stderr) + return 3 + + +def main(argv=None): + def interrupted(signum, frame): + raise KeyboardInterrupt + + # Docker stops one-shot jobs with SIGTERM; let the collector finalize coverage. + previous = signal.signal(signal.SIGTERM, interrupted) + try: + try: + # argparse can echo arbitrary argument values; keep configuration errors private. + with redirect_stderr(io.StringIO()): + args = parse_collection_args(argv, description=__doc__) + except KeyboardInterrupt: + return 130 + except SystemExit as exc: + if exc.code == 0: + return 0 + print("Invalid pipeline arguments or collection configuration.", file=sys.stderr) + return 3 + except Exception: + print("Could not load collection configuration.", file=sys.stderr) + return 3 + return run_pipeline(args) + finally: + signal.signal(signal.SIGTERM, previous) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_browser_pipeline.py b/tests/test_browser_pipeline.py new file mode 100644 index 0000000..1ab47e8 --- /dev/null +++ b/tests/test_browser_pipeline.py @@ -0,0 +1,202 @@ +"""Opt-in headed Chromium tests against a disposable *_test PostgreSQL database.""" + +from contextlib import redirect_stderr, redirect_stdout +from decimal import Decimal +import gzip +import io +import json +import os +from pathlib import Path +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import patch + +from psycopg import sql + +from car_traker import database, pipeline +from car_traker.collect import search_url +import test_database_integration as database_tests + + +@unittest.skipUnless( + os.environ.get("RUN_BROWSER_TESTS") == "1" and os.environ.get("TEST_DATABASE_URL"), + "RUN_BROWSER_TESTS=1 and TEST_DATABASE_URL are required", +) +class BrowserPipelineTests(unittest.TestCase): + tables = database_tests.DatabaseIntegrationTests.tables + + @classmethod + def setUpClass(cls): + cls.conn = database_tests.DatabaseIntegrationTests.connect() + cls.addClassCleanup(cls.conn.close) + database.migrate(cls.conn) + + def setUp(self): + self.conn.execute(sql.SQL("TRUNCATE {} RESTART IDENTITY").format( + sql.SQL(", ").join(sql.Identifier("silver", table) for table in self.tables) + )) + temporary = TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.directory = Path(temporary.name) + self.config = self.directory / "searches.toml" + self.config.write_text( + '[[searches]]\nfamily = "synthetic"\ntext = "synthetic car"\n', + encoding="utf-8", + ) + self.output = self.directory / "raw" + self.search = {"list_id": 123, "price": ["12000"]} + self.detail = { + "list_id": 123, "price": ["12500.50"], + "body": "Synthetic private description.\nRetain this raw text exactly.", + "location": {"lat": 48.125, "lng": 2.375}, + } + + def run_browser_pipeline(self, *, detail_status=200, import_status="succeeded"): + # Import only when opted in; never replace the browser, collector or importer. + from playwright.sync_api import BrowserContext + + original_new_page = BrowserContext.new_page + search = search_url("synthetic car", 1) + detail = "https://www.leboncoin.fr/ad/voitures/123" + requests, responses, unexpected = [], [], [] + + def handle(route): + url = route.request.url + requests.append(url) + if url == search: + status = 200 + props = {"searchData": {"ads": [self.search], "total": 1, "max_pages": 1}} + elif url == detail: + status = detail_status + # Even the 403 has valid data: ignoring HTTP status must fail this test. + props = {"ad": self.detail} + else: + unexpected.append(url) + route.abort() + return + data = json.dumps({"props": {"pageProps": props}}) + route.fulfill( + status=status, content_type="text/html", + body=f'', + ) + + def new_page(context, *args, **kwargs): + page = original_new_page(context, *args, **kwargs) + page.route("**/*", handle) + page.on("response", lambda response: responses.append((response.url, response.status))) + return page + + stdout, stderr = io.StringIO(), io.StringIO() + # Preserve DISPLAY/browser paths, but discard inherited libpq/homelab settings. + environment = {key: value for key, value in os.environ.items() if not key.startswith("PG")} + environment["DATABASE_URL"] = os.environ["TEST_DATABASE_URL"] + with patch.dict(os.environ, environment, clear=True), patch.object( + BrowserContext, "new_page", new_page, + ), patch("car_traker.collect.time.sleep"), redirect_stdout(stdout), redirect_stderr(stderr): + code = pipeline.main([ + "--headed", "--config", str(self.config), "--output", str(self.output), + "--delay", "1", "--max-pages", "1", "--max-details", "1", + ]) + + self.assertEqual(unexpected, []) + self.assertEqual(requests, [search, detail]) + self.assertEqual(responses, [(search, 200), (detail, detail_status)]) + self.assertNotIn(self.detail["body"], stdout.getvalue() + stderr.getvalue()) + self.assertNotIn("48.125", stdout.getvalue() + stderr.getvalue()) + self.assertNotIn("2.375", stdout.getvalue() + stderr.getvalue()) + paths = list(self.output.rglob("manifest.json")) + self.assertEqual(len(paths), 1) + path = paths[0] + manifest = json.loads(path.read_text(encoding="utf-8")) + state = json.loads((path.parent / "pipeline.json").read_text(encoding="utf-8")) + self.assertEqual(state["collection_status"], manifest["status"]) + self.assertEqual(state["import_status"], import_status) + if import_status == "succeeded": + self.assertEqual(state["import_summary"]["status"], "imported") + self.assertEqual(state["import_summary"]["source_status"], manifest["status"]) + self.assertEqual(state["import_summary"]["rejected"], 0) + self.assertEqual(self.conn.execute( + "SELECT run_id::text, source_status, manifest FROM silver.collection_runs" + ).fetchall(), [(manifest["run_id"], manifest["status"], manifest)]) + self.assertEqual(self.conn.execute( + "SELECT family, status, pages_observed, records_reported FROM silver.search_coverage" + ).fetchall(), [("synthetic", "complete", 1, 1)]) + with gzip.open(path.parent / manifest["searches"][0]["pages"][0]["file"], "rt") as stream: + records = [json.loads(line) for line in stream] + self.assertEqual(len(records), 1) + self.assertEqual(records[0]["record_type"], "search") + self.assertEqual(records[0]["payload"], self.search) + return code, path, manifest + + def test_complete_run_preserves_raw_and_typed_detail_and_replays(self): + code, path, manifest = self.run_browser_pipeline() + self.assertEqual(code, 0) + self.assertEqual(manifest["status"], "complete") + with gzip.open(path.parent / manifest["details"]["listings"]["123"]["file"], "rt") as stream: + records = [json.loads(line) for line in stream] + self.assertEqual(len(records), 1) + self.assertEqual(records[0]["record_type"], "detail") + self.assertEqual(records[0]["payload"], self.detail) + self.assertEqual(self.conn.execute( + "SELECT source_listing_id, record_type, price_eur, currency, description, " + "latitude, longitude, raw_payload FROM silver.observations ORDER BY record_type" + ).fetchall(), [ + ("123", "detail", Decimal("12500.50"), "EUR", self.detail["body"], + 48.125, 2.375, self.detail), + ("123", "search", Decimal("12000"), "EUR", None, None, None, self.search), + ]) + self.assertEqual(self.conn.execute( + "SELECT source_listing_id, status FROM silver.detail_coverage" + ).fetchall(), [("123", "complete")]) + before = self.state() + self.assertEqual({table: len(rows) for table, rows in before.items()}, { + "collection_runs": 1, "raw_files": 2, "search_coverage": 1, + "detail_coverage": 1, "listings": 1, "observations": 2, "rejected_records": 0, + }) + with redirect_stdout(io.StringIO()): + replay = database.import_run(self.conn, path) + self.assertEqual(replay, { + "run_id": manifest["run_id"], "status": "already_imported", "rejected": 0, + }) + self.assertEqual(self.state(), before) + + def test_blocked_detail_imports_partial_run_and_retains_search(self): + code, path, manifest = self.run_browser_pipeline(detail_status=403) + self.assertEqual(code, 1) + self.assertEqual(manifest["status"], "partial") + self.assertEqual(manifest["error_type"], "SourceAccessError") + self.assertEqual(manifest["details"]["status"], "partial") + self.assertEqual(manifest["details"]["listings"]["123"]["http_status"], 403) + self.assertFalse((path.parent / "detail-123.jsonl.gz").exists()) + self.assertEqual(self.conn.execute( + "SELECT source_listing_id, status, reason, http_status FROM silver.detail_coverage" + ).fetchall(), [("123", "failed", "SourceAccessError", 403)]) + self.assertEqual(self.conn.execute( + "SELECT source_listing_id, record_type, price_eur, raw_payload FROM silver.observations" + ).fetchall(), [("123", "search", Decimal("12000"), self.search)]) + self.assertEqual(self.conn.execute( + "SELECT record_type, expected_records, accepted_records, rejected_records FROM silver.raw_files" + ).fetchall(), [("search", 1, 1, 0)]) + self.assertEqual(self.conn.execute("SELECT count(*) FROM silver.rejected_records").fetchone(), (0,)) + + def state(self): + return {table: self.conn.execute(sql.SQL( + "SELECT to_jsonb(t) FROM silver.{} t ORDER BY to_jsonb(t)::text" + ).format(sql.Identifier(table))).fetchall() for table in self.tables} + + def test_interrupt_after_commit_is_unknown_and_replay_is_safe(self): + def committed_then_interrupted(conn, path): + database.import_run(conn, path) + raise KeyboardInterrupt + + with patch.object(pipeline, "import_run", side_effect=committed_then_interrupted): + code, path, _ = self.run_browser_pipeline(import_status="unknown") + self.assertEqual(code, 130) + before = self.state() + self.assertEqual(len(before["observations"]), 2) + self.assertEqual(database.import_run(self.conn, path)["status"], "already_imported") + self.assertEqual(self.state(), before) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_collect_browser.py b/tests/test_collect_browser.py new file mode 100644 index 0000000..dfacf9d --- /dev/null +++ b/tests/test_collect_browser.py @@ -0,0 +1,43 @@ +from argparse import Namespace +import io +from pathlib import Path +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import patch + +from car_traker.collect import collect_browser + + +class BrowserCleanupTests(unittest.TestCase): + def setUp(self): + temporary = TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.path = Path(temporary.name) / "manifest.json" + self.path.write_text('{"status":"complete"}') + self.args = Namespace(headed=False, searches=[], output=self.path.parent, + max_pages=1, delay=1, search_only=True, max_details=None) + self.playwright = self.enterContext(patch("playwright.sync_api.sync_playwright")) + self.browser = self.playwright.return_value.__enter__.return_value.chromium.launch.return_value + self.collect = self.enterContext(patch("car_traker.collect.collect", return_value=(self.path, "complete"))) + self.stderr = self.enterContext(patch("sys.stderr", new_callable=io.StringIO)) + + def test_browser_close_failure_retains_finalized_result(self): + self.browser.close.side_effect = RuntimeError("SYNTHETIC_SECRET") + self.assertEqual(collect_browser(self.args), (self.path, "complete")) + self.assertTrue((self.path.parent / "browser_cleanup.json").exists()) + self.assertNotIn("SYNTHETIC_SECRET", self.stderr.getvalue()) + + def test_driver_cleanup_failure_retains_finalized_result(self): + self.playwright.return_value.__exit__.side_effect = RuntimeError("cleanup failed") + self.assertEqual(collect_browser(self.args), (self.path, "complete")) + + def test_collection_failure_is_not_suppressed(self): + self.collect.side_effect = OSError("write failed") + with self.assertRaises(OSError): + collect_browser(self.args) + + def test_cleanup_interruption_remains_cancellation_with_manifest_path(self): + self.browser.close.side_effect = KeyboardInterrupt + with self.assertRaises(KeyboardInterrupt): + collect_browser(self.args) + self.assertIn(str(self.path), self.stderr.getvalue()) diff --git a/tests/test_container_entrypoint.py b/tests/test_container_entrypoint.py new file mode 100644 index 0000000..9c4aad2 --- /dev/null +++ b/tests/test_container_entrypoint.py @@ -0,0 +1,57 @@ +import io +import os +import select +import signal +import subprocess +import sys +import unittest +from unittest.mock import patch + +from car_traker import container_entrypoint + + +class ContainerEntrypointTests(unittest.TestCase): + def test_requires_a_command(self): + with patch("sys.argv", ["entrypoint"]), patch("sys.stderr", new_callable=io.StringIO): + self.assertEqual(container_entrypoint.main(), 3) + + +@unittest.skipUnless(os.environ.get("RUN_BROWSER_TESTS") == "1", "Xvfb container tests are opt-in") +class VirtualDisplayTests(unittest.TestCase): + def test_child_exit_code_is_preserved(self): + result = subprocess.run([ + sys.executable, "-m", "car_traker.container_entrypoint", + sys.executable, "-c", "raise SystemExit(7)", + ], capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 7, result.stderr) + + def test_sigterm_waits_for_child_finalization(self): + command = """ +import signal, sys, time +def stop(signum, frame): + print('cleanup-start', flush=True) + time.sleep(1) + print('cleanup-done', flush=True) + sys.exit(130) +signal.signal(signal.SIGTERM, stop) +print('ready', flush=True) +while True: + signal.pause() +""" + process = subprocess.Popen([ + sys.executable, "-m", "car_traker.container_entrypoint", + sys.executable, "-c", command, + ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + ready, _, _ = select.select([process.stdout], [], [], 15) + self.assertTrue(ready, "Child did not become ready") + self.assertEqual(process.stdout.readline().strip(), "ready") + process.send_signal(signal.SIGTERM) + stdout, stderr = process.communicate(timeout=15) + self.assertEqual(process.returncode, 130, stderr) + self.assertIn("cleanup-start", stdout) + self.assertIn("cleanup-done", stdout) + finally: + if process.poll() is None: + process.terminate() + process.communicate(timeout=15) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..924c866 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,222 @@ +import argparse +from contextlib import redirect_stderr, redirect_stdout +import fcntl +import io +import json +import os +import signal +from pathlib import Path +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +from car_traker import pipeline + + +class PipelineTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.output = Path(self.directory.name) / "raw" + self.args = argparse.Namespace(output=self.output, searches=[]) + self.stdout, self.stderr = io.StringIO(), io.StringIO() + self.enterContext(redirect_stdout(self.stdout)) + self.enterContext(redirect_stderr(self.stderr)) + self.enterContext(patch.dict(os.environ, {"PGHOST": "homelab"}, clear=True)) + self.preflight, self.import_connection = MagicMock(), MagicMock() + self.preflight.__enter__.return_value = self.preflight + self.import_connection.__enter__.return_value = self.import_connection + self.connect = self.enterContext(patch.object( + pipeline.psycopg, "connect", side_effect=[self.preflight, self.import_connection], + )) + self.collect = self.enterContext(patch.object(pipeline, "collect_browser", side_effect=self.fixture)) + self.summary = {"status": "imported", "accepted": 1, "rejected": 0} + self.importer = self.enterContext(patch.object(pipeline, "import_run", return_value=self.summary)) + self.path = self.output / "returned-run" / "manifest.json" + self.raw = self.path.parent / "search-001.jsonl.gz" + + def fixture(self, args, status="complete", error_type=None): + self.preflight.__exit__.assert_called_once() + self.path.parent.mkdir(parents=True, exist_ok=True) + manifest = {"status": status} + if error_type: + manifest["error_type"] = error_type + self.path.write_text(json.dumps(manifest), encoding="utf-8") + self.raw.write_bytes(b"raw snapshot") + self.manifest_bytes = self.path.read_bytes() + return self.path, status + + def state(self): + return json.loads((self.path.parent / "pipeline.json").read_text()) + + def test_collection_statuses_import_exact_manifest(self): + for status, code in (("complete", 0), ("partial", 1), ("failed", 1)): + with self.subTest(status=status): + self.preflight.reset_mock() + self.connect.side_effect = [self.preflight, self.import_connection] + self.collect.side_effect = lambda args: self.fixture(args, status) + stale = self.output / "zz-newer-run" / "manifest.json" + stale.parent.mkdir(parents=True, exist_ok=True) + stale.write_text('{"status": "complete"}') + output_mode = self.output.stat().st_mode + self.assertEqual(pipeline.run_pipeline(self.args), code) + self.assertEqual(self.output.stat().st_mode, output_mode) + self.importer.assert_called_with(self.import_connection, self.path) + self.assertEqual(self.path.read_bytes(), self.manifest_bytes) + self.assertEqual(self.state()["collection_status"], status) + self.assertEqual(self.state()["import_status"], "succeeded") + self.preflight.execute.assert_called_once_with("SELECT 1 FROM silver.collection_runs LIMIT 0") + for call in self.connect.call_args_list: + self.assertEqual(call.args, ("",)) + self.assertEqual(call.kwargs, {"connect_timeout": 10, "autocommit": True}) + + def test_new_output_directory_is_private(self): + self.assertEqual(pipeline.run_pipeline(self.args), 0) + self.assertEqual(self.output.stat().st_mode & 0o777, 0o700) + + def test_rejections_take_priority_over_partial_collection(self): + for status in ("complete", "partial", "failed"): + with self.subTest(status=status): + self.preflight.reset_mock() + self.connect.side_effect = [self.preflight, self.import_connection] + self.collect.side_effect = lambda args: self.fixture(args, status) + self.summary["rejected"] = 1 + self.assertEqual(pipeline.run_pipeline(self.args), 2) + + def test_import_failure_preserves_raw_and_reports_promptly_without_secrets(self): + def fail(conn, path): + self.assertIn(str(path), self.stdout.getvalue()) + raise pipeline.psycopg.OperationalError("password authentication failed TOP_SECRET") + self.importer.side_effect = fail + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.assertEqual(self.raw.read_bytes(), b"raw snapshot") + self.assertEqual(self.path.read_bytes(), self.manifest_bytes) + self.assertEqual(self.state()["import_status"], "unknown") + self.assertEqual(self.state()["error_type"], "OperationalError") + self.assertNotIn("TOP_SECRET", self.stderr.getvalue() + self.stdout.getvalue() + json.dumps(self.state())) + + def test_preflight_failure_never_collects(self): + self.preflight.execute.side_effect = pipeline.psycopg.OperationalError("TOP_SECRET") + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.collect.assert_not_called() + self.importer.assert_not_called() + self.preflight.__exit__.assert_called_once() + self.assertNotIn("TOP_SECRET", self.stderr.getvalue()) + + def test_finalized_interruption_is_not_imported(self): + self.collect.side_effect = lambda args: self.fixture(args, "partial", "KeyboardInterrupt") + self.assertEqual(pipeline.run_pipeline(self.args), 130) + self.importer.assert_not_called() + self.assertEqual(self.state()["import_status"], "not_attempted") + self.assertEqual(self.state()["error_type"], "KeyboardInterrupt") + self.assertEqual(self.path.read_bytes(), self.manifest_bytes) + + def test_interruption_before_collector_returns(self): + self.collect.side_effect = KeyboardInterrupt + self.assertEqual(pipeline.run_pipeline(self.args), 130) + self.importer.assert_not_called() + + def test_interruption_during_import_exits_connection_with_interrupt(self): + self.importer.side_effect = KeyboardInterrupt + self.assertEqual(pipeline.run_pipeline(self.args), 130) + self.assertIs(self.import_connection.__exit__.call_args.args[0], KeyboardInterrupt) + self.assertEqual(self.state()["import_status"], "unknown") + + def test_lock_refusal_and_release_keeps_inode(self): + self.output.mkdir() + lock_path = self.output / ".pipeline.lock" + with lock_path.open("a") as lock: + inode = lock_path.stat().st_ino + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.collect.assert_not_called() + self.connect.assert_not_called() + self.collect.side_effect = RuntimeError("TOP_SECRET browser crash") + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.assertEqual(lock_path.stat().st_ino, inode) + with lock_path.open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + self.importer.assert_not_called() + self.assertNotIn("TOP_SECRET", self.stderr.getvalue()) + + def test_no_implicit_database(self): + with patch.dict(os.environ, {"PGUSER": "user", "PGDATABASE": "db"}, clear=True): + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.connect.assert_not_called() + self.collect.assert_not_called() + + def test_lock_is_held_through_preflight_collection_and_import(self): + def assert_locked(): + with (self.output / ".pipeline.lock").open("a") as lock: + with self.assertRaises(BlockingIOError): + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + + def collect(args): + assert_locked() + return self.fixture(args) + + def import_run(conn, path): + assert_locked() + return self.summary + + self.preflight.execute.side_effect = lambda query: assert_locked() + self.collect.side_effect = collect + self.importer.side_effect = import_run + self.assertEqual(pipeline.run_pipeline(self.args), 0) + with (self.output / ".pipeline.lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + + def test_explicit_database_configuration(self): + for env, conninfo in (({"DATABASE_URL": "postgresql://configured"}, "postgresql://configured"), + ({"PGSERVICE": "homelab"}, ""), ({"PGHOSTADDR": "192.0.2.1"}, "")): + with self.subTest(env=env), patch.dict(os.environ, env, clear=True): + self.preflight.reset_mock() + self.connect.side_effect = [self.preflight, self.import_connection] + self.assertEqual(pipeline.run_pipeline(self.args), 0) + self.connect.assert_called_with(conninfo, connect_timeout=10, autocommit=True) + + def test_input_error_message_is_not_logged(self): + self.importer.side_effect = pipeline.InputError("TOP_SECRET") + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.assertNotIn("TOP_SECRET", self.stderr.getvalue() + json.dumps(self.state())) + + def test_sidecar_storage_failure_prevents_import(self): + with patch.object(pipeline, "write_json", side_effect=OSError("TOP_SECRET")): + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.importer.assert_not_called() + self.assertTrue(self.raw.exists()) + self.assertNotIn("TOP_SECRET", self.stderr.getvalue()) + + def test_main_delegates_parser_and_runner(self): + with patch.object(pipeline, "parse_collection_args", return_value=self.args) as parser: + with patch.object(pipeline, "run_pipeline", return_value=2) as run: + self.assertEqual(pipeline.main(["--search-only"]), 2) + parser.assert_called_once_with(["--search-only"], description=pipeline.__doc__) + run.assert_called_once_with(self.args) + + def test_main_sanitizes_config_failures_and_handles_interrupt(self): + for error, code in ((ValueError("TOP_SECRET"), 3), (SystemExit(2), 3), + (KeyboardInterrupt(), 130), (SystemExit(0), 0)): + with self.subTest(error=type(error).__name__): + def fail(*args, **kwargs): + print("TOP_SECRET", file=pipeline.sys.stderr) + raise error + with patch.object(pipeline, "parse_collection_args", side_effect=fail): + self.assertEqual(pipeline.main([]), code) + self.assertNotIn("TOP_SECRET", self.stderr.getvalue()) + + def test_sigterm_is_handled_and_previous_handler_restored(self): + previous = signal.getsignal(signal.SIGTERM) + + def terminated(args): + os.kill(os.getpid(), signal.SIGTERM) + + self.collect.side_effect = terminated + with patch.object(pipeline, "parse_collection_args", return_value=self.args): + self.assertEqual(pipeline.main([]), 130) + self.importer.assert_not_called() + self.assertIs(signal.getsignal(signal.SIGTERM), previous) + + +if __name__ == "__main__": + unittest.main()