apps/quiz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

overhaul: single-session deployment + redesigned frontend

Browse Source 
Backend simplification:
- The server now loads ONE pool JSON from $QUIZ_POOL_PATH at startup and
  upserts a single canonical session. The session id comes from the pool
  JSON's optional "session_id" field, falling back to $QUIZ_SESSION_ID.
- The multi-quiz / multi-session CRUD API is gone:
    DELETED  GET/POST /admin/api/quizzes
    DELETED  POST    /admin/api/quizzes/upload
    DELETED  GET/POST /admin/api/sessions
    DELETED  GET     /admin/login (HTML stub)
    DELETED  GET     /admin/api/sessions/{sid}/csv (replaced by /admin/api/csv)
  Replaced with a single-session control surface:
    GET  /admin/                — serves admin.html unconditionally
    GET  /admin/api/state       — admin-gated; pool meta + state + QR + join URL
    POST /admin/api/reset       — admin-gated; wipe submissions + back to lobby
    POST /admin/logout          — clear admin cookie
    GET  /admin/api/csv         — single-session results
    WS   /ws/instructor/{sid}   — kept; new commands "next" + "reset"
- Instructor "Next" button is now a single state-driving command
  (RoomManager.advance_to_next): from lobby it opens Q0; from question_open
  it closes the current Q and opens the next; from question_closed it
  opens the next; if past the last question it ends the session.
- New RoomManager.reset wipes submissions, participants, and per-question
  state, then broadcasts a clean lobby.
- Student GET / now redirects to /?sid=<canonical> when no sid is given,
  so the QR / share URL is fully deterministic.

Frontend rewrite (functional baseline; visual polish to follow):
- /admin/ is now a single SPA: GET /admin/api/state decides login form
  vs dashboard. No separate /admin/login URL bar.
- Admin dashboard is state-driven with one primary action per state.
  QR code, join URL, and live participant list are always visible on the
  left so the operator can leave the page on a projector.
- Student answer buttons are big and tappable; reveal screen highlights
  correct/wrong choice + shows score, total, and rank.
- Static admin/student SPAs share a CSS palette with light/dark support.

Tests rewritten around the single canonical session id.
The auto-bootstrapped session lets each test fixture skip the old
quiz/session creation dance. 39/39 tests pass.

Cleanup:
- Deleted CODEX_PROMPT.md, IMPLEMENTATION_REPORT.md, NOTES.md, SPEC.md,
  static/observer.html (obsolete codex-build artifacts and the unused
  observer view).
- .gitignore now blocks /pool.json (the runtime file the operator drops
  on the server) and the leftover .codex_done / codex_run.log / etc.
- bootstrap.sh seeds /opt/quiz/pool.json from examples/pool_example.json
  on first deploy so a fresh box reaches a usable state without manual
  intervention; .env now includes QUIZ_POOL_PATH.
This commit is contained in:
ameer
2026-05-02 21:13:54 +08:00
parent 32c531247d
commit e7a2f0387b
29 changed files with 1696 additions and 1533 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
tests/conftest.py
View File

@@ -2,6 +2,8 @@
from __future__ import annotations
import json
import pytest
from fastapi.testclient import TestClient
@@ -9,12 +11,16 @@ from app.config import Settings
from app.main import create_app
CANONICAL_SID = "main"
@pytest.fixture
def sample_pool():
return {
"title": "Sample Quiz",
"score_fn": "linear_decay",
"time_limit_default": 2,
"session_id": CANONICAL_SID,
"questions": [
{
"id": "q1",
@@ -57,35 +63,30 @@ def sample_pool():
@pytest.fixture
def client(tmp_path):
def client(tmp_path, sample_pool):
pool_path = tmp_path / "pool.json"
pool_path.write_text(json.dumps(sample_pool))
settings = Settings(
db_path=str(tmp_path / "quiz.db"),
secret_key="test-secret",
admin_password="admin-pass",
public_url="http://testserver",
pool_path=str(pool_path),
default_session_id=CANONICAL_SID,
)
app = create_app(settings)
with TestClient(app) as test_client:
yield test_client
@pytest.fixture
def sid() -> str:
return CANONICAL_SID
def admin_login(client: TestClient) -> None:
response = client.post("/admin/login", json={"password": "admin-pass"})
assert response.status_code == 200
def create_quiz(client: TestClient, pool: dict) -> int:
admin_login(client)
response = client.post("/admin/api/quizzes", json={"pool_json": pool})
assert response.status_code == 200, response.text
return response.json()["quiz_id"]
def create_session(client: TestClient, pool: dict) -> str:
quiz_id = create_quiz(client, pool)
response = client.post("/admin/api/sessions", json={"quiz_id": quiz_id})
assert response.status_code == 200, response.text
return response.json()["sid"]
def join_student(client: TestClient, sid: str, student_id: str = "s1", name: str = "Student One") -> dict:

72
tests/test_api_admin.py
View File

@@ -1,42 +1,56 @@
from conftest import admin_login, create_quiz, create_session, join_student
from conftest import admin_login, join_student
def test_admin_login_required_and_quiz_session_crud(client, sample_pool):
assert client.get("/admin/").status_code == 401
def test_admin_state_requires_login(client):
# /admin/api/state is the canonical "am I logged in" probe used by the SPA.
assert client.get("/admin/api/state").status_code == 401
assert client.post("/admin/login", json={"password": "wrong"}).status_code == 401
admin_login(client)
assert client.get("/admin/").status_code == 200
quiz_id = create_quiz(client, sample_pool)
quizzes = client.get("/admin/api/quizzes").json()["quizzes"]
assert any(item["id"] == quiz_id for item in quizzes)
response = client.post("/admin/api/sessions", json={"quiz_id": quiz_id})
def test_admin_state_after_login_includes_pool_meta_and_qr(client, sid):
admin_login(client)
response = client.get("/admin/api/state")
assert response.status_code == 200
payload = response.json()
assert len(payload["sid"]) == 6
assert payload["join_url"].endswith(f"?sid={payload['sid']}")
assert payload["sid"] == sid
assert payload["state"] == "lobby"
assert payload["join_url"].endswith(f"?sid={sid}")
assert payload["qr_url"].startswith("data:image/svg+xml;base64,")
sessions = client.get("/admin/api/sessions").json()["sessions"]
assert sessions[0]["sid"] == payload["sid"]
assert payload["pool_meta"]["question_count"] == 5
assert payload["pool_meta"]["score_fn"] == "linear_decay"
def test_quiz_upload_and_csv_export(client, sample_pool):
sid = create_session(client, sample_pool)
join_student(client, sid, "s1", "Student One")
csv_response = client.get(f"/admin/api/sessions/{sid}/csv")
assert csv_response.status_code == 200
assert "student_id,name,question_idx" in csv_response.text
upload = client.post(
"/admin/api/quizzes/upload",
files={"file": ("pool.json", __import__("json").dumps(sample_pool), "application/json")},
)
assert upload.status_code == 200
def test_admin_html_served_without_auth_gate(client):
# The HTML shell is unauthed; the SPA decides login vs dashboard from
# the /admin/api/state response. Anything else would force a separate
# /admin/login page back into the URL bar.
response = client.get("/admin/")
assert response.status_code == 200
assert "<title>Quiz Admin</title>" in response.text
def test_invalid_quiz_and_session_errors(client):
def test_csv_endpoint_is_admin_only_and_serves_results(client, sid):
assert client.get("/admin/api/csv").status_code == 401
admin_login(client)
assert client.post("/admin/api/quizzes", json={"pool_json": {"title": "bad", "questions": []}}).status_code == 400
assert client.post("/admin/api/sessions", json={"quiz_id": 999}).status_code == 404
join_student(client, sid)
response = client.get("/admin/api/csv")
assert response.status_code == 200
assert "student_id,name,question_idx" in response.text
def test_admin_logout_clears_cookie(client):
admin_login(client)
assert client.get("/admin/api/state").status_code == 200
client.post("/admin/logout")
assert client.get("/admin/api/state").status_code == 401
def test_admin_reset_clears_participants_and_state(client, sid):
admin_login(client)
join_student(client, sid, "s1", "First")
join_student(client, sid, "s2", "Second")
response = client.post("/admin/api/reset")
assert response.status_code == 200
state = client.get("/admin/api/state").json()
assert state["state"] == "lobby"
assert state["current_question_idx"] is None

16
tests/test_api_student.py
View File

@@ -1,8 +1,7 @@
from conftest import create_session, join_student
from conftest import join_student
def test_session_metadata_join_me_and_stats(client, sample_pool):
sid = create_session(client, sample_pool)
def test_session_metadata_join_me_and_stats(client, sid):
metadata = client.get(f"/api/session/{sid}").json()
assert metadata["title"] == "Sample Quiz"
assert metadata["state"] == "lobby"
@@ -22,8 +21,15 @@ def test_session_metadata_join_me_and_stats(client, sample_pool):
assert stats["top5"][0]["name"] == "Updated Name"
def test_root_without_sid_redirects_to_canonical(client, sid):
response = client.get("/", follow_redirects=False)
assert response.status_code == 302
assert response.headers["location"] == f"/?sid={sid}"
def test_invalid_session_and_missing_cookie_paths(client):
assert client.get("/?sid=BAD").status_code == 200
assert "Ask your instructor" in client.get("/?sid=BAD").text
response = client.get("/?sid=BAD")
assert response.status_code == 404
assert "Ask your instructor" in response.text
assert client.get("/api/session/BAD").status_code == 404
assert client.get("/api/session/BAD/me").status_code == 401

8
tests/test_csv_export.py
View File

@@ -1,8 +1,8 @@
from conftest import create_session, join_student
from conftest import admin_login, join_student
def test_csv_export_contains_one_row_per_submission(client, sample_pool):
sid = create_session(client, sample_pool)
def test_csv_export_contains_one_row_per_submission(client, sid):
admin_login(client)
join_student(client, sid, "s1", "Student One")
rooms = client.app.state.rooms
client.portal.call(rooms.open_question, sid, 0, 2)
@@ -10,7 +10,7 @@ def test_csv_export_contains_one_row_per_submission(client, sample_pool):
assert ack["type"] == "submit_ack"
client.portal.call(rooms.close_question, sid)
response = client.get(f"/admin/api/sessions/{sid}/csv")
response = client.get("/admin/api/csv")
lines = response.text.strip().splitlines()
assert lines[0] == "sid,student_id,name,question_idx,answer,elapsed_ms,score,status"
assert len(lines) == 2

8
tests/test_late_join.py
View File

@@ -1,10 +1,9 @@
from fastapi.testclient import TestClient
from conftest import create_session, join_student
from conftest import join_student
def test_late_join_during_open_gets_reduced_remaining_and_can_score(client, sample_pool):
sid = create_session(client, sample_pool)
def test_late_join_during_open_gets_reduced_remaining_and_can_score(client, sid):
rooms = client.app.state.rooms
client.portal.call(rooms.open_question, sid, 0, 2)
@@ -20,8 +19,7 @@ def test_late_join_during_open_gets_reduced_remaining_and_can_score(client, samp
assert ws.receive_json()["score"] > 0
def test_join_after_closed_gets_missed_row(client, sample_pool):
sid = create_session(client, sample_pool)
def test_join_after_closed_gets_missed_row(client, sid):
rooms = client.app.state.rooms
client.portal.call(rooms.open_question, sid, 0, 1)
client.portal.call(rooms.close_question, sid)

43
tests/test_load_simulation.py
View File

@@ -1,10 +1,9 @@
import time
from conftest import create_session, join_student
from conftest import admin_login, join_student
def test_load_simulation_50_students_full_quiz_and_autoclose(client, sample_pool):
sid = create_session(client, sample_pool)
def test_load_simulation_50_students_full_quiz(client, sid, sample_pool):
"""50 students answer 5 questions; instructor drives transitions via the
single 'advance_to_next' WS command."""
rooms = client.app.state.rooms
sockets = []
try:
@@ -14,33 +13,27 @@ def test_load_simulation_50_students_full_quiz_and_autoclose(client, sample_pool
sockets.append(ws)
assert ws.receive_json()["type"] == "state"
# Start: opens Q0 from lobby.
client.portal.call(rooms.advance_to_next, sid)
for ws in sockets:
assert ws.receive_json()["type"] == "question_open"
for question_idx in range(5):
client.portal.call(rooms.open_question, sid, question_idx, 1)
for ws in sockets:
assert ws.receive_json()["type"] == "question_open"
for idx, ws in enumerate(sockets):
answer = sample_pool["questions"][question_idx]["correct"] if idx % 3 else "A"
ws.send_json({"type": "submit", "question_idx": question_idx, "answer": answer})
assert ws.receive_json()["type"] == "submit_ack"
if question_idx == 4:
started = time.monotonic()
for ws in sockets:
message = ws.receive_json()
assert message["type"] == "question_closed"
assert time.monotonic() - started < 2
else:
client.portal.call(rooms.close_question, sid)
for ws in sockets:
assert ws.receive_json()["type"] == "question_closed"
client.portal.call(rooms.next_question, sid)
for ws in sockets:
assert ws.receive_json()["type"] == "between_questions"
client.portal.call(rooms.end_session, sid)
for ws in sockets:
assert ws.receive_json()["type"] == "session_ended"
client.portal.call(rooms.advance_to_next, sid)
for ws in sockets:
first = ws.receive_json()
assert first["type"] == "question_closed"
second = ws.receive_json()
expected_next = "question_open" if question_idx < 4 else "session_ended"
assert second["type"] == expected_next
csv_lines = client.get(f"/admin/api/sessions/{sid}/csv").text.strip().splitlines()
admin_login(client)
csv_lines = client.get("/admin/api/csv").text.strip().splitlines()
assert len(csv_lines) == 1 + 50 * 5
stats = client.get(f"/api/session/{sid}/stats?question_idx=4").json()
assert stats["top5"]

5
tests/test_reconnect.py
View File

@@ -1,8 +1,7 @@
from conftest import create_session, join_student
from conftest import join_student
def test_reconnect_replays_existing_submit_ack(client, sample_pool):
sid = create_session(client, sample_pool)
def test_reconnect_replays_existing_submit_ack(client, sid):
join_student(client, sid, "s1", "Student One")
rooms = client.app.state.rooms
client.portal.call(rooms.open_question, sid, 0, 2)

86
tests/test_state_machine.py
View File

@@ -1,45 +1,81 @@
from conftest import create_session, join_student
from conftest import join_student
def test_full_lifecycle_with_three_students(client, sample_pool):
sid = create_session(client, sample_pool)
def test_full_lifecycle_via_advance_and_close(client, sid):
"""End-to-end: 3 students, instructor drives via advance_to_next which
closes the open question and opens the next in a single step."""
rooms = client.app.state.rooms
sockets = []
for idx in range(3):
join_student(client, sid, f"s{idx}", f"Student {idx}")
ws = client.websocket_connect(f"/ws/student/{sid}").__enter__()
sockets.append(ws)
assert ws.receive_json()["state"] == "lobby"
try:
client.portal.call(rooms.open_question, sid, 0, 2)
for idx in range(3):
join_student(client, sid, f"s{idx}", f"Student {idx}")
ws = client.websocket_connect(f"/ws/student/{sid}").__enter__()
sockets.append(ws)
assert ws.receive_json()["state"] == "lobby"
# Start: opens Q0.
client.portal.call(rooms.advance_to_next, sid)
for ws in sockets:
assert ws.receive_json()["type"] == "question_open"
for idx, ws in enumerate(sockets):
ws.send_json({"type": "submit", "question_idx": 0, "answer": "B" if idx < 2 else "A"})
assert ws.receive_json()["type"] == "submit_ack"
client.portal.call(rooms.close_question, sid)
# Advance: closes Q0 and opens Q1 in one step.
client.portal.call(rooms.advance_to_next, sid)
for ws in sockets:
assert ws.receive_json()["type"] == "question_closed"
session = client.portal.call(rooms.get_session, sid)
assert session["state"] == "question_closed"
client.portal.call(rooms.next_question, sid)
for ws in sockets:
assert ws.receive_json()["type"] == "between_questions"
assert client.portal.call(rooms.get_session, sid)["state"] == "between_questions"
client.portal.call(rooms.open_question, sid, 1, 2)
for ws in sockets:
assert ws.receive_json()["type"] == "question_open"
assert client.portal.call(rooms.get_session, sid)["state"] == "question_open"
assert client.portal.call(rooms.get_session, sid)["current_question_idx"] == 1
# End the session early.
client.portal.call(rooms.end_session, sid)
for ws in sockets:
message = ws.receive_json()
assert message["type"] in {"question_closed", "session_ended"}
if message["type"] == "question_closed":
first = ws.receive_json()
# end_session closes the open question, then sends session_ended.
if first["type"] == "question_closed":
assert ws.receive_json()["type"] == "session_ended"
else:
assert first["type"] == "session_ended"
assert client.portal.call(rooms.get_session, sid)["state"] == "finished"
finally:
for ws in locals().get("sockets", []):
for ws in sockets:
ws.__exit__(None, None, None)
def test_explicit_close_then_advance_skips_redundant_close(client, sid):
"""If the instructor closes manually first, the next advance just opens
the following question (no double-close broadcast)."""
rooms = client.app.state.rooms
join_student(client, sid, "s1", "Solo")
with client.websocket_connect(f"/ws/student/{sid}") as ws:
assert ws.receive_json()["state"] == "lobby"
client.portal.call(rooms.open_question, sid, 0, 2)
assert ws.receive_json()["type"] == "question_open"
client.portal.call(rooms.close_question, sid)
assert ws.receive_json()["type"] == "question_closed"
client.portal.call(rooms.advance_to_next, sid)
assert ws.receive_json()["type"] == "question_open"
assert client.portal.call(rooms.get_session, sid)["current_question_idx"] == 1
def test_reset_clears_participants_and_returns_to_lobby(client, sid):
rooms = client.app.state.rooms
join_student(client, sid, "s1", "First")
join_student(client, sid, "s2", "Second")
client.portal.call(rooms.open_question, sid, 0, 2)
client.portal.call(rooms.submit_answer, sid, "s1", 0, "B")
client.portal.call(rooms.close_question, sid)
client.portal.call(rooms.reset, sid)
session = client.portal.call(rooms.get_session, sid)
assert session["state"] == "lobby"
assert session["current_question_idx"] is None
# Participants and submissions are wiped.
board = client.portal.call(rooms.leaderboard, sid)
assert board == []

65
tests/test_ws_admin.py
View File

@@ -1,20 +1,42 @@
import pytest
from starlette.websockets import WebSocketDisconnect
from conftest import admin_login, create_session, join_student
from conftest import admin_login, join_student
def test_instructor_ws_requires_admin_cookie(client, sample_pool):
sid = create_session(client, sample_pool)
client.cookies.clear()
def test_instructor_ws_requires_admin_cookie(client, sid):
with pytest.raises(WebSocketDisconnect) as exc:
with client.websocket_connect(f"/ws/instructor/{sid}"):
pass
assert exc.value.code == 4001
def test_instructor_controls_transition_and_broadcast(client, sample_pool):
sid = create_session(client, sample_pool)
def test_instructor_next_command_drives_full_loop(client, sid):
"""The 'next' WS message drives the entire lifecycle:
lobby → opens Q0 → closes Q0 + opens Q1 → ... → closes last + ends."""
join_student(client, sid, "s1", "Student One")
admin_login(client)
with client.websocket_connect(f"/ws/student/{sid}") as student_ws:
assert student_ws.receive_json()["type"] == "state"
with client.websocket_connect(f"/ws/instructor/{sid}") as admin_ws:
# Drain lobby snapshot.
assert admin_ws.receive_json()["type"] == "state"
assert admin_ws.receive_json()["type"] == "lobby_update"
# First "next" opens Q0 from lobby.
admin_ws.send_json({"type": "next"})
assert student_ws.receive_json()["type"] == "question_open"
admin_open = admin_ws.receive_json()
assert admin_open["type"] == "question_open"
assert admin_ws.receive_json()["type"] == "live_histogram"
# Second "next" closes Q0 and opens Q1.
admin_ws.send_json({"type": "next"})
student_msgs = [student_ws.receive_json() for _ in range(2)]
assert {m["type"] for m in student_msgs} == {"question_closed", "question_open"}
def test_instructor_close_then_next_emits_clean_open(client, sid):
join_student(client, sid, "s1", "Student One")
admin_login(client)
with client.websocket_connect(f"/ws/student/{sid}") as student_ws:
@@ -23,16 +45,33 @@ def test_instructor_controls_transition_and_broadcast(client, sample_pool):
assert admin_ws.receive_json()["type"] == "state"
assert admin_ws.receive_json()["type"] == "lobby_update"
admin_ws.send_json({"type": "open_question", "question_idx": 0, "time_limit": 2})
student_open = student_ws.receive_json()
assert student_open["type"] == "question_open"
admin_open = admin_ws.receive_json()
assert admin_open["type"] == "question_open"
assert student_ws.receive_json()["type"] == "question_open"
assert admin_ws.receive_json()["type"] == "question_open"
assert admin_ws.receive_json()["type"] == "live_histogram"
admin_ws.send_json({"type": "close_question"})
assert student_ws.receive_json()["type"] == "question_closed"
messages = [admin_ws.receive_json(), admin_ws.receive_json()]
assert {msg["type"] for msg in messages} == {"question_closed", "full_leaderboard"}
admin_msgs = [admin_ws.receive_json(), admin_ws.receive_json()]
assert {m["type"] for m in admin_msgs} == {"question_closed", "full_leaderboard"}
admin_ws.send_json({"type": "next"})
assert student_ws.receive_json()["type"] == "between_questions"
assert student_ws.receive_json()["type"] == "question_open"
def test_reset_command_returns_session_to_lobby(client, sid):
join_student(client, sid, "s1", "Student One")
admin_login(client)
with client.websocket_connect(f"/ws/instructor/{sid}") as admin_ws:
assert admin_ws.receive_json()["type"] == "state"
assert admin_ws.receive_json()["type"] == "lobby_update"
admin_ws.send_json({"type": "open_question", "question_idx": 0, "time_limit": 2})
assert admin_ws.receive_json()["type"] == "question_open"
assert admin_ws.receive_json()["type"] == "live_histogram"
admin_ws.send_json({"type": "reset"})
# After reset, the instructor receives a state=lobby snapshot + lobby_update.
msgs = []
while len(msgs) < 2:
msgs.append(admin_ws.receive_json())
types = [m["type"] for m in msgs]
assert "state" in types

8
tests/test_ws_student.py
View File

@@ -1,19 +1,17 @@
import pytest
from starlette.websockets import WebSocketDisconnect
from conftest import create_session, join_student
from conftest import join_student
def test_student_ws_requires_cookie(client, sample_pool):
sid = create_session(client, sample_pool)
def test_student_ws_requires_cookie(client, sid):
with pytest.raises(WebSocketDisconnect) as exc:
with client.websocket_connect(f"/ws/student/{sid}"):
pass
assert exc.value.code == 4001
def test_student_ws_initial_state_submit_and_closed_reject(client, sample_pool):
sid = create_session(client, sample_pool)
def test_student_ws_initial_state_submit_and_closed_reject(client, sid):
join_student(client, sid, "s1", "Student One")
with client.websocket_connect(f"/ws/student/{sid}") as ws:
state = ws.receive_json()