diff --git a/routers/exam/__init__.py b/routers/exam/__init__.py index 35e6488..3aa820d 100644 --- a/routers/exam/__init__.py +++ b/routers/exam/__init__.py @@ -8,9 +8,11 @@ from fastapi import APIRouter from routers.exam.templates import router as templates_router from routers.exam.batches import router as batches_router +from routers.exam.bank import router as bank_router router = APIRouter() router.include_router(templates_router) router.include_router(batches_router) +router.include_router(bank_router) __all__ = ["router"] diff --git a/routers/exam/bank.py b/routers/exam/bank.py new file mode 100644 index 0000000..ff97640 --- /dev/null +++ b/routers/exam/bank.py @@ -0,0 +1,152 @@ +"""Question bank + custom-paper assembly (/api/exam/bank, /api/exam/custom-papers) — mode 3. + +Build-your-own-from-the-spec. The BANK is a browsable index over the leaf questions across the templates +the caller can see (RLS-scoped, institute-wide); a CUSTOM PAPER is a normal exam_template whose questions +are COPIES of the selected bank questions, so it flows through the existing setup / marking / projection +pipeline unchanged. Copy-on-assemble (v1) avoids decoupling question identity from a template; a shared- +question model is Phase-2. Design: ~/cc/ideas/2026-07-02-mode3-question-bank-design.md. + +All access is as-the-user (E1): the bank query returns only questions in templates RLS lets the caller see, +and a custom paper is created owned by the caller + institute-scoped (writes pass the same RLS with-check). +""" +from __future__ import annotations + +import os +import uuid +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel + +from modules.database.services.exam_projection import project_template_safe +from modules.logger_tool import initialise_logger +from routers.exam.dependencies import ExamContext, get_exam_context + +logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True) + +router = APIRouter() + + +def _rows(result: Any) -> List[Dict[str, Any]]: + data = getattr(result, "data", None) + if not data: + return [] + return data if isinstance(data, list) else [data] + + +def _first(result: Any) -> Optional[Dict[str, Any]]: + rows = _rows(result) + return rows[0] if rows else None + + +@router.get("/bank") +async def list_bank( + spec_ref: Optional[str] = None, + subject: Optional[str] = None, + ctx: ExamContext = Depends(get_exam_context), +) -> Dict[str, Any]: + """Leaf questions across the caller's institute templates, with their source-paper context. + + Filter by `spec_ref` (spec point) and/or `subject`. Facets over the full visible set drive the UI + filters (so filtering by one axis doesn't hide the others' options). This is the spec-planning surface: + a teacher picks a spec point → gets every question tagged it across their papers → assembles a paper. + """ + sel = ( + "id, template_id, label, max_marks, answer_type, spec_ref, bounds, page, " + "exam_templates(id, title, subject, exam_code)" + ) + query = ctx.supabase.table("exam_questions").select(sel).eq("is_container", False) + if spec_ref: + query = query.eq("spec_ref", spec_ref) + rows = _rows(query.execute()) + + facets_spec: Dict[str, int] = {} + facets_subject: Dict[str, int] = {} + items: List[Dict[str, Any]] = [] + for r in rows: + tmpl = r.get("exam_templates") or {} + subj = tmpl.get("subject") + if r.get("spec_ref"): + facets_spec[r["spec_ref"]] = facets_spec.get(r["spec_ref"], 0) + 1 + if subj: + facets_subject[subj] = facets_subject.get(subj, 0) + 1 + if subject and subj != subject: + continue + items.append({ + "id": r["id"], "template_id": r["template_id"], "label": r.get("label"), + "max_marks": r.get("max_marks"), "answer_type": r.get("answer_type"), + "spec_ref": r.get("spec_ref"), "bounds": r.get("bounds"), "page": r.get("page"), + "paper": {"id": tmpl.get("id"), "title": tmpl.get("title"), + "subject": subj, "exam_code": tmpl.get("exam_code")}, + }) + return {"questions": items, "n": len(items), + "facets": {"spec_ref": facets_spec, "subject": facets_subject}} + + +class CustomPaperRequest(BaseModel): + title: str + subject: Optional[str] = None + institute_id: Optional[str] = None + question_ids: List[str] + + +@router.post("/custom-papers") +async def create_custom_paper( + body: CustomPaperRequest, + ctx: ExamContext = Depends(get_exam_context), +) -> Dict[str, Any]: + """Assemble the selected bank questions into a new template (copy-on-assemble), then project it.""" + if not body.question_ids: + raise HTTPException(status_code=400, detail="question_ids is required") + institute_id = ctx.resolve_institute(body.institute_id) + + # RLS makes this return only leaf questions the caller may see; unknown/forbidden ids are silently dropped. + srcs = _rows( + ctx.supabase.table("exam_questions").select("*") + .in_("id", body.question_ids).eq("is_container", False).execute() + ) + by_id = {s["id"]: s for s in srcs} + ordered = [by_id[qid] for qid in body.question_ids if qid in by_id] # preserve the caller's order + if not ordered: + raise HTTPException(status_code=404, detail="No accessible questions for the given ids") + + template_id = str(uuid.uuid4()) + ctx.supabase.table("exam_templates").insert({ + "id": template_id, "title": body.title, "subject": body.subject, + "institute_id": institute_id, "teacher_id": ctx.user_id, "status": "draft", + }).execute() + + id_map: Dict[str, str] = {} + q_rows: List[Dict[str, Any]] = [] + for order, s in enumerate(ordered): + new_id = str(uuid.uuid4()) + id_map[s["id"]] = new_id + q_rows.append({ + "id": new_id, "template_id": template_id, "parent_id": None, + "label": s.get("label"), "order": order, "max_marks": s.get("max_marks") or 0, + "answer_type": s.get("answer_type"), "mcq_options": s.get("mcq_options"), + "mark_scheme": s.get("mark_scheme") or {}, "is_container": False, + "spec_ref": s.get("spec_ref"), "bounds": s.get("bounds"), "page": s.get("page"), + "source": "manual", "confirmed": True, + }) + ctx.supabase.table("exam_questions").insert(q_rows).execute() + + # Copy each selected question's response areas onto its new copy (geometry travels with the question). + ra_rows: List[Dict[str, Any]] = [] + if id_map: + for ra in _rows( + ctx.supabase.table("exam_response_areas").select("*") + .in_("question_id", list(id_map.keys())).execute() + ): + ra_rows.append({ + "id": str(uuid.uuid4()), "question_id": id_map[ra["question_id"]], "template_id": template_id, + "page": ra.get("page"), "bounds": ra.get("bounds"), "kind": ra.get("kind"), + "response_form": ra.get("response_form"), "context_type": ra.get("context_type"), + "source": "manual", "confirmed": True, + }) + if ra_rows: + ctx.supabase.table("exam_response_areas").insert(ra_rows).execute() + + project_template_safe(template_id) # a custom paper is a normal paper in the graph + return {"id": template_id, "title": body.title, "subject": body.subject, + "n_questions": len(q_rows), "n_response_areas": len(ra_rows)} diff --git a/tests/test_exam_bank.py b/tests/test_exam_bank.py new file mode 100644 index 0000000..434c37c --- /dev/null +++ b/tests/test_exam_bank.py @@ -0,0 +1,134 @@ +"""Tests for the question-bank + custom-paper assembly router (/api/exam/bank, /custom-papers) — mode 3. + +Same FakeSupabase + dependency-override pattern as test_exam_templates.py; the as-user RLS itself is +verified against the live dev DB (.94), not here. +""" +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +import routers.exam.bank as bank_mod +from routers.exam.bank import router +from routers.exam.dependencies import ExamContext, get_exam_context + + +TEACHER = "00000000-0000-0000-0000-000000000001" +INST_A = "10000000-0000-0000-0000-000000000001" + + +@pytest.fixture(autouse=True) +def _stub_projection(monkeypatch): + monkeypatch.setattr(bank_mod, "project_template_safe", lambda tid: None) + + +class FakeResult: + def __init__(self, data): + self.data = data + + +class FakeQuery: + def __init__(self, store, table): + self.store = store + self.table = table + self.rows = list(store.get(table, [])) + self._op = None + self._payload = None + + def select(self, *_a, **_k): + self._op = "select" + return self + + def insert(self, payload): + self._op = "insert" + self._payload = payload + return self + + def eq(self, key, value): + self.rows = [r for r in self.rows if r.get(key) == value] + return self + + def in_(self, key, values): + values = set(values) + self.rows = [r for r in self.rows if r.get(key) in values] + return self + + def execute(self): + backing = self.store.setdefault(self.table, []) + if self._op == "insert": + payloads = self._payload if isinstance(self._payload, list) else [self._payload] + inserted = [] + for p in payloads: + row = dict(p) + row.setdefault("id", f"gen-{self.table}-{len(backing)}") + backing.append(row) + inserted.append(row) + return FakeResult(inserted) + return FakeResult(self.rows) + + +class FakeSupabase: + def __init__(self, store): + self.store = store + + def table(self, name): + return FakeQuery(self.store, name) + + +def make_client(store, user_id=TEACHER, institute_ids=(INST_A,)): + app = FastAPI() + app.include_router(router, prefix="/api/exam") + app.dependency_overrides[get_exam_context] = lambda: ExamContext(user_id, "tok", FakeSupabase(store), list(institute_ids)) + return TestClient(app), store + + +def test_bank_lists_leaf_questions_with_facets_excluding_containers(): + store = {"exam_questions": [ + {"id": "c1", "template_id": "t1", "label": "Q1", "is_container": True, "spec_ref": None, + "exam_templates": {"id": "t1", "title": "Physics 1H", "subject": "physics", "exam_code": "AQA-PHYS-8463"}}, + {"id": "q1", "template_id": "t1", "label": "01.1", "is_container": False, "max_marks": 3, "spec_ref": "4.2", + "exam_templates": {"id": "t1", "title": "Physics 1H", "subject": "physics", "exam_code": "AQA-PHYS-8463"}}, + {"id": "q2", "template_id": "t1", "label": "01.2", "is_container": False, "max_marks": 2, "spec_ref": "4.2", + "exam_templates": {"id": "t1", "title": "Physics 1H", "subject": "physics", "exam_code": "AQA-PHYS-8463"}}, + ]} + c, _ = make_client(store) + body = c.get("/api/exam/bank").json() + labels = {q["label"] for q in body["questions"]} + assert labels == {"01.1", "01.2"} # container excluded + assert body["facets"]["spec_ref"] == {"4.2": 2} + assert body["questions"][0]["paper"]["exam_code"] == "AQA-PHYS-8463" + + +def test_bank_filters_by_spec_ref(): + store = {"exam_questions": [ + {"id": "q1", "template_id": "t1", "label": "a", "is_container": False, "spec_ref": "4.2", "exam_templates": {"subject": "physics"}}, + {"id": "q2", "template_id": "t1", "label": "b", "is_container": False, "spec_ref": "4.5", "exam_templates": {"subject": "physics"}}, + ]} + c, _ = make_client(store) + body = c.get("/api/exam/bank?spec_ref=4.2").json() + assert [q["id"] for q in body["questions"]] == ["q1"] + + +def test_custom_paper_copies_selected_questions_in_order(): + store = {"exam_questions": [ + {"id": "q1", "template_id": "t1", "label": "01.1", "is_container": False, "max_marks": 3, "spec_ref": "4.2", "answer_type": "short"}, + {"id": "q2", "template_id": "t2", "label": "05.1", "is_container": False, "max_marks": 6, "spec_ref": "4.5", "answer_type": "written"}, + ]} + c, store = make_client(store) + r = c.post("/api/exam/custom-papers", json={"title": "Electricity mini-test", "subject": "physics", "question_ids": ["q2", "q1"]}) + assert r.status_code == 200 + body = r.json() + assert body["n_questions"] == 2 + # a new template row was created, owned + institute-scoped + new_tpls = [t for t in store["exam_templates"] if t["id"] == body["id"]] + assert new_tpls and new_tpls[0]["teacher_id"] == TEACHER and new_tpls[0]["institute_id"] == INST_A + # two COPIES exist under the new template, order preserving the requested [q2, q1], with fresh ids + copies = sorted([q for q in store["exam_questions"] if q["template_id"] == body["id"]], key=lambda q: q["order"]) + assert [q["label"] for q in copies] == ["05.1", "01.1"] + assert all(q["id"] not in ("q1", "q2") for q in copies) # not the originals + assert copies[0]["max_marks"] == 6 and copies[0]["spec_ref"] == "4.5" + + +def test_custom_paper_requires_accessible_questions(): + c, _ = make_client({"exam_questions": []}) + assert c.post("/api/exam/custom-papers", json={"title": "x", "question_ids": ["nope"]}).status_code == 404 + assert c.post("/api/exam/custom-papers", json={"title": "x", "question_ids": []}).status_code == 400