From a9f61d86451e6365e000e0a7c618a3922170df28 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 28 Oct 2026 12:02:28 +1100 Subject: [PATCH 1/2] feat(uploads): keep every upload's original file and rows, insert-only (M3b, #107) Migration 0007 adds per-tenant uploads, upload_blobs and source_rows. Each upload attempt records its filename, SHA-256, size, uploader, mode, outcome and counts, with the file's bytes and parsed rows. All three tables reject UPDATE, DELETE and TRUNCATE. Overwriting or deleting a migrated file no longer loses its original (the product owner moved retiring overwrite to M5, 28 Oct). New routes: GET /api/csv/uploads (list_files) and GET /api/csv/uploads/{id}/original (read_rows), which returns the exact bytes with an X-Content-SHA256 header. /api/csv/files now names each file's uploader. VCRM: BR-23 verified; BR-24 partial until the runtime role (M3d). CI replica 325/0/0, evals 26/26, lint and health clean. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019YPoXzBSj89ZHgw7nfTmbB --- alembic/versions/0007_preserved_uploads.py | 84 ++++++++++++++++ api/routers/csv_routes.py | 55 +++++++++- api/services/audit.py | 4 + api/services/originals.py | 65 ++++++++++++ docs/ROLES.md | 6 +- docs/VCRM.md | 12 +-- tests/test_migrations.py | 2 +- tests/test_preserved_uploads.py | 112 +++++++++++++++++++++ tests/test_tenant_isolation.py | 7 +- 9 files changed, 329 insertions(+), 18 deletions(-) create mode 100644 alembic/versions/0007_preserved_uploads.py create mode 100644 api/services/originals.py create mode 100644 tests/test_preserved_uploads.py diff --git a/alembic/versions/0007_preserved_uploads.py b/alembic/versions/0007_preserved_uploads.py new file mode 100644 index 0000000..14acf95 --- /dev/null +++ b/alembic/versions/0007_preserved_uploads.py @@ -0,0 +1,84 @@ +"""Keep every upload's original file and parsed rows, insert-only (M3b, #107; BR-23, BR-24). + +- ``uploads``: one row per upload attempt. Who uploaded it, when, the file's + SHA-256 and size, the mode, the outcome (imported, partly_imported, failed, + duplicate or rejected), the row counts, and the registry row it produced. +- ``upload_blobs``: the file's bytes exactly as received. +- ``source_rows``: the parsed rows as text; row 0 is the header. + +All three reject UPDATE, DELETE and TRUNCATE, like audit_log (0006). Deleting +or overwriting a migrated file removes its table and registry row, never its +original. ``csv_file_id`` is a plain value, not a foreign key, for that reason. + +Runs per tenant, like 0003. + +Revision ID: 0007 +Revises: 0006 +Create Date: 2026-10-28 +""" + +from alembic import op + +from api.config import settings + +revision = "0007" +down_revision = "0006" +branch_labels = None +depends_on = None + +_TABLES = ("uploads", "upload_blobs", "source_rows") + + +def _schema() -> str: + return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA) + + +def upgrade() -> None: + s = _schema() + op.execute(f"""CREATE TABLE {s}.uploads ( + id BIGSERIAL PRIMARY KEY, + file_name TEXT NOT NULL, + sha256 TEXT NOT NULL, + byte_size INTEGER NOT NULL, + uploader TEXT, + mode TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN + ('imported', 'partly_imported', 'failed', 'duplicate', 'rejected')), + total_rows INTEGER, + imported_rows INTEGER, + failed_rows INTEGER, + csv_file_id BIGINT, + uploaded_at TIMESTAMPTZ NOT NULL DEFAULT now())""") + op.execute(f"CREATE INDEX uploads_csv_file_id ON {s}.uploads (csv_file_id)") + op.execute(f"""CREATE TABLE {s}.upload_blobs ( + upload_id BIGINT PRIMARY KEY REFERENCES {s}.uploads (id), + content BYTEA NOT NULL)""") + op.execute(f"""CREATE TABLE {s}.source_rows ( + upload_id BIGINT NOT NULL REFERENCES {s}.uploads (id), + row_number INTEGER NOT NULL, + cells TEXT[] NOT NULL, + PRIMARY KEY (upload_id, row_number))""") + # One guard function for every insert-only table, naming the table it stopped. + op.execute(f""" + CREATE OR REPLACE FUNCTION {s}.audit_reject_change() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + RAISE EXCEPTION '% is append-only: % is not allowed', TG_TABLE_NAME, TG_OP + USING ERRCODE = 'insufficient_privilege'; + END $$""") + for t in _TABLES: + op.execute(f"CREATE TRIGGER {t}_no_change BEFORE UPDATE OR DELETE ON {s}.{t} " + f"FOR EACH ROW EXECUTE FUNCTION {s}.audit_reject_change()") + op.execute(f"CREATE TRIGGER {t}_no_truncate BEFORE TRUNCATE ON {s}.{t} " + f"FOR EACH STATEMENT EXECUTE FUNCTION {s}.audit_reject_change()") + + +def downgrade() -> None: + s = _schema() + for t in reversed(_TABLES): + op.execute(f"DROP TABLE IF EXISTS {s}.{t}") + op.execute(f""" + CREATE OR REPLACE FUNCTION {s}.audit_reject_change() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + RAISE EXCEPTION 'audit_log is append-only: % is not allowed', TG_OP + USING ERRCODE = 'insufficient_privilege'; + END $$""") diff --git a/api/routers/csv_routes.py b/api/routers/csv_routes.py index 55853b9..fe676f0 100644 --- a/api/routers/csv_routes.py +++ b/api/routers/csv_routes.py @@ -3,8 +3,9 @@ from __future__ import annotations import logging +from urllib.parse import quote -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Response from psycopg2 import sql from pydantic import BaseModel, Field @@ -15,6 +16,7 @@ from api.services.audit import coerce_file_id, record_audit, write_audit from api.services.csv_parse import build_preview from api.services.dynamic_loader import upload_dynamic +from api.services.originals import record_upload from api.services.te_loader import match_te_table, upload_te logger = logging.getLogger(__name__) @@ -80,6 +82,10 @@ def upload(req: UploadRequest) -> dict: else: raise HTTPException(422, f"Unknown mode '{req.mode}' — use 'dynamic' or 'te'") + # M3b (#107): keep the original and its outcome, whatever the outcome was. + upload_id = record_upload(req.fileName, req.content, req.mode, result) + result["uploadId"] = None if upload_id is None else str(upload_id) + # O-04: record the load so audit_log covers creations, not just deletions. # Only a genuine load is audited: "duplicate_file" and "error" outcomes # changed nothing, so logging them would make the trail read as if data had @@ -101,10 +107,12 @@ def list_files() -> list[dict]: with conn.cursor() as cur: cur.execute( sql.SQL( - "SELECT id, file_name, table_name, mode, row_count, column_names, created_at, " - "total_rows, duplicate_rows_skipped, failed_rows " - "FROM {}.csv_files ORDER BY created_at DESC, id DESC" - ).format(sql.Identifier(uploads_schema())) + "SELECT f.id, f.file_name, f.table_name, f.mode, f.row_count, f.column_names, f.created_at, " + "f.total_rows, f.duplicate_rows_skipped, f.failed_rows, u.uploader, u.id " + "FROM {s}.csv_files f LEFT JOIN LATERAL (SELECT uploader, id FROM {s}.uploads " + " WHERE csv_file_id = f.id ORDER BY id DESC LIMIT 1) u ON true " + "ORDER BY f.created_at DESC, f.id DESC" + ).format(s=sql.Identifier(uploads_schema())) ) return [ { @@ -118,11 +126,48 @@ def list_files() -> list[dict]: "total_rows": r[7], "duplicate_rows_skipped": r[8], "failed_rows": r[9], + "uploader": r[10], + "upload_id": None if r[11] is None else str(r[11]), } for r in cur.fetchall() ] +@router.get("/uploads", dependencies=[Depends(allow("list_files"))]) +def list_uploads(limit: int = 100, offset: int = 0) -> list[dict]: + """Every upload attempt and its outcome, newest first (M3b, #107; BR-23).""" + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL( + "SELECT id, file_name, sha256, byte_size, uploader, mode, status, total_rows, imported_rows, " + "failed_rows, csv_file_id, uploaded_at FROM {}.uploads ORDER BY id DESC LIMIT %s OFFSET %s" + ).format(sql.Identifier(uploads_schema())), (max(1, min(limit, 500)), max(0, offset))) + keys = ("id", "file_name", "sha256", "byte_size", "uploader", "mode", "status", "total_rows", + "imported_rows", "failed_rows", "csv_file_id", "uploaded_at") + out = [dict(zip(keys, r)) for r in cur.fetchall()] + for u in out: + u["id"] = str(u["id"]) + u["csv_file_id"] = None if u["csv_file_id"] is None else str(u["csv_file_id"]) + u["uploaded_at"] = u["uploaded_at"].isoformat() + return out + + +@router.get("/uploads/{upload_id}/original", dependencies=[Depends(allow("read_rows"))]) +def upload_original(upload_id: int) -> Response: + """The file exactly as it was uploaded (M3b; BR-24).""" + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL( + "SELECT u.file_name, u.sha256, b.content FROM {s}.uploads u " + "JOIN {s}.upload_blobs b ON b.upload_id = u.id WHERE u.id = %s" + ).format(s=sql.Identifier(uploads_schema())), (upload_id,)) + row = cur.fetchone() + if row is None: + raise HTTPException(404, "No such upload") + name, digest, content = row + return Response(bytes(content), media_type="text/csv", headers={ + "Content-Disposition": f"attachment; filename*=UTF-8''{quote(name)}", + "X-Content-SHA256": digest}) + + @router.get("/tables/{table_name}/rows", dependencies=[Depends(allow("read_rows"))]) def table_rows(table_name: str, limit: int = 50) -> dict: if not table_name.startswith("csv_") or len(table_name) > 64: diff --git a/api/services/audit.py b/api/services/audit.py index 0910280..c67e9ce 100644 --- a/api/services/audit.py +++ b/api/services/audit.py @@ -45,6 +45,10 @@ _actor: ContextVar[str | None] = ContextVar("audit_actor", default=None) +def current_actor() -> str | None: + return _actor.get() + + @contextmanager def acting_as(actor: str): token = _actor.set(actor) diff --git a/api/services/originals.py b/api/services/originals.py new file mode 100644 index 0000000..2aee2d8 --- /dev/null +++ b/api/services/originals.py @@ -0,0 +1,65 @@ +"""Keep each upload's original file and parsed rows (M3b, #107; BR-23, BR-24). + +Every upload attempt is recorded in ``uploads`` with its outcome, and its bytes +and parsed rows go to ``upload_blobs`` and ``source_rows``. All three tables are +insert-only (migration 0007), so overwriting or deleting a migrated file never +loses its original. + +Known limitation, the same as record_audit: the loaders commit their own +transaction, so this runs after the load. A failure here is logged at ERROR +and doesn't fail the upload. M6's single-transaction import (ADR-0002) closes it. +""" + +from __future__ import annotations + +import hashlib +import logging + +from psycopg2 import sql +from psycopg2.extras import execute_values + +from api.db import Conn +from api.services.audit import coerce_file_id, current_actor +from api.services.csv_parse import parse_csv +from api.tenancy import uploads_schema + +logger = logging.getLogger(__name__) + +_STATUS = {"duplicate_file": "duplicate", "invalid_structure": "rejected", "error": "failed"} + + +def outcome(result: dict) -> str: + status = result.get("status") + if status == "ok": + return "partly_imported" if result.get("failedRows") else "imported" + return _STATUS.get(status, "failed") + + +def record_upload(file_name: str, content: str, mode: str, result: dict) -> int | None: + """Store the original and its outcome. Returns the upload id, or None if it couldn't be stored.""" + data = content.encode("utf-8") + ok = result.get("status") == "ok" + try: + with Conn() as conn: + with conn.cursor() as cur: + s = sql.Identifier(uploads_schema()) + cur.execute(sql.SQL( + "INSERT INTO {}.uploads (file_name, sha256, byte_size, uploader, mode, status, " + "total_rows, imported_rows, failed_rows, csv_file_id) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id").format(s), + (file_name, hashlib.sha256(data).hexdigest(), len(data), current_actor(), mode, + outcome(result), result.get("totalRows"), result.get("insertedRows"), + result.get("failedRows"), coerce_file_id(result.get("fileId")) if ok else None)) + upload_id = cur.fetchone()[0] + cur.execute(sql.SQL("INSERT INTO {}.upload_blobs (upload_id, content) VALUES (%s, %s)").format(s), + (upload_id, data)) + rows = parse_csv(content) # ponytail: second parse after the loader's; share it if uploads get slow + if rows: + execute_values(cur, sql.SQL("INSERT INTO {}.source_rows (upload_id, row_number, cells) VALUES %s") + .format(s).as_string(cur), [(upload_id, n, r) for n, r in enumerate(rows)], + page_size=1000) + conn.commit() + return upload_id + except Exception as exc: # noqa: BLE001 — the load has already committed; see the module docstring + logger.error("Failed to store the original of %r: %s", file_name, exc) + return None diff --git a/docs/ROLES.md b/docs/ROLES.md index 366b089..5ac0a76 100644 --- a/docs/ROLES.md +++ b/docs/ROLES.md @@ -6,11 +6,11 @@ Every member of a tenant has one role (ADR-0003). This table is enforced by `PER |---|---|---|---|---|---| | preview | `POST /api/csv/preview` | ✓ | | ✓ | | | upload | `POST /api/csv/upload` | ✓ | | ✓ | | -| list_files | `GET /api/csv/files` | ✓ | ✓ | ✓ | ✓ | -| read_rows | `GET /api/csv/tables/{table}/rows` | ✓ | ✓ | ✓ | ✓ | +| list_files | `GET /api/csv/files`, `GET /api/csv/uploads` | ✓ | ✓ | ✓ | ✓ | +| read_rows | `GET /api/csv/tables/{table}/rows`, `GET /api/csv/uploads/{id}/original` | ✓ | ✓ | ✓ | ✓ | | list_te | `GET /api/te/tables` | ✓ | ✓ | ✓ | ✓ | | delete_file | `DELETE /api/csv/files/{id}` | | | ✓ | | -| read_audit | `GET /api/audit/log` | | | ✓ | ✓ | +| read_audit | `GET /api/audit/log`, `GET /api/audit/verify` | | | ✓ | ✓ | | reset_password | `POST /api/auth/users/{username}/password` | | | ✓ | | A role that isn't allowed gets **403**. Two kinds of request are outside this table: diff --git a/docs/VCRM.md b/docs/VCRM.md index 433fd08..ef5f876 100644 --- a/docs/VCRM.md +++ b/docs/VCRM.md @@ -142,8 +142,8 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified" | ID | Status | Reason | Planned verification | Milestone | |----|--------|--------|----------------------|-----------| -| BR-23 | ⚠️ Partial | Registry has UNIQUE file hash and name and a migration history (M1: #87); still no uploader, tenant or status | Integration + DB constraint tests | M1, M3 | -| BR-24 | ❌ Not started | Original never stored; overwrite drops data | DB negative tests | M3 | +| BR-23 | ✅ | Every upload attempt gets an `uploads` row (M3b, #107): id, original filename, SHA-256, uploader, timestamp and outcome (imported, partly_imported, failed, duplicate, rejected). The tenant is the schema the row lives in (decision 6), not a column. Registry UNIQUE hash and name from M1 (#87). `test_preserved_uploads.py` | Integration + DB constraint tests | M1, M3 | +| BR-24 | ⚠️ Partial | Original bytes (`upload_blobs`) and parsed rows (`source_rows`) kept for every upload, including failed ones and files later overwritten or deleted (M3b, #107). Triggers reject UPDATE, DELETE and TRUNCATE (`test_preserved_uploads.py`, 6 tests). Still to do: the least-privilege runtime role (M3d); edits kept separate arrive with M4 | DB negative tests | M3 | | BR-25 | ⚠️ Partial | 10-row preview sample; 200-row cap on table view | API paging + e2e | M4 | | BR-26 | ❌ Not started | No code | API + e2e | M4 | | BR-27 | ❌ Not started | No code | API + DB tests | M4 | @@ -178,12 +178,12 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified" | Status | Count | Requirements | |--------|-------|-------------| -| ✅ Verified | 19 | BR-01, BR-03, BR-04, BR-05, BR-06, BR-07, BR-08, BR-09, BR-10, BR-11, BR-12, BR-13, BR-14, BR-15, BR-16, BR-17, BR-18, BR-19, BR-20 | -| ⚠️ Partial | 11 | BR-23, BR-25, BR-28, BR-29, BR-30, BR-31, BR-33, BR-34, BR-35, BR-36, BR-37 (ART, added 2026-10-26) | -| ❌ Not started | 5 | BR-24, BR-26, BR-27, BR-32 (ART, added 2026-10-26); BR-38 (added 2026-10-27) | +| ✅ Verified | 20 | BR-01, BR-03, BR-04, BR-05, BR-06, BR-07, BR-08, BR-09, BR-10, BR-11, BR-12, BR-13, BR-14, BR-15, BR-16, BR-17, BR-18, BR-19, BR-20, BR-23 (M3b) | +| ⚠️ Partial | 11 | BR-24, BR-25, BR-28, BR-29, BR-30, BR-31, BR-33, BR-34, BR-35, BR-36, BR-37 (ART, added 2026-10-26) | +| ❌ Not started | 4 | BR-26, BR-27, BR-32 (ART, added 2026-10-26); BR-38 (added 2026-10-27) | | ⏸ Deferred by decision | 3 | BR-02, BR-21, BR-22 | -**Headline (2026-10-27):** with BR-38 added (cross-platform offline install, #128), **19 of 38 (50 %)** are verified, 3 are deferred by decision, and the other 16 are partial or not started. +**Headline (2026-10-28):** with BR-23 verified by M3b (#107), **20 of 38 (53 %)** are verified, 3 are deferred by decision, and the other 15 are partial or not started. **Headline (2026-10-26):** with the 15 ART requirements added, **19 of 37 (51 %)** are verified, 3 are deferred by decision, and the other 15 are partial or not started. diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 7fb54af..d6f9079 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -148,7 +148,7 @@ def setUp(self): def test_bootstrap_brings_the_schema_to_head(self): with Conn() as conn, conn.cursor() as cur: cur.execute(f'SELECT version_num FROM "{settings.UPLOADS_SCHEMA}".alembic_version') - self.assertEqual(cur.fetchone()[0], "0006") + self.assertEqual(cur.fetchone()[0], "0007") def test_duplicate_hash_is_rejected_by_the_database(self): insert = (f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files (file_name, file_hash, table_name, mode) ' diff --git a/tests/test_preserved_uploads.py b/tests/test_preserved_uploads.py new file mode 100644 index 0000000..497d707 --- /dev/null +++ b/tests/test_preserved_uploads.py @@ -0,0 +1,112 @@ +"""Every upload keeps its original file and parsed rows, insert-only (M3b, #107; BR-23, BR-24). + +Integration: needs a live PostgreSQL. Fails (never skips) without one. +""" + +from __future__ import annotations + +import hashlib +import os +import unittest +import uuid + +import psycopg2.errors +import pytest +from fastapi.testclient import TestClient +from psycopg2 import sql + +from api.config import settings +from api.db import Conn +from api.main import app + + +@pytest.mark.integration +class PreservedUploads(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls._orig_allow_destructive = settings.allow_destructive + settings.allow_destructive = True + cls.ctx = TestClient(app) + cls.client = cls.ctx.__enter__() + + @classmethod + def tearDownClass(cls): + settings.allow_destructive = cls._orig_allow_destructive + cls.ctx.__exit__(None, None, None) + + def setUp(self): + self.tag = f"{os.getpid()}_{uuid.uuid4().hex[:8]}" + self.name = f"orig_{self.tag}.csv" + + def tearDown(self): + for f in self.client.get("/api/csv/files").json(): + if f["file_name"] == self.name: + self.client.delete(f"/api/csv/files/{f['id']}") + + def _upload(self, content: str, types: list[str] | None = None, overwrite: bool = False) -> dict: + body = {"fileName": self.name, "content": content, "overwrite": overwrite} + if types: + body["types"] = types + return self.client.post("/api/csv/upload", json=body).json() + + def _upload_row(self, upload_id: str) -> dict: + return next(u for u in self.client.get("/api/csv/uploads").json() if u["id"] == upload_id) + + def _source_rows(self, upload_id: str) -> list[list[str]]: + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL("SELECT cells FROM {}.source_rows WHERE upload_id = %s ORDER BY row_number") + .format(sql.Identifier(settings.UPLOADS_SCHEMA)), (int(upload_id),)) + return [r[0] for r in cur.fetchall()] + + def test_original_bytes_and_rows_are_kept_with_the_outcome(self): + content = f'name,note\n{self.tag},"a, b"\nsecond,"line\nbreak"\n' + res = self._upload(content) + self.assertEqual(res["status"], "ok", res) + row = self._upload_row(res["uploadId"]) + self.assertEqual((row["status"], row["uploader"], row["total_rows"], row["imported_rows"]), + ("imported", "api_key", 2, 2)) + self.assertEqual(row["csv_file_id"], res["fileId"]) + original = self.client.get(f"/api/csv/uploads/{res['uploadId']}/original") + self.assertEqual(original.content, content.encode("utf-8")) + self.assertEqual(original.headers["x-content-sha256"], hashlib.sha256(content.encode()).hexdigest()) + self.assertEqual(self._source_rows(res["uploadId"]), + [["name", "note"], [self.tag, "a, b"], ["second", "line\nbreak"]]) + [listed] = [f for f in self.client.get("/api/csv/files").json() if f["file_name"] == self.name] + self.assertEqual((listed["uploader"], listed["upload_id"]), ("api_key", res["uploadId"])) + + def test_a_failed_upload_still_keeps_its_original(self): + content = f"Date,Ref\n27 Oct 26,{self.tag}\n" + res = self._upload(content, ["timestamptz", "text"]) + self.assertEqual(res["status"], "error", res) + row = self._upload_row(res["uploadId"]) + self.assertEqual((row["status"], row["csv_file_id"], row["failed_rows"]), ("failed", None, 1)) + self.assertEqual(self.client.get(f"/api/csv/uploads/{res['uploadId']}/original").content, content.encode()) + + def test_partly_failed_and_duplicate_outcomes(self): + res = self._upload(f"n,t\n1,{self.tag}\nx,{self.tag}\n", ["int8", "text"]) + self.assertEqual(self._upload_row(res["uploadId"])["status"], "partly_imported") + again = self._upload(f"n,t\n1,{self.tag}\nx,{self.tag}\n", ["int8", "text"]) + self.assertEqual(again["status"], "duplicate_file", again) + self.assertEqual(self._upload_row(again["uploadId"])["status"], "duplicate") + + def test_overwrite_keeps_the_replaced_original(self): + first = self._upload(f"v\nold_{self.tag}\n") + second = self._upload(f"v\nnew_{self.tag}\n", overwrite=True) + self.assertEqual(second["status"], "ok", second) + self.assertEqual(self.client.get(f"/api/csv/uploads/{first['uploadId']}/original").content, + f"v\nold_{self.tag}\n".encode()) + self.assertEqual(self._source_rows(first["uploadId"]), [["v"], [f"old_{self.tag}"]]) + + def test_preserved_tables_reject_update_and_delete(self): + res = self._upload(f"v\n{self.tag}\n") + uid = int(res["uploadId"]) + for table, column in (("uploads", "file_name"), ("upload_blobs", "content"), ("source_rows", "cells")): + for statement in (f"UPDATE {{}}.{table} SET {column} = {column} WHERE {'id' if table == 'uploads' else 'upload_id'} = %s", + f"DELETE FROM {{}}.{table} WHERE {'id' if table == 'uploads' else 'upload_id'} = %s"): + with self.subTest(statement=statement), Conn() as conn, conn.cursor() as cur: + with self.assertRaises(psycopg2.errors.InsufficientPrivilege): + cur.execute(sql.SQL(statement).format(sql.Identifier(settings.UPLOADS_SCHEMA)), (uid,)) + conn.rollback() + + def test_unknown_upload_is_404(self): + self.assertEqual(self.client.get("/api/csv/uploads/999999999/original").status_code, 404) diff --git a/tests/test_tenant_isolation.py b/tests/test_tenant_isolation.py index 0af647c..bbd4a01 100644 --- a/tests/test_tenant_isolation.py +++ b/tests/test_tenant_isolation.py @@ -16,7 +16,7 @@ from fastapi.testclient import TestClient from psycopg2 import sql -from api.config import settings +from api.config import TE_TABLES, settings from api.db import Conn from api.main import app from api.services import accounts @@ -172,7 +172,8 @@ def test_tenant_te_tables_keep_their_foreign_keys_and_triggers(self): schema = schema_for(self.slugs["b"]) with Conn() as conn, conn.cursor() as cur: cur.execute("SELECT count(*) FROM information_schema.table_constraints " - "WHERE table_schema = %s AND constraint_type = 'FOREIGN KEY'", (schema,)) + "WHERE table_schema = %s AND constraint_type = 'FOREIGN KEY' AND table_name = ANY(%s)", + (schema, TE_TABLES)) # the T&E tables only; 0007's upload tables have their own tenant_fks = cur.fetchone()[0] cur.execute("SELECT count(*) FROM information_schema.table_constraints " "WHERE table_schema = %s AND constraint_type = 'FOREIGN KEY'", (settings.TE_SCHEMA,)) @@ -198,7 +199,7 @@ def test_tenant_schema_is_at_head_with_its_own_registry(self): with Conn() as conn, conn.cursor() as cur: cur.execute(sql.SQL("SELECT version_num FROM {}.alembic_version").format( sql.Identifier(schema_for(self.slugs["b"])))) - self.assertEqual(cur.fetchone()[0], "0006") + self.assertEqual(cur.fetchone()[0], "0007") cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s", (schema_for(self.slugs["b"]),)) self.assertTrue({"csv_files", "audit_log"} <= {r[0] for r in cur.fetchall()}) From 472f9f86251e6b3ac827508370e024477706bb8c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 28 Oct 2026 12:07:24 +1100 Subject: [PATCH 2/2] feat(uploads): keep each upload's row errors and step log on the server (M3c backend, #107) The uploads row now stores why rows failed (capped at API_MAX_ROW_ERRORS) and the step log, inside the same insert-only row. New route GET /api/csv/uploads/{id} (list_files) returns them, so any session in the tenant can read them, not only the browser that uploaded the file. Product owner's scope addition on #107, 27 Oct. CI replica 326/0/0, evals 26/26, lint clean. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019YPoXzBSj89ZHgw7nfTmbB --- alembic/versions/0007_preserved_uploads.py | 5 ++++- api/routers/csv_routes.py | 15 +++++++++++++++ api/services/originals.py | 11 +++++++---- tests/test_preserved_uploads.py | 11 +++++++++++ 4 files changed, 37 insertions(+), 5 deletions(-) diff --git a/alembic/versions/0007_preserved_uploads.py b/alembic/versions/0007_preserved_uploads.py index 14acf95..f0f1fe4 100644 --- a/alembic/versions/0007_preserved_uploads.py +++ b/alembic/versions/0007_preserved_uploads.py @@ -2,7 +2,8 @@ - ``uploads``: one row per upload attempt. Who uploaded it, when, the file's SHA-256 and size, the mode, the outcome (imported, partly_imported, failed, - duplicate or rejected), the row counts, and the registry row it produced. + duplicate or rejected), the row counts, the registry row it produced, and + why rows failed plus the step log (M3c; row errors capped at API_MAX_ROW_ERRORS). - ``upload_blobs``: the file's bytes exactly as received. - ``source_rows``: the parsed rows as text; row 0 is the header. @@ -48,6 +49,8 @@ def upgrade() -> None: imported_rows INTEGER, failed_rows INTEGER, csv_file_id BIGINT, + row_errors JSONB, + logs JSONB, uploaded_at TIMESTAMPTZ NOT NULL DEFAULT now())""") op.execute(f"CREATE INDEX uploads_csv_file_id ON {s}.uploads (csv_file_id)") op.execute(f"""CREATE TABLE {s}.upload_blobs ( diff --git a/api/routers/csv_routes.py b/api/routers/csv_routes.py index fe676f0..898abb3 100644 --- a/api/routers/csv_routes.py +++ b/api/routers/csv_routes.py @@ -151,6 +151,21 @@ def list_uploads(limit: int = 100, offset: int = 0) -> list[dict]: return out +@router.get("/uploads/{upload_id}", dependencies=[Depends(allow("list_files"))]) +def upload_detail(upload_id: int) -> dict: + """One upload's outcome, why its rows failed, and its step log (M3c, #107). + Stored on the server, so any session in the tenant can read them.""" + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL( + "SELECT status, total_rows, imported_rows, failed_rows, row_errors, logs, uploader, file_name " + "FROM {}.uploads WHERE id = %s").format(sql.Identifier(uploads_schema())), (upload_id,)) + row = cur.fetchone() + if row is None: + raise HTTPException(404, "No such upload") + keys = ("status", "totalRows", "importedRows", "failedRows", "rowErrors", "logs", "uploader", "fileName") + return {"id": str(upload_id), **dict(zip(keys, row))} + + @router.get("/uploads/{upload_id}/original", dependencies=[Depends(allow("read_rows"))]) def upload_original(upload_id: int) -> Response: """The file exactly as it was uploaded (M3b; BR-24).""" diff --git a/api/services/originals.py b/api/services/originals.py index 2aee2d8..60f88fb 100644 --- a/api/services/originals.py +++ b/api/services/originals.py @@ -16,8 +16,9 @@ import logging from psycopg2 import sql -from psycopg2.extras import execute_values +from psycopg2.extras import Json, execute_values +from api.config import settings from api.db import Conn from api.services.audit import coerce_file_id, current_actor from api.services.csv_parse import parse_csv @@ -45,11 +46,13 @@ def record_upload(file_name: str, content: str, mode: str, result: dict) -> int s = sql.Identifier(uploads_schema()) cur.execute(sql.SQL( "INSERT INTO {}.uploads (file_name, sha256, byte_size, uploader, mode, status, " - "total_rows, imported_rows, failed_rows, csv_file_id) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id").format(s), + "total_rows, imported_rows, failed_rows, csv_file_id, row_errors, logs) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id").format(s), (file_name, hashlib.sha256(data).hexdigest(), len(data), current_actor(), mode, outcome(result), result.get("totalRows"), result.get("insertedRows"), - result.get("failedRows"), coerce_file_id(result.get("fileId")) if ok else None)) + result.get("failedRows"), coerce_file_id(result.get("fileId")) if ok else None, + Json((result.get("rowErrors") or [])[:settings.MAX_ROW_ERRORS_REPORTED]), + Json(result.get("logs") or []))) upload_id = cur.fetchone()[0] cur.execute(sql.SQL("INSERT INTO {}.upload_blobs (upload_id, content) VALUES (%s, %s)").format(s), (upload_id, data)) diff --git a/tests/test_preserved_uploads.py b/tests/test_preserved_uploads.py index 497d707..2aa352d 100644 --- a/tests/test_preserved_uploads.py +++ b/tests/test_preserved_uploads.py @@ -82,6 +82,16 @@ def test_a_failed_upload_still_keeps_its_original(self): self.assertEqual((row["status"], row["csv_file_id"], row["failed_rows"]), ("failed", None, 1)) self.assertEqual(self.client.get(f"/api/csv/uploads/{res['uploadId']}/original").content, content.encode()) + def test_failure_reasons_and_log_are_readable_from_another_session(self): + """M3c (#107): the reasons rows failed live on the server, not only in the uploading browser.""" + res = self._upload(f"n,t\n1,{self.tag}\nx,{self.tag}\n", ["int8", "text"]) + other = TestClient(app) # a separate client with no state shared with the uploader; no + detail = other.get(f"/api/csv/uploads/{res['uploadId']}").json() # lifespan, so the pool stays up + self.assertEqual((detail["status"], detail["failedRows"]), ("partly_imported", 1)) + self.assertEqual(detail["rowErrors"], res["rowErrors"]) + self.assertEqual(detail["rowErrors"][0]["rowNumber"], 2) + self.assertEqual([entry["step"] for entry in detail["logs"]], [entry["step"] for entry in res["logs"]]) + def test_partly_failed_and_duplicate_outcomes(self): res = self._upload(f"n,t\n1,{self.tag}\nx,{self.tag}\n", ["int8", "text"]) self.assertEqual(self._upload_row(res["uploadId"])["status"], "partly_imported") @@ -110,3 +120,4 @@ def test_preserved_tables_reject_update_and_delete(self): def test_unknown_upload_is_404(self): self.assertEqual(self.client.get("/api/csv/uploads/999999999/original").status_code, 404) + self.assertEqual(self.client.get("/api/csv/uploads/999999999").status_code, 404)