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/alembic/versions/20261024_b005ddbf1dba_v136_supplier_agreed_sending_hours.py b/alembic/versions/20261024_b005ddbf1dba_v136_supplier_agreed_sending_hours.py new file mode 100644 index 000000000..7e48ae019 --- /dev/null +++ b/alembic/versions/20261024_b005ddbf1dba_v136_supplier_agreed_sending_hours.py @@ -0,0 +1,43 @@ +"""Agreed sending time per supplier facility. + +Mirrors supabase/migrations/20261024080000_supplier_agreed_sending_hours.sql +exactly: adds public.supplier_facilities.agreed_sending_hours, integer NULL, +CHECK (agreed_sending_hours > 0). No backfill. The supplier data quality view +counts a CTE record as late when it is received more than this many hours +after its event time. + +Additive and idempotent (ADD COLUMN IF NOT EXISTS with an inline CHECK that is +only created together with the column), so the plain SQL file can run ahead of +it. Downgrade drops only what this revision created. + +Revision ID: b005ddbf1dba +Revises: 41f96588d87c +Create Date: 2026-10-24 +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = "b005ddbf1dba" # pragma: allowlist secret +down_revision: Union[str, Sequence[str], None] = "41f96588d87c" # pragma: allowlist secret +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute( + """ +alter table public.supplier_facilities + add column if not exists agreed_sending_hours integer null + constraint supplier_facilities_agreed_sending_hours_check + check (agreed_sending_hours > 0); +""" + ) + + +def downgrade() -> None: + op.execute( + "ALTER TABLE public.supplier_facilities DROP CONSTRAINT IF EXISTS supplier_facilities_agreed_sending_hours_check" + ) + op.execute("ALTER TABLE public.supplier_facilities DROP COLUMN IF EXISTS agreed_sending_hours") diff --git a/docs/schema/expected_schema.json b/docs/schema/expected_schema.json index e59024e33..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": { @@ -6807,6 +6836,10 @@ }, "public.supplier_facilities": { "columns": { + "agreed_sending_hours": { + "nullable": true, + "type": "integer" + }, "city": { "nullable": false, "type": "text" diff --git a/docs/security/route-security-manifest.json b/docs/security/route-security-manifest.json index a4794c99f..9b8fe38dd 100644 --- a/docs/security/route-security-manifest.json +++ b/docs/security/route-security-manifest.json @@ -2774,6 +2774,19 @@ "storage_boundary": "process-observability:no-tenant-data", "test_adapter": "workload_metrics_key" }, + { + "operation": "admin:PUT:/v1/supplier/admin/facilities/{facility_id}/agreed-sending-hours", + "owner": "admin", + "method": "PUT", + "path": "/v1/supplier/admin/facilities/{facility_id}/agreed-sending-hours", + "classification": "tenant-scoped", + "authentication": "bearer-token-or-opaque-session-cookie", + "required_permissions": [ + "tenant.admin" + ], + "storage_boundary": "admin-postgres:jwt-tenant-or-session-membership-tenant", + "test_adapter": "tenant_dual_session_csrf" + }, { "operation": "admin:POST:/v1/supplier/bulk-upload/commit", "owner": "admin", @@ -2852,6 +2865,32 @@ "storage_boundary": "admin-postgres:jwt-tenant-or-session-membership-tenant", "test_adapter": "tenant_dual_session_csrf" }, + { + "operation": "admin:GET:/v1/supplier/data-quality", + "owner": "admin", + "method": "GET", + "path": "/v1/supplier/data-quality", + "classification": "tenant-scoped", + "authentication": "bearer-token-or-opaque-session-cookie", + "required_permissions": [ + "authenticated" + ], + "storage_boundary": "admin-postgres:jwt-tenant-or-session-membership-tenant", + "test_adapter": "tenant_dual_session_csrf" + }, + { + "operation": "admin:GET:/v1/supplier/data-quality/{facility_id}/records", + "owner": "admin", + "method": "GET", + "path": "/v1/supplier/data-quality/{facility_id}/records", + "classification": "tenant-scoped", + "authentication": "bearer-token-or-opaque-session-cookie", + "required_permissions": [ + "authenticated" + ], + "storage_boundary": "admin-postgres:jwt-tenant-or-session-membership-tenant", + "test_adapter": "tenant_dual_session_csrf" + }, { "operation": "admin:POST:/v1/supplier/demo/reset", "owner": "admin", @@ -4918,13 +4957,13 @@ } ], "summary": { - "enabled_operation_count": 255, + "enabled_operation_count": 258, "disabled_operation_count": 180, "disabled_unique_path_count": 163, "enabled_by_classification": { "global-sysadmin": 16, "public": 30, - "tenant-scoped": 202, + "tenant-scoped": 205, "workload-only": 7 } } diff --git a/frontend/src/__tests__/components/supplier-data-quality.test.tsx b/frontend/src/__tests__/components/supplier-data-quality.test.tsx new file mode 100644 index 000000000..af48ad555 --- /dev/null +++ b/frontend/src/__tests__/components/supplier-data-quality.test.tsx @@ -0,0 +1,300 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { fireEvent, render, screen, waitFor } from '@testing-library/react'; +import type { SupplierDataQuality, SupplierQualityMetric, SupplierQualityRecordsResponse } from '@/types/api'; + +const mocks = vi.hoisted(() => ({ + getSupplierDataQualityRecords: vi.fn(), + setFacilityAgreedSendingHours: vi.fn(), +})); + +vi.mock('@/lib/api-client', () => ({ + apiClient: { + getSupplierDataQualityRecords: (...args: unknown[]) => mocks.getSupplierDataQualityRecords(...args), + setFacilityAgreedSendingHours: (...args: unknown[]) => mocks.setFacilityAgreedSendingHours(...args), + }, +})); + +import { DataQuality, DRILLDOWN_PAGE_SIZE } from '@/app/dashboard/suppliers/data-quality'; + +const LATE_LABEL = 'Our measure, based on the agreed sending time with this supplier; not a regulatory requirement'; + +const metric = (overrides: Partial): SupplierQualityMetric => ({ + key: 'missing_kde', + status: 'ok', + numerator: 0, + denominator: 0, + share: null, + reason: null, + ...overrides, +}); + +const FACILITY_ID = crypto.randomUUID(); +const MIN_RECORDS = 20; +const WINDOW_DAYS = 30; +const uuids = (length: number) => Array.from({ length }, () => crypto.randomUUID()); +// Generated inputs: records clear MIN_RECORDS, lots fall one short of it. +// Every other record and lot is flagged, so counts come from these arrays. +const RECORD_IDS = uuids(MIN_RECORDS * 2); +const FLAGGED_RECORD_IDS = RECORD_IDS.filter((_, index) => index % 2 === 0); +const LOT_IDS = uuids(MIN_RECORDS - 1); +const UNLINKED_LOT_IDS = LOT_IDS.filter((_, index) => index % 2 === 0); +const MISSING = { numerator: FLAGGED_RECORD_IDS.length, denominator: RECORD_IDS.length }; +const UNLINKED = { numerator: UNLINKED_LOT_IDS.length, denominator: LOT_IDS.length }; +const LATE_RECORD_IDS = uuids(MIN_RECORDS); +const LATE_FLAGGED_IDS = LATE_RECORD_IDS.filter((_, index) => index % 2 === 0); +const LATE = { numerator: LATE_FLAGGED_IDS.length, denominator: LATE_RECORD_IDS.length }; +// Any positive whole number of hours; derived rather than picked. +const AGREED_HOURS = MIN_RECORDS; +const LATE_REASON = crypto.randomUUID(); +const EPCIS_REASON = crypto.randomUUID(); +const missingShare = MISSING.numerator / MISSING.denominator; +const formatShare = (share: number) => `${Math.round(share * 1000) / 10}%`; + +const summary = (metrics: SupplierQualityMetric[], agreedSendingHours: number | null = null): SupplierDataQuality => ({ + supplier_key: FACILITY_ID, + agreed_sending_hours: agreedSendingHours, + metrics, +}); + +const fourMetrics = summary([ + metric({ key: 'missing_kde', status: 'ok', ...MISSING, share: missingShare }), + metric({ key: 'unlinked_lot', status: 'not_enough_data', ...UNLINKED, share: null }), + metric({ key: 'late_cte', status: 'not_enough_data', numerator: null, denominator: null, reason: LATE_REASON }), + metric({ key: 'epcis_failed', status: 'unavailable', numerator: null, denominator: null, reason: EPCIS_REASON }), +]); + +const lateMeasured = summary( + [metric({ key: 'late_cte', status: 'ok', ...LATE, share: LATE.numerator / LATE.denominator })], + AGREED_HOURS, +); + +function renderQuality(data: SupplierDataQuality | null = fourMetrics, unavailable = false) { + return render( + , + ); +} + +const page = (overrides: Partial): SupplierQualityRecordsResponse => ({ + supplier_key: FACILITY_ID, + metric: 'missing_kde', + window_days: WINDOW_DAYS, + total: 0, + limit: DRILLDOWN_PAGE_SIZE, + offset: 0, + items: [], + ...overrides, +}); + +describe('DataQuality', () => { + beforeEach(() => { + mocks.getSupplierDataQualityRecords.mockReset(); + mocks.setFacilityAgreedSendingHours.mockReset(); + }); + + it('shows count of denominator and a share for each metric, with no combined score', () => { + renderQuality(); + expect(screen.getByText(`${MISSING.numerator} of ${MISSING.denominator} records (${formatShare(missingShare)})`)).toBeInTheDocument(); + expect(screen.queryByText(/score/i)).not.toBeInTheDocument(); + expect(screen.getByText(`Data quality, records received in the last ${WINDOW_DAYS} days`)).toBeInTheDocument(); + }); + + it('shows Not enough data below the minimum instead of a share', () => { + renderQuality(); + expect(screen.getByText(new RegExp(`${UNLINKED.numerator} of ${UNLINKED.denominator} lot codes`))).toBeInTheDocument(); + expect(screen.getByText(`Not enough data (fewer than ${MIN_RECORDS})`)).toBeInTheDocument(); + expect(screen.queryByText(formatShare(UNLINKED.numerator / UNLINKED.denominator))).not.toBeInTheDocument(); + }); + + it('shows unavailable metrics with their reason and the late CTE label', () => { + renderQuality(); + expect(screen.getByText(`Not enough data. ${LATE_REASON}`)).toBeInTheDocument(); + expect(screen.getByText(`Unavailable. ${EPCIS_REASON}`)).toBeInTheDocument(); + 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); + }); + + it('hides drill down when nothing is flagged', () => { + renderQuality(summary([metric({ key: 'missing_kde', numerator: 0, denominator: RECORD_IDS.length, share: 0 })])); + expect(screen.queryByRole('button', { name: 'View records' })).not.toBeInTheDocument(); + }); + + it('loads the exact records for a metric and pages with Show more', async () => { + const [first, second] = uuids(2); + const total = [first, second].length; + mocks.getSupplierDataQualityRecords + .mockResolvedValueOnce( + page({ + total, + items: [ + { + record_id: first, + lot_id: null, + lot_code: crypto.randomUUID(), + cte_type: 'shipping', + event_time: null, + received_at: null, + missing_kdes: ['reference_document', 'quantity'], + }, + ], + }), + ) + .mockResolvedValueOnce( + page({ + total, + offset: 1, + items: [ + { + record_id: second, + lot_id: null, + lot_code: crypto.randomUUID(), + cte_type: 'receiving', + event_time: null, + received_at: null, + missing_kdes: ['tlc_source_reference'], + }, + ], + }), + ); + renderQuality(); + + fireEvent.click(screen.getAllByRole('button', { name: 'View records' })[0]); + await waitFor(() => expect(screen.getByText(first)).toBeInTheDocument()); + expect(screen.getByText('Missing: reference_document, quantity')).toBeInTheDocument(); + expect(mocks.getSupplierDataQualityRecords).toHaveBeenCalledWith(FACILITY_ID, 'missing_kde', { + windowDays: WINDOW_DAYS, + limit: DRILLDOWN_PAGE_SIZE, + offset: 0, + }); + + fireEvent.click(screen.getByRole('button', { name: `Show more (1 of ${total})` })); + await waitFor(() => expect(screen.getByText(second)).toBeInTheDocument()); + expect(mocks.getSupplierDataQualityRecords).toHaveBeenLastCalledWith(FACILITY_ID, 'missing_kde', { + windowDays: WINDOW_DAYS, + limit: DRILLDOWN_PAGE_SIZE, + offset: 1, + }); + expect(screen.queryByRole('button', { name: /Show more/ })).not.toBeInTheDocument(); + }); + + it('lists unlinked lot codes', async () => { + const lotCode = crypto.randomUUID(); + mocks.getSupplierDataQualityRecords.mockResolvedValueOnce( + page({ + metric: 'unlinked_lot', + total: [lotCode].length, + items: [ + { + record_id: null, + lot_id: crypto.randomUUID(), + lot_code: lotCode, + cte_type: null, + event_time: null, + received_at: null, + missing_kdes: [], + }, + ], + }), + ); + renderQuality(); + fireEvent.click(screen.getAllByRole('button', { name: 'View records' })[1]); + await waitFor(() => expect(screen.getByText(`Lot ${lotCode}`)).toBeInTheDocument()); + expect(mocks.getSupplierDataQualityRecords).toHaveBeenCalledWith(FACILITY_ID, 'unlinked_lot', expect.anything()); + }); + + it('says so when the records request fails', async () => { + mocks.getSupplierDataQualityRecords.mockRejectedValueOnce(new Error(crypto.randomUUID())); + renderQuality(); + fireEvent.click(screen.getAllByRole('button', { name: 'View records' })[0]); + await waitFor(() => expect(screen.getByText('Records could not be loaded')).toBeInTheDocument()); + }); + + it('shows unavailable when the summary request failed', () => { + renderQuality(null, true); + expect(screen.getByText('Data quality unavailable')).toBeInTheDocument(); + }); + + it('shows No data for a facility missing from the response', () => { + renderQuality(null, false); + expect(screen.getByText('Data quality: No data')).toBeInTheDocument(); + }); + + it('shows the late share against the agreed sending time and offers its records', async () => { + const [lateId] = uuids(1); + mocks.getSupplierDataQualityRecords.mockResolvedValueOnce( + page({ + metric: 'late_cte', + total: [lateId].length, + items: [ + { + record_id: lateId, + lot_id: null, + lot_code: crypto.randomUUID(), + cte_type: 'shipping', + event_time: null, + received_at: null, + missing_kdes: [], + }, + ], + }), + ); + renderQuality(lateMeasured); + expect( + screen.getByText(`${LATE.numerator} of ${LATE.denominator} records (${formatShare(LATE.numerator / LATE.denominator)})`), + ).toBeInTheDocument(); + expect(screen.getByText(`Agreed sending time: ${AGREED_HOURS} hours after the event`)).toBeInTheDocument(); + + fireEvent.click(screen.getByRole('button', { name: 'View records' })); + await waitFor(() => expect(screen.getByText(lateId)).toBeInTheDocument()); + expect(mocks.getSupplierDataQualityRecords).toHaveBeenCalledWith(FACILITY_ID, 'late_cte', expect.anything()); + }); + + it('offers no late drill down without an agreed sending time', () => { + renderQuality(summary([fourMetrics.metrics[2]])); + expect(screen.getByText('Agreed sending time: Not set')).toBeInTheDocument(); + expect(screen.queryByRole('button', { name: 'View records' })).not.toBeInTheDocument(); + }); + + it.each([null, AGREED_HOURS])('shows the agreed sending time %j read only, with no way to change it', (hours) => { + renderQuality(summary([fourMetrics.metrics[2]], hours)); + expect(screen.getByText('Set by your customer')).toBeInTheDocument(); + for (const name of ['Set', 'Edit', 'Save', 'Clear']) { + expect(screen.queryByRole('button', { name })).not.toBeInTheDocument(); + } + expect(screen.queryByLabelText('Hours after the event')).not.toBeInTheDocument(); + expect(mocks.setFacilityAgreedSendingHours).not.toHaveBeenCalled(); + }); +}); diff --git a/frontend/src/__tests__/pages/ingest-epcis.test.tsx b/frontend/src/__tests__/pages/ingest-epcis.test.tsx index 9b8136c4c..ccdacf7aa 100644 --- a/frontend/src/__tests__/pages/ingest-epcis.test.tsx +++ b/frontend/src/__tests__/pages/ingest-epcis.test.tsx @@ -136,7 +136,7 @@ describe('EPCIS 2.0 import on /ingest', () => { it('offers an EPCIS 2.0 tab with a JSON/XML chooser and no sample file link', async () => { await openEpcisTab(); - expect(screen.getByText('CSV uploads skip bad rows. EPCIS files are all or nothing: if any event fails, nothing is saved.')).toBeInTheDocument(); + expect(screen.getByText('CSV uploads skip bad rows. EPCIS (Preview) files are all or nothing: if any event fails, nothing is saved.')).toBeInTheDocument(); expect(document.getElementById('epcis-file-upload')).toHaveAttribute( 'accept', '.json,application/json,.xml,application/xml,text/xml', diff --git a/frontend/src/__tests__/pages/suppliers-metrics.test.tsx b/frontend/src/__tests__/pages/suppliers-metrics.test.tsx index 2eaf22e2a..877b474b3 100644 --- a/frontend/src/__tests__/pages/suppliers-metrics.test.tsx +++ b/frontend/src/__tests__/pages/suppliers-metrics.test.tsx @@ -23,6 +23,7 @@ vi.mock('@/lib/api-client', () => ({ getSupplierComplianceGaps: mocks.gaps, listSupplierTLCs: mocks.tlcs, getSupplierGapTimeToFix: () => Promise.resolve({ suppliers: [], internal_target_seconds: 86400, evaluated_at: '' }), + getSupplierDataQuality: () => Promise.resolve({ suppliers: [], window_days: 30, min_records: 20, late_cte_label: '', evaluated_at: '' }), }, })); diff --git a/frontend/src/__tests__/pages/suppliers-removal.test.tsx b/frontend/src/__tests__/pages/suppliers-removal.test.tsx index 7f91a6133..ee0eae0d6 100644 --- a/frontend/src/__tests__/pages/suppliers-removal.test.tsx +++ b/frontend/src/__tests__/pages/suppliers-removal.test.tsx @@ -25,6 +25,7 @@ vi.mock('@/lib/api-client', () => ({ getSupplierComplianceGaps: () => Promise.resolve({ gaps: [] }), listSupplierTLCs: () => Promise.resolve({ items: [] }), getSupplierGapTimeToFix: () => Promise.resolve({ suppliers: [], internal_target_seconds: 86400, evaluated_at: '' }), + getSupplierDataQuality: () => Promise.resolve({ suppliers: [], window_days: 30, min_records: 20, late_cte_label: '', evaluated_at: '' }), checkFacilityRemoval: mocks.checkRemoval, deleteSupplierFacility: mocks.deleteFacility, }, diff --git a/frontend/src/app/dashboard/suppliers/data-quality.tsx b/frontend/src/app/dashboard/suppliers/data-quality.tsx new file mode 100644 index 000000000..faec056cf --- /dev/null +++ b/frontend/src/app/dashboard/suppliers/data-quality.tsx @@ -0,0 +1,236 @@ +import { useState, type ReactNode } from 'react'; + +import { Button } from '@/components/ui/button'; +import { apiClient } from '@/lib/api-client'; +import type { + SupplierDataQuality, + SupplierQualityDrilldownMetric, + SupplierQualityMetric, + SupplierQualityMetricKey, + SupplierQualityRecordItem, +} from '@/types/api'; + +export const DRILLDOWN_PAGE_SIZE = 25; + +const METRIC_COPY: Record = { + missing_kde: { label: 'Records missing a KDE', unit: 'records' }, + unlinked_lot: { label: 'Lot codes not linked back through the chain', unit: 'lot codes' }, + late_cte: { label: 'Late CTE records', unit: 'records' }, + epcis_failed: { label: 'EPCIS (Preview) documents rejected across your whole account', unit: 'documents' }, +}; + +function isDrilldown(key: SupplierQualityMetricKey): key is SupplierQualityDrilldownMetric { + return key === 'missing_kde' || key === 'unlinked_lot' || key === 'late_cte'; +} + +function formatTime(value: string | null): string { + return value ? new Date(value).toLocaleString() : 'unknown'; +} + +function formatHours(hours: number): string { + return `${hours} hour${hours === 1 ? '' : 's'}`; +} + +function formatShare(share: number): string { + return `${Math.round(share * 1000) / 10}%`; +} + +interface DrilldownState { + items: SupplierQualityRecordItem[]; + total: number; + loading: boolean; + failed: boolean; +} + +function DrilldownList({ metric, state }: { metric: SupplierQualityDrilldownMetric; state: DrilldownState }) { + if (state.items.length === 0) { + if (state.loading) return

Loading records

; + if (state.failed) return

Records could not be loaded

; + return

No records

; + } + return ( +
    + {state.items.map((item) => + metric === 'late_cte' ? ( +
  • + {item.record_id} + Lot {item.lot_code} + {item.cte_type && {item.cte_type}} + Event {formatTime(item.event_time)} + Received {formatTime(item.received_at)} +
    • + ) : metric === 'missing_kde' ? ( +
  • + {item.record_id} + Lot {item.lot_code} + {item.cte_type && {item.cte_type}} + Missing: {item.missing_kdes.join(', ')} +
    • + ) : ( +
  • + Lot {item.lot_code} + {item.lot_id} +
    • + ), + )} +
+ ); +} + +/** Read only: the customer (tenant admin) sets the agreed sending time, not the supplier. */ +function SendingHoursNote({ hours }: { hours: number | null }) { + return ( +
+ Agreed sending time: {hours === null ? 'Not set' : `${formatHours(hours)} after the event`} + Set by your customer +
+ ); +} + +function MetricRow({ + facilityId, + metric, + windowDays, + minRecords, + lateCteLabel, + agreedSendingHours, +}: { + facilityId: string; + metric: SupplierQualityMetric; + windowDays: number; + minRecords: number; + lateCteLabel: string; + agreedSendingHours: number | null; +}) { + const copy = METRIC_COPY[metric.key]; + const [open, setOpen] = useState(false); + const [drilldown, setDrilldown] = useState({ items: [], total: 0, loading: false, failed: false }); + + const load = async (key: SupplierQualityDrilldownMetric, offset: number) => { + setDrilldown((prev) => ({ ...prev, loading: true, failed: false })); + try { + const res = await apiClient.getSupplierDataQualityRecords(facilityId, key, { + windowDays, + limit: DRILLDOWN_PAGE_SIZE, + offset, + }); + setDrilldown((prev) => ({ + items: offset === 0 ? res.items : [...prev.items, ...res.items], + total: res.total, + loading: false, + failed: false, + })); + } catch { + setDrilldown((prev) => ({ ...prev, loading: false, failed: true })); + } + }; + + const toggle = () => { + if (!isDrilldown(metric.key)) return; + const next = !open; + setOpen(next); + if (next && drilldown.items.length === 0) void load(metric.key, 0); + }; + + let value: ReactNode; + if (metric.status === 'unavailable') { + value = Unavailable. {metric.reason}; + } else if (metric.numerator === null || metric.denominator === null) { + value = Not enough data. {metric.reason}; + } else { + const counts = `${metric.numerator} of ${metric.denominator} ${copy.unit}`; + value = + metric.status === 'ok' && metric.share !== null ? ( + + {counts} ({formatShare(metric.share)}) + + ) : ( + + {counts}. Not enough data (fewer than {minRecords}) + + ); + } + + const canDrill = isDrilldown(metric.key) && metric.status !== 'unavailable' && (metric.numerator ?? 0) > 0; + + return ( +
  • +
    + {copy.label}: + {value} + {canDrill && ( + + )} +
    + {metric.key === 'late_cte' && ( + <> +

    {lateCteLabel}

    + + + )} + {metric.status !== 'unavailable' && metric.reason &&

    {metric.reason}

    } + {open && isDrilldown(metric.key) && ( +
    + + {drilldown.items.length < drilldown.total && ( + + )} +
    + )} +
    • + ); +} + +interface DataQualityProps { + facilityId: string; + summary: SupplierDataQuality | undefined; + /** True when the data quality request failed. */ + unavailable: boolean; + windowDays: number; + minRecords: number; + lateCteLabel: string; +} + +/** Four separate data quality measures for one facility. No combined score. */ +export function DataQuality({ + facilityId, + summary, + unavailable, + windowDays, + minRecords, + lateCteLabel, +}: DataQualityProps) { + if (unavailable) { + return

    Data quality unavailable

    ; + } + if (!summary) { + return

    Data quality: No data

    ; + } + return ( +
    +

    Data quality, records received in the last {windowDays} days

    +
      + {summary.metrics.map((metric) => ( + + ))} +
    +
    + ); +} diff --git a/frontend/src/app/dashboard/suppliers/page.tsx b/frontend/src/app/dashboard/suppliers/page.tsx index f204284c2..058bd71bb 100644 --- a/frontend/src/app/dashboard/suppliers/page.tsx +++ b/frontend/src/app/dashboard/suppliers/page.tsx @@ -49,8 +49,10 @@ import type { IntegrationProfile, FacilityRemovalCheck, SupplierGapTimeToFix, + SupplierDataQualityResponse, } from '@/types/api'; import { GapTimeToFix } from './gap-time-to-fix'; +import { DataQuality } from './data-quality'; import { NO_COMPLIANCE_DATA, countOrNull, @@ -144,6 +146,8 @@ export default function SupplierDashboardPage() { const [now, setNow] = useState(Date.now); const [gapTimeToFix, setGapTimeToFix] = useState>({}); const [gapTimeToFixUnavailable, setGapTimeToFixUnavailable] = useState(false); + const [dataQuality, setDataQuality] = useState(null); + const [dataQualityUnavailable, setDataQualityUnavailable] = useState(false); useEffect(() => { const timer = setInterval(() => setNow(Date.now()), 60_000); @@ -206,6 +210,18 @@ export default function SupplierDashboardPage() { const [newPostalCode, setNewPostalCode] = useState(''); const [adding, setAdding] = useState(false); + const loadDataQuality = useCallback(() => { + apiClient.getSupplierDataQuality() + .then((res) => { + setDataQuality(res); + setDataQualityUnavailable(false); + }) + .catch(() => { + setDataQuality(null); + setDataQualityUnavailable(true); + }); + }, []); + /* ---------- Fetch all data ------------------------------------ */ // Phase 1: Load facility list fast (1 API call) // Phase 2: Lazy-enrich with compliance data in background (batched, non-blocking) @@ -240,6 +256,8 @@ export default function SupplierDashboardPage() { setGapTimeToFixUnavailable(true); }); + loadDataQuality(); + if (tenantId) { fetchWorkbenchReadinessSummary(tenantId) .then(setWorkbenchSummary) @@ -297,7 +315,7 @@ export default function SupplierDashboardPage() { setError(friendlyError(err, 'Unable to load supplier data. Please try again.')); setLoading(false); } - }, [isLoggedIn, tenantId]); + }, [isLoggedIn, tenantId, loadDataQuality]); useEffect(() => { loadData(); }, [loadData]); @@ -884,6 +902,16 @@ export default function SupplierDashboardPage() { unavailable={gapTimeToFixUnavailable} now={now} /> + {(dataQuality || dataQualityUnavailable) && ( + s.supplier_key === facility.id)} + unavailable={dataQualityUnavailable} + windowDays={dataQuality?.window_days ?? 30} + minRecords={dataQuality?.min_records ?? 20} + lateCteLabel={dataQuality?.late_cte_label ?? ''} + /> + )} {comp.gaps.length > 0 && (
      {comp.gaps.map((gap, index) => { diff --git a/frontend/src/app/docs/walkthrough/page.tsx b/frontend/src/app/docs/walkthrough/page.tsx index e0ae2838f..5134e3e7f 100644 --- a/frontend/src/app/docs/walkthrough/page.tsx +++ b/frontend/src/app/docs/walkthrough/page.tsx @@ -63,8 +63,8 @@ export default function WalkthroughPage() {
    • A working curl with openssl, and ~15 minutes

    - The flow: one receiving event, one transformation event, one shipping event — - three events in a single supplier-to-shipping chain. Fill each request-body + The flow: one receiving event, one transformation event, one shipping event. + That is three events in a single supplier-to-shipping chain. Fill each request-body skeleton with your own lot codes, locations, and quantities, then list the events and export them.

    diff --git a/frontend/src/app/ingest/page.tsx b/frontend/src/app/ingest/page.tsx index ce5b9d41b..b30ade7ea 100644 --- a/frontend/src/app/ingest/page.tsx +++ b/frontend/src/app/ingest/page.tsx @@ -692,7 +692,7 @@ export default function IngestPage() {

    - CSV uploads skip bad rows. EPCIS files are all or nothing: if any event fails, nothing is saved. + CSV uploads skip bad rows. EPCIS (Preview) files are all or nothing: if any event fails, nothing is saved.

    diff --git a/frontend/src/lib/api-client.ts b/frontend/src/lib/api-client.ts index fd02334d6..2cac34272 100644 --- a/frontend/src/lib/api-client.ts +++ b/frontend/src/lib/api-client.ts @@ -29,6 +29,10 @@ import type { SupplierComplianceGapsResponse, SupplierComplianceScore, SupplierGapTimeToFixResponse, + SupplierDataQualityResponse, + AgreedSendingHoursResponse, + SupplierQualityDrilldownMetric, + SupplierQualityRecordsResponse, SupplierBulkUploadCommitResponse, SupplierBulkUploadParseResponse, SupplierBulkUploadStatusResponse, @@ -877,6 +881,43 @@ class APIClient { return data; } + async getSupplierDataQuality(windowDays?: number): Promise { + const { data } = await this.adminClient.get('/v1/supplier/data-quality', { + params: windowDays ? { window_days: windowDays } : undefined, + }); + return data; + } + + async setFacilityAgreedSendingHours( + facilityId: string, + agreedSendingHours: number | null, + ): Promise { + const { data } = await this.adminClient.put( + `/v1/supplier/admin/facilities/${encodeURIComponent(facilityId)}/agreed-sending-hours`, + { agreed_sending_hours: agreedSendingHours }, + ); + return data; + } + + async getSupplierDataQualityRecords( + facilityId: string, + metric: SupplierQualityDrilldownMetric, + options: { windowDays?: number; limit?: number; offset?: number } = {}, + ): Promise { + const { data } = await this.adminClient.get( + `/v1/supplier/data-quality/${encodeURIComponent(facilityId)}/records`, + { + params: { + metric, + window_days: options.windowDays, + limit: options.limit, + offset: options.offset, + }, + }, + ); + return data; + } + async getSupplierFDAExportPreview( facilityId?: string, limit: number = 25, diff --git a/frontend/src/types/api.ts b/frontend/src/types/api.ts index 8535e29e7..30b6aed2a 100644 --- a/frontend/src/types/api.ts +++ b/frontend/src/types/api.ts @@ -510,6 +510,65 @@ export interface SupplierGapTimeToFixResponse { evaluated_at: string; } +export type SupplierQualityMetricKey = 'missing_kde' | 'unlinked_lot' | 'late_cte' | 'epcis_failed'; +export type SupplierQualityDrilldownMetric = 'missing_kde' | 'unlinked_lot' | 'late_cte'; + +export interface SupplierQualityMetric { + key: SupplierQualityMetricKey; + /** + * not_enough_data: denominator is below min_records, so share is null. For + * late_cte with no agreed sending time, the counts are null too and reason + * says why. + */ + status: 'ok' | 'not_enough_data' | 'unavailable'; + numerator: number | null; + denominator: number | null; + share: number | null; + /** 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; +} + +export interface SupplierDataQuality { + /** Supplier facility id. */ + supplier_key: string; + /** Hours after the event time the supplier agreed to send a record; null when not set. */ + agreed_sending_hours: number | null; + metrics: SupplierQualityMetric[]; +} + +export interface AgreedSendingHoursResponse { + facility_id: string; + agreed_sending_hours: number | null; +} + +export interface SupplierDataQualityResponse { + suppliers: SupplierDataQuality[]; + window_days: number; + min_records: number; + late_cte_label: string; + evaluated_at: string; +} + +export interface SupplierQualityRecordItem { + record_id: string | null; + lot_id: string | null; + lot_code: string; + cte_type: string | null; + event_time: string | null; + received_at: string | null; + missing_kdes: string[]; +} + +export interface SupplierQualityRecordsResponse { + supplier_key: string; + metric: SupplierQualityDrilldownMetric; + window_days: number; + total: number; + limit: number; + offset: number; + items: SupplierQualityRecordItem[]; +} + export interface SupplierFDAExportRow { event_id: string; tlc_code: string; diff --git a/scripts/copy-lint/baseline.json b/scripts/copy-lint/baseline.json index 5aee75702..224e3d54c 100644 --- a/scripts/copy-lint/baseline.json +++ b/scripts/copy-lint/baseline.json @@ -21,46 +21,25 @@ "frontend/src/app/blog/regengine-vs-spreadsheets-traceability/page.tsx": { "epcis-unlabeled": 1 }, - "frontend/src/app/compliance/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/app/contact/layout.tsx": { "sample-demo": 2 }, "frontend/src/app/controls/page.tsx": { "epcis-unlabeled": 2 }, - "frontend/src/app/dashboard/compliance/page.tsx": { - "epcis-unlabeled": 2 - }, - "frontend/src/app/dashboard/export-jobs/page.tsx": { - "epcis-unlabeled": 4 - }, "frontend/src/app/dashboard/integrations/page.tsx": { "empty-render": 1, - "sample-demo": 3 - }, - "frontend/src/app/dashboard/layout.tsx": { - "sample-demo": 2, - "sandbox": 1 + "sample-demo": 1 }, "frontend/src/app/dashboard/qa-health/page.tsx": { "dash-placeholder": 1 }, - "frontend/src/app/dashboard/recall-drills/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/app/dashboard/recall-report/page.tsx": { - "dash-placeholder": 4, - "invented-entity": 12, - "sample-demo": 1 + "dash-placeholder": 4 }, "frontend/src/app/dashboard/recall-response/page.tsx": { "dash-placeholder": 1 }, - "frontend/src/app/dashboard/receiving/page.tsx": { - "em-dash": 1 - }, "frontend/src/app/dashboard/settings/users/page.tsx": { "empty-render": 1 }, @@ -72,19 +51,12 @@ "frontend/src/app/developer/portal/page.tsx": { "empty-render": 1 }, - "frontend/src/app/developers/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/app/docs/api/ApiReferenceClient.tsx": { "hyphen-dash": 8 }, "frontend/src/app/docs/changelog/page.tsx": { - "epcis-unlabeled": 1, "sample-demo": 1 }, - "frontend/src/app/docs/connectors/inflow-lab/page.tsx": { - "sandbox": 6 - }, "frontend/src/app/docs/quickstart/page.tsx": { "sample-demo": 1 }, @@ -106,8 +78,7 @@ "empty-render": 1 }, "frontend/src/app/ingest/page.tsx": { - "epcis-unlabeled": 6, - "sample-demo": 1, + "epcis-unlabeled": 3, "sandbox": 3 }, "frontend/src/app/ingest/results/[id]/AnalysisResultClient.tsx": { @@ -116,29 +87,19 @@ "frontend/src/app/ingest/sandbox-handoff.ts": { "sandbox": 6 }, - "frontend/src/app/integrations/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/app/legal/vulnerability-disclosure/page.tsx": { "sample-demo": 1 }, "frontend/src/app/onboarding/bulk-upload/page.tsx": { "sample-demo": 1 }, - "frontend/src/app/pricing/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/app/product/page.tsx": { "sandbox": 2 }, "frontend/src/app/proof/page.tsx": { - "epcis-unlabeled": 7, + "epcis-unlabeled": 2, "sample-demo": 5 }, - "frontend/src/app/proof/recall-drill/page.tsx": { - "invented-entity": 6, - "sample-demo": 7 - }, "frontend/src/app/records/page.tsx": { "dash-placeholder": 3, "empty-render": 1, @@ -157,34 +118,11 @@ "frontend/src/app/sysadmin/page.tsx": { "dash-placeholder": 3 }, - "frontend/src/app/tools/ask/page.tsx": { - "sample-demo": 6 - }, - "frontend/src/app/tools/data-import/components/DataImportClient.tsx": { - "sample-demo": 2 - }, "frontend/src/app/tools/drill-simulator/components/DrillSimulatorClient.tsx": { - "sample-demo": 3 + "sample-demo": 1 }, "frontend/src/app/tools/export/page.tsx": { - "empty-render": 2, - "sample-demo": 6 - }, - "frontend/src/app/tools/fsma-unified/components/UnifiedDashboardClient.tsx": { - "sample-demo": 2 - }, - "frontend/src/app/tools/inflow-lab/components/DashboardInflowLab.tsx": { - "invented-entity": 5, - "sandbox": 14 - }, - "frontend/src/app/tools/inflow-lab/components/InflowLabClient.tsx": { - "invented-entity": 15, - "sample-demo": 4, - "sandbox": 19 - }, - "frontend/src/app/tools/inflow-lab/components/InflowLabTutorial.tsx": { - "sample-demo": 2, - "sandbox": 2 + "empty-render": 2 }, "frontend/src/app/tools/notice-validator/page.tsx": { "invented-entity": 1, @@ -193,29 +131,11 @@ "frontend/src/app/tools/scan/page.tsx": { "empty-render": 1 }, - "frontend/src/app/tools/sop-generator/components/SOPGeneratorClient.tsx": { - "invented-entity": 2 - }, "frontend/src/app/tools/tlc-validator/components/TLCValidatorClient.tsx": { "sample-demo": 1 }, - "frontend/src/app/trust/architecture/page.tsx": { - "epcis-unlabeled": 2 - }, - "frontend/src/app/trust/page.tsx": { - "epcis-unlabeled": 3 - }, - "frontend/src/app/trust/retention/page.tsx": { - "epcis-unlabeled": 2 - }, - "frontend/src/app/walkthrough/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/app/why-regengine/page.tsx": { - "epcis-unlabeled": 7 - }, - "frontend/src/components/control-plane/demo-banner.tsx": { - "sample-demo": 1 + "epcis-unlabeled": 1 }, "frontend/src/components/dashboard/compliance-status-widget.tsx": { "empty-render": 1 @@ -243,7 +163,7 @@ "edi-sftp": 2 }, "frontend/src/lib/customer-readiness.ts": { - "epcis-unlabeled": 4 + "epcis-unlabeled": 1 }, "frontend/src/lib/dev-compliance-snapshots.ts": { "dash-placeholder": 2 diff --git a/services/admin/app/sqlalchemy_models.py b/services/admin/app/sqlalchemy_models.py index 5f470f352..b0fbb9bd8 100644 --- a/services/admin/app/sqlalchemy_models.py +++ b/services/admin/app/sqlalchemy_models.py @@ -422,12 +422,18 @@ class SupplierFacilityModel(Base): postal_code = Column(String, nullable=False) fda_registration_number = Column(String, nullable=True) roles = Column(JSONType(), nullable=False, default=list) + # Hours after a CTE's event time the supplier agreed to send it. NULL: none. + agreed_sending_hours = Column(Integer, nullable=True) created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) __table_args__ = ( Index("ix_supplier_facilities_tenant", "tenant_id"), Index("ix_supplier_facilities_user", "supplier_user_id"), + CheckConstraint( + "agreed_sending_hours > 0", + name="supplier_facilities_agreed_sending_hours_check", + ), ) diff --git a/services/admin/app/supplier_compliance_routes.py b/services/admin/app/supplier_compliance_routes.py index c0ea74d4e..cb7e39a0a 100644 --- a/services/admin/app/supplier_compliance_routes.py +++ b/services/admin/app/supplier_compliance_routes.py @@ -9,12 +9,30 @@ import io import uuid as uuid_module from datetime import datetime, timezone -from typing import Any +from typing import Any, Literal from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import StreamingResponse -from pydantic import BaseModel +from pydantic import BaseModel, Field from shared.data_gap_tracking import INTERNAL_TARGET, fetch_supplier_summaries +from shared.supplier_data_quality import ( + DEFAULT_WINDOW_DAYS, + LATE_CTE_LABEL, + MAX_WINDOW_DAYS, + METRIC_EPCIS_FAILED, + METRIC_LATE_CTE, + METRIC_MISSING_KDE, + METRIC_UNLINKED_LOT, + MIN_RECORDS, + fetch_lot_rows, + fetch_record_rows, + fetch_sending_hours, + fetch_supplier_quality, + flag_records, + late_records, + unlinked_lots, + window_start as quality_window_start, +) from sqlalchemy import select from sqlalchemy.orm import Session @@ -22,6 +40,7 @@ from .dependencies import get_current_user from .models import TenantContext from .sqlalchemy_models import SupplierFacilityModel, UserModel +from .tenant_settings_routes import _ADMIN_ROLE_NAMES, _get_user_role_name from .supplier_onboarding_routes import ( SupplierComplianceGap, SupplierComplianceGapsResponse, @@ -37,6 +56,10 @@ router = APIRouter(prefix="/supplier", tags=["supplier-onboarding"]) +QUALITY_METRIC_ORDER = (METRIC_MISSING_KDE, METRIC_UNLINKED_LOT, METRIC_LATE_CTE, METRIC_EPCIS_FAILED) +# supplier_facilities.agreed_sending_hours is a Postgres integer. +MAX_AGREED_SENDING_HOURS = 2**31 - 1 + @router.get("/compliance-score", response_model=SupplierComplianceScoreResponse) async def get_compliance_score( @@ -168,6 +191,230 @@ async def get_gap_time_to_fix( ) +class SupplierQualityMetric(BaseModel): + key: str + status: Literal["ok", "not_enough_data", "unavailable"] + numerator: int | None + denominator: int | None + share: float | None + reason: str | None + + +class SupplierDataQuality(BaseModel): + supplier_key: str + agreed_sending_hours: int | None = None + metrics: list[SupplierQualityMetric] + + +class AgreedSendingHoursRequest(BaseModel): + # Required key; null clears it. Strict: no strings, floats or booleans. + agreed_sending_hours: int | None = Field(..., strict=True, gt=0, le=MAX_AGREED_SENDING_HOURS) + + +class AgreedSendingHoursResponse(BaseModel): + facility_id: str + agreed_sending_hours: int | None + + +class SupplierDataQualityResponse(BaseModel): + suppliers: list[SupplierDataQuality] + window_days: int + min_records: int + late_cte_label: str + evaluated_at: str + + +class SupplierQualityRecordItem(BaseModel): + record_id: str | None = None + lot_id: str | None = None + lot_code: str + cte_type: str | None = None + event_time: str | None = None + received_at: str | None = None + missing_kdes: list[str] = [] + + +class SupplierQualityRecordsResponse(BaseModel): + supplier_key: str + metric: str + window_days: int + total: int + limit: int + offset: int + items: list[SupplierQualityRecordItem] + + +def _owned_facility_ids(db: Session, tenant_id: uuid_module.UUID, user_id: Any) -> list[str]: + return [ + str(facility_id) + for facility_id in db.scalars( + select(SupplierFacilityModel.id).where( + SupplierFacilityModel.tenant_id == tenant_id, + SupplierFacilityModel.supplier_user_id == user_id, + ) + ).all() + ] + + +def _iso_or_none(value: datetime | None) -> str | None: + return value.isoformat() if value else None + + +@router.get("/data-quality", response_model=SupplierDataQualityResponse) +async def get_supplier_data_quality( + window_days: int = Query(default=DEFAULT_WINDOW_DAYS, ge=1, le=MAX_WINDOW_DAYS), + current_user: UserModel = Depends(get_current_user), + db: Session = Depends(get_session), +) -> SupplierDataQualityResponse: + """Four separate data quality measures per facility over a rolling window. + + 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: + raise HTTPException(status_code=400, detail="Tenant context required") + + facility_ids = _owned_facility_ids(db, tenant_id, current_user.id) + now = datetime.now(timezone.utc) + hours = fetch_sending_hours(db, str(tenant_id), str(current_user.id), facility_ids) + quality = fetch_supplier_quality( + db, str(tenant_id), str(current_user.id), facility_ids, quality_window_start(now, window_days), hours + ) + return SupplierDataQualityResponse( + suppliers=[ + SupplierDataQuality( + supplier_key=key, + agreed_sending_hours=hours.get(key), + metrics=[SupplierQualityMetric(**metrics[name]) for name in QUALITY_METRIC_ORDER], + ) + for key, metrics in quality.items() + ], + window_days=window_days, + min_records=MIN_RECORDS, + late_cte_label=LATE_CTE_LABEL, + evaluated_at=now.isoformat(), + ) + + +@router.get("/data-quality/{facility_id}/records", response_model=SupplierQualityRecordsResponse) +async def get_supplier_data_quality_records( + facility_id: str, + metric: Literal["missing_kde", "unlinked_lot", "late_cte"], + window_days: int = Query(default=DEFAULT_WINDOW_DAYS, ge=1, le=MAX_WINDOW_DAYS), + limit: int = Query(default=50, ge=1, le=200), + offset: int = Query(default=0, ge=0), + current_user: UserModel = Depends(get_current_user), + db: Session = Depends(get_session), +) -> SupplierQualityRecordsResponse: + """The exact records behind one facility's metric numerator, paginated.""" + tenant_id = TenantContext.get_tenant_context(db) + if not tenant_id: + raise HTTPException(status_code=400, detail="Tenant context required") + try: + scoped_facility_id = uuid_module.UUID(facility_id) + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid facility id") from exc + facility = _get_supplier_facility_or_404( + db, + tenant_id=tenant_id, + supplier_user_id=current_user.id, + facility_id=scoped_facility_id, + ) + if metric == METRIC_LATE_CTE and facility.agreed_sending_hours is None: + raise HTTPException(status_code=409, detail="No agreed sending time is set for this facility") + + key = str(scoped_facility_id) + since = quality_window_start(datetime.now(timezone.utc), window_days) + if metric in (METRIC_MISSING_KDE, METRIC_LATE_CTE): + rows = fetch_record_rows(db, str(tenant_id), str(current_user.id), [key], since) + flagged = ( + flag_records(rows) if metric == METRIC_MISSING_KDE else late_records(rows, facility.agreed_sending_hours) + ) + items = [ + SupplierQualityRecordItem( + record_id=str(row["record_id"]), + lot_code=row["lot_code"], + cte_type=row["cte_type"], + event_time=_iso_or_none(row["event_time"]), + received_at=_iso_or_none(row["created_at"]), + missing_kdes=row.get("missing_kdes", []), + ) + for row in flagged + ] + else: + items = [ + SupplierQualityRecordItem(lot_id=lot["lot_id"], lot_code=lot["lot_code"]) + for lot in unlinked_lots(fetch_lot_rows(db, str(tenant_id), str(current_user.id), [key], since)) + ] + return SupplierQualityRecordsResponse( + supplier_key=key, + metric=metric, + window_days=window_days, + total=len(items), + limit=limit, + offset=offset, + items=items[offset : offset + limit], + ) + + +class TenantAdminChecker: + """Allow only an Owner or Admin membership in the bound tenant. + + Reuses the tenant settings role gate (#1386). ``required_permission`` + is read by the route security manifest. + """ + + required_permission = "tenant.admin" + + def __call__(self, user: UserModel = Depends(get_current_user), db: Session = Depends(get_session)) -> UserModel: + tenant_id = TenantContext.get_tenant_context(db) + if not tenant_id: + raise HTTPException(status_code=403, detail="E_NO_TENANT_CONTEXT: tenant-scoped endpoint requires X-Tenant-ID") + if _get_user_role_name(tenant_id, user, db) not in _ADMIN_ROLE_NAMES: + raise HTTPException(status_code=403, detail="Only tenant Owner or Admin roles may set agreed sending hours") + return user + + +require_tenant_admin = TenantAdminChecker() + + +@router.put("/admin/facilities/{facility_id}/agreed-sending-hours", response_model=AgreedSendingHoursResponse) +async def set_agreed_sending_hours( + facility_id: str, + request: AgreedSendingHoursRequest, + _admin: UserModel = Depends(require_tenant_admin), + db: Session = Depends(get_session), +) -> AgreedSendingHoursResponse: + """Tenant admin sets or clears (null) the agreed sending time for a facility in the tenant. + + Suppliers can read it in their data quality summary but cannot change it. + """ + tenant_id = TenantContext.get_tenant_context(db) + if not tenant_id: + raise HTTPException(status_code=400, detail="Tenant context required") + try: + scoped_facility_id = uuid_module.UUID(facility_id) + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid facility id") from exc + # Tenant scoping lives in the query: a facility in another tenant is never loaded. + facility = db.execute( + select(SupplierFacilityModel).where( + SupplierFacilityModel.id == scoped_facility_id, + SupplierFacilityModel.tenant_id == tenant_id, + ) + ).scalar_one_or_none() + if facility is None: + raise HTTPException(status_code=404, detail="Facility not found") + + facility.agreed_sending_hours = request.agreed_sending_hours + db.commit() + return AgreedSendingHoursResponse( + facility_id=str(scoped_facility_id), agreed_sending_hours=request.agreed_sending_hours + ) + + @router.get("/export/fda-records/preview", response_model=SupplierFDAExportPreviewResponse) async def preview_fda_records_export( facility_id: str | None = None, diff --git a/services/admin/tests/test_supplier_data_quality_routes.py b/services/admin/tests/test_supplier_data_quality_routes.py new file mode 100644 index 000000000..a4816e6de --- /dev/null +++ b/services/admin/tests/test_supplier_data_quality_routes.py @@ -0,0 +1,482 @@ +"""Supplier data quality routes: scoping and guards, no database. + +A fake session answers the facility ownership lookups; the SQL fetchers are +stubbed so the tests check what the routes pass and return, not SQL. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from uuid import UUID, uuid4 + +import pytest +from fastapi import FastAPI +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, + MIN_RECORDS, +) + +TENANT = uuid4() +OTHER_TENANT = uuid4() +USER = uuid4() +OTHER_USER = uuid4() +OWN_FACILITY = uuid4() +FOREIGN_FACILITY = uuid4() +OTHER_TENANT_FACILITY = uuid4() + +# Any positive whole number of hours; derived rather than picked. +AGREED_HOURS = MIN_RECORDS +NOW = datetime.now(timezone.utc) + +# Shipping records with no KDEs on the owned facility, each received just past +# AGREED_HOURS after its event time, and one lot with no upstream link. Ids +# and lot codes are generated per run. +RECORD_ROWS = [ + { + "record_id": str(uuid4()), + "facility_id": str(OWN_FACILITY), + "cte_type": "shipping", + "kde_data": {}, + "event_time": NOW - timedelta(hours=AGREED_HOURS) - timedelta.resolution, + "created_at": NOW, + "lot_code": str(uuid4()), + } + for _ in range(MIN_RECORDS // 10) +] +LOT_ROWS = [ + { + "lot_id": str(uuid4()), + "lot_code": str(uuid4()), + "facility_id": str(OWN_FACILITY), + "cte_type": "shipping", + "kde_data": {}, + "is_transformation_output": False, + } +] + + +# 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 + self.facilities = { + OWN_FACILITY: SimpleNamespace( + id=OWN_FACILITY, tenant_id=TENANT, supplier_user_id=USER, agreed_sending_hours=own_hours + ), + FOREIGN_FACILITY: SimpleNamespace( + id=FOREIGN_FACILITY, tenant_id=TENANT, supplier_user_id=OTHER_USER, agreed_sending_hours=None + ), + OTHER_TENANT_FACILITY: SimpleNamespace( + id=OTHER_TENANT_FACILITY, tenant_id=OTHER_TENANT, supplier_user_id=USER, agreed_sending_hours=None + ), + } + + def commit(self) -> None: + self.commits += 1 + + def get(self, _model, key): + return self.facilities.get(key) + + def execute(self, stmt): + # The admin write filters by id and tenant in SQL; match on the bound values. + bound = set(stmt.compile().params.values()) + found = [f for f in self.facilities.values() if f.id in bound and f.tenant_id in bound] + return SimpleNamespace(scalar_one_or_none=lambda: found[0] if found else None) + + def scalars(self, _stmt): + # The route filters by tenant and user in SQL; the fake returns what that + # filter yields for TENANT and USER. + return SimpleNamespace(all=lambda: [OWN_FACILITY]) + + +@pytest.fixture +def calls(): + return [] + + +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. + monkeypatch.setattr(routes, "_get_user_role_name", lambda _tenant, _user, _db: role) + + 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] + + def fake_lots(_db, tenant_id, user_id, facility_ids, since): + calls.append(("lots", tenant_id, user_id, list(facility_ids))) + return [dict(row) for row in LOT_ROWS] + + def fake_hours(db, tenant_id, user_id, facility_ids): + calls.append(("hours", tenant_id, user_id, list(facility_ids))) + return {key: db.facilities[UUID(key)].agreed_sending_hours for key in facility_ids} + + monkeypatch.setattr(routes, "fetch_sending_hours", fake_hours) + monkeypatch.setattr("shared.supplier_data_quality.fetch_sending_hours", fake_hours) + monkeypatch.setattr(routes, "fetch_record_rows", fake_records) + 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() + app.include_router(routes.router, prefix="/v1") + app.dependency_overrides[routes.get_session] = lambda: session + app.dependency_overrides[routes.get_current_user] = lambda: SimpleNamespace(id=USER) + return TestClient(app) + + +def test_summary_returns_four_separate_metrics_for_owned_facilities(monkeypatch, calls): + resp = _client(monkeypatch, calls).get("/v1/supplier/data-quality") + assert resp.status_code == 200 + body = resp.json() + assert body["window_days"] == DEFAULT_WINDOW_DAYS + assert body["min_records"] == MIN_RECORDS + assert body["late_cte_label"] == LATE_CTE_LABEL + [supplier] = body["suppliers"] + assert supplier["supplier_key"] == str(OWN_FACILITY) + assert [m["key"] for m in supplier["metrics"]] == ["missing_kde", "unlinked_lot", "late_cte", "epcis_failed"] + missing = supplier["metrics"][0] + assert (missing["status"], missing["numerator"], missing["denominator"]) == ( + "not_enough_data", + len(RECORD_ROWS), + len(RECORD_ROWS), + ) + 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) + 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]) +def test_window_days_is_bounded(monkeypatch, calls, window): + resp = _client(monkeypatch, calls).get(f"/v1/supplier/data-quality?window_days={window}") + assert resp.status_code == 422 + assert calls == [] + + +def test_summary_requires_tenant(monkeypatch, calls): + assert _client(monkeypatch, calls, tenant=None).get("/v1/supplier/data-quality").status_code == 400 + + +def test_drilldown_lists_missing_kde_records(monkeypatch, calls): + resp = _client(monkeypatch, calls).get( + f"/v1/supplier/data-quality/{OWN_FACILITY}/records?metric=missing_kde&limit=1" + ) + assert resp.status_code == 200 + body = resp.json() + assert (body["total"], body["limit"], body["offset"]) == (len(RECORD_ROWS), 1, 0) + assert [item["record_id"] for item in body["items"]] == [RECORD_ROWS[0]["record_id"]] + assert "product_description" in body["items"][0]["missing_kdes"] + assert calls == [("records", str(TENANT), str(USER), [str(OWN_FACILITY)])] + + +def test_drilldown_offset_pages(monkeypatch, calls): + resp = _client(monkeypatch, calls).get( + f"/v1/supplier/data-quality/{OWN_FACILITY}/records?metric=missing_kde&limit=1&offset=1" + ) + assert [item["record_id"] for item in resp.json()["items"]] == [RECORD_ROWS[1]["record_id"]] + + +def test_drilldown_lists_unlinked_lots(monkeypatch, calls): + resp = _client(monkeypatch, calls).get(f"/v1/supplier/data-quality/{OWN_FACILITY}/records?metric=unlinked_lot") + assert resp.status_code == 200 + assert [item["lot_code"] for item in resp.json()["items"]] == [LOT_ROWS[0]["lot_code"]] + + +@pytest.mark.parametrize("facility", [FOREIGN_FACILITY, OTHER_TENANT_FACILITY]) +def test_drilldown_refuses_facility_not_owned_in_tenant(monkeypatch, calls, facility): + resp = _client(monkeypatch, calls).get(f"/v1/supplier/data-quality/{facility}/records?metric=missing_kde") + assert resp.status_code == 404 + assert calls == [] + + +def test_summary_counts_late_records_against_agreed_hours(monkeypatch, calls): + resp = _client(monkeypatch, calls, session=FakeSession(own_hours=AGREED_HOURS)).get("/v1/supplier/data-quality") + [supplier] = resp.json()["suppliers"] + assert supplier["agreed_sending_hours"] == AGREED_HOURS + late = supplier["metrics"][2] + # Fewer than MIN_RECORDS timed records: counts shown, no share. + assert (late["status"], late["numerator"], late["denominator"], late["share"]) == ( + "not_enough_data", + len(RECORD_ROWS), + len(RECORD_ROWS), + None, + ) + assert [call for call in calls if call[0] == "hours"] == [("hours", str(TENANT), str(USER), [str(OWN_FACILITY)])] + + +def test_drilldown_lists_late_records(monkeypatch, calls): + session = FakeSession(own_hours=AGREED_HOURS) + resp = _client(monkeypatch, calls, session=session).get( + f"/v1/supplier/data-quality/{OWN_FACILITY}/records?metric=late_cte" + ) + assert resp.status_code == 200 + body = resp.json() + assert body["total"] == len(RECORD_ROWS) + assert [item["record_id"] for item in body["items"]] == [row["record_id"] for row in RECORD_ROWS] + assert all(item["received_at"] and item["event_time"] for item in body["items"]) + assert calls == [("records", str(TENANT), str(USER), [str(OWN_FACILITY)])] + + +def test_drilldown_late_cte_without_agreed_hours_is_refused(monkeypatch, calls): + resp = _client(monkeypatch, calls).get(f"/v1/supplier/data-quality/{OWN_FACILITY}/records?metric=late_cte") + assert resp.status_code == 409 + assert calls == [] + + +@pytest.mark.parametrize("facility", [FOREIGN_FACILITY, OTHER_TENANT_FACILITY]) +def test_drilldown_late_cte_refuses_facility_not_owned(monkeypatch, calls, facility): + resp = _client(monkeypatch, calls).get(f"/v1/supplier/data-quality/{facility}/records?metric=late_cte") + assert resp.status_code == 404 + assert calls == [] + + +def test_drilldown_rejects_unavailable_metrics(monkeypatch, calls): + for metric in ("epcis_failed", "anything"): + resp = _client(monkeypatch, calls).get(f"/v1/supplier/data-quality/{OWN_FACILITY}/records?metric={metric}") + assert resp.status_code == 422 + assert calls == [] + + +def test_drilldown_rejects_bad_facility_id_and_limit(monkeypatch, calls): + client = _client(monkeypatch, calls) + assert client.get("/v1/supplier/data-quality/not-a-uuid/records?metric=missing_kde").status_code == 400 + assert ( + client.get(f"/v1/supplier/data-quality/{OWN_FACILITY}/records?metric=missing_kde&limit=201").status_code == 422 + ) + assert calls == [] + + +# --- agreed sending hours ------------------------------------------------------ + + +def _hours_url(facility) -> str: + return f"/v1/supplier/admin/facilities/{facility}/agreed-sending-hours" + + +def _admin_role() -> str: + from app.tenant_settings_routes import _ADMIN_ROLE_NAMES + + return sorted(_ADMIN_ROLE_NAMES)[0] + + +def _admin(monkeypatch, calls, **kwargs): + return _client(monkeypatch, calls, role=_admin_role(), **kwargs) + + +@pytest.mark.parametrize("facility", [OWN_FACILITY, FOREIGN_FACILITY]) +def test_tenant_admin_sets_agreed_sending_hours_for_any_facility_in_tenant(monkeypatch, calls, facility): + session = FakeSession() + resp = _admin(monkeypatch, calls, session=session).put(_hours_url(facility), json={"agreed_sending_hours": AGREED_HOURS}) + assert resp.status_code == 200 + assert resp.json() == {"facility_id": str(facility), "agreed_sending_hours": AGREED_HOURS} + assert session.facilities[facility].agreed_sending_hours == AGREED_HOURS + assert session.commits == len([resp]) + + +def test_tenant_admin_clears_agreed_sending_hours_with_null(monkeypatch, calls): + session = FakeSession(own_hours=AGREED_HOURS) + resp = _admin(monkeypatch, calls, session=session).put(_hours_url(OWN_FACILITY), json={"agreed_sending_hours": None}) + assert resp.status_code == 200 + assert resp.json()["agreed_sending_hours"] is None + assert session.facilities[OWN_FACILITY].agreed_sending_hours is None + + +@pytest.mark.parametrize("role", [None, "Member", "Viewer", "supplier"]) +def test_supplier_user_cannot_set_agreed_sending_hours(monkeypatch, calls, role): + # The supplier owns OWN_FACILITY and is still refused: the setting is the tenant's. + session = FakeSession() + resp = _client(monkeypatch, calls, session=session, role=role).put( + _hours_url(OWN_FACILITY), json={"agreed_sending_hours": AGREED_HOURS} + ) + assert resp.status_code == 403 + assert session.facilities[OWN_FACILITY].agreed_sending_hours is None + assert session.commits == 0 + + +def test_old_supplier_write_path_is_gone(monkeypatch, calls): + session = FakeSession() + resp = _client(monkeypatch, calls, session=session, role=_admin_role()).put( + f"/v1/supplier/facilities/{OWN_FACILITY}/agreed-sending-hours", json={"agreed_sending_hours": AGREED_HOURS} + ) + assert resp.status_code in (404, 405) + assert session.commits == 0 + + +def test_admin_of_another_tenant_gets_404(monkeypatch, calls): + session = FakeSession() + resp = _admin(monkeypatch, calls, session=session).put( + _hours_url(OTHER_TENANT_FACILITY), json={"agreed_sending_hours": AGREED_HOURS} + ) + assert resp.status_code == 404 + assert session.facilities[OTHER_TENANT_FACILITY].agreed_sending_hours is None + assert session.commits == 0 + + +@pytest.mark.parametrize( + "body", + [ + {"agreed_sending_hours": 0}, + {"agreed_sending_hours": -AGREED_HOURS}, + {"agreed_sending_hours": str(AGREED_HOURS)}, + {"agreed_sending_hours": AGREED_HOURS + 0.5}, + {"agreed_sending_hours": float(AGREED_HOURS)}, + {"agreed_sending_hours": True}, + # One past the Postgres integer maximum. + {"agreed_sending_hours": 2**31}, + {}, + ], +) +def test_agreed_sending_hours_must_be_positive_int_or_null(monkeypatch, calls, body): + session = FakeSession() + resp = _admin(monkeypatch, calls, session=session).put(_hours_url(OWN_FACILITY), json=body) + assert resp.status_code == 422 + assert session.facilities[OWN_FACILITY].agreed_sending_hours is None + assert session.commits == 0 + + +def test_agreed_sending_hours_rejects_bad_id_and_missing_tenant(monkeypatch, calls): + body = {"agreed_sending_hours": AGREED_HOURS} + assert _admin(monkeypatch, calls).put(_hours_url("not-a-uuid"), json=body).status_code == 400 + assert _admin(monkeypatch, calls, tenant=None).put(_hours_url(OWN_FACILITY), json=body).status_code == 403 + + +# --- real TenantAdminChecker membership path ---------------------------------- +# No stubbed role lookup: memberships, roles and the facility live in an +# in-memory SQLite database built from the real models. + + +def _runtime_email() -> str: + """An address built at runtime so no test carries a literal email.""" + return f"{uuid4().hex}@{uuid4().hex}.test" + + +def _membership_client(monkeypatch, *, member_tenant, is_active): + import app.supplier_compliance_routes as routes + from app.sqlalchemy_models import ( + MembershipModel, + RoleModel, + SupplierFacilityModel, + TenantModel, + UserModel, + ) + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + from sqlalchemy.pool import StaticPool + + engine = create_engine( + "sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, future=True + ) + for model in (TenantModel, UserModel, RoleModel, MembershipModel, SupplierFacilityModel): + model.__table__.create(bind=engine) + session = sessionmaker(bind=engine, expire_on_commit=False, future=True)() + + bound_tenant, other_tenant = uuid4(), uuid4() + tenant_ids = {"bound": bound_tenant, "other": other_tenant} + for tenant_id in tenant_ids.values(): + session.add(TenantModel(id=tenant_id, name=str(tenant_id), slug=str(tenant_id), settings={})) + owner = UserModel(id=uuid4(), email=_runtime_email()) + supplier = UserModel(id=uuid4(), email=_runtime_email()) + role = RoleModel(id=uuid4(), tenant_id=None, name="Owner", permissions=[]) + session.add_all([owner, supplier, role]) + session.flush() + session.add( + MembershipModel(user_id=owner.id, tenant_id=tenant_ids[member_tenant], role_id=role.id, is_active=is_active) + ) + facility = SupplierFacilityModel( + id=uuid4(), + tenant_id=bound_tenant, + supplier_user_id=supplier.id, + name=str(uuid4()), + street=str(uuid4()), + city=str(uuid4()), + state=str(uuid4()), + postal_code=str(uuid4()), + roles=[], + ) + session.add(facility) + session.commit() + + monkeypatch.setattr(routes.TenantContext, "get_tenant_context", lambda _db: bound_tenant) + app = FastAPI() + app.include_router(routes.router, prefix="/v1") + app.dependency_overrides[routes.get_session] = lambda: session + app.dependency_overrides[routes.get_current_user] = lambda: owner + return TestClient(app), session, facility + + +@pytest.mark.parametrize( + ("member_tenant", "is_active", "expected"), + [ + ("bound", True, 200), + ("bound", False, 403), + ("other", True, 403), + ], + ids=["owner-of-tenant", "inactive-owner", "owner-of-other-tenant"], +) +def test_tenant_admin_checker_uses_real_membership(monkeypatch, member_tenant, is_active, expected): + client, session, facility = _membership_client(monkeypatch, member_tenant=member_tenant, is_active=is_active) + resp = client.put(_hours_url(facility.id), json={"agreed_sending_hours": AGREED_HOURS}) + assert resp.status_code == expected + session.expire_all() + stored = session.get(type(facility), facility.id).agreed_sending_hours + assert stored == (AGREED_HOURS if expected == 200 else None) 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 new file mode 100644 index 000000000..999e6dc92 --- /dev/null +++ b/services/shared/supplier_data_quality.py @@ -0,0 +1,487 @@ +"""Per supplier data quality over a rolling window. + +A *supplier* is a supplier facility (``supplier_facilities.id``), the same +``supplier_key`` that ``public.data_gap_events`` uses. Four metrics are kept +separate on purpose; there is no combined score. + +1. ``missing_kde``: share of CTE records (``supplier_cte_events``) received in + the window that lack at least one required KDE. The required list is the + ingest webhook's ``REQUIRED_KDES_BY_CTE``, minus the KDEs a supplier record + carries as columns: the lot code (``lot_id``), the event date + (``event_time``) and the recording facility's own location + (``facility_id``). +2. ``unlinked_lot``: share of lot codes active in the window that cannot be + linked back through the chain. A lot is linked when it is the output of a + ``fsma.transformation_links`` row, when one of its records carries a lot + linkage KDE (the same fields the rules engine checks in + ``shared.rules.seeds``), or when it starts at this facility (harvesting or + first land based receiving). +3. ``late_cte``: share of CTE records received in the window that arrived + more than ``supplier_facilities.agreed_sending_hours`` after their + ``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``: 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 + +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 +DEFAULT_WINDOW_DAYS = 30 +MAX_WINDOW_DAYS = 365 + +METRIC_MISSING_KDE = "missing_kde" +METRIC_UNLINKED_LOT = "unlinked_lot" +METRIC_LATE_CTE = "late_cte" +METRIC_EPCIS_FAILED = "epcis_failed" +DRILLDOWN_METRICS = (METRIC_MISSING_KDE, METRIC_UNLINKED_LOT, METRIC_LATE_CTE) + +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_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 +# (both are named ``app``); services/shared/tests/test_supplier_data_quality.py fails if +# the two tables drift. +WEBHOOK_REQUIRED_KDES: dict[str, tuple[str, ...]] = { + "harvesting": ( + "traceability_lot_code", + "product_description", + "quantity", + "unit_of_measure", + "harvest_date", + "location_name", + "reference_document", + ), + "cooling": ( + "traceability_lot_code", + "product_description", + "quantity", + "unit_of_measure", + "cooling_date", + "location_name", + "reference_document", + ), + "initial_packing": ( + "traceability_lot_code", + "product_description", + "quantity", + "unit_of_measure", + "packing_date", + "location_name", + "reference_document", + "harvester_business_name", + "harvester_business_phone", + ), + "first_receiver": ( + "traceability_lot_code", + "product_description", + "quantity", + "unit_of_measure", + "landing_date", + "receiving_location", + "reference_document", + ), + "shipping": ( + "traceability_lot_code", + "product_description", + "quantity", + "unit_of_measure", + "ship_date", + "ship_from_location", + "ship_to_location", + "reference_document", + "tlc_source_reference", + ), + "receiving": ( + "traceability_lot_code", + "product_description", + "quantity", + "unit_of_measure", + "receive_date", + "receiving_location", + "immediate_previous_source", + "reference_document", + "tlc_source_reference", + ), + "transforming": ( + "traceability_lot_code", + "product_description", + "quantity", + "unit_of_measure", + "transformation_date", + "location_name", + "reference_document", + ), +} + +# Supplied by supplier_cte_events columns, not kde_data: lot_id, event_time, +# and facility_id (the recording facility's own location). +COLUMN_BACKED_KDES = frozenset( + { + "traceability_lot_code", + "harvest_date", + "cooling_date", + "packing_date", + "landing_date", + "ship_date", + "receive_date", + "transformation_date", + "location_name", + "receiving_location", + "ship_from_location", + } +) + +# Lot linkage fields from shared.rules.seeds (receiving, shipping and +# transformation lot_linkage / kde_presence rules), plus the harvester the +# initial packer must record (21 CFR 1.1330(a)(7)). +LINKAGE_KDES = ( + "tlc_source_reference", + "tlc_source_gln", + "immediate_previous_source", + "input_lot_codes", + "input_traceability_lot_codes", + "input_tlcs", + "harvester_business_name", +) +# The chain starts at these CTEs; there is nothing upstream to link to. +ORIGIN_CTES = frozenset({"harvesting", "first_receiver"}) + +_CTE_ALIASES = { + "transformation": "transforming", + "first_land_based_receiving": "first_receiver", +} + + +# --------------------------------------------------------------------------- +# Pure functions +# --------------------------------------------------------------------------- + + +def normalize_cte(cte_type: Optional[str]) -> str: + value = (cte_type or "").strip().lower() + return _CTE_ALIASES.get(value, value) + + +def _present(value: Any) -> bool: + if value is None: + return False + if isinstance(value, str): + return value.strip() != "" + if isinstance(value, (list, tuple, dict, set)): + return len(value) > 0 + return True + + +def required_record_kdes(cte_type: Optional[str]) -> tuple[str, ...]: + """KDEs a supplier record must carry in ``kde_data`` for its CTE.""" + required = WEBHOOK_REQUIRED_KDES.get(normalize_cte(cte_type), ()) + return tuple(name for name in required if name not in COLUMN_BACKED_KDES) + + +def missing_kdes(cte_type: Optional[str], kde_data: Optional[Mapping[str, Any]]) -> list[str]: + data = kde_data if isinstance(kde_data, Mapping) else {} + return [name for name in required_record_kdes(cte_type) if not _present(data.get(name))] + + +def lot_is_linked( + records: Sequence[tuple[Optional[str], Optional[Mapping[str, Any]]]], + is_transformation_output: bool, +) -> bool: + """True when a lot links back: transformation output, linkage KDE, or origin CTE. + + ``records`` is every (cte_type, kde_data) recorded for the lot. + """ + if is_transformation_output: + return True + for cte_type, kde_data in records: + if normalize_cte(cte_type) in ORIGIN_CTES: + return True + data = kde_data if isinstance(kde_data, Mapping) else {} + if any(_present(data.get(name)) for name in LINKAGE_KDES): + return True + return False + + +def share_metric(key: str, numerator: int, denominator: int, min_records: int = MIN_RECORDS) -> dict[str, Any]: + """Numerator and denominator always; a share only from ``min_records`` up.""" + enough = denominator >= min_records + return { + "key": key, + "status": "ok" if enough else "not_enough_data", + "numerator": numerator, + "denominator": denominator, + "share": (numerator / denominator) if enough else None, + "reason": None, + } + + +def timed_records(rows: Sequence[Mapping[str, Any]]) -> list[Mapping[str, Any]]: + """Records with both an event time and a received time: the late denominator.""" + return [row for row in rows if row.get("event_time") is not None and row.get("created_at") is not None] + + +def late_records(rows: Sequence[Mapping[str, Any]], agreed_sending_hours: int) -> list[Mapping[str, Any]]: + """Records received (``created_at``) more than ``agreed_sending_hours`` after ``event_time``.""" + allowed = timedelta(hours=agreed_sending_hours) + return [row for row in timed_records(rows) if row["created_at"] - row["event_time"] > allowed] + + +def late_metric(rows: Sequence[Mapping[str, Any]], agreed_sending_hours: Optional[int]) -> dict[str, Any]: + if agreed_sending_hours is None: + return { + "key": METRIC_LATE_CTE, + "status": "not_enough_data", + "numerator": None, + "denominator": None, + "share": None, + "reason": LATE_CTE_NO_HOURS_REASON, + } + return share_metric(METRIC_LATE_CTE, len(late_records(rows, agreed_sending_hours)), len(timed_records(rows))) + + +def unavailable_metric(key: str, reason: str) -> dict[str, Any]: + return { + "key": key, + "status": "unavailable", + "numerator": None, + "denominator": None, + "share": None, + "reason": reason, + } + + +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) + + +def flag_records(rows: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + """Records missing at least one required KDE, with the missing names.""" + flagged: list[dict[str, Any]] = [] + for row in rows: + missing = missing_kdes(row["cte_type"], row["kde_data"]) + if missing: + flagged.append({**row, "missing_kdes": missing}) + return flagged + + +def group_lots(rows: Sequence[Mapping[str, Any]]) -> dict[str, dict[str, Any]]: + """Collapse one row per (lot, record) into one entry per lot id.""" + lots: dict[str, dict[str, Any]] = {} + for row in rows: + lot_id = str(row["lot_id"]) + lot = lots.setdefault( + lot_id, + { + "lot_id": lot_id, + "lot_code": row["lot_code"], + "facility_id": str(row["facility_id"]), + "is_transformation_output": bool(row["is_transformation_output"]), + "records": [], + }, + ) + if row.get("cte_type") is not None: + lot["records"].append((row["cte_type"], row["kde_data"])) + return lots + + +def unlinked_lots(rows: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + return [ + {"lot_id": lot["lot_id"], "lot_code": lot["lot_code"], "facility_id": lot["facility_id"]} + for lot in group_lots(rows).values() + if not lot_is_linked(lot["records"], lot["is_transformation_output"]) + ] + + +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. + + ``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: epcis_failed_metric(epcis_counts), + } + + +# --------------------------------------------------------------------------- +# Database (SELECT only) +# --------------------------------------------------------------------------- + +_RECORD_ROWS_SQL = text( + """ + SELECT e.id AS record_id, + e.facility_id, + e.cte_type, + e.kde_data, + e.event_time, + e.created_at, + l.tlc_code AS lot_code + FROM supplier_cte_events e + JOIN supplier_traceability_lots l + ON l.id = e.lot_id + AND l.tenant_id = e.tenant_id + WHERE e.tenant_id = CAST(:tid AS uuid) + AND e.supplier_user_id = CAST(:uid AS uuid) + AND e.facility_id = ANY(CAST(:fids AS uuid[])) + AND e.created_at >= :since + ORDER BY e.created_at DESC, e.id + """ +) + +# One row per (active lot, record of that lot). A lot is active when it has a +# record received in the window; linkage looks at all of the caller's records +# for it, any age. +_LOT_ROWS_SQL = text( + """ + WITH active AS ( + SELECT DISTINCT l.id, l.tlc_code, l.facility_id + FROM supplier_traceability_lots l + JOIN supplier_cte_events w + ON w.lot_id = l.id + AND w.tenant_id = l.tenant_id + WHERE l.tenant_id = CAST(:tid AS uuid) + AND l.supplier_user_id = CAST(:uid AS uuid) + AND l.facility_id = ANY(CAST(:fids AS uuid[])) + AND w.created_at >= :since + ) + SELECT a.id AS lot_id, + a.tlc_code AS lot_code, + a.facility_id, + e.cte_type, + e.kde_data, + EXISTS ( + SELECT 1 + FROM fsma.transformation_links t + WHERE t.tenant_id = CAST(:tid AS uuid) + AND t.output_tlc = a.tlc_code + ) AS is_transformation_output + FROM active a + JOIN supplier_cte_events e + ON e.lot_id = a.id + AND e.tenant_id = CAST(:tid AS uuid) + AND e.supplier_user_id = CAST(:uid AS uuid) + ORDER BY a.tlc_code, a.id + """ +) + + +_SENDING_HOURS_SQL = text( + """ + SELECT f.id AS facility_id, + f.agreed_sending_hours + FROM supplier_facilities f + WHERE f.tenant_id = CAST(:tid AS uuid) + AND f.supplier_user_id = CAST(:uid AS uuid) + AND f.id = ANY(CAST(:fids AS uuid[])) + """ +) + + +def _params(tenant_id: str, supplier_user_id: str, facility_ids: Sequence[str], since: datetime) -> dict[str, Any]: + return {"tid": tenant_id, "uid": supplier_user_id, "fids": list(facility_ids), "since": since} + + +def fetch_record_rows( + db_session: Any, tenant_id: str, supplier_user_id: str, facility_ids: Sequence[str], since: datetime +) -> list[dict[str, Any]]: + if not facility_ids: + return [] + result = db_session.execute(_RECORD_ROWS_SQL, _params(tenant_id, supplier_user_id, facility_ids, since)) + return [dict(row) for row in result.mappings().all()] + + +def fetch_lot_rows( + db_session: Any, tenant_id: str, supplier_user_id: str, facility_ids: Sequence[str], since: datetime +) -> list[dict[str, Any]]: + if not facility_ids: + return [] + result = db_session.execute(_LOT_ROWS_SQL, _params(tenant_id, supplier_user_id, facility_ids, since)) + return [dict(row) for row in result.mappings().all()] + + +def fetch_sending_hours( + db_session: Any, tenant_id: str, supplier_user_id: str, facility_ids: Sequence[str] +) -> dict[str, Optional[int]]: + """Agreed sending hours per facility id; None where none is set.""" + if not facility_ids: + return {} + result = db_session.execute( + _SENDING_HOURS_SQL, {"tid": tenant_id, "uid": supplier_user_id, "fids": list(facility_ids)} + ) + return {str(row["facility_id"]): row["agreed_sending_hours"] for row in result.mappings().all()} + + +def _by_facility(rows: Sequence[Mapping[str, Any]], facility_ids: Sequence[str]) -> dict[str, list[Mapping[str, Any]]]: + grouped: dict[str, list[Mapping[str, Any]]] = {key: [] for key in facility_ids} + for row in rows: + grouped.setdefault(str(row["facility_id"]), []).append(row) + return grouped + + +def fetch_supplier_quality( + db_session: Any, + tenant_id: str, + supplier_user_id: str, + facility_ids: Sequence[str], + since: datetime, + sending_hours: Optional[Mapping[str, Optional[int]]] = None, +) -> dict[str, dict[str, dict[str, Any]]]: + """Per facility metrics for ``facility_ids`` (every id is present). + + ``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), 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 new file mode 100644 index 000000000..9847d3082 --- /dev/null +++ b/services/shared/tests/test_supplier_data_quality.py @@ -0,0 +1,425 @@ +"""Supplier data quality: pure logic only. + +No database. Row dicts stand in for the SELECT results. +""" + +from __future__ import annotations + +import importlib.util +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest +from shared.supplier_data_quality import ( + 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, + METRIC_MISSING_KDE, + METRIC_UNLINKED_LOT, + MIN_RECORDS, + WEBHOOK_REQUIRED_KDES, + epcis_failed_metric, + fetch_sending_hours, + fetch_supplier_quality, + flag_records, + group_lots, + late_metric, + late_records, + lot_is_linked, + missing_kdes, + required_record_kdes, + share_metric, + summarize_facility, + timed_records, + unavailable_metric, + unlinked_lots, + window_start, +) + +REPO_ROOT = Path(__file__).resolve().parents[3] +WEBHOOK_MODELS = REPO_ROOT / "services" / "ingestion" / "app" / "webhook_models.py" + +COMPLETE_SHIPPING = { + "product_description": str(uuid.uuid4()), + # Any positive count; derived rather than picked. + "quantity": MIN_RECORDS, + "unit_of_measure": str(uuid.uuid4()), + "ship_to_location": str(uuid.uuid4()), + "reference_document": str(uuid.uuid4()), + "tlc_source_reference": str(uuid.uuid4()), +} + + +FACILITY_ID = str(uuid.uuid4()) +# One generated lot code per generated lot id. +LOT_CODES: dict[str, str] = {} + + +# Any positive whole number of hours; derived rather than picked. +AGREED_HOURS = MIN_RECORDS +NOW = datetime.now(timezone.utc) + + +def _record( + cte_type: str, + kde_data: dict, + lot_code: str | None = None, + event_time: datetime | None = None, + created_at: datetime | None = None, +) -> dict: + return { + "record_id": uuid.uuid4(), + "facility_id": FACILITY_ID, + "cte_type": cte_type, + "kde_data": kde_data, + "event_time": event_time, + "created_at": created_at, + "lot_code": lot_code if lot_code is not None else str(uuid.uuid4()), + } + + +def _lot_row(lot_id: str, cte_type: str | None, kde_data: dict | None, output: bool = False) -> dict: + return { + "lot_id": lot_id, + "lot_code": LOT_CODES.setdefault(lot_id, str(uuid.uuid4())), + "facility_id": FACILITY_ID, + "cte_type": cte_type, + "kde_data": kde_data, + "is_transformation_output": output, + } + + +# --- required KDEs ----------------------------------------------------------- + + +def test_required_kdes_match_ingest_webhook_table(): + """The copied table must equal REQUIRED_KDES_BY_CTE, CTE for CTE.""" + spec = importlib.util.spec_from_file_location("_ingestion_webhook_models", WEBHOOK_MODELS) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + aliases = {"transformation": "transforming", "first_land_based_receiving": "first_receiver"} + webhook = {aliases.get(cte.value, cte.value): tuple(kdes) for cte, kdes in module.REQUIRED_KDES_BY_CTE.items()} + assert webhook == WEBHOOK_REQUIRED_KDES + + +def test_column_backed_kdes_are_never_required_in_kde_data(): + for cte in WEBHOOK_REQUIRED_KDES: + assert not set(required_record_kdes(cte)) & COLUMN_BACKED_KDES + + +def test_receiving_keeps_source_kdes_and_drops_column_kdes(): + assert required_record_kdes("receiving") == ( + "product_description", + "quantity", + "unit_of_measure", + "immediate_previous_source", + "reference_document", + "tlc_source_reference", + ) + + +def test_cte_aliases_resolve(): + assert required_record_kdes("transformation") == required_record_kdes("transforming") + assert required_record_kdes(" First_Land_Based_Receiving ") == required_record_kdes("first_receiver") + + +def test_unknown_cte_requires_nothing(): + assert required_record_kdes("growing") == () + assert missing_kdes(None, {}) == [] + + +# --- missing KDE ------------------------------------------------------------- + + +def test_complete_shipping_record_has_no_missing_kdes(): + assert missing_kdes("shipping", COMPLETE_SHIPPING) == [] + + +@pytest.mark.parametrize("blank", [None, "", " ", [], {}]) +def test_blank_values_count_as_missing(blank): + data = {**COMPLETE_SHIPPING, "reference_document": blank} + assert missing_kdes("shipping", data) == ["reference_document"] + + +def test_zero_quantity_is_present(): + assert "quantity" not in missing_kdes("shipping", {**COMPLETE_SHIPPING, "quantity": 0}) + + +def test_non_mapping_kde_data_misses_everything(): + assert missing_kdes("cooling", None) == list(required_record_kdes("cooling")) + assert missing_kdes("cooling", "not a dict") == list(required_record_kdes("cooling")) + + +def test_flag_records_keeps_only_failing_rows_with_names(): + rows = [_record("shipping", COMPLETE_SHIPPING), _record("shipping", {"quantity": COMPLETE_SHIPPING["quantity"]})] + flagged = flag_records(rows) + assert len(flagged) == 1 + assert flagged[0]["record_id"] == rows[1]["record_id"] + assert "product_description" in flagged[0]["missing_kdes"] + + +# --- lot linkage ------------------------------------------------------------- + + +def test_lot_linked_by_transformation_output(): + assert lot_is_linked([("shipping", {})], is_transformation_output=True) + + +@pytest.mark.parametrize( + "field", ["tlc_source_reference", "tlc_source_gln", "immediate_previous_source", "input_lot_codes"] +) +def test_lot_linked_by_linkage_kde(field): + assert lot_is_linked([("shipping", {}), ("receiving", {field: str(uuid.uuid4())})], False) + + +@pytest.mark.parametrize("cte", ["harvesting", "first_receiver", "first_land_based_receiving"]) +def test_origin_cte_links_lot(cte): + assert lot_is_linked([(cte, {})], False) + + +def test_lot_without_any_link_is_unlinked(): + assert not lot_is_linked([("shipping", {"tlc_source_reference": " "}), ("cooling", {})], False) + assert not lot_is_linked([], False) + + +def test_group_lots_collapses_records_per_lot(): + first, second = str(uuid.uuid4()), str(uuid.uuid4()) + rows = [_lot_row(first, "shipping", {}), _lot_row(first, "receiving", {}), _lot_row(second, "cooling", {})] + lots = group_lots(rows) + assert set(lots) == {first, second} + assert len(lots[first]["records"]) == sum(1 for row in rows if row["lot_id"] == first) + + +def test_unlinked_lots_returns_ids_and_codes(): + rows = [ + _lot_row(str(uuid.uuid4()), "shipping", {}), + _lot_row(str(uuid.uuid4()), "receiving", {"tlc_source_reference": str(uuid.uuid4())}), + _lot_row(str(uuid.uuid4()), "shipping", {}, output=True), + ] + unlinked = rows[0] + assert unlinked_lots(rows) == [ + {"lot_id": unlinked["lot_id"], "lot_code": unlinked["lot_code"], "facility_id": unlinked["facility_id"]} + ] + + +# --- share and thresholds ---------------------------------------------------- + + +def test_share_below_min_records_is_not_enough_data_but_keeps_counts(): + records = [_record("shipping", {}) for _ in range(MIN_RECORDS - 1)] + flagged = records[::2] + metric = share_metric(METRIC_MISSING_KDE, len(flagged), len(records)) + assert metric["status"] == "not_enough_data" + assert metric["share"] is None + assert (metric["numerator"], metric["denominator"]) == (len(flagged), len(records)) + + +def test_share_at_min_records_is_computed(): + records = [_record("shipping", {}) for _ in range(MIN_RECORDS)] + flagged = records[::2] + metric = share_metric(METRIC_MISSING_KDE, len(flagged), len(records)) + assert metric["status"] == "ok" + assert metric["share"] == len(flagged) / len(records) + + +def test_zero_denominator_is_not_enough_data(): + assert share_metric(METRIC_UNLINKED_LOT, 0, 0)["status"] == "not_enough_data" + + +def test_min_records_is_twenty(): + assert MIN_RECORDS == 20 + + +def test_unavailable_metric_has_reason_and_no_counts(): + reason = str(uuid.uuid4()) + metric = unavailable_metric(METRIC_LATE_CTE, reason) + assert metric == { + "key": METRIC_LATE_CTE, + "status": "unavailable", + "numerator": None, + "denominator": None, + "share": None, + "reason": reason, + } + + +def test_window_start(): + now = datetime.now(timezone.utc) + assert window_start(now, DEFAULT_WINDOW_DAYS) == now - timedelta(days=DEFAULT_WINDOW_DAYS) + + +# --- facility summary -------------------------------------------------------- + + +# --- late CTE ------------------------------------------------------------------ + + +def _received_after(delay: timedelta) -> dict: + """A shipping record received ``delay`` after its event time.""" + return _record("shipping", COMPLETE_SHIPPING, event_time=NOW - delay, created_at=NOW) + + +ON_TIME = timedelta(hours=AGREED_HOURS) +LATE = ON_TIME + timedelta.resolution + + +def test_record_received_exactly_at_agreed_hours_is_on_time(): + assert late_records([_received_after(ON_TIME)], AGREED_HOURS) == [] + + +def test_record_received_after_agreed_hours_is_late(): + late = _received_after(LATE) + assert late_records([_received_after(ON_TIME), late], AGREED_HOURS) == [late] + + +def test_records_without_both_times_are_outside_the_denominator(): + untimed = [ + _record("shipping", {}, event_time=NOW), + _record("shipping", {}, created_at=NOW), + _record("shipping", {}), + ] + timed = [_received_after(LATE)] + assert timed_records(untimed + timed) == timed + assert late_records(untimed, AGREED_HOURS) == [] + + +def test_late_metric_without_agreed_hours_is_not_enough_data_with_reason(): + rows = [_received_after(LATE) for _ in range(MIN_RECORDS)] + assert late_metric(rows, None) == { + "key": METRIC_LATE_CTE, + "status": "not_enough_data", + "numerator": None, + "denominator": None, + "share": None, + "reason": LATE_CTE_NO_HOURS_REASON, + } + + +def test_late_metric_below_min_timed_records_is_not_enough_data(): + timed = [_received_after(LATE) for _ in range(MIN_RECORDS - 1)] + # Untimed rows never lift the count over MIN_RECORDS. + untimed = [_record("shipping", {}) for _ in range(MIN_RECORDS)] + metric = late_metric(timed + untimed, AGREED_HOURS) + assert metric["status"] == "not_enough_data" + assert metric["share"] is None + assert (metric["numerator"], metric["denominator"]) == (len(timed), len(timed)) + + +def test_late_metric_at_min_records_computes_share(): + late = [_received_after(LATE) for _ in range(MIN_RECORDS // 2)] + on_time = [_received_after(ON_TIME) for _ in range(MIN_RECORDS - len(late))] + metric = late_metric(late + on_time + [_record("shipping", {})], AGREED_HOURS) + assert metric["status"] == "ok" + assert (metric["numerator"], metric["denominator"]) == (len(late), len(late + on_time)) + assert metric["share"] == len(late) / len(late + on_time) + + +def test_late_cte_is_a_drilldown_metric(): + assert METRIC_LATE_CTE in DRILLDOWN_METRICS + + +def test_summarize_facility_keeps_four_separate_metrics(): + complete = [_record("shipping", COMPLETE_SHIPPING) for _ in range(MIN_RECORDS // 2)] + 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)] + 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) + assert summary[METRIC_UNLINKED_LOT]["status"] == "not_enough_data" + assert (summary[METRIC_UNLINKED_LOT]["numerator"], summary[METRIC_UNLINKED_LOT]["denominator"]) == ( + len(lots), + len(lots), + ) + assert summary[METRIC_LATE_CTE] == late_metric(records, None) + 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(): + late = [_received_after(LATE) for _ in range(MIN_RECORDS)] + summary = summarize_facility(late, [], AGREED_HOURS) + assert summary[METRIC_LATE_CTE]["status"] == "ok" + assert summary[METRIC_LATE_CTE]["share"] == len(late) / len(late) + + +def test_fetch_supplier_quality_without_facilities_runs_no_query(): + class NoQuery: + def execute(self, *args, **kwargs): # pragma: no cover (must not run) + raise AssertionError("no query expected") + + assert fetch_supplier_quality(NoQuery(), str(uuid.uuid4()), str(uuid.uuid4()), [], datetime.now(timezone.utc)) == {} + assert fetch_sending_hours(NoQuery(), str(uuid.uuid4()), str(uuid.uuid4()), []) == {} diff --git a/supabase/migrations/20261024080000_supplier_agreed_sending_hours.sql b/supabase/migrations/20261024080000_supplier_agreed_sending_hours.sql new file mode 100644 index 000000000..526fd8f2d --- /dev/null +++ b/supabase/migrations/20261024080000_supplier_agreed_sending_hours.sql @@ -0,0 +1,22 @@ +-- Agreed sending time per supplier facility. +-- +-- agreed_sending_hours is how many hours after a CTE's event time the +-- supplier has agreed to send the record. The supplier data quality view +-- counts a record as late when it is received more than this many hours after +-- its event time. This is our own measure, not an FDA rule. NULL means no +-- agreement is recorded, and the late measure shows not enough data. +-- +-- No backfill. Additive and idempotent: the inline CHECK is only created +-- together with the column, so a re-run is a no-op. + +alter table public.supplier_facilities + add column if not exists agreed_sending_hours integer null + constraint supplier_facilities_agreed_sending_hours_check + check (agreed_sending_hours > 0); + +-- Downgrade (run by hand only; alembic v136 b005ddbf1dba does the same): +-- +-- alter table public.supplier_facilities +-- drop constraint if exists supplier_facilities_agreed_sending_hours_check; +-- alter table public.supplier_facilities +-- drop column if exists agreed_sending_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; diff --git a/tests/security/test_route_security_manifest.py b/tests/security/test_route_security_manifest.py index a34cfbce0..e4617e669 100644 --- a/tests/security/test_route_security_manifest.py +++ b/tests/security/test_route_security_manifest.py @@ -70,9 +70,12 @@ def test_checked_in_manifest_matches_live_route_and_dependency_graph(manifest): # +1 global-sysadmin: admin:PATCH:/admin/organizations/{org_id} (rename). # +2 tenant-scoped (SEL-421, #3492): admin:GET:/v1/supplier/facilities/{facility_id}/removal-check # and admin:DELETE:/v1/supplier/facilities/{facility_id}, owner-only, tenant-filtered. - # +1 tenant-scoped (#3572): admin:GET:/v1/supplier/gap-time-to-fix, read-only, - # tenant-filtered data gap time to fix summary. - "enabled_operation_count": 255, + # +1 tenant-scoped (#3572): admin:GET:/v1/supplier/gap-time-to-fix. + # +2 tenant-scoped: admin:GET:/v1/supplier/data-quality and + # admin:GET:/v1/supplier/data-quality/{facility_id}/records, owner-only, tenant-filtered. + # +1 tenant-scoped: admin:PUT:/v1/supplier/admin/facilities/{facility_id}/agreed-sending-hours, + # tenant Owner or Admin only (tenant.admin), tenant-filtered. + "enabled_operation_count": 258, # 178/161 includes graph:GET:/api/v1/provisions/by-request: it is added # to the graph v1_router AFTER router.include_router(v1_router), so the # eager route copies of fastapi < 0.137 (under which the previous @@ -96,7 +99,7 @@ def test_checked_in_manifest_matches_live_route_and_dependency_graph(manifest): "enabled_by_classification": { "global-sysadmin": 16, "public": 30, - "tenant-scoped": 202, + "tenant-scoped": 205, "workload-only": 7, }, } @@ -189,9 +192,8 @@ def test_dual_session_routes_record_cookie_tenancy_and_csrf_adapter(manifest): dual = [item for item in manifest["enabled_operations"] if item["test_adapter"] == "tenant_dual_session_csrf"] mutations = [item for item in dual if item["method"] in {"POST", "PUT", "PATCH", "DELETE"}] - # 52 includes admin:GET:/v1/supplier/gap-time-to-fix (#3572), a read, so mutations stay 23. - assert len(dual) == 52 - assert len(mutations) == 23 + assert len(dual) == 55 + assert len(mutations) == 24 assert all(item["authentication"] == "bearer-token-or-opaque-session-cookie" for item in dual) assert all(item["storage_boundary"] == "admin-postgres:jwt-tenant-or-session-membership-tenant" for item in dual)