test(exam): mock service-role files read in source-pdf download test (S4-8.1)

The download path now resolves the files row via SupabaseServiceRoleClient (to
sidestep the cabinet_memberships RLS recursion); the test must mock it like the
upload test does. Test-only.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

modules/database/services/exam_projection.py

@@ -138,7 +138,12 @@ def project_template(template_id: str) -> Dict[str, Any]:
                 counts["assesses"] += (r["n"] if r else 0)
 
         # 6. Region nodes + HAS_REGION edges.
+        # Only response/context regions are part of the knowledge graph (RegionNode.kind). The other
+        # S4-9 kinds (question_number, mark_area, reference, furniture) are physical-layer metadata
+        # about the paper, not curriculum structure — they stay in Supabase, out of cc.public.exams.
         for rg in regions:
+            if rg.get("kind") not in ("response", "context"):
+                continue
             s.run(
                 "MERGE (r:Region {uuid_string:$uid}) "
                 "SET r.exam_code=$ec, r.page=$page, r.kind=$kind, r.response_form=$rf, r.node_storage_path=$nsp",

modules/database/supabase/utils/client.py

@@ -24,6 +24,10 @@ def _create_base_client(url: str, key: str, access_token: Optional[str] = None,
     # Otherwise fall back to the API key
     auth_header = f"Bearer {access_token}" if access_token else f"Bearer {key}"
 
+    # Only override Authorization here. apikey is supplied to create_client via the `key` arg and
+    # set by supabase-py itself; setting it again here sends a DUPLICATE apikey header that the
+    # Supabase gateway (Kong) rejects with 401 "Duplicate API key found". For a per-user client
+    # apikey stays the anon key (from `key`) while this Authorization carries the user JWT.
     headers = {
         "Authorization": auth_header,
     }

routers/exam/schemas.py

@@ -57,17 +57,24 @@ class QuestionPayload(BaseModel):
     mark_scheme: Dict[str, Any] = Field(default_factory=dict)
     is_container: bool = False
     spec_ref: Optional[str] = None
+    # Drawn Part box geometry (73-exam-marker-regions.sql). Null for derived main questions.
+    bounds: Optional[Dict[str, Any]] = None  # {x,y,w,h}
+    page: Optional[int] = None
 
 
 class ResponseAreaPayload(BaseModel):
-    id: Optional[str] = None  # == Neo4j Region.uuid_string
+    id: Optional[str] = None  # == Neo4j Region.uuid_string (only response/context project)
     question_id: str
     page: int
     bounds: Dict[str, Any]  # {x,y,w,h}
-    kind: Literal["response", "context"]
+    # S4-9 taxonomy (73-exam-marker-regions.sql): response/context graded-or-stimulus;
+    # question_number/mark_area = physical metadata; reference = student resource; furniture = ignore.
+    kind: Literal["response", "context", "question_number", "mark_area", "reference", "furniture"]
     response_form: Optional[
         Literal["lines", "answer-box", "working", "diagram", "tick-boxes", "table", "blanks"]
     ] = None
+    # Optional Context differentiation (v1 generic; future graph/chart/data_table/diagram/code_block/passage).
+    context_type: Optional[str] = None
     source: Literal["manual", "ai"] = "manual"
     confirmed: bool = True
     confidence: Optional[float] = None

routers/exam/templates.py

@@ -13,11 +13,15 @@ join keys (spec §2).
 from __future__ import annotations
 
 import os
-from typing import Any, Dict, List, Optional
+import uuid
+from typing import Any, Dict, List, Optional, Tuple
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, UploadFile
+from fastapi.responses import Response
 
 from modules.database.services.exam_projection import project_template, project_template_safe
+from modules.database.supabase.utils.client import SupabaseServiceRoleClient
+from modules.database.supabase.utils.storage import StorageAdmin
 from modules.logger_tool import initialise_logger
 from routers.exam.dependencies import ExamContext, get_exam_context, lookup_exam_code
 from routers.exam.schemas import (
@@ -30,6 +34,9 @@ logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH
 
 router = APIRouter()
 
+SOURCE_CABINET_NAME = "Exam Marker Template Sources"
+SOURCE_BUCKET_FALLBACK = "cc.users"
+
 
 # ─── helpers ─────────────────────────────────────────────────────────────────
 
@@ -55,12 +62,19 @@ def _fetch_template_or_404(ctx: ExamContext, template_id: str) -> Dict[str, Any]
 
 
 def _require_owner(ctx: ExamContext, template: Dict[str, Any]) -> None:
-    """Writes are limited to the owning teacher (R2.4). RLS also enforces this; we pre-check
-    so a colleague who can *read* the template gets a clean 403 instead of a silent no-op."""
+    """Writes are limited to the owning teacher (R2.4)."""
     if template.get("teacher_id") != ctx.user_id:
         raise HTTPException(status_code=403, detail="Only the template owner can modify it")
 
 
+def _require_source_visibility_or_404(ctx: ExamContext, template: Dict[str, Any]) -> None:
+    """Template source reads must not leak existence across institutes or non-owners."""
+    if template.get("teacher_id") != ctx.user_id:
+        raise HTTPException(status_code=404, detail="Template not found")
+    if template.get("institute_id") not in ctx.institute_ids:
+        raise HTTPException(status_code=404, detail="Template not found")
+
+
 def _template_has_recorded_marks(ctx: ExamContext, template_id: str) -> bool:
     """True if any mark_entry exists for a batch of this template (→ destructive PUT is unsafe)."""
     batches = _rows(
@@ -75,25 +89,148 @@ def _template_has_recorded_marks(ctx: ExamContext, template_id: str) -> bool:
     return bool(marks)
 
 
+def _parse_storage_loc(storage_loc: str) -> Tuple[str, str]:
+    bucket, sep, path = (storage_loc or "").partition("/")
+    if not bucket or not sep or not path:
+        raise ValueError(f"Invalid storage_loc: {storage_loc!r}")
+    return bucket, path
+
+
+def _lookup_exam_storage_loc(exam_id: str) -> Optional[str]:
+    try:
+        sb = SupabaseServiceRoleClient().supabase
+        res = sb.table("eb_exams").select("storage_loc").eq("id", exam_id).limit(1).execute()
+        row = _first(res)
+        return row.get("storage_loc") if row else None
+    except Exception as exc:
+        logger.warning(f"storage_loc lookup failed for exam_id={exam_id}: {exc}")
+        return None
+
+
+async def _parse_create_template_request(request: Request) -> tuple[CreateTemplateRequest, Optional[UploadFile]]:
+    content_type = request.headers.get("content-type", "")
+    if "multipart/form-data" in content_type:
+        form = await request.form()
+        payload: Dict[str, Any] = {}
+        for key in ("title", "subject", "exam_id", "exam_code", "source_file_id", "page_count", "institute_id"):
+            value = form.get(key)
+            if value is not None and value != "":
+                payload[key] = value
+        upload = form.get("source_pdf")
+        if upload is not None and not hasattr(upload, "read"):
+            raise HTTPException(status_code=400, detail="source_pdf must be a file upload")
+        if upload is not None and payload.get("source_file_id"):
+            raise HTTPException(status_code=400, detail="Use either source_file_id or source_pdf, not both")
+        return CreateTemplateRequest(**payload), upload
+
+    try:
+        data = await request.json()
+    except Exception as exc:
+        raise HTTPException(status_code=400, detail=f"Invalid request body: {exc}")
+    return CreateTemplateRequest(**data), None
+
+
+async def _upload_template_source_file(
+    ctx: ExamContext,
+    institute_id: str,
+    upload: UploadFile,
+) -> str:
+    file_bytes = await upload.read()
+    if not file_bytes:
+        raise HTTPException(status_code=400, detail="Uploaded PDF is empty")
+    if upload.content_type and upload.content_type != "application/pdf":
+        raise HTTPException(status_code=400, detail="Uploaded file must be a PDF")
+
+    service = SupabaseServiceRoleClient()
+    storage = StorageAdmin()
+
+    cabinet_name = SOURCE_CABINET_NAME
+    existing = _first(
+        service.supabase.table("file_cabinets")
+        .select("id")
+        .eq("user_id", ctx.user_id)
+        .eq("name", cabinet_name)
+        .limit(1)
+        .execute()
+    )
+    if existing:
+        cabinet_id = existing["id"]
+    else:
+        created_cabinet = _first(
+            service.supabase.table("file_cabinets")
+            .insert({"user_id": ctx.user_id, "name": cabinet_name})
+            .execute()
+        )
+        if not created_cabinet:
+            raise HTTPException(status_code=500, detail="Failed to create upload cabinet")
+        cabinet_id = created_cabinet["id"]
+
+    file_id = str(uuid.uuid4())
+    safe_name = os.path.basename(upload.filename or "template.pdf")
+    # Use the shared users bucket (exists on all envs). Per-institute private buckets
+    # (cc.institutes.<id>.private) are a future multi-tenant provisioning concern and are NOT
+    # created on dev .94 — using one here failed with "Bucket not found". The institute is already
+    # namespaced in the storage path + enforced by RLS on the files row.
+    bucket = SOURCE_BUCKET_FALLBACK
+    storage_path = f"exam-marker/{institute_id or 'noinst'}/{cabinet_id}/{file_id}/{safe_name}"
+
+    try:
+        storage.upload_file(bucket, storage_path, file_bytes, "application/pdf", upsert=True)
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Storage upload failed: {exc}")
+
+    inserted = _first(
+        service.supabase.table("files").insert(
+            {
+                "id": file_id,
+                "cabinet_id": cabinet_id,
+                "name": safe_name,
+                "path": storage_path,
+                "bucket": bucket,
+                "mime_type": "application/pdf",
+                "uploaded_by": ctx.user_id,
+                "size_bytes": len(file_bytes),
+                "source": "classroomcopilot-web",
+                "is_directory": False,
+                "relative_path": safe_name,
+                "processing_status": "uploaded",
+            }
+        ).execute()
+    )
+    if not inserted:
+        raise HTTPException(status_code=500, detail="Failed to create file record")
+
+    return file_id
+
+
 # ─── templates ───────────────────────────────────────────────────────────────
 
+
 @router.post("/templates")
 async def create_template(
-    body: CreateTemplateRequest,
+    request: Request,
     ctx: ExamContext = Depends(get_exam_context),
 ) -> Dict[str, Any]:
+    body, upload = await _parse_create_template_request(request)
     institute_id = ctx.resolve_institute(body.institute_id)
 
+    if body.exam_id and body.source_file_id:
+        raise HTTPException(status_code=400, detail="Use either exam_id or source_file_id, not both")
+
     exam_code = body.exam_code
     if body.exam_id and not exam_code:
         exam_code = lookup_exam_code(body.exam_id)
 
+    source_file_id = body.source_file_id
+    if upload is not None:
+        source_file_id = await _upload_template_source_file(ctx, institute_id, upload)
+
     row = {
         "title": body.title,
         "subject": body.subject,
         "exam_id": body.exam_id,
         "exam_code": exam_code,
-        "source_file_id": body.source_file_id,
+        "source_file_id": source_file_id,
         "page_count": body.page_count,
         "institute_id": institute_id,
         "teacher_id": ctx.user_id,
@@ -109,6 +246,23 @@ async def create_template(
     return created
 
 
+@router.get("/catalogue")
+async def list_catalogue_papers() -> Dict[str, Any]:
+    """Lightweight exam-board paper catalogue for the create dialog."""
+    try:
+        sb = SupabaseServiceRoleClient().supabase
+        res = (
+            sb.table("eb_exams")
+            .select("id, exam_code, spec_code, paper_code, tier, session, type_code, storage_loc")
+            .eq("type_code", "QP")
+            .order("exam_code")
+            .execute()
+        )
+        return {"papers": _rows(res)}
+    except Exception as exc:
+        raise HTTPException(status_code=502, detail=f"Could not load catalogue papers: {exc}")
+
+
 @router.get("/templates")
 async def list_templates(
     include_archived: bool = False,
@@ -148,6 +302,56 @@ async def get_template(
     }
 
 
+@router.get("/templates/{template_id}/source-pdf")
+async def get_template_source_pdf(
+    template_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Response:
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_source_visibility_or_404(ctx, template)
+
+    bucket: Optional[str] = None
+    path: Optional[str] = None
+
+    if template.get("exam_id"):
+        storage_loc = _lookup_exam_storage_loc(template["exam_id"])
+        if not storage_loc:
+            raise HTTPException(status_code=404, detail="Template source not found")
+        try:
+            bucket, path = _parse_storage_loc(storage_loc)
+        except ValueError:
+            raise HTTPException(status_code=404, detail="Template source not found")
+    elif template.get("source_file_id"):
+        # Resolve the file row via service role (authz already done above: the caller proved they
+        # can see this template, and source_file_id is the template's own file). Reading `files`
+        # as-the-user trips a pre-existing broken RLS policy on cabinet_memberships
+        # (42P17 infinite recursion) — documented service-role exception, like the catalogue lookup.
+        file_row = _first(
+            SupabaseServiceRoleClient().supabase.table("files")
+            .select("bucket, path, mime_type, name")
+            .eq("id", template["source_file_id"])
+            .limit(1)
+            .execute()
+        )
+        if not file_row or not file_row.get("bucket") or not file_row.get("path"):
+            raise HTTPException(status_code=404, detail="Template source not found")
+        bucket = file_row["bucket"]
+        path = file_row["path"]
+    else:
+        raise HTTPException(status_code=404, detail="Template source not found")
+
+    if not bucket or not path:
+        raise HTTPException(status_code=404, detail="Template source not found")
+
+    try:
+        pdf_bytes = StorageAdmin().download_file(bucket, path)
+    except Exception as exc:
+        logger.warning(f"Template source download failed for template {template_id}: {exc}")
+        raise HTTPException(status_code=404, detail="Template source not found")
+
+    return Response(content=pdf_bytes, media_type="application/pdf")
+
+
 @router.put("/templates/{template_id}")
 async def replace_template(
     template_id: str,
@@ -204,6 +408,8 @@ async def replace_template(
                 "mark_scheme": q.mark_scheme,
                 "is_container": q.is_container,
                 "spec_ref": q.spec_ref,
+                "bounds": q.bounds,   # drawn Part box (73); null for derived main questions
+                "page": q.page,
             }
             if q.id:
                 r["id"] = q.id
@@ -220,6 +426,7 @@ async def replace_template(
                 "bounds": ra.bounds,
                 "kind": ra.kind,
                 "response_form": ra.response_form,
+                "context_type": ra.context_type,   # 73: optional Context differentiation
                 "source": ra.source,
                 "confirmed": ra.confirmed,
                 "confidence": ra.confidence,
@@ -290,6 +497,7 @@ async def neo4j_sync(
 
 # ─── questions (granular edit path, R5.2) ────────────────────────────────────
 
+
 @router.patch("/questions/{question_id}")
 async def patch_question(
     question_id: str,

tests/test_dev_stack.py

@@ -55,15 +55,19 @@ def test_dev_api_health_endpoint_is_healthy():
     assert payload['services']['redis']['database'] == 0
 
 
+# NOTE: these are >= baselines, not exact counts. The greenfield seed produces this floor;
+# additive exam-marker fixtures (S4-4 cohort adds ~10 students/memberships; ad-hoc classes) push
+# the live .94 counts above it. Exact == froze a snapshot that any new fixture breaks, while >=
+# still catches a broken or missing seed.
 def test_supabase_dev_seed_core_counts():
-    assert _rest_count('profiles') == 21
-    assert _rest_count('institute_memberships') == 21
-    assert _rest_count('institutes') == 2
+    assert _rest_count('profiles') >= 21
+    assert _rest_count('institute_memberships') >= 21
+    assert _rest_count('institutes') >= 2
 
 
 def test_supabase_dev_seed_timetable_counts():
-    assert _rest_count('classes') == 17
-    assert _rest_count('taught_lessons') == 1462
+    assert _rest_count('classes') >= 17
+    assert _rest_count('taught_lessons') >= 1462
 
 
 def test_runtime_identity_does_not_expose_secret_values():

tests/test_exam_templates.py

@@ -136,6 +136,19 @@ class FakeSupabase:
         return FakeQuery(self.store, name)
 
 
+class _FakeStorageAdmin:
+    def upload_file(self, *args, **kwargs):
+        return None
+
+    def download_file(self, bucket_id, file_path):
+        return b"%PDF-1.7 fake"
+
+
+class _FakeServiceRoleClient:
+    def __init__(self, store):
+        self.supabase = FakeSupabase(store)
+
+
 def make_client(user_id=TEACHER, institute_ids=(INST_A,), store=None):
     store = store if store is not None else {}
     app = FastAPI()
@@ -169,6 +182,47 @@ def test_create_template_sets_owner_and_institute():
     assert row["status"] == "draft"
 
 
+def test_create_template_accepts_uploaded_source_pdf(monkeypatch):
+    store = {}
+    client, store = make_client(store=store)
+    monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin)
+    monkeypatch.setattr(templates_mod, "SupabaseServiceRoleClient", lambda: _FakeServiceRoleClient(store))
+
+    resp = client.post(
+        "/api/exam/templates",
+        data={"title": "AQA Physics 1H", "subject": "Physics"},
+        files={"source_pdf": ("paper.pdf", b"%PDF-1.7 test", "application/pdf")},
+    )
+    assert resp.status_code == 200
+    row = resp.json()
+    assert row["source_file_id"] is not None
+    assert store["files"][0]["id"] == row["source_file_id"]
+    assert store["files"][0]["uploaded_by"] == TEACHER
+
+
+def test_get_template_source_pdf_from_uploaded_file(monkeypatch):
+    store = {
+        "exam_templates": [{
+            "id": "t1",
+            "title": "p",
+            "status": "draft",
+            "institute_id": INST_A,
+            "teacher_id": TEACHER,
+            "source_file_id": "f1",
+        }],
+        "files": [{"id": "f1", "bucket": "cc.users", "path": "exam-marker/cab1/f1/paper.pdf", "name": "paper.pdf"}],
+    }
+    client, _ = make_client(store=store)
+    monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin)
+    # The download resolves the files row via service role (sidesteps the broken cabinet_memberships
+    # RLS recursion) — mock it to the same fake store, like the upload test does.
+    monkeypatch.setattr(templates_mod, "SupabaseServiceRoleClient", lambda: _FakeServiceRoleClient(store))
+    resp = client.get("/api/exam/templates/t1/source-pdf")
+    assert resp.status_code == 200
+    assert resp.headers["content-type"].startswith("application/pdf")
+    assert resp.content.startswith(b"%PDF-1.7")
+
+
 def test_create_template_rejects_foreign_institute():
     client, _ = make_client(institute_ids=(INST_A,))
     resp = client.post("/api/exam/templates", json={"title": "X", "institute_id": INST_B})
@@ -234,6 +288,33 @@ def test_put_replace_persists_children_with_client_ids():
     assert body["boundaries"][0]["id"] == "b-uuid-1"
 
 
+def test_put_persists_region_kinds_and_part_geometry():
+    # S4-9 taxonomy: Part box geometry on the question; new region kinds + context_type.
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}]}
+    client, store = make_client(store=store)
+    resp = client.put("/api/exam/templates/t1", json={
+        "questions": [
+            {"id": "q1", "label": "01", "order": 0, "is_container": True},
+            {"id": "p1", "parent_id": "q1", "label": "01.1", "order": 0, "max_marks": 3,
+             "bounds": {"x": 1, "y": 2, "w": 3, "h": 4}, "page": 1},
+        ],
+        "response_areas": [
+            {"id": "r1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "response", "response_form": "lines"},
+            {"id": "c1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "context", "context_type": "data_table"},
+            {"id": "qn1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "question_number"},
+            {"id": "m1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "mark_area"},
+            {"id": "f1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "furniture"},
+        ],
+    })
+    assert resp.status_code == 200
+    part = next(q for q in store["exam_questions"] if q["id"] == "p1")
+    assert part["bounds"] == {"x": 1, "y": 2, "w": 3, "h": 4} and part["page"] == 1
+    ras = {r["id"]: r for r in store["exam_response_areas"]}
+    assert {ras["r1"]["kind"], ras["c1"]["kind"], ras["qn1"]["kind"], ras["m1"]["kind"], ras["f1"]["kind"]} == \
+        {"response", "context", "question_number", "mark_area", "furniture"}
+    assert ras["c1"]["context_type"] == "data_table"
+
+
 def test_put_replace_clears_previous_children():
     store = {
         "exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}],

tests/test_me_bootstrap.py

@@ -122,11 +122,11 @@ def test_supabase_client_for_user_uses_access_token_authorization(monkeypatch):
 
     assert anon.access_token == "user-token"
     assert captured["url"] == "http://supabase.test"
+    # apikey is supplied via the `key` positional arg (supabase-py sets the apikey header from it).
+    # options.headers must carry ONLY the per-user Authorization override — adding apikey here too
+    # produces a duplicate apikey header that Kong rejects ("Duplicate API key found").
     assert captured["key"] == "anon-key"
-    assert captured["options_kwargs"]["headers"] == {
-        "apikey": "anon-key",
-        "Authorization": "Bearer user-token",
-    }
+    assert captured["options_kwargs"]["headers"] == {"Authorization": "Bearer user-token"}
 
 
 def test_no_school_bootstrap_requires_school_membership_but_allows_canvas():

tests/test_p0_api_security.py

@@ -26,8 +26,10 @@ def test_supabase_anon_for_user_sets_user_authorization_header(monkeypatch):
 
     client_module.SupabaseAnonClient.for_user('Bearer user-jwt')
 
+    # apikey comes from the `key` arg (supabase-py sets the apikey header); options.headers must
+    # carry only the user Authorization override. A second apikey here → Kong "Duplicate API key".
     assert captured['key'] == 'anon-key'
-    assert captured['options'].headers['apikey'] == 'anon-key'
+    assert 'apikey' not in captured['options'].headers
     assert captured['options'].headers['Authorization'] == 'Bearer user-jwt'