diff --git a/alembic/versions/20261024_a3f7ce361b0e_v137_epcis_document_receipts.py b/alembic/versions/20261024_a3f7ce361b0e_v137_epcis_document_receipts.py new file mode 100644 index 000000000..17bcba325 --- /dev/null +++ b/alembic/versions/20261024_a3f7ce361b0e_v137_epcis_document_receipts.py @@ -0,0 +1,110 @@ +"""EPCIS document receipts: fsma.epcis_document_receipts. + +Mirrors supabase/migrations/20261024090000_epcis_document_receipts.sql exactly: +same table, columns, constraints, index, RLS policy and grants. One row per +EPCIS document the ingest routes receive, accepted or rejected, keyed by the +SHA-256 of the raw request body. error_summary is a short sanitized summary, +never payload content. + +Additive and idempotent (IF NOT EXISTS, drop-if-exists-then-create policy, +role-guarded grants), so the plain SQL file can run ahead of it. Downgrade +drops only what this revision created. + +Revision ID: a3f7ce361b0e +Revises: b005ddbf1dba +Create Date: 2026-10-24 +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = "a3f7ce361b0e" # pragma: allowlist secret +down_revision: Union[str, Sequence[str], None] = "b005ddbf1dba" # pragma: allowlist secret +branch_labels = None +depends_on = None + +_UPGRADE_STATEMENTS: tuple[str, ...] = ( + """ +CREATE TABLE IF NOT EXISTS fsma.epcis_document_receipts ( + id uuid primary key default gen_random_uuid(), + tenant_id uuid not null, + document_sha256 text not null, + received_at timestamptz not null default now(), + status text not null, + error_summary text, + constraint epcis_document_receipts_status_check + check (status in ('accepted', 'rejected')), + constraint epcis_document_receipts_sha256_check + check (document_sha256 ~ '^[0-9a-f]{64}$'), + constraint epcis_document_receipts_error_summary_length_check + check (error_summary is null or char_length(error_summary) <= 200) +); +""", + """ +-- Read path: rejected share per tenant over a rolling window. +create index if not exists idx_epcis_document_receipts_tenant_received + on fsma.epcis_document_receipts (tenant_id, received_at); +""", + """ +ALTER TABLE fsma.epcis_document_receipts ENABLE ROW LEVEL SECURITY; +""", + """ +ALTER TABLE fsma.epcis_document_receipts FORCE ROW LEVEL SECURITY; +""", + """ +-- CASE form, not OR: public.get_tenant_context() raises when app.tenant_id is +-- unset and Postgres does not order OR operands, so the sysadmin arm is tested +-- first (same predicate shape as alembic v096, v134 and v135). +drop policy if exists tenant_isolation_epcis_document_receipts on fsma.epcis_document_receipts; +""", + """ +create policy tenant_isolation_epcis_document_receipts on fsma.epcis_document_receipts + for all + using ( + case + when (select current_setting('regengine.is_sysadmin', true)) = 'true' + and current_user = 'regengine_sysadmin' then true + else tenant_id = (select public.get_tenant_context()) + end + ) + with check ( + case + when (select current_setting('regengine.is_sysadmin', true)) = 'true' + and current_user = 'regengine_sysadmin' then true + else tenant_id = (select public.get_tenant_context()) + end + ); +""", + """ +-- Runtime role allowlist (alembic v111): new tables are inaccessible to the +-- app role until a migration names them. Insert and select only: a receipt is +-- evidence and is never updated or deleted by the app. +do $$ +begin + if exists (select 1 from pg_roles where rolname = 'anon') then + revoke all on fsma.epcis_document_receipts from anon; + end if; + if exists (select 1 from pg_roles where rolname = 'authenticated') then + revoke all on fsma.epcis_document_receipts from authenticated; + end if; + if exists (select 1 from pg_roles where rolname = 'regengine') then + grant select, insert on fsma.epcis_document_receipts to regengine; + end if; + if exists (select 1 from pg_roles where rolname = 'regengine_sysadmin') then + grant select on fsma.epcis_document_receipts to regengine_sysadmin; + end if; +end $$; +""", +) + + +def upgrade() -> None: + for statement in _UPGRADE_STATEMENTS: + op.execute(statement) + + +def downgrade() -> None: + op.execute("DROP POLICY IF EXISTS tenant_isolation_epcis_document_receipts ON fsma.epcis_document_receipts") + op.execute("DROP INDEX IF EXISTS fsma.idx_epcis_document_receipts_tenant_received") + op.execute("DROP TABLE IF EXISTS fsma.epcis_document_receipts") diff --git a/docs/schema/expected_schema.json b/docs/schema/expected_schema.json index 95c52ff48..73e13227f 100644 --- a/docs/schema/expected_schema.json +++ b/docs/schema/expected_schema.json @@ -1474,6 +1474,35 @@ } ] }, + "fsma.epcis_document_receipts": { + "columns": { + "document_sha256": { + "nullable": false, + "type": "text" + }, + "error_summary": { + "nullable": true, + "type": "text" + }, + "id": { + "nullable": false, + "type": "uuid" + }, + "received_at": { + "nullable": false, + "type": "timestamp" + }, + "status": { + "nullable": false, + "type": "text" + }, + "tenant_id": { + "nullable": false, + "type": "uuid" + } + }, + "foreign_keys": [] + }, "fsma.evidence_attachments": { "columns": { "created_at": { diff --git a/frontend/src/__tests__/components/supplier-data-quality.test.tsx b/frontend/src/__tests__/components/supplier-data-quality.test.tsx index bddece96a..af48ad555 100644 --- a/frontend/src/__tests__/components/supplier-data-quality.test.tsx +++ b/frontend/src/__tests__/components/supplier-data-quality.test.tsx @@ -119,6 +119,29 @@ describe('DataQuality', () => { expect(screen.getByText(LATE_LABEL)).toBeInTheDocument(); }); + it('shows a computed EPCIS share with its tenant wide reason and no drill down', () => { + const receipts = uuids(MIN_RECORDS); + const rejected = receipts.filter((_, index) => index % 2 === 0); + const share = rejected.length / receipts.length; + const tenantReason = crypto.randomUUID(); + renderQuality( + summary([ + metric({ + key: 'epcis_failed', + status: 'ok', + numerator: rejected.length, + denominator: receipts.length, + share, + reason: tenantReason, + }), + ]), + ); + expect(screen.getByText('EPCIS (Preview) documents rejected across your whole account:')).toBeInTheDocument(); + expect(screen.getByText(`${rejected.length} of ${receipts.length} documents (${formatShare(share)})`)).toBeInTheDocument(); + expect(screen.getByText(tenantReason)).toBeInTheDocument(); + expect(screen.queryByRole('button', { name: 'View records' })).not.toBeInTheDocument(); + }); + it('offers drill down only for computable metrics with flagged records', () => { renderQuality(); expect(screen.getAllByRole('button', { name: 'View records' })).toHaveLength(2); diff --git a/frontend/src/app/dashboard/suppliers/data-quality.tsx b/frontend/src/app/dashboard/suppliers/data-quality.tsx index 86c37f491..faec056cf 100644 --- a/frontend/src/app/dashboard/suppliers/data-quality.tsx +++ b/frontend/src/app/dashboard/suppliers/data-quality.tsx @@ -16,7 +16,7 @@ const METRIC_COPY: Record )} + {metric.status !== 'unavailable' && metric.reason &&

{metric.reason}

} {open && isDrilldown(metric.key) && (
diff --git a/frontend/src/types/api.ts b/frontend/src/types/api.ts index 80f723da9..30b6aed2a 100644 --- a/frontend/src/types/api.ts +++ b/frontend/src/types/api.ts @@ -524,7 +524,7 @@ export interface SupplierQualityMetric { numerator: number | null; denominator: number | null; share: number | null; - /** Set when status is unavailable, or late_cte has no agreed sending time. */ + /** Set when status is unavailable, when late_cte has no agreed sending time, and on epcis_failed to say the count is tenant wide. */ reason: string | null; } diff --git a/services/admin/app/supplier_compliance_routes.py b/services/admin/app/supplier_compliance_routes.py index 594301a7a..cb7e39a0a 100644 --- a/services/admin/app/supplier_compliance_routes.py +++ b/services/admin/app/supplier_compliance_routes.py @@ -268,7 +268,9 @@ async def get_supplier_data_quality( ) -> SupplierDataQualityResponse: """Four separate data quality measures per facility over a rolling window. - Scoped to the tenant and to facilities owned by the calling user. + Scoped to the tenant and to facilities owned by the calling user, except + epcis_failed: EPCIS receipts carry no facility, so that share is tenant + wide and repeated on every facility. """ tenant_id = TenantContext.get_tenant_context(db) if not tenant_id: diff --git a/services/admin/tests/test_supplier_data_quality_routes.py b/services/admin/tests/test_supplier_data_quality_routes.py index 638c0d610..a4816e6de 100644 --- a/services/admin/tests/test_supplier_data_quality_routes.py +++ b/services/admin/tests/test_supplier_data_quality_routes.py @@ -15,6 +15,8 @@ from fastapi.testclient import TestClient from shared.supplier_data_quality import ( DEFAULT_WINDOW_DAYS, + EPCIS_TENANT_SCOPE_REASON, + EPCIS_UNAVAILABLE_REASON, LATE_CTE_LABEL, LATE_CTE_NO_HOURS_REASON, MAX_WINDOW_DAYS, @@ -60,6 +62,11 @@ ] +# Tenant wide EPCIS receipts: enough for a share, every other one rejected. +RECEIPTS = [uuid4() for _ in range(MIN_RECORDS)] +REJECTED_RECEIPTS = RECEIPTS[::2] + + class FakeSession: def __init__(self, own_hours: int | None = None) -> None: self.commits = 0 @@ -98,7 +105,14 @@ def calls(): return [] -def _client(monkeypatch, calls, tenant=TENANT, session=None, role=None): +def _client( + monkeypatch, + calls, + tenant=TENANT, + session=None, + role=None, + receipt_counts=(len(REJECTED_RECEIPTS), len(RECEIPTS)), +): import app.supplier_compliance_routes as routes # Membership role of the caller in the bound tenant; None means no membership. @@ -106,6 +120,10 @@ def _client(monkeypatch, calls, tenant=TENANT, session=None, role=None): session = session if session is not None else FakeSession() + def fake_receipt_counts(_db, tenant_id, since): + calls.append(("receipts", tenant_id)) + return receipt_counts + def fake_records(_db, tenant_id, user_id, facility_ids, since): calls.append(("records", tenant_id, user_id, list(facility_ids))) return [dict(row) for row in RECORD_ROWS] @@ -124,6 +142,7 @@ def fake_hours(db, tenant_id, user_id, facility_ids): monkeypatch.setattr(routes, "fetch_lot_rows", fake_lots) monkeypatch.setattr("shared.supplier_data_quality.fetch_record_rows", fake_records) monkeypatch.setattr("shared.supplier_data_quality.fetch_lot_rows", fake_lots) + monkeypatch.setattr("shared.supplier_data_quality.fetch_receipt_counts", fake_receipt_counts) monkeypatch.setattr(routes.TenantContext, "get_tenant_context", lambda _db: tenant) app = FastAPI() @@ -152,9 +171,39 @@ def test_summary_returns_four_separate_metrics_for_owned_facilities(monkeypatch, assert supplier["agreed_sending_hours"] is None late = supplier["metrics"][2] assert (late["status"], late["numerator"], late["reason"]) == ("not_enough_data", None, LATE_CTE_NO_HOURS_REASON) - assert supplier["metrics"][3]["status"] == "unavailable" - assert all(call[1:3] == (str(TENANT), str(USER)) for call in calls) - assert all(call[3] == [str(OWN_FACILITY)] for call in calls) + epcis = supplier["metrics"][3] + assert (epcis["status"], epcis["numerator"], epcis["denominator"], epcis["share"], epcis["reason"]) == ( + "ok", + len(REJECTED_RECEIPTS), + len(RECEIPTS), + len(REJECTED_RECEIPTS) / len(RECEIPTS), + EPCIS_TENANT_SCOPE_REASON, + ) + scoped = [call for call in calls if call[0] != "receipts"] + assert all(call[1:3] == (str(TENANT), str(USER)) for call in scoped) + assert all(call[3] == [str(OWN_FACILITY)] for call in scoped) + # Receipts carry no facility: one tenant scoped count, no user or facility filter. + assert [call for call in calls if call[0] == "receipts"] == [("receipts", str(TENANT))] + + +def test_epcis_under_min_records_is_not_enough_data(monkeypatch, calls): + receipts = RECEIPTS[: MIN_RECORDS - 1] + rejected = receipts[::2] + client = _client(monkeypatch, calls, receipt_counts=(len(rejected), len(receipts))) + [supplier] = client.get("/v1/supplier/data-quality").json()["suppliers"] + epcis = supplier["metrics"][3] + assert (epcis["status"], epcis["numerator"], epcis["denominator"], epcis["share"]) == ( + "not_enough_data", + len(rejected), + len(receipts), + None, + ) + + +def test_epcis_unreadable_receipts_is_unavailable(monkeypatch, calls): + [supplier] = _client(monkeypatch, calls, receipt_counts=None).get("/v1/supplier/data-quality").json()["suppliers"] + epcis = supplier["metrics"][3] + assert (epcis["status"], epcis["reason"]) == ("unavailable", EPCIS_UNAVAILABLE_REASON) @pytest.mark.parametrize("window", [0, MAX_WINDOW_DAYS + 1, -1]) diff --git a/services/ingestion/app/epcis/router.py b/services/ingestion/app/epcis/router.py index 2978f9abc..91c4c5819 100644 --- a/services/ingestion/app/epcis/router.py +++ b/services/ingestion/app/epcis/router.py @@ -6,12 +6,20 @@ from __future__ import annotations +import json from datetime import datetime, timezone from typing import Optional from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request from fastapi.responses import JSONResponse from pydantic import BaseModel, Field +from shared.epcis_receipts import ( + RECEIPT_ACCEPTED, + RECEIPT_REJECTED, + record_epcis_receipt, + summarize_error, + summarize_failed_events, +) from shared.tenant_context import TenantContext from ..shared.tenant_resolution import resolve_tenant_id @@ -81,6 +89,31 @@ def _resolve_authenticated_tenant( return resolved_tenant +def _receipt_for_error(tenant_id: Optional[str], raw_body: bytes, exc: BaseException) -> None: + """Rejected receipt for a document whose ingest raised. Never raises.""" + if tenant_id: + record_epcis_receipt(tenant_id, raw_body, RECEIPT_REJECTED, summarize_error(exc)) + + +def _receipt_for_response(tenant_id: Optional[str], raw_body: bytes, response: JSONResponse) -> None: + """Receipt for a document that got a response: rejected from HTTP 400 up. + + A partial batch (207) is accepted, with the failed event count as summary. + """ + if not tenant_id: + return + try: + content = json.loads(response.body) + failed = content.get("failed") if isinstance(content, dict) else None + total = content.get("total") if isinstance(content, dict) else None + counts_are_ints = isinstance(failed, int) and isinstance(total, int) + summary = summarize_failed_events(failed, total) if counts_are_ints and failed else None + except (ValueError, TypeError): + summary = None + status = RECEIPT_REJECTED if response.status_code >= 400 else RECEIPT_ACCEPTED + record_epcis_receipt(tenant_id, raw_body, status, summary) + + class BatchIngestRequest(BaseModel): events: list[dict] = Field(default_factory=list) @@ -135,9 +168,17 @@ async def ingest_epcis_event( ctx: TenantContext = Depends(require_tenant_context), ): resolved_tenant = ctx.tenant_id + # FastAPI already read and cached the body to parse ``event``. + raw_body = await request.body() - payload, status_code = _ingest_single_event(resolved_tenant, event) - return JSONResponse(content=payload, status_code=status_code) + try: + payload, status_code = _ingest_single_event(resolved_tenant, event) + response = JSONResponse(content=payload, status_code=status_code) + except Exception as exc: + _receipt_for_error(resolved_tenant, raw_body, exc) + raise + _receipt_for_response(resolved_tenant, raw_body, response) + return response @router.post("/events/batch", summary="Batch ingest EPCIS events") @@ -159,6 +200,28 @@ async def ingest_epcis_batch( x_regengine_api_key: Optional[str] = Header(default=None, alias="X-RegEngine-API-Key"), _: TenantContext | None = Depends(require_tenant_context), ): + # Without a dependency context (direct calls) the tenant is resolved + # inside _ingest_batch_document, so no tenant is known up front and no + # receipt is filed. + receipt_tenant = _.tenant_id if _ is not None else None + raw_body = await request.body() if receipt_tenant else b"" + try: + response = _ingest_batch_document(request, body, mode, x_tenant_id, x_regengine_api_key, _) + except Exception as exc: + _receipt_for_error(receipt_tenant, raw_body, exc) + raise + _receipt_for_response(receipt_tenant, raw_body, response) + return response + + +def _ingest_batch_document( + request: Request, + body: BatchIngestRequest, + mode: Optional[str], + x_tenant_id: Optional[str], + x_regengine_api_key: Optional[str], + _: TenantContext | None, +) -> JSONResponse: if not body.events: raise HTTPException(status_code=400, detail="Batch ingest requires at least one event") @@ -378,9 +441,17 @@ async def ingest_epcis_xml( # Stream the body with a hard cap. RequestSizeLimitMiddleware only enforces # the Content-Length header, so a chunked upload would otherwise buffer an # unbounded XML document into memory here before parsing (DoS). - raw_body = await read_request_body_with_limit( - request, max_bytes=MAX_FILE_SIZE_BYTES, label="EPCIS XML" - ) + raw_body = await read_request_body_with_limit(request, max_bytes=MAX_FILE_SIZE_BYTES, label="EPCIS XML") + try: + response = _ingest_xml_document(resolved_tenant, raw_body, mode) + except Exception as exc: + _receipt_for_error(resolved_tenant, raw_body, exc) + raise + _receipt_for_response(resolved_tenant, raw_body, response) + return response + + +def _ingest_xml_document(resolved_tenant: str, raw_body: bytes, mode: str) -> JSONResponse: if not raw_body: raise HTTPException(status_code=400, detail="Empty XML payload") diff --git a/services/ingestion/tests/test_epcis_document_receipts.py b/services/ingestion/tests/test_epcis_document_receipts.py new file mode 100644 index 000000000..d3d390747 --- /dev/null +++ b/services/ingestion/tests/test_epcis_document_receipts.py @@ -0,0 +1,293 @@ +"""EPCIS ingest routes write one document receipt per request body. + +Persistence is stubbed and ``record_epcis_receipt`` is captured, so the tests +check what each route files (tenant, SHA-256 of the raw body, status, summary) +and that a failing receipt write leaves the ingest response unchanged. Ids and +payload values are generated per run. +""" + +from __future__ import annotations + +import importlib +import json +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient + +service_dir = Path(__file__).parent.parent +sys.path.insert(0, str(service_dir)) + +router_mod = importlib.import_module("app.epcis.router") +from app.webhook_compat import _verify_api_key # noqa: E402 +from shared.epcis_receipts import ( # noqa: E402 + MAX_ERROR_SUMMARY_LENGTH, + RECEIPT_ACCEPTED, + RECEIPT_REJECTED, + document_sha256, +) +from shared.supplier_data_quality import MIN_RECORDS # noqa: E402 + +from services.ingestion.tests.tenant_context_overrides import install_tenant_context_override # noqa: E402 + +TENANT = str(uuid4()) +# Batch size comes from the spec constant, not a picked number. +BATCH_SIZE = MIN_RECORDS + + +@pytest.fixture +def receipts(monkeypatch: pytest.MonkeyPatch) -> list[dict]: + filed: list[dict] = [] + + def _capture(tenant_id, raw_body, status, error_summary=None, session_factory=None): + filed.append({"tenant_id": tenant_id, "raw_body": raw_body, "status": status, "error_summary": error_summary}) + return True + + monkeypatch.setattr(router_mod, "record_epcis_receipt", _capture) + return filed + + +@pytest.fixture +def client() -> TestClient: + app = FastAPI() + app.include_router(router_mod.router) + app.dependency_overrides[_verify_api_key] = lambda: None + install_tenant_context_override(app, default_tenant=TENANT, enforce_claims=True) + with TestClient(app) as c: + yield c + + +def _json_body(payload: dict) -> bytes: + return json.dumps(payload).encode() + + +def _post_json(client: TestClient, path: str, payload: dict): + raw = _json_body(payload) + return raw, client.post(path, content=raw, headers={"Content-Type": "application/json"}) + + +def _event() -> dict: + return {"id": str(uuid4()), "value": str(uuid4())} + + +def _assert_receipt(receipt: dict, raw: bytes, status: str) -> None: + assert receipt["tenant_id"] == TENANT + assert receipt["raw_body"] == raw + assert document_sha256(receipt["raw_body"]) == document_sha256(raw) + assert receipt["status"] == status + + +# ── single event ───────────────────────────────────────────────────────────── + + +def test_accepted_event_files_accepted_receipt_with_raw_body(client, receipts, monkeypatch): + created_id = str(uuid4()) + monkeypatch.setattr(router_mod, "_ingest_single_event", lambda t, e: ({"cte_id": created_id}, 201)) + raw, resp = _post_json(client, "/api/v1/epcis/events", _event()) + + assert resp.status_code == 201 + assert resp.json() == {"cte_id": created_id} + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_ACCEPTED) + assert receipt["error_summary"] is None + + +def test_validation_failure_files_rejected_receipt_without_payload(client, receipts, monkeypatch): + event = _event() + + def _reject(_tenant, payload): + # The message quotes the payload, as real validation messages can. + raise HTTPException( + status_code=422, + detail={"error": "fsma_validation_failed", "message": f"bad value {payload['value']}"}, + ) + + monkeypatch.setattr(router_mod, "_ingest_single_event", _reject) + raw, resp = _post_json(client, "/api/v1/epcis/events", event) + + assert resp.status_code == 422 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_REJECTED) + assert receipt["error_summary"] == "HTTPException 422 fsma_validation_failed" + assert event["value"] not in receipt["error_summary"] + + +def test_unexpected_error_files_rejected_receipt_with_class_only(client, receipts, monkeypatch): + event = _event() + + def _boom(_tenant, payload): + raise RuntimeError(payload["value"]) + + monkeypatch.setattr(router_mod, "_ingest_single_event", _boom) + raw = _json_body(event) + with pytest.raises(RuntimeError): + client.post("/api/v1/epcis/events", content=raw, headers={"Content-Type": "application/json"}) + + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_REJECTED) + assert receipt["error_summary"] == RuntimeError.__name__ + + +def test_receipt_write_failure_does_not_change_the_response(client, monkeypatch): + created_id = str(uuid4()) + monkeypatch.setattr(router_mod, "_ingest_single_event", lambda t, e: ({"cte_id": created_id}, 201)) + + def _broken_session(): + raise ConnectionError(str(uuid4())) + + # The real writer runs; only the session factory fails. + monkeypatch.setattr("shared.database.get_db_safe", _broken_session) + _, resp = _post_json(client, "/api/v1/epcis/events", _event()) + + assert resp.status_code == 201 + assert resp.json() == {"cte_id": created_id} + + +def test_receipt_write_failure_keeps_the_rejection(client, monkeypatch): + def _reject(_tenant, _payload): + raise HTTPException(status_code=422, detail={"error": "fsma_validation_failed"}) + + monkeypatch.setattr(router_mod, "_ingest_single_event", _reject) + monkeypatch.setattr("shared.database.get_db_safe", lambda: None) + _, resp = _post_json(client, "/api/v1/epcis/events", _event()) + + assert resp.status_code == 422 + assert resp.json()["detail"] == {"error": "fsma_validation_failed"} + + +# ── batch ──────────────────────────────────────────────────────────────────── + + +def test_atomic_batch_files_one_accepted_receipt(client, receipts, monkeypatch): + events = [_event() for _ in range(BATCH_SIZE)] + monkeypatch.setattr( + router_mod, "_ingest_batch_events_db_atomic", lambda t, evs: [({"cte_id": str(uuid4())}, 201) for _ in evs] + ) + raw, resp = _post_json(client, "/api/v1/epcis/events/batch?mode=atomic", {"events": events}) + + assert resp.status_code == 201 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_ACCEPTED) + + +def test_atomic_batch_validation_failure_files_rejected_receipt(client, receipts, monkeypatch): + def _reject(_tenant, _events): + raise HTTPException(status_code=400, detail={"error": "batch_validation_failed", "errors": _events}) + + monkeypatch.setattr(router_mod, "_ingest_batch_events_db_atomic", _reject) + events = [_event() for _ in range(BATCH_SIZE)] + raw, resp = _post_json(client, "/api/v1/epcis/events/batch?mode=atomic", {"events": events}) + + assert resp.status_code == 400 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_REJECTED) + assert receipt["error_summary"] == "HTTPException 400 batch_validation_failed" + assert all(event["value"] not in receipt["error_summary"] for event in events) + + +def test_partial_batch_with_some_failures_is_accepted_with_failed_count(client, receipts, monkeypatch): + events = [_event() for _ in range(BATCH_SIZE)] + failing = {event["id"] for event in events[::2]} + + def _single(_tenant, event): + if event["id"] in failing: + raise HTTPException(status_code=422, detail=event["value"]) + return {"cte_id": str(uuid4())}, 201 + + monkeypatch.setattr(router_mod, "_ingest_single_event", _single) + raw, resp = _post_json(client, "/api/v1/epcis/events/batch?mode=partial", {"events": events}) + + assert resp.status_code == 207 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_ACCEPTED) + assert receipt["error_summary"] == f"{len(failing)} of {len(events)} events failed" + + +def test_partial_batch_with_every_event_failing_is_rejected(client, receipts, monkeypatch): + events = [_event() for _ in range(BATCH_SIZE)] + + def _single(_tenant, event): + raise HTTPException(status_code=422, detail=event["value"]) + + monkeypatch.setattr(router_mod, "_ingest_single_event", _single) + raw, resp = _post_json(client, "/api/v1/epcis/events/batch?mode=partial", {"events": events}) + + assert resp.status_code == 400 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_REJECTED) + assert receipt["error_summary"] == f"{len(events)} of {len(events)} events failed" + + +def test_empty_batch_files_rejected_receipt(client, receipts): + raw, resp = _post_json(client, "/api/v1/epcis/events/batch", {"events": []}) + + assert resp.status_code == 400 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_REJECTED) + assert receipt["error_summary"] == "HTTPException 400" + + +# ── XML ────────────────────────────────────────────────────────────────────── + + +def _post_xml(client: TestClient, raw: bytes, mode: str = "atomic"): + return client.post( + f"/api/v1/epcis/events/xml?mode={mode}", content=raw, headers={"Content-Type": "application/xml"} + ) + + +def test_xml_document_files_accepted_receipt(client, receipts, monkeypatch): + raw = uuid4().hex.encode() + events = [_event() for _ in range(BATCH_SIZE)] + monkeypatch.setattr(router_mod, "_is_xml_content", lambda body: True) + monkeypatch.setattr(router_mod, "_parse_epcis_xml", lambda body: events) + monkeypatch.setattr( + router_mod, "_ingest_batch_events_db_atomic", lambda t, evs: [({"cte_id": str(uuid4())}, 201) for _ in evs] + ) + + assert _post_xml(client, raw).status_code == 201 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_ACCEPTED) + + +def test_xml_without_events_files_rejected_receipt(client, receipts, monkeypatch): + raw = uuid4().hex.encode() + monkeypatch.setattr(router_mod, "_is_xml_content", lambda body: True) + monkeypatch.setattr(router_mod, "_parse_epcis_xml", lambda body: []) + + assert _post_xml(client, raw).status_code == 422 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_REJECTED) + assert receipt["error_summary"] == "HTTPException 422" + + +def test_non_xml_body_files_rejected_receipt(client, receipts, monkeypatch): + raw = uuid4().hex.encode() + monkeypatch.setattr(router_mod, "_is_xml_content", lambda body: False) + + assert _post_xml(client, raw).status_code == 400 + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_REJECTED) + assert len(receipt["error_summary"]) <= MAX_ERROR_SUMMARY_LENGTH + + +def test_invalid_mode_is_rejected_before_the_body_is_read(client, receipts): + assert _post_xml(client, uuid4().hex.encode(), mode=uuid4().hex).status_code == 400 + assert receipts == [] + + +@pytest.mark.parametrize( + "counts", [{"failed": str(BATCH_SIZE), "total": BATCH_SIZE}, {"failed": BATCH_SIZE, "total": None}] +) +def test_partial_response_with_non_int_counts_files_no_summary(receipts, counts): + from fastapi.responses import JSONResponse + + raw = uuid4().hex.encode() + router_mod._receipt_for_response(TENANT, raw, JSONResponse(counts, status_code=207)) + + [receipt] = receipts + _assert_receipt(receipt, raw, RECEIPT_ACCEPTED) + assert receipt["error_summary"] is None diff --git a/services/shared/epcis_receipts.py b/services/shared/epcis_receipts.py new file mode 100644 index 000000000..cc144644e --- /dev/null +++ b/services/shared/epcis_receipts.py @@ -0,0 +1,162 @@ +"""EPCIS document receipts: ``fsma.epcis_document_receipts``. + +One row per EPCIS document an ingest route receives, accepted or rejected, +keyed by the SHA-256 of the raw request body. ``error_summary`` is built by +:func:`summarize_error` and never carries payload content: only the exception +class, the HTTP status, and the machine error code of a dict ``detail``. Free +text details are dropped, because messages can interpolate payload values. A +failed event count may be added by the caller. + +Receipts are tenant scoped. An EPCIS upload is not linked to a supplier +facility, so the supplier data quality ``epcis_failed`` metric is a tenant +wide share (see ``shared.supplier_data_quality``). + +Writing a receipt is fail open: :func:`record_epcis_receipt` never raises, so +a receipt failure cannot change an ingest response. +""" + +from __future__ import annotations + +import hashlib +import logging +import re +import uuid +from datetime import datetime +from typing import Any, Callable, Mapping, Optional + +from sqlalchemy import text + +logger = logging.getLogger(__name__) + +RECEIPT_ACCEPTED = "accepted" +RECEIPT_REJECTED = "rejected" +RECEIPT_STATUSES = (RECEIPT_ACCEPTED, RECEIPT_REJECTED) +# Matches the error_summary length check in alembic v137. +MAX_ERROR_SUMMARY_LENGTH = 200 + +# A machine error code such as ``batch_validation_failed``. The EPCIS ingest +# path sets ``detail["error"]`` from string literals only. +_SAFE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") + +_INSERT_SQL = text( + """ + INSERT INTO fsma.epcis_document_receipts (tenant_id, document_sha256, status, error_summary) + VALUES (CAST(:tenant_id AS uuid), :document_sha256, :status, :error_summary) + """ +) + +_COUNTS_SQL = text( + """ + SELECT count(*) FILTER (WHERE status = 'rejected') AS rejected, + count(*) AS total + FROM fsma.epcis_document_receipts + WHERE tenant_id = CAST(:tid AS uuid) + AND received_at >= :since + """ +) + + +def document_sha256(raw_body: bytes) -> str: + return hashlib.sha256(raw_body).hexdigest() + + +def _truncate(summary: str) -> str: + return summary[:MAX_ERROR_SUMMARY_LENGTH] + + +def summarize_error(exc: BaseException) -> str: + """Exception class, HTTP status and the dict detail's error code. + + ``str(exc)`` and string details are never used: either can quote the + payload. + """ + parts = [type(exc).__name__] + status_code = getattr(exc, "status_code", None) + if isinstance(status_code, int): + parts.append(str(status_code)) + detail = getattr(exc, "detail", None) + code = detail.get("error") if isinstance(detail, Mapping) else None + if isinstance(code, str) and _SAFE_CODE.match(code): + parts.append(code) + return _truncate(" ".join(parts)) + + +def summarize_failed_events(failed: int, total: int) -> str: + return _truncate(f"{failed} of {total} events failed") + + +def record_epcis_receipt( + tenant_id: Any, + raw_body: bytes, + status: str, + error_summary: Optional[str] = None, + session_factory: Optional[Callable[[], Any]] = None, +) -> bool: + """Insert one receipt in its own transaction. Never raises. + + Returns True when the row was committed. A tenant id that is not a UUID + is skipped: the column is uuid and the RLS check casts the tenant GUC. + """ + try: + tenant = str(uuid.UUID(str(tenant_id))) + except (ValueError, AttributeError, TypeError): + logger.warning("epcis_receipt_skipped reason=tenant_not_uuid") + return False + if status not in RECEIPT_STATUSES: + logger.warning("epcis_receipt_skipped reason=unknown_status") + return False + + session = None + try: + if session_factory is None: + from shared.database import get_db_safe # noqa: PLC0415 (no engine setup at import) + + session_factory = get_db_safe + session = session_factory() + if session is None: + logger.warning("epcis_receipt_write_failed tenant=%s error_class=NoSession", tenant) + return False + from shared.tenant_context import set_tenant_guc # noqa: PLC0415 + + set_tenant_guc(session, tenant) + session.execute( + _INSERT_SQL, + { + "tenant_id": tenant, + "document_sha256": document_sha256(raw_body), + "status": status, + "error_summary": _truncate(error_summary) if error_summary else None, + }, + ) + session.commit() + return True + except Exception as exc: # noqa: BLE001 -- receipts must never change the ingest response + logger.warning("epcis_receipt_write_failed tenant=%s error_class=%s", tenant, type(exc).__name__) + if session is not None: + try: + session.rollback() + except Exception: # noqa: BLE001 + pass + return False + finally: + if session is not None: + try: + session.close() + except Exception: # noqa: BLE001 + pass + + +def fetch_receipt_counts(db_session: Any, tenant_id: str, since: datetime) -> Optional[tuple[int, int]]: + """(rejected, total) receipts for the tenant since ``since``. + + Runs in a savepoint and returns None when the table cannot be read (for + example before alembic v137 is applied), so the caller's transaction and + the other metrics are unaffected. + """ + try: + with db_session.begin_nested(): + row = db_session.execute(_COUNTS_SQL, {"tid": tenant_id, "since": since}).mappings().one() + except Exception as exc: # noqa: BLE001 -- one unreadable metric must not fail the summary + logger.warning("epcis_receipt_counts_failed error_class=%s", type(exc).__name__) + return None + return int(row["rejected"]), int(row["total"]) diff --git a/services/shared/supplier_data_quality.py b/services/shared/supplier_data_quality.py index db91316f8..999e6dc92 100644 --- a/services/shared/supplier_data_quality.py +++ b/services/shared/supplier_data_quality.py @@ -21,11 +21,17 @@ ``event_time`` (received is ``created_at``). The denominator is the records in the window with both times. Our measure against the agreed sending time, not an FDA rule. With no agreed hours set, the status is not enough data. -4. ``epcis_failed``: unavailable. EPCIS documents are not stored as files with - a validation result, and nothing ties an EPCIS upload to a facility. - -Read only. Every query filters on tenant, on the calling supplier user and on -that user's facility ids. +4. ``epcis_failed``: share of EPCIS documents received in the window that were + rejected (``fsma.epcis_document_receipts``, written by the EPCIS ingest + routes). Receipts are tenant scoped: an EPCIS upload carries no supplier + facility, so this cannot be mapped per facility. The tenant wide share is + computed once and shown on every facility, with + ``EPCIS_TENANT_SCOPE_REASON`` saying so. It is unavailable only when the + receipts table cannot be read. + +Read only. The record and lot queries filter on tenant, on the calling +supplier user and on that user's facility ids. The EPCIS receipt count +filters on tenant only (see above). """ from __future__ import annotations @@ -33,6 +39,7 @@ from datetime import datetime, timedelta from typing import Any, Mapping, Optional, Sequence +from shared.epcis_receipts import fetch_receipt_counts from sqlalchemy import text MIN_RECORDS = 20 @@ -47,10 +54,11 @@ LATE_CTE_LABEL = "Our measure, based on the agreed sending time with this supplier; not a regulatory requirement" LATE_CTE_NO_HOURS_REASON = "No agreed sending time is set for this supplier, so late records cannot be measured." -EPCIS_UNAVAILABLE_REASON = ( - "EPCIS uploads are not stored as files with a validation result linked to a facility, " - "so failing files cannot be counted." +EPCIS_TENANT_SCOPE_REASON = ( + "Counted for your whole account, not for this facility: an EPCIS upload is not linked " + "to a facility, so every facility shows the same count." ) +EPCIS_UNAVAILABLE_REASON = "EPCIS document receipts could not be read." # Copy of services/ingestion/app/webhook_models.py REQUIRED_KDES_BY_CTE, keyed # by supplier CTE name. The admin service cannot import the ingestion package @@ -265,6 +273,14 @@ def unavailable_metric(key: str, reason: str) -> dict[str, Any]: } +def epcis_failed_metric(counts: Optional[tuple[int, int]]) -> dict[str, Any]: + """Tenant wide rejected share from (rejected, total) receipt counts.""" + if counts is None: + return unavailable_metric(METRIC_EPCIS_FAILED, EPCIS_UNAVAILABLE_REASON) + rejected, total = counts + return {**share_metric(METRIC_EPCIS_FAILED, rejected, total), "reason": EPCIS_TENANT_SCOPE_REASON} + + def window_start(now: datetime, window_days: int) -> datetime: return now - timedelta(days=window_days) @@ -311,14 +327,18 @@ def summarize_facility( record_rows: Sequence[Mapping[str, Any]], lot_rows: Sequence[Mapping[str, Any]], agreed_sending_hours: Optional[int] = None, + epcis_counts: Optional[tuple[int, int]] = None, ) -> dict[str, dict[str, Any]]: - """The four metrics for one facility, keyed by metric.""" + """The four metrics for one facility, keyed by metric. + + ``epcis_counts`` is the tenant wide (rejected, total), not per facility. + """ lot_count = len(group_lots(lot_rows)) return { METRIC_MISSING_KDE: share_metric(METRIC_MISSING_KDE, len(flag_records(record_rows)), len(record_rows)), METRIC_UNLINKED_LOT: share_metric(METRIC_UNLINKED_LOT, len(unlinked_lots(lot_rows)), lot_count), METRIC_LATE_CTE: late_metric(record_rows, agreed_sending_hours), - METRIC_EPCIS_FAILED: unavailable_metric(METRIC_EPCIS_FAILED, EPCIS_UNAVAILABLE_REASON), + METRIC_EPCIS_FAILED: epcis_failed_metric(epcis_counts), } @@ -449,13 +469,19 @@ def fetch_supplier_quality( ``sending_hours`` is fetched when the caller has not already done so. """ + if not facility_ids: + return {} records = _by_facility( fetch_record_rows(db_session, tenant_id, supplier_user_id, facility_ids, since), facility_ids ) lots = _by_facility(fetch_lot_rows(db_session, tenant_id, supplier_user_id, facility_ids, since), facility_ids) + epcis_counts = fetch_receipt_counts(db_session, tenant_id, since) hours = ( sending_hours if sending_hours is not None else fetch_sending_hours(db_session, tenant_id, supplier_user_id, facility_ids) ) - return {key: summarize_facility(records[key], lots[key], hours.get(key)) for key in facility_ids} + return { + key: summarize_facility(records[key], lots[key], hours.get(key), epcis_counts) + for key in facility_ids + } diff --git a/services/shared/tests/test_epcis_receipts.py b/services/shared/tests/test_epcis_receipts.py new file mode 100644 index 000000000..6f8dfab36 --- /dev/null +++ b/services/shared/tests/test_epcis_receipts.py @@ -0,0 +1,202 @@ +"""EPCIS document receipts: hashing, sanitized summaries, fail open writes. + +No database. Fake sessions record what the writer and the count reader send. +""" + +from __future__ import annotations + +import hashlib +import uuid +from contextlib import nullcontext +from datetime import datetime, timezone + +import pytest +from fastapi import HTTPException +from shared.epcis_receipts import ( + MAX_ERROR_SUMMARY_LENGTH, + RECEIPT_ACCEPTED, + RECEIPT_REJECTED, + RECEIPT_STATUSES, + document_sha256, + fetch_receipt_counts, + record_epcis_receipt, + summarize_error, + summarize_failed_events, +) + + +class FakeSession: + def __init__(self, fail_on_insert: bool = False) -> None: + self.statements: list[tuple[str, dict]] = [] + self.fail_on_insert = fail_on_insert + self.committed = False + self.rolled_back = False + self.closed = False + + def execute(self, statement, params=None): + sql = str(statement) + if self.fail_on_insert and "INSERT" in sql: + raise RuntimeError(str(uuid.uuid4())) + self.statements.append((sql, dict(params or {}))) + + def commit(self): + self.committed = True + + def rollback(self): + self.rolled_back = True + + def close(self): + self.closed = True + + +def _raw() -> bytes: + return uuid.uuid4().hex.encode() + + +# --- hashing and summaries --------------------------------------------------- + + +def test_document_sha256_is_hex_digest_of_raw_body(): + raw = _raw() + assert document_sha256(raw) == hashlib.sha256(raw).hexdigest() + + +def test_statuses_match_the_table_check_constraint(): + assert RECEIPT_STATUSES == ("accepted", "rejected") + + +def test_summary_keeps_class_status_and_error_code_only(): + secret = str(uuid.uuid4()) + exc = HTTPException(status_code=422, detail={"error": "fsma_validation_failed", "message": secret}) + assert summarize_error(exc) == "HTTPException 422 fsma_validation_failed" + + +@pytest.mark.parametrize( + "detail_template", + ["Unmapped bizStep: {}", "EPCIS event '{}' not found", "{}", "Unsupported event type {}"], +) +def test_summary_drops_string_details(detail_template): + # Plain words included: a string detail can interpolate a payload value + # with no punctuation around it. + value = uuid.uuid4().hex + summary = summarize_error(HTTPException(status_code=400, detail=detail_template.format(value))) + assert value not in summary + assert summary == "HTTPException 400" + + +def test_summary_drops_an_error_code_that_is_not_a_code(): + value = f"{uuid.uuid4()} {uuid.uuid4()}" + summary = summarize_error(HTTPException(status_code=400, detail={"error": value})) + assert summary == "HTTPException 400" + + +def test_summary_of_plain_exception_is_class_name_only(): + value = str(uuid.uuid4()) + assert summarize_error(ValueError(value)) == ValueError.__name__ + + +def test_summary_is_truncated(): + long_name = "E" * (MAX_ERROR_SUMMARY_LENGTH * 2) + exc = type(long_name, (Exception,), {})() + assert len(summarize_error(exc)) == MAX_ERROR_SUMMARY_LENGTH + + +def test_failed_event_summary_counts_only(): + failed, total = len(RECEIPT_STATUSES), MAX_ERROR_SUMMARY_LENGTH + assert summarize_failed_events(failed, total) == f"{failed} of {total} events failed" + + +# --- writer ------------------------------------------------------------------ + + +def test_record_sets_tenant_guc_then_inserts_and_commits(): + tenant = str(uuid.uuid4()) + raw = _raw() + session = FakeSession() + + assert record_epcis_receipt(tenant, raw, RECEIPT_ACCEPTED, session_factory=lambda: session) is True + + (guc_sql, guc_params), (insert_sql, insert_params) = session.statements + assert "set_config('app.tenant_id'" in guc_sql + assert tenant in guc_params.values() + assert "fsma.epcis_document_receipts" in insert_sql + assert insert_params == { + "tenant_id": tenant, + "document_sha256": document_sha256(raw), + "status": RECEIPT_ACCEPTED, + "error_summary": None, + } + assert session.committed and session.closed + + +def test_record_truncates_error_summary(): + session = FakeSession() + summary = "x" * (MAX_ERROR_SUMMARY_LENGTH * 2) + record_epcis_receipt(str(uuid.uuid4()), _raw(), RECEIPT_REJECTED, summary, session_factory=lambda: session) + assert len(session.statements[-1][1]["error_summary"]) == MAX_ERROR_SUMMARY_LENGTH + + +def test_record_never_raises_on_insert_failure(): + session = FakeSession(fail_on_insert=True) + assert record_epcis_receipt(str(uuid.uuid4()), _raw(), RECEIPT_REJECTED, session_factory=lambda: session) is False + assert session.rolled_back and session.closed and not session.committed + + +def test_record_never_raises_when_session_factory_fails(): + def _broken(): + raise ConnectionError(str(uuid.uuid4())) + + assert record_epcis_receipt(str(uuid.uuid4()), _raw(), RECEIPT_ACCEPTED, session_factory=_broken) is False + + +def test_record_without_session_is_skipped(): + assert record_epcis_receipt(str(uuid.uuid4()), _raw(), RECEIPT_ACCEPTED, session_factory=lambda: None) is False + + +def test_record_skips_non_uuid_tenant_without_opening_a_session(): + def _must_not_open(): # pragma: no cover (must not run) + raise AssertionError("no session expected") + + assert record_epcis_receipt(uuid.uuid4().hex[:8], _raw(), RECEIPT_ACCEPTED, session_factory=_must_not_open) is False + + +def test_record_rejects_unknown_status(): + session = FakeSession() + status = uuid.uuid4().hex + assert record_epcis_receipt(str(uuid.uuid4()), _raw(), status, session_factory=lambda: session) is False + assert session.statements == [] + + +# --- count reader ------------------------------------------------------------ + + +class CountSession: + def __init__(self, row: dict | None = None, error: Exception | None = None) -> None: + self.row = row + self.error = error + self.params: dict | None = None + + def begin_nested(self): + return nullcontext() + + def execute(self, statement, params): + if self.error: + raise self.error + self.params = params + row = self.row + return type("Result", (), {"mappings": lambda _self: type("M", (), {"one": lambda _m: row})()})() + + +def test_counts_are_tenant_and_window_scoped(): + tenant = str(uuid.uuid4()) + since = datetime.now(timezone.utc) + rejected, total = len(RECEIPT_STATUSES), MAX_ERROR_SUMMARY_LENGTH + session = CountSession(row={"rejected": rejected, "total": total}) + + assert fetch_receipt_counts(session, tenant, since) == (rejected, total) + assert session.params == {"tid": tenant, "since": since} + + +def test_unreadable_receipts_table_returns_none(): + session = CountSession(error=RuntimeError(str(uuid.uuid4()))) + assert fetch_receipt_counts(session, str(uuid.uuid4()), datetime.now(timezone.utc)) is None diff --git a/services/shared/tests/test_supplier_data_quality.py b/services/shared/tests/test_supplier_data_quality.py index 0197ae499..9847d3082 100644 --- a/services/shared/tests/test_supplier_data_quality.py +++ b/services/shared/tests/test_supplier_data_quality.py @@ -15,6 +15,8 @@ COLUMN_BACKED_KDES, DEFAULT_WINDOW_DAYS, DRILLDOWN_METRICS, + EPCIS_TENANT_SCOPE_REASON, + EPCIS_UNAVAILABLE_REASON, LATE_CTE_NO_HOURS_REASON, METRIC_EPCIS_FAILED, METRIC_LATE_CTE, @@ -22,6 +24,7 @@ METRIC_UNLINKED_LOT, MIN_RECORDS, WEBHOOK_REQUIRED_KDES, + epcis_failed_metric, fetch_sending_hours, fetch_supplier_quality, flag_records, @@ -326,7 +329,9 @@ def test_summarize_facility_keeps_four_separate_metrics(): incomplete = [_record("shipping", {}) for _ in range(MIN_RECORDS - len(complete))] records = complete + incomplete lots = [_lot_row(str(uuid.uuid4()), "shipping", {}) for _ in range(MIN_RECORDS // 2)] - summary = summarize_facility(records, lots) + receipts = [uuid.uuid4() for _ in range(MIN_RECORDS)] + rejected = receipts[::2] + summary = summarize_facility(records, lots, epcis_counts=(len(rejected), len(receipts))) assert set(summary) == {METRIC_MISSING_KDE, METRIC_UNLINKED_LOT, METRIC_LATE_CTE, METRIC_EPCIS_FAILED} assert summary[METRIC_MISSING_KDE]["share"] == len(incomplete) / len(records) @@ -336,7 +341,72 @@ def test_summarize_facility_keeps_four_separate_metrics(): len(lots), ) assert summary[METRIC_LATE_CTE] == late_metric(records, None) - assert summary[METRIC_EPCIS_FAILED]["status"] == "unavailable" + assert summary[METRIC_EPCIS_FAILED]["share"] == len(rejected) / len(receipts) + + +# --- EPCIS failed (tenant wide) ---------------------------------------------- + + +def test_epcis_failed_share_from_receipts_says_it_is_tenant_wide(): + receipts = [uuid.uuid4() for _ in range(MIN_RECORDS)] + rejected = receipts[::2] + metric = epcis_failed_metric((len(rejected), len(receipts))) + assert metric == { + "key": METRIC_EPCIS_FAILED, + "status": "ok", + "numerator": len(rejected), + "denominator": len(receipts), + "share": len(rejected) / len(receipts), + "reason": EPCIS_TENANT_SCOPE_REASON, + } + + +def test_epcis_failed_under_min_records_is_not_enough_data(): + receipts = [uuid.uuid4() for _ in range(MIN_RECORDS - 1)] + rejected = receipts[::2] + metric = epcis_failed_metric((len(rejected), len(receipts))) + assert metric["status"] == "not_enough_data" + assert metric["share"] is None + assert (metric["numerator"], metric["denominator"]) == (len(rejected), len(receipts)) + assert metric["reason"] == EPCIS_TENANT_SCOPE_REASON + + +def test_epcis_failed_without_receipts_is_not_enough_data(): + assert epcis_failed_metric((0, 0))["status"] == "not_enough_data" + + +def test_epcis_failed_is_unavailable_when_receipts_cannot_be_read(): + assert epcis_failed_metric(None) == unavailable_metric(METRIC_EPCIS_FAILED, EPCIS_UNAVAILABLE_REASON) + + +def test_epcis_reasons_use_no_dashes(): + for reason in (EPCIS_TENANT_SCOPE_REASON, EPCIS_UNAVAILABLE_REASON): + assert not any(dash in reason for dash in ("\u2013", "\u2014", " - ")) + + +def test_fetch_supplier_quality_repeats_tenant_epcis_counts_on_every_facility(monkeypatch): + import shared.supplier_data_quality as quality + + tenant = str(uuid.uuid4()) + facilities = [str(uuid.uuid4()) for _ in range(MIN_RECORDS // 10)] + receipts = [uuid.uuid4() for _ in range(MIN_RECORDS)] + rejected = receipts[::2] + calls = [] + + def fake_counts(_db, tenant_id, since): + calls.append(tenant_id) + return len(rejected), len(receipts) + + monkeypatch.setattr(quality, "fetch_record_rows", lambda *args: []) + monkeypatch.setattr(quality, "fetch_lot_rows", lambda *args: []) + monkeypatch.setattr(quality, "fetch_receipt_counts", fake_counts) + result = fetch_supplier_quality( + object(), tenant, str(uuid.uuid4()), facilities, datetime.now(timezone.utc), sending_hours={} + ) + + assert calls == [tenant] + assert set(result) == set(facilities) + assert {result[key][METRIC_EPCIS_FAILED]["share"] for key in facilities} == {len(rejected) / len(receipts)} def test_summarize_facility_measures_late_records_with_agreed_hours(): diff --git a/supabase/migrations/20261024090000_epcis_document_receipts.sql b/supabase/migrations/20261024090000_epcis_document_receipts.sql new file mode 100644 index 000000000..2ca99c03c --- /dev/null +++ b/supabase/migrations/20261024090000_epcis_document_receipts.sql @@ -0,0 +1,81 @@ +-- EPCIS document receipts. +-- +-- One row per EPCIS document the ingest routes receive (POST +-- /api/v1/epcis/events, /events/batch and /events/xml), accepted or rejected. +-- document_sha256 is the SHA-256 of the raw request body. error_summary is a +-- short sanitized summary (exception class, HTTP status, error code or failed +-- event count); it never holds payload content or secrets. +-- +-- Receipts are tenant scoped. An EPCIS upload is not linked to a supplier +-- facility, so the supplier data quality metric epcis_failed is a tenant wide +-- share shown on every facility. +-- +-- Idempotent: safe to re-run. Downgrade block at the end of this file. + +CREATE TABLE IF NOT EXISTS fsma.epcis_document_receipts ( + id uuid primary key default gen_random_uuid(), + tenant_id uuid not null, + document_sha256 text not null, + received_at timestamptz not null default now(), + status text not null, + error_summary text, + constraint epcis_document_receipts_status_check + check (status in ('accepted', 'rejected')), + constraint epcis_document_receipts_sha256_check + check (document_sha256 ~ '^[0-9a-f]{64}$'), + constraint epcis_document_receipts_error_summary_length_check + check (error_summary is null or char_length(error_summary) <= 200) +); + +-- Read path: rejected share per tenant over a rolling window. +create index if not exists idx_epcis_document_receipts_tenant_received + on fsma.epcis_document_receipts (tenant_id, received_at); + +ALTER TABLE fsma.epcis_document_receipts ENABLE ROW LEVEL SECURITY; +ALTER TABLE fsma.epcis_document_receipts FORCE ROW LEVEL SECURITY; + +-- CASE form, not OR: public.get_tenant_context() raises when app.tenant_id is +-- unset and Postgres does not order OR operands, so the sysadmin arm is tested +-- first (same predicate shape as alembic v096, v134 and v135). +drop policy if exists tenant_isolation_epcis_document_receipts on fsma.epcis_document_receipts; +create policy tenant_isolation_epcis_document_receipts on fsma.epcis_document_receipts + for all + using ( + case + when (select current_setting('regengine.is_sysadmin', true)) = 'true' + and current_user = 'regengine_sysadmin' then true + else tenant_id = (select public.get_tenant_context()) + end + ) + with check ( + case + when (select current_setting('regengine.is_sysadmin', true)) = 'true' + and current_user = 'regengine_sysadmin' then true + else tenant_id = (select public.get_tenant_context()) + end + ); + +-- Runtime role allowlist (alembic v111): new tables are inaccessible to the +-- app role until a migration names them. Insert and select only: a receipt is +-- evidence and is never updated or deleted by the app. +do $$ +begin + if exists (select 1 from pg_roles where rolname = 'anon') then + revoke all on fsma.epcis_document_receipts from anon; + end if; + if exists (select 1 from pg_roles where rolname = 'authenticated') then + revoke all on fsma.epcis_document_receipts from authenticated; + end if; + if exists (select 1 from pg_roles where rolname = 'regengine') then + grant select, insert on fsma.epcis_document_receipts to regengine; + end if; + if exists (select 1 from pg_roles where rolname = 'regengine_sysadmin') then + grant select on fsma.epcis_document_receipts to regengine_sysadmin; + end if; +end $$; + +-- Downgrade (run by hand only; drops every receipt): +-- +-- DROP POLICY IF EXISTS tenant_isolation_epcis_document_receipts ON fsma.epcis_document_receipts; +-- DROP INDEX IF EXISTS fsma.idx_epcis_document_receipts_tenant_received; +-- DROP TABLE IF EXISTS fsma.epcis_document_receipts;