168 lines
6.4 KiB
Python
168 lines
6.4 KiB
Python
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
import routers.markbook as markbook_mod
|
|
from routers.exam.dependencies import ExamContext
|
|
from routers.markbook import router
|
|
|
|
TEACHER = "00000000-0000-0000-0000-000000000001"
|
|
INST = "10000000-0000-0000-0000-000000000001"
|
|
CLASS = "c-1"
|
|
|
|
|
|
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._filters = []
|
|
self._op = None
|
|
self._payload = None
|
|
self._limit = None
|
|
|
|
def select(self, *_a, **_k):
|
|
self._op = "select"; return self
|
|
|
|
def insert(self, payload):
|
|
self._op = "insert"; self._payload = payload; return self
|
|
|
|
def upsert(self, payload, **_kwargs):
|
|
self._op = "upsert"; self._payload = payload; return self
|
|
|
|
def eq(self, k, v):
|
|
self._filters.append(("eq", k, v)); self.rows = [r for r in self.rows if r.get(k) == v]; return self
|
|
|
|
def in_(self, k, vals):
|
|
vals = set(vals); self._filters.append(("in", k, vals)); self.rows = [r for r in self.rows if r.get(k) in vals]; return self
|
|
|
|
def order(self, *_a, **_k):
|
|
return self
|
|
|
|
def limit(self, n):
|
|
self._limit = n; return self
|
|
|
|
def _match(self, row):
|
|
for op, k, v in self._filters:
|
|
if op == "eq" and row.get(k) != v:
|
|
return False
|
|
if op == "in" and row.get(k) not in v:
|
|
return False
|
|
return True
|
|
|
|
def execute(self):
|
|
backing = self.store.setdefault(self.table, [])
|
|
if self._op in ("insert", "upsert"):
|
|
payloads = self._payload if isinstance(self._payload, list) else [self._payload]
|
|
out = []
|
|
for p in payloads:
|
|
row = dict(p)
|
|
if self._op == "upsert" and self.table == "assessment_marks":
|
|
existing = next((r for r in backing if r.get("assessment_id") == row.get("assessment_id") and r.get("student_id") == row.get("student_id")), None)
|
|
if existing:
|
|
existing.update(row); out.append(existing); continue
|
|
if self._op == "upsert" and row.get("id") is not None:
|
|
existing = next((r for r in backing if r.get("id") == row["id"]), None)
|
|
if existing:
|
|
existing.update(row); out.append(existing); continue
|
|
row.setdefault("id", f"gen-{self.table}-{len(backing)}")
|
|
backing.append(row); out.append(row)
|
|
return FakeResult(out)
|
|
rows = self.rows[: self._limit] if self._limit is not None else self.rows
|
|
return FakeResult(rows)
|
|
|
|
|
|
class FakeSupabase:
|
|
def __init__(self, store):
|
|
self.store = store
|
|
|
|
def table(self, name):
|
|
return FakeQuery(self.store, name)
|
|
|
|
|
|
def make_client(store):
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/markbook")
|
|
from routers.exam.dependencies import get_exam_context
|
|
app.dependency_overrides[get_exam_context] = lambda: ExamContext(TEACHER, "tok", FakeSupabase(store), [INST])
|
|
return TestClient(app)
|
|
|
|
|
|
def base_store(**extra):
|
|
store = {
|
|
"classes": [{"id": CLASS, "name": "10A Maths", "institute_id": INST}],
|
|
"class_students": [
|
|
{"class_id": CLASS, "student_id": "s1", "status": "active"},
|
|
{"class_id": CLASS, "student_id": "s2", "status": "active"},
|
|
{"class_id": CLASS, "student_id": "s3", "status": "inactive"},
|
|
],
|
|
}
|
|
store.update(extra)
|
|
return store
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def names(monkeypatch):
|
|
monkeypatch.setattr(markbook_mod, "resolve_student_names", lambda ids: {sid: {"s1": "Alice", "s2": "Bob"}.get(sid, sid) for sid in ids})
|
|
|
|
|
|
def test_create_assessment_sets_class_and_tenant():
|
|
store = base_store()
|
|
c = make_client(store)
|
|
r = c.post(f"/api/markbook/classes/{CLASS}/assessments", json={"title": "Homework 1", "date": "2026-09-10", "max_marks": 20})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["class_id"] == CLASS
|
|
assert body["tenant_id"] == INST
|
|
assert body["title"] == "Homework 1"
|
|
assert body["max_marks"] == 20
|
|
|
|
|
|
def test_grid_reuses_active_roster_and_computes_summaries():
|
|
store = base_store(
|
|
class_assessments=[
|
|
{"id": "a1", "class_id": CLASS, "tenant_id": INST, "title": "HW", "date": "2026-09-01", "max_marks": 10},
|
|
{"id": "a2", "class_id": CLASS, "tenant_id": INST, "title": "Quiz", "date": "2026-09-08", "max_marks": 20},
|
|
],
|
|
assessment_marks=[
|
|
{"assessment_id": "a1", "student_id": "s1", "mark": 8},
|
|
{"assessment_id": "a2", "student_id": "s1", "mark": 18},
|
|
{"assessment_id": "a1", "student_id": "s2", "mark": 6},
|
|
],
|
|
)
|
|
body = make_client(store).get(f"/api/markbook/classes/{CLASS}/grid").json()
|
|
assert [s["student_name"] for s in body["students"]] == ["Alice", "Bob"]
|
|
alice = body["students"][0]
|
|
assert alice["marks"] == {"a1": 8, "a2": 18}
|
|
assert alice["total"] == 26
|
|
assert alice["percentage"] == 86.7
|
|
assert body["assessment_summaries"][0]["average_mark"] == 7.0
|
|
assert body["summary"]["entered_mark_count"] == 3
|
|
|
|
|
|
def test_mark_upsert_rejects_over_max_and_inactive_students():
|
|
store = base_store(class_assessments=[{"id": "a1", "class_id": CLASS, "tenant_id": INST, "title": "HW", "max_marks": 10}])
|
|
c = make_client(store)
|
|
assert c.put(f"/api/markbook/classes/{CLASS}/assessments/a1/marks/s1", json={"mark": 11}).status_code == 422
|
|
assert c.put(f"/api/markbook/classes/{CLASS}/assessments/a1/marks/s3", json={"mark": 5}).status_code == 404
|
|
ok = c.put(f"/api/markbook/classes/{CLASS}/assessments/a1/marks/s1", json={"mark": 9})
|
|
assert ok.status_code == 200
|
|
assert ok.json()["mark"]["mark"] == 9
|
|
|
|
|
|
def test_csv_export_has_roster_rows_and_assessment_columns():
|
|
store = base_store(
|
|
class_assessments=[{"id": "a1", "class_id": CLASS, "tenant_id": INST, "title": "HW", "max_marks": 10}],
|
|
assessment_marks=[{"assessment_id": "a1", "student_id": "s1", "mark": 8}],
|
|
)
|
|
text = make_client(store).get(f"/api/markbook/classes/{CLASS}/csv").text
|
|
lines = text.strip().splitlines()
|
|
assert lines[0] == "row,student_name,student_id,HW,total,percentage"
|
|
assert lines[1].startswith("1,Alice,s1,8,8")
|
|
assert lines[2].startswith("2,Bob,s2,,,")
|