From 18c127613db50e6f99702c22d8f2280dfd61524b Mon Sep 17 00:00:00 2001 From: oshrizak <63424207+oshrizak@users.noreply.github.com> Date: Wed, 17 Jun 2026 19:17:54 -0700 Subject: [PATCH] feat(api): add by-job-id PII approve/deny endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /api/v1/documents/{job_id}/pii/approve POST /api/v1/documents/{job_id}/pii/deny The token-based equivalent at /api/v1/approval/{token}/decision works for the operator-driven flow that already exists. These by-job-id variants exist so machine clients (the reflow-canvas-lti Canvas connector) can forward a faculty decision after a status poll — they already authenticated with an API key and they already know the job_id, so a token round-trip isn't needed. Body shape mirrors ApprovalDecisionInput minus `decision` (implicit from the URL): `{justification?, reviewed_by}`. Response is the same shape as the token endpoint returns. Status codes: 200 approval recorded; quick_approve/quick_deny ran; background task continues with S3 cleanup / queue dispatch. 404 job_id not found. 409 job exists but isn't in `awaiting_approval` — race with another reviewer in a parallel tab, or a timeout sweeper already decided. The Canvas LTI connector surfaces this verbatim to faculty. 422 body validation (justification < 10 chars when present, reviewed_by missing) — FastAPI handles automatically. The 409 pre-check is intentional: quick_approve/quick_deny set the job status unconditionally, which is safe for the token-based endpoint (the token proves the job was in awaiting_approval at issue time) but unsafe by job_id where two concurrent calls could otherwise both succeed and overwrite each other. Tests cover happy approve, happy deny, 404, and 409, all against the live testcontainers stack via the integration tier. Verified locally with `pytest -m integration` inside the api-gateway image. --- src/api/documents.py | 199 ++++++++++++++++++ .../api/test_pii_decision_by_job_id.py | 196 +++++++++++++++++ 2 files changed, 395 insertions(+) create mode 100644 tests/integration/api/test_pii_decision_by_job_id.py diff --git a/src/api/documents.py b/src/api/documents.py index fe9584b..23dfa3c 100644 --- a/src/api/documents.py +++ b/src/api/documents.py @@ -20,6 +20,7 @@ get_storage_service, ) from ..services import JobService, QueueService, S3URLService, StorageService +from ..services.approval_service import ApprovalService from ..services.document_processing_service import DocumentProcessingService from ..services.metrics_service import jobs_submitted_total from .schemas import ( @@ -711,3 +712,201 @@ async def get_ledger( processing_duration_ms=0, final_markdown_url=final_markdown_url, ) + + +# --------------------------------------------------------------------------- +# PII approval / denial — by-job-id endpoints. +# +# The token-based equivalent lives at /api/v1/approval/{token}/decision in +# api.approval. These by-job-id variants exist so machine clients (e.g. the +# reflow-canvas-lti connector) can forward a faculty decision without having +# to juggle an approval token — they already authenticated with an API key +# and they already know the job_id from a prior status poll. +# --------------------------------------------------------------------------- + + +class PIIDecisionInput(BaseModel): + """Input for the by-job-id PII decision endpoints.""" + + justification: str | None = Field( + None, + min_length=10, + max_length=1000, + description="Optional explanation for the decision (10–1000 chars when provided).", + ) + reviewed_by: str = Field( + ..., + min_length=3, + description="Reviewer identifier (email or stable user id).", + ) + + +class PIIDecisionResponse(BaseModel): + """Response for the by-job-id PII decision endpoints.""" + + message: str + job_id: str + decision: Literal["approved", "denied"] + + +@router.post( + "/{job_id}/pii/approve", + response_model=PIIDecisionResponse, + summary="Approve PII gate for a document (by job_id)", + description=( + "Records a faculty approval of the PII findings on a job that is in " + "``awaiting_approval`` status, then resumes processing. Symmetric " + "counterpart of ``POST /{job_id}/pii/deny``. Intended for connector " + "consumption (the Canvas LTI connector forwards a faculty decision " + "here after the panorama PII gate)." + ), +) +async def approve_pii( + job_id: str, + body: PIIDecisionInput, + background_tasks: BackgroundTasks, + redis_client: Any = Depends(get_redis_client), + storage_service: StorageService = Depends(get_storage_service), + s3_url_service: S3URLService = Depends(get_s3_url_service), +) -> PIIDecisionResponse: + return await _process_pii_decision( + job_id=job_id, + decision="approved", + body=body, + background_tasks=background_tasks, + redis_client=redis_client, + storage_service=storage_service, + s3_url_service=s3_url_service, + ) + + +@router.post( + "/{job_id}/pii/deny", + response_model=PIIDecisionResponse, + summary="Deny PII gate for a document (by job_id)", + description=( + "Records a faculty denial of the PII findings on a job that is in " + "``awaiting_approval`` status, then cleans up the document. Symmetric " + "counterpart of ``POST /{job_id}/pii/approve``." + ), +) +async def deny_pii( + job_id: str, + body: PIIDecisionInput, + background_tasks: BackgroundTasks, + redis_client: Any = Depends(get_redis_client), + storage_service: StorageService = Depends(get_storage_service), + s3_url_service: S3URLService = Depends(get_s3_url_service), +) -> PIIDecisionResponse: + return await _process_pii_decision( + job_id=job_id, + decision="denied", + body=body, + background_tasks=background_tasks, + redis_client=redis_client, + storage_service=storage_service, + s3_url_service=s3_url_service, + ) + + +async def _process_pii_decision( + *, + job_id: str, + decision: Literal["approved", "denied"], + body: PIIDecisionInput, + background_tasks: BackgroundTasks, + redis_client: Any, + storage_service: StorageService, + s3_url_service: S3URLService, +) -> PIIDecisionResponse: + """Shared body for approve_pii + deny_pii. + + Pre-validates the job's current status (404 if missing, 409 if not + awaiting_approval) before invoking the approval_service. That contract is + what the Canvas LTI connector relies on to surface "another instructor + decided in a parallel tab" as a 409 to the operator. The token-based + sibling endpoint in api.approval lets quick_approve / quick_deny set the + status unconditionally; that's safe there because the token already + proves the job is in awaiting_approval, but it isn't safe by job_id. + + Connector contract pinned by: + https://github.com/oshrizak/reflow-canvas-lti — the Canvas LTI + connector tries this endpoint first when forwarding a faculty PII + decision; on 404/405 it falls back to + ``POST /api/v1/approval/{token}/decision`` so the connector keeps + working against Core deployments that pre-date this PR. Both + branches are covered in the connector's + ``tests/integration/test_pii_decision.py`` + (``test_pii_approve_prefers_by_job_id_endpoint`` + + ``test_pii_approve_falls_back_to_token_endpoint_on_405``). + Net effect once this PR merges + Core ships: connector drops the + approval-token round-trip and submits decisions in one POST. + """ + job_service = JobService(redis_client) + queue_service = QueueService(redis_client) + approval_service = ApprovalService( + redis_client=redis_client, + s3_client=None, # Lazy-loaded inside the background task on the denial path. + job_service=job_service, + queue_service=queue_service, + storage_service=storage_service, + s3_url_service=s3_url_service, + ) + + job = await job_service.get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail=f"Job {job_id} not found") + + current_status = str(job.get("status") or "") + if current_status != "awaiting_approval": + raise HTTPException( + status_code=409, + detail=( + f"Job {job_id} is in {current_status!r}, not 'awaiting_approval' " + f"— a decision was already recorded." + ), + ) + + s3_key = job.get("s3_key", "") + justification = body.justification or "" + + try: + if decision == "approved": + await approval_service.quick_approve(job_id) + background_tasks.add_task( + approval_service.process_approval_background, + job_id=job_id, + s3_key=s3_key, + justification=justification, + reviewed_by=body.reviewed_by, + ) + return PIIDecisionResponse( + message="Job approved - processing started", + job_id=job_id, + decision="approved", + ) + + await approval_service.quick_deny(job_id) + background_tasks.add_task( + approval_service.process_denial_background, + job_id=job_id, + s3_key=s3_key, + justification=justification, + reviewed_by=body.reviewed_by, + ) + return PIIDecisionResponse( + message="Job denied - cleanup started", + job_id=job_id, + decision="denied", + ) + except ValueError as exc: + # Defensive: approval_service can raise ValueError on data checks + # (e.g. missing s3_key). The job exists but isn't in a decidable + # state — surface as 409 not 500. + raise HTTPException(status_code=409, detail=str(exc)) from exc + except Exception as exc: # noqa: BLE001 + logger.exception("PII decision (%s) failed for job=%s", decision, job_id) + raise HTTPException( + status_code=500, + detail=f"Failed to process PII decision: {exc}", + ) from exc diff --git a/tests/integration/api/test_pii_decision_by_job_id.py b/tests/integration/api/test_pii_decision_by_job_id.py new file mode 100644 index 0000000..e67b40a --- /dev/null +++ b/tests/integration/api/test_pii_decision_by_job_id.py @@ -0,0 +1,196 @@ +"""Integration tests for the by-job-id PII decision endpoints. + +Covers POST /api/v1/documents/{job_id}/pii/approve and +POST /api/v1/documents/{job_id}/pii/deny — the variants the Canvas LTI +connector consumes. The token-based equivalent at +/api/v1/approval/{token}/decision is covered by test_approval_flow.py. +""" + +from unittest.mock import AsyncMock, patch + +import pytest +from httpx import ASGITransport, AsyncClient +from src.dependencies import ( + get_redis_client, + get_s3_url_service, + get_storage_service, +) +from src.main import app + + +@pytest.fixture +def job_id(): + return "550e8400-e29b-41d4-a716-446655440000" + + +@pytest.fixture +def awaiting_approval_job(job_id): + return { + "job_id": job_id, + "s3_key": "temp/test-doc.pdf", + "status": "awaiting_approval", + } + + +@pytest.fixture +def decided_job(job_id): + """Job that already moved past awaiting_approval — should trigger 409.""" + return { + "job_id": job_id, + "s3_key": "temp/test-doc.pdf", + "status": "processing_queued", + } + + +@pytest.fixture +def post_body(): + return { + "justification": "Author bylines and citations — not student PII.", + "reviewed_by": "faculty@example.edu", + } + + +def _override_deps(mock_redis: AsyncMock) -> None: + """Wire dependency_overrides for redis + storage + s3 url services.""" + app.dependency_overrides[get_redis_client] = lambda: mock_redis + app.dependency_overrides[get_storage_service] = lambda: AsyncMock() + app.dependency_overrides[get_s3_url_service] = lambda: AsyncMock() + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_approve_pii_happy_path( + job_id, awaiting_approval_job, post_body, api_key_headers +): + """Happy path: existing awaiting_approval job → 200, status flipped to processing_queued.""" + mock_redis = AsyncMock() + _override_deps(mock_redis) + mock_job_service = AsyncMock() + mock_job_service.get_job.return_value = awaiting_approval_job + mock_approval_service = AsyncMock() + + try: + with ( + patch("src.api.documents.JobService", return_value=mock_job_service), + patch("src.api.documents.ApprovalService", return_value=mock_approval_service), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + f"/api/v1/documents/{job_id}/pii/approve", + json=post_body, + headers=api_key_headers, + ) + + assert response.status_code == 200 + body = response.json() + assert body["decision"] == "approved" + assert body["job_id"] == job_id + assert "approved" in body["message"].lower() + mock_approval_service.quick_approve.assert_awaited_once_with(job_id) + finally: + app.dependency_overrides.clear() + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_deny_pii_happy_path( + job_id, awaiting_approval_job, post_body, api_key_headers +): + """Happy path: deny → 200, quick_deny called once.""" + mock_redis = AsyncMock() + _override_deps(mock_redis) + mock_job_service = AsyncMock() + mock_job_service.get_job.return_value = awaiting_approval_job + mock_approval_service = AsyncMock() + + try: + with ( + patch("src.api.documents.JobService", return_value=mock_job_service), + patch("src.api.documents.ApprovalService", return_value=mock_approval_service), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + f"/api/v1/documents/{job_id}/pii/deny", + json=post_body, + headers=api_key_headers, + ) + + assert response.status_code == 200 + body = response.json() + assert body["decision"] == "denied" + assert body["job_id"] == job_id + assert "denied" in body["message"].lower() + mock_approval_service.quick_deny.assert_awaited_once_with(job_id) + finally: + app.dependency_overrides.clear() + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_approve_pii_job_not_found(job_id, post_body, api_key_headers): + """JobService.get_job returns None → 404.""" + mock_redis = AsyncMock() + _override_deps(mock_redis) + mock_job_service = AsyncMock() + mock_job_service.get_job.return_value = None + + try: + with patch("src.api.documents.JobService", return_value=mock_job_service): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + f"/api/v1/documents/{job_id}/pii/approve", + json=post_body, + headers=api_key_headers, + ) + + assert response.status_code == 404 + assert job_id in response.json()["detail"] + finally: + app.dependency_overrides.clear() + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_approve_pii_already_decided_returns_409( + job_id, decided_job, post_body, api_key_headers +): + """Job exists but isn't in awaiting_approval → 409. + + This is the contract the Canvas LTI connector relies on to surface + "another instructor already decided in a parallel tab" as a 409 to the + operator. Without this pre-check, ``quick_approve`` would silently + overwrite the already-decided state. + """ + mock_redis = AsyncMock() + _override_deps(mock_redis) + mock_job_service = AsyncMock() + mock_job_service.get_job.return_value = decided_job + mock_approval_service = AsyncMock() + + try: + with ( + patch("src.api.documents.JobService", return_value=mock_job_service), + patch("src.api.documents.ApprovalService", return_value=mock_approval_service), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + f"/api/v1/documents/{job_id}/pii/approve", + json=post_body, + headers=api_key_headers, + ) + + assert response.status_code == 409 + assert "processing_queued" in response.json()["detail"] + mock_approval_service.quick_approve.assert_not_awaited() + finally: + app.dependency_overrides.clear() + +