feat: add category-based custom tests and permission-safe sharing

This commit is contained in:
Daniel 2026-09-07 01:17:42 +02:00
parent 2cdfe71707
commit affd7177b5
19 changed files with 1003 additions and 212 deletions

View file

@ -0,0 +1,21 @@
"""Optional category hierarchy; existing categories remain roots."""
from alembic import op
import sqlalchemy as sa
revision = "c82d19e4a601"
down_revision = "5f8c1c2a9d40"
branch_labels = None
depends_on = None
def upgrade():
# create_all may already have created this column on a fresh install.
if "parent_id" not in {c["name"] for c in sa.inspect(op.get_bind()).get_columns("question_categories")}:
op.add_column("question_categories", sa.Column("parent_id", sa.Integer(), nullable=True))
op.create_foreign_key("fk_question_categories_parent", "question_categories", "question_categories",
["parent_id"], ["id"], ondelete="RESTRICT")
def downgrade():
op.drop_constraint("fk_question_categories_parent", "question_categories", type_="foreignkey")
op.drop_column("question_categories", "parent_id")

View file

@ -8,6 +8,7 @@ class QuestionCategory(Base):
__tablename__ = "question_categories"
id = Column(Integer, primary_key=True, index=True)
parent_id = Column(Integer, ForeignKey("question_categories.id", ondelete="RESTRICT", name="fk_question_categories_parent"), nullable=True)
name = Column(String, nullable=False)
description = Column(Text, nullable=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)

View file

@ -22,16 +22,13 @@ from app.schemas.attempt import (
DashboardStats,
QuizStats,
)
from app.utils.quiz_access import can_access_quiz, general_quiz_visibility, require_quiz_access
from app.utils.auth import get_current_user
from app.utils.quiz_questions import get_quiz_questions
router = APIRouter()
def can_access_quiz(quiz: Quiz, user: User) -> bool:
return bool(user.is_moderator or quiz.user_id == user.id or quiz.is_published == 1)
@router.post("/start", response_model=AttemptResponse)
def start_attempt(
quiz_id: int,
@ -42,7 +39,7 @@ def start_attempt(
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
if not quiz:
raise HTTPException(status_code=404, detail="Quiz not found")
if not can_access_quiz(quiz, current_user):
if not can_access_quiz(db, quiz, current_user):
raise HTTPException(status_code=403, detail="This quiz is private")
# Enforce max_attempts
@ -118,6 +115,9 @@ def submit_attempt(
if attempt.completed_at:
raise HTTPException(status_code=400, detail="Attempt already submitted")
quiz = db.query(Quiz).filter(Quiz.id == attempt.quiz_id).first()
require_quiz_access(db, quiz, current_user)
# Get all questions for this quiz via junction table
questions = {q.id: q for q in get_quiz_questions(db, attempt.quiz_id)}
@ -244,6 +244,7 @@ class ProgressSave(BaseModel):
def save_progress(
data: ProgressSave,
request: Request = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Save in-progress quiz answers to Redis (survives logout/browser change).
@ -251,6 +252,13 @@ def save_progress(
Also records the latest active browser session for diagnostics. Resuming
from another browser is allowed; the newest browser takes over the attempt.
"""
quiz = db.query(Quiz).filter(Quiz.id == data.quiz_id).first()
require_quiz_access(db, quiz, current_user)
attempt = db.query(QuizAttempt).filter(QuizAttempt.id == data.attempt_id,
QuizAttempt.quiz_id == data.quiz_id, QuizAttempt.user_id == current_user.id,
QuizAttempt.completed_at.is_(None)).first()
if not attempt:
raise HTTPException(404, "Active attempt not found")
try:
import redis as redis_lib, json as _json
from app.config import settings
@ -292,6 +300,8 @@ def get_progress(
Allows another browser/device to resume the attempt; the newest browser
takes over the soft activity marker instead of blocking with a 409.
Auto-submits timed quizzes if timer has expired."""
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
require_quiz_access(db, quiz, current_user)
try:
import redis as redis_lib, json as _json
from app.config import settings
@ -425,6 +435,8 @@ def get_in_progress_attempt(
current_user: User = Depends(get_current_user),
):
"""Return the latest incomplete attempt for a quiz, or null."""
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
require_quiz_access(db, quiz, current_user)
attempt = db.query(QuizAttempt).filter(
QuizAttempt.quiz_id == quiz_id,
QuizAttempt.user_id == current_user.id,
@ -455,7 +467,7 @@ def get_in_progress_attempts(
.filter(
QuizAttempt.user_id == current_user.id,
QuizAttempt.completed_at.is_(None),
Quiz.course_id.is_(None), # exclude course quizzes
general_quiz_visibility(current_user), # exclude unavailable and course quizzes
)
.order_by(QuizAttempt.started_at.desc())
.all()
@ -594,6 +606,7 @@ def get_attempt(
raise HTTPException(status_code=404, detail="Attempt not found")
quiz = db.query(Quiz).filter(Quiz.id == attempt.quiz_id).first()
require_quiz_access(db, quiz, current_user)
is_course_quiz = quiz and quiz.course_id is not None
review_allowed = not is_course_quiz or (quiz.allow_review == 1)

View file

@ -2,7 +2,6 @@ from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, EmailStr
from sqlalchemy import or_
from sqlalchemy.orm import Session
from app.config import settings
@ -12,6 +11,7 @@ from app.models.email_verification import EmailVerification
from app.models.quiz import Quiz
from app.models.user import User
from app.schemas.auth import Token
from app.utils.quiz_access import general_quiz_visibility
from app.utils.auth import create_access_token, get_current_user, verify_password
from app.utils.quiz_questions import get_quiz_questions
@ -99,8 +99,7 @@ def _mobile_login_rate_limit(client_ip: str):
def _visible_quizzes_query(db: Session, current_user: User):
query = db.query(Quiz).filter(Quiz.deleted_at.is_(None), Quiz.course_id.is_(None))
if not current_user.is_moderator:
query = query.filter(or_(Quiz.is_published == 1, Quiz.user_id == current_user.id))
query = query.filter(general_quiz_visibility(current_user))
return query

View file

@ -1,152 +1,111 @@
"""Question category management — organise bank questions by topic/subject."""
"""Question category management; saved quiz membership is never changed here."""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.question import Question
from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.models.user import User
from app.services.quiz_builder import (bank_query, category_descendants, category_breadcrumbs,
validate_parent, GenerateTestRequest, generate_test)
from app.utils.auth import get_current_user, require_moderator
router = APIRouter()
class QCatCreate(BaseModel):
name: str
name: str = Field(min_length=1, max_length=200)
description: str | None = None
parent_id: int | None = None
class QCatResponse(BaseModel):
id: int
name: str
description: str | None
parent_id: int | None = None
breadcrumbs: list[dict] = Field(default_factory=list)
question_count: int = 0
model_config = {"from_attributes": True}
def category_response(db, user, cat, cats):
ids = category_descendants(cats, [cat.id])
return QCatResponse(id=cat.id, name=cat.name, description=cat.description, parent_id=cat.parent_id,
breadcrumbs=category_breadcrumbs(cats, cat.id),
question_count=bank_query(db, user).filter(Question.question_category_id.in_(ids)).count())
@router.get("/", response_model=list[QCatResponse])
def list_question_categories(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
def list_question_categories(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
cats = db.query(QuestionCategory).order_by(QuestionCategory.name).all()
return [
QCatResponse(
id=c.id, name=c.name, description=c.description,
question_count=db.query(Question).filter(Question.question_category_id == c.id).count(),
)
for c in cats
]
# ponytail: one count per category; aggregate counts if the taxonomy becomes large.
return [category_response(db, current_user, cat, cats) for cat in cats]
def validate_category(db, data, cat_id=None):
# Serialize hierarchy writes so concurrent moves cannot create a cycle.
cats = db.query(QuestionCategory).order_by(QuestionCategory.id).with_for_update().all()
if not data.name.strip():
raise HTTPException(400, "Name cannot be empty")
if any(c.name == data.name.strip() and c.id != cat_id for c in cats):
raise HTTPException(400, "Category already exists")
validate_parent(cats, cat_id, data.parent_id)
return cats
@router.post("/", response_model=QCatResponse)
def create_question_category(
data: QCatCreate,
db: Session = Depends(get_db),
current_user: User = Depends(require_moderator),
):
if not data.name.strip():
raise HTTPException(status_code=400, detail="Name cannot be empty")
existing = db.query(QuestionCategory).filter(QuestionCategory.name == data.name.strip()).first()
if existing:
raise HTTPException(status_code=400, detail="Category already exists")
cat = QuestionCategory(name=data.name.strip(), description=data.description, user_id=current_user.id)
def create_question_category(data: QCatCreate, db: Session = Depends(get_db),
current_user: User = Depends(require_moderator)):
cats = validate_category(db, data)
cat = QuestionCategory(name=data.name.strip(), description=data.description,
parent_id=data.parent_id, user_id=current_user.id)
db.add(cat)
db.commit()
db.refresh(cat)
return QCatResponse(id=cat.id, name=cat.name, description=cat.description, question_count=0)
return category_response(db, current_user, cat, cats + [cat])
@router.patch("/{cat_id}", response_model=QCatResponse)
def update_question_category(
cat_id: int,
data: QCatCreate,
db: Session = Depends(get_db),
current_user: User = Depends(require_moderator),
):
cat = db.query(QuestionCategory).filter(QuestionCategory.id == cat_id).first()
def update_question_category(cat_id: int, data: QCatCreate, db: Session = Depends(get_db),
current_user: User = Depends(require_moderator)):
cats = validate_category(db, data, cat_id)
cat = next((c for c in cats if c.id == cat_id), None)
if not cat:
raise HTTPException(status_code=404, detail="Category not found")
raise HTTPException(404, "Category not found")
cat.name = data.name.strip()
cat.description = data.description
if "parent_id" in data.model_fields_set:
cat.parent_id = data.parent_id
db.commit()
count = db.query(Question).filter(Question.question_category_id == cat_id).count()
return QCatResponse(id=cat.id, name=cat.name, description=cat.description, question_count=count)
return category_response(db, current_user, cat, cats)
@router.delete("/{cat_id}", status_code=204)
def delete_question_category(
cat_id: int,
move_to: int | None = None, # optional: move questions to this category instead of uncategorizing
db: Session = Depends(get_db),
current_user: User = Depends(require_moderator),
):
cat = db.query(QuestionCategory).filter(QuestionCategory.id == cat_id).first()
def delete_question_category(cat_id: int, move_to: int | None = None, db: Session = Depends(get_db),
current_user: User = Depends(require_moderator)):
cats = db.query(QuestionCategory).order_by(QuestionCategory.id).with_for_update().all()
cat = next((c for c in cats if c.id == cat_id), None)
if not cat:
raise HTTPException(status_code=404, detail="Category not found")
if move_to is not None:
target = db.query(QuestionCategory).filter(QuestionCategory.id == move_to).first()
if not target:
raise HTTPException(status_code=404, detail="Target category not found")
db.query(Question).filter(Question.question_category_id == cat_id).update(
{"question_category_id": move_to}
)
else:
db.query(Question).filter(Question.question_category_id == cat_id).update(
{"question_category_id": None}
)
raise HTTPException(404, "Category not found")
if any(c.parent_id == cat_id for c in cats):
raise HTTPException(400, "Move child categories before deleting this category")
validate_parent(cats, cat_id, move_to)
db.query(Question).filter(Question.question_category_id == cat_id).update({"question_category_id": move_to})
db.delete(cat)
db.commit()
@router.post("/{cat_id}/create-quiz")
def create_quiz_from_question_category(
cat_id: int,
title: str,
mode: str = "timed",
time_limit_minutes: int | None = None,
db: Session = Depends(get_db),
current_user: User = Depends(require_moderator),
):
"""Create a new quiz from all questions in a question category."""
cat = db.query(QuestionCategory).filter(QuestionCategory.id == cat_id).first()
if not cat:
raise HTTPException(status_code=404, detail="Category not found")
if not title.strip():
raise HTTPException(status_code=400, detail="Title is required")
if mode not in ("timed", "learning"):
raise HTTPException(status_code=400, detail="Mode must be timed or learning")
source_questions = db.query(Question).filter(
Question.question_category_id == cat_id
).order_by(Question.id).all()
if not source_questions:
raise HTTPException(status_code=400, detail="This category has no questions")
first_quiz = db.query(Quiz).filter(Quiz.id == source_questions[0].source_quiz_id).first() if source_questions[0].source_quiz_id else None
if not first_quiz:
raise HTTPException(status_code=400, detail="Cannot determine source section — questions must have an origin quiz")
from app.models.quiz_question_link import QuizQuestionLink
new_quiz = Quiz(
section_id=first_quiz.section_id,
user_id=current_user.id,
title=title.strip(),
questions_count=len(source_questions),
mode=mode,
time_limit_minutes=time_limit_minutes,
)
db.add(new_quiz)
db.flush()
# Reference existing questions via junction — no copies, edits propagate
for pos, sq in enumerate(source_questions):
db.add(QuizQuestionLink(quiz_id=new_quiz.id, question_id=sq.id, position=pos))
db.commit()
db.refresh(new_quiz)
return {"id": new_quiz.id, "title": new_quiz.title, "questions_count": new_quiz.questions_count}
def create_quiz_from_question_category(cat_id: int, title: str, mode: str = "timed",
time_limit_minutes: int | None = None, count: int | None = None,
db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
ids = category_descendants(db.query(QuestionCategory).all(), [cat_id])
available = bank_query(db, current_user).filter(Question.question_category_id.in_(ids)).count()
if mode not in ("timed", "learning") or not title.strip() or len(title) > 200 or (time_limit_minutes is not None and time_limit_minutes <= 0):
raise HTTPException(400, "Provide a title, valid mode, and positive time limit")
count = available if count is None else count
if not 1 <= count <= 200:
raise HTTPException(400, "Select between 1 and 200 available questions using Create Custom Test")
return generate_test(db, current_user, GenerateTestRequest(title=title, mode=mode,
time_limit_minutes=time_limit_minutes, count=count, category_ids=[cat_id]))

View file

@ -5,6 +5,7 @@ import logging
import os
import re
import uuid
from typing import Literal
logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form
@ -19,11 +20,23 @@ from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.models.user import User
from app.models.favorite import Favorite
from app.services.quiz_builder import (bank_query, category_descendants, filtered_bank_query,
CreateFromBankRequest, GenerateTestRequest, create_saved_test, generate_test)
from app.utils.auth import get_current_user, require_moderator
router = APIRouter()
def parse_category_ids(value):
try:
ids = [int(part.strip()) for part in value.split(",")]
except ValueError:
raise HTTPException(400, "Category IDs must be comma-separated integers")
if any(cid <= 0 for cid in ids):
raise HTTPException(400, "Category IDs must be positive")
return ids
@router.delete("/{question_id}", status_code=204)
def delete_question(
question_id: int,
@ -137,15 +150,15 @@ def get_bank_ids(
current_user: User = Depends(get_current_user),
):
"""Return just IDs for all matching questions (for server-side select-all)."""
query = db.query(Question.id)
query = bank_query(db, current_user).with_entities(Question.id)
if quiz_id:
query = query.filter(Question.source_quiz_id == quiz_id)
if category_ids:
cat_id_list = [int(c.strip()) for c in category_ids.split(",") if c.strip().isdigit()]
cat_id_list = parse_category_ids(category_ids)
if cat_id_list:
query = query.filter(Question.question_category_id.in_(cat_id_list))
query = query.filter(Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), cat_id_list)))
elif category_id is not None:
query = query.filter(Question.question_category_id == category_id)
query = query.filter(Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), [category_id])))
if uncategorized:
query = query.filter(Question.question_category_id.is_(None))
if favorites_only:
@ -196,26 +209,19 @@ def get_question_bank(
current_user: User = Depends(get_current_user),
):
"""List all questions across all quizzes. Supports keyword filter and quiz filter."""
query = bank_query(db, current_user)
if my_questions:
query = db.query(Question).filter(Question.user_id == current_user.id)
else:
query = db.query(Question).filter(
or_(
Question.is_shared == 1,
Question.is_shared.is_(None), # legacy questions without is_shared
Question.user_id == current_user.id, # always show own questions
)
)
query = query.filter(Question.user_id == current_user.id)
if quiz_id:
query = query.filter(Question.source_quiz_id == quiz_id)
if category_ids:
cat_id_list = [int(c.strip()) for c in category_ids.split(",") if c.strip().isdigit()]
cat_id_list = parse_category_ids(category_ids)
if cat_id_list:
query = query.filter(Question.question_category_id.in_(cat_id_list))
query = query.filter(Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), cat_id_list)))
elif category_id is not None:
query = query.filter(Question.question_category_id == category_id)
query = query.filter(Question.question_category_id.in_(category_descendants(db.query(QuestionCategory).all(), [category_id])))
if uncategorized:
query = query.filter(Question.question_category_id.is_(None))
@ -443,62 +449,27 @@ def list_question_images(
return [{"image_path": path, "url": f"/uploads/{path}"} for path in paths]
class CreateFromBankRequest(BaseModel):
title: str
question_ids: list[int]
mode: str = "timed"
time_limit_minutes: int | None = None
@router.post("/from-bank")
def create_quiz_from_bank(
data: CreateFromBankRequest,
@router.get("/builder/count")
def count_builder_questions(
category_ids: list[int] = Query(default=[]),
state: Literal["all", "unused", "incorrect", "bookmarked"] = "all",
is_shared: bool = False,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Create a new quiz referencing existing bank questions (no copying — edits propagate)."""
if not data.title.strip():
raise HTTPException(status_code=400, detail="Title is required")
if not data.question_ids:
raise HTTPException(status_code=400, detail="Select at least one question")
if data.mode not in ("timed", "learning"):
raise HTTPException(status_code=400, detail="Mode must be timed or learning")
return {"count": filtered_bank_query(db, current_user, category_ids, state, is_shared).count()}
source_questions = db.query(Question).filter(Question.id.in_(data.question_ids)).all()
if not source_questions:
raise HTTPException(status_code=404, detail="No valid questions found")
# Preserve caller's requested order
id_order = {qid: i for i, qid in enumerate(data.question_ids)}
source_questions.sort(key=lambda q: id_order.get(q.id, len(data.question_ids)))
# Find a section_id from the source questions' origin quiz
first_src_quiz_id = source_questions[0].source_quiz_id
first_quiz = db.query(Quiz).filter(Quiz.id == first_src_quiz_id).first() if first_src_quiz_id else None
section_id = first_quiz.section_id if first_quiz else None
@router.post("/builder")
def create_builder_quiz(data: GenerateTestRequest, db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)):
return generate_test(db, current_user, data)
is_mod = current_user.role in ("admin", "moderator")
new_quiz = Quiz(
section_id=section_id,
user_id=current_user.id,
title=data.title.strip(),
questions_count=len(source_questions),
mode=data.mode,
time_limit_minutes=data.time_limit_minutes,
is_published=1 if is_mod else 0,
is_shared=1 if is_mod else 0,
)
db.add(new_quiz)
db.flush()
# Reference existing questions via junction (no copies)
from app.models.quiz_question_link import QuizQuestionLink
for pos, sq in enumerate(source_questions):
db.add(QuizQuestionLink(quiz_id=new_quiz.id, question_id=sq.id, position=pos))
db.commit()
db.refresh(new_quiz)
return {"id": new_quiz.id, "title": new_quiz.title, "questions_count": new_quiz.questions_count}
@router.post("/from-bank")
def create_quiz_from_bank(data: CreateFromBankRequest, db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)):
return create_saved_test(db, current_user, data, data.question_ids)
class BulkCategoryRequest(BaseModel):
@ -690,11 +661,11 @@ def export_qti(
"""Export questions as QTI 2.1 XML."""
if question_ids:
ids = [int(x.strip()) for x in question_ids.split(",") if x.strip().isdigit()]
questions = db.query(Question).filter(Question.id.in_(ids)).all()
questions = bank_query(db, current_user).filter(Question.id.in_(ids)).all()
if len(questions) != len(set(ids)):
raise HTTPException(400, "Some questions are missing, private, or unavailable")
else:
questions = db.query(Question).filter(
or_(Question.is_shared == 1, Question.is_shared.is_(None), Question.user_id == current_user.id)
).limit(500).all()
questions = bank_query(db, current_user).limit(500).all()
items_xml = []
for q in questions:

View file

@ -12,16 +12,13 @@ from app.models.attempt import QuizAttempt
from app.models.user import User
from app.schemas.quiz import QuizCreate, QuizUpdate, QuizResponse, QuizDetail, QuizLearningDetail, QuizReview
from app.services import quiz_service
from app.utils.quiz_access import can_access_quiz, general_quiz_visibility, require_quiz_access, set_quiz_shared
from app.utils.auth import get_current_user, require_moderator
from app.utils.quiz_questions import get_quiz_questions, question_in_quiz, remove_question_from_quiz
router = APIRouter()
def can_access_quiz(quiz: Quiz, user: User) -> bool:
return bool(user.is_moderator or quiz.user_id == user.id or quiz.is_published == 1)
@router.post("/")
def create_quiz(
quiz_data: QuizCreate,
@ -166,7 +163,7 @@ def search_quizzes(
def _ensure_quiz(quiz_id: int, match_type: str):
if quiz_id not in results:
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
if not quiz or quiz.course_id is not None or not can_access_quiz(quiz, current_user):
if not quiz or quiz.course_id is not None or not can_access_quiz(db, quiz, current_user):
return False
results[quiz_id] = {
"quiz_id": quiz.id,
@ -184,8 +181,7 @@ def search_quizzes(
# ── Title search ─────────────────────────────────────────────
if mode in ("title", "all"):
title_query = db.query(Quiz).filter(Quiz.title.ilike(f"%{phrase}%"), Quiz.course_id.is_(None))
if not current_user.is_moderator:
title_query = title_query.filter(or_(Quiz.is_published == 1, Quiz.user_id == current_user.id))
title_query = title_query.filter(general_quiz_visibility(current_user))
for quiz in title_query.limit(30).all():
_ensure_quiz(quiz.id, "title")
@ -276,10 +272,9 @@ def list_quizzes(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""List quizzes. Moderators see all; regular users only see published."""
"""List accessible general quizzes (owned, published, or shared)."""
q = db.query(Quiz).filter(Quiz.deleted_at.is_(None), Quiz.course_id.is_(None))
if not current_user.is_moderator:
q = q.filter(or_(Quiz.is_published == 1, Quiz.user_id == current_user.id))
q = q.filter(general_quiz_visibility(current_user))
return q.order_by(Quiz.created_at.desc()).all()
@ -296,10 +291,12 @@ def get_quiz(
quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first()
if not quiz:
raise HTTPException(status_code=404, detail="Quiz not found")
if not can_access_quiz(quiz, current_user):
if not can_access_quiz(db, quiz, current_user):
raise HTTPException(status_code=403, detail="This quiz is private")
if study or quiz.mode == "learning":
if quiz.mode != "learning":
require_quiz_access(db, quiz, current_user, review=True)
result = QuizLearningDetail.model_validate(quiz)
else:
result = QuizDetail.model_validate(quiz)
@ -355,7 +352,7 @@ def shuffle_quiz(
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
if not quiz:
raise HTTPException(status_code=404, detail="Quiz not found")
if not can_access_quiz(quiz, current_user):
if not can_access_quiz(db, quiz, current_user):
raise HTTPException(status_code=403, detail="This quiz is private")
questions = get_quiz_questions(db, quiz_id)
@ -399,6 +396,8 @@ def review_quiz(
if not quiz:
raise HTTPException(status_code=404, detail="Quiz not found")
require_quiz_access(db, quiz, current_user, review=True)
has_attempt = db.query(QuizAttempt).filter(
QuizAttempt.quiz_id == quiz_id,
QuizAttempt.user_id == current_user.id,
@ -594,3 +593,12 @@ def permanently_delete_quiz(
db.delete(quiz)
db.commit()
@router.patch("/{quiz_id}/share")
def share_quiz(quiz_id: int, shared: bool = Query(...), db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)):
quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first()
if not quiz:
raise HTTPException(404, "Quiz not found")
return set_quiz_shared(db, quiz, current_user, shared)

View file

@ -0,0 +1,151 @@
"""Permission-safe, saved general-bank tests and category selection."""
import random
from typing import Literal
from fastapi import HTTPException
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import func, or_, select
from sqlalchemy.orm import aliased
from app.models.attempt import AttemptAnswer, QuizAttempt
from app.models.favorite import Favorite
from app.models.question import Question
from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.utils.quiz_questions import add_questions_to_quiz
def category_descendants(categories, selected):
parents = {c.id: c.parent_id for c in categories}
result = set(selected)
if result - parents.keys():
raise HTTPException(400, "Category not found")
while True:
expanded = result | {cid for cid, parent in parents.items() if parent in result}
if expanded == result:
return result
result = expanded
def category_breadcrumbs(categories, category_id):
by_id = {c.id: c for c in categories}
path, seen = [], set()
while category_id in by_id and category_id not in seen:
seen.add(category_id)
cat = by_id[category_id]
path.append({"id": cat.id, "name": cat.name})
category_id = cat.parent_id
return list(reversed(path))
def validate_parent(categories, category_id, parent_id):
if parent_id is None:
return
if parent_id not in {c.id for c in categories}:
raise HTTPException(400, "Parent category not found")
if category_id is not None and parent_id in category_descendants(categories, [category_id]):
raise HTTPException(400, "A category cannot be its own parent or a descendant's child")
def general_question_predicate():
source = aliased(Quiz)
return ~select(source.id).where(source.id == Question.source_quiz_id, source.course_id.isnot(None)).exists()
def shareable_question_predicate():
return general_question_predicate() & or_(Question.is_shared == 1, Question.is_shared.is_(None))
def bank_question_predicate(user):
return general_question_predicate() & or_(
Question.is_shared == 1, Question.is_shared.is_(None), Question.user_id == user.id,
)
def bank_query(db, user):
return db.query(Question).filter(bank_question_predicate(user))
def filtered_bank_query(db, user, category_ids=(), state="all", shared=False):
query = bank_query(db, user)
if category_ids:
ids = category_descendants(db.query(QuestionCategory).all(), category_ids)
query = query.filter(Question.question_category_id.in_(ids))
if shared:
query = query.filter(shareable_question_predicate())
if state == "bookmarked":
query = query.filter(Question.id.in_(select(Favorite.question_id).where(Favorite.user_id == user.id)))
elif state in ("unused", "incorrect"):
# Latest completed, nonexpired general-bank answer; deterministic ties.
answers = db.query(
AttemptAnswer.question_id.label("question_id"), AttemptAnswer.is_correct.label("is_correct"),
func.row_number().over(partition_by=AttemptAnswer.question_id, order_by=(
QuizAttempt.completed_at.desc(), QuizAttempt.id.desc(), AttemptAnswer.id.desc(),
)).label("rank"),
).join(QuizAttempt, AttemptAnswer.attempt_id == QuizAttempt.id).join(Quiz, QuizAttempt.quiz_id == Quiz.id).filter(
QuizAttempt.user_id == user.id, QuizAttempt.completed_at.isnot(None),
or_(QuizAttempt.expired == 0, QuizAttempt.expired.is_(None)), Quiz.course_id.is_(None),
).subquery()
if state == "unused":
query = query.filter(~Question.id.in_(select(answers.c.question_id)))
else:
query = query.filter(Question.id.in_(select(answers.c.question_id).where(
answers.c.rank == 1, answers.c.is_correct.is_(False),
)))
elif state != "all":
raise HTTPException(400, "Invalid question state")
return query
class TestOptions(BaseModel):
title: str = Field(min_length=1, max_length=200)
mode: Literal["timed", "learning"] = "timed"
time_limit_minutes: int | None = Field(default=None, gt=0)
is_shared: bool = False
@field_validator("title")
@classmethod
def nonblank_title(cls, value):
if not value.strip():
raise ValueError("Title is required")
return value.strip()
class CreateFromBankRequest(TestOptions):
question_ids: list[int] = Field(min_length=1)
class GenerateTestRequest(TestOptions):
category_ids: list[int] = Field(default_factory=list)
state: Literal["all", "unused", "incorrect", "bookmarked"] = "all"
count: int = Field(ge=1, le=200)
expected_count: int | None = Field(default=None, ge=0)
def create_saved_test(db, user, data, question_ids):
ids = list(dict.fromkeys(question_ids))
if not 1 <= len(ids) <= 200:
raise HTTPException(400, "Select between 1 and 200 questions")
query = bank_query(db, user).filter(Question.id.in_(ids))
if data.is_shared:
query = query.filter(shareable_question_predicate())
if query.count() != len(ids):
raise HTTPException(400, "Some questions are missing, private, or unavailable for this test")
quiz = Quiz(user_id=user.id, title=data.title, mode=data.mode,
time_limit_minutes=data.time_limit_minutes if data.mode == "timed" else None,
questions_count=len(ids), is_published=0, is_shared=int(data.is_shared))
db.add(quiz)
db.flush()
add_questions_to_quiz(db, quiz.id, ids)
db.commit()
db.refresh(quiz)
return {"id": quiz.id, "title": quiz.title, "questions_count": quiz.questions_count}
def generate_test(db, user, data):
ids = [row[0] for row in filtered_bank_query(db, user, data.category_ids, data.state, data.is_shared).with_entities(Question.id).all()]
if data.expected_count is not None and data.expected_count != len(ids):
raise HTTPException(409, "Available count changed. Refresh the count and try again")
if len(ids) < data.count:
raise HTTPException(400, f"Only {len(ids)} questions available; requested {data.count}")
return create_saved_test(db, user, data, random.sample(ids, data.count))

View file

@ -0,0 +1,66 @@
"""One general-quiz visibility rule for web, attempts and mobile.
Course access is deliberately separate: publication/sharing never grants enrollment.
"""
from fastapi import HTTPException
from sqlalchemy import or_, select
from app.models.course import Course, CourseEnrollment
from app.models.question import Question
from app.models.quiz import Quiz
from app.models.quiz_question_link import QuizQuestionLink
from app.services.quiz_builder import shareable_question_predicate, bank_question_predicate
def quiz_shareable_predicate(user=None):
allowed = bank_question_predicate(user) if user is not None else shareable_question_predicate()
return ~select(QuizQuestionLink.quiz_id).join(Question, Question.id == QuizQuestionLink.question_id).where(
QuizQuestionLink.quiz_id == Quiz.id, ~allowed,
).exists()
def general_quiz_visibility(user):
privileged = (Quiz.user_id == user.id) & quiz_shareable_predicate(user)
if user.is_moderator:
privileged = True
return (Quiz.course_id.is_(None) & Quiz.deleted_at.is_(None) & or_(
privileged,
(or_(Quiz.is_published == 1, Quiz.is_shared == 1) & quiz_shareable_predicate()),
))
def can_access_quiz(db, quiz, user):
if not quiz or quiz.deleted_at is not None:
return False
if quiz.course_id is None:
return db.query(Quiz.id).filter(Quiz.id == quiz.id, general_quiz_visibility(user)).first() is not None
course = db.query(Course).filter(Course.id == quiz.course_id).first()
if not course:
return False
if user.is_moderator or course.user_id == user.id:
return True
return db.query(CourseEnrollment.id).filter(
CourseEnrollment.course_id == course.id, CourseEnrollment.user_id == user.id,
).first() is not None
def require_quiz_access(db, quiz, user, review=False):
if not can_access_quiz(db, quiz, user):
raise HTTPException(403, "This quiz is private or no longer available")
if review and quiz.course_id is not None and quiz.allow_review != 1 and not user.is_moderator and quiz.user_id != user.id:
raise HTTPException(403, "Review is not allowed for this course quiz")
def set_quiz_shared(db, quiz, user, shared):
if quiz.user_id != user.id and not user.is_moderator:
raise HTTPException(403, "Only the owner or a moderator can change sharing")
if quiz.course_id is not None:
raise HTTPException(400, "Course quizzes cannot be shared in the general bank")
if shared and not db.query(Quiz.id).filter(Quiz.id == quiz.id, quiz_shareable_predicate()).first():
raise HTTPException(400, "This test contains private or course-only questions")
quiz.is_shared = int(shared)
# Explicit revocation must also revoke legacy publication.
if not shared:
quiz.is_published = 0
db.commit()
return {"id": quiz.id, "is_shared": quiz.is_shared, "is_published": quiz.is_published}

View file

@ -0,0 +1,47 @@
"""Migration graph and PostgreSQL DDL checks without a database connection."""
import importlib.util
import io
from pathlib import Path
import unittest
from unittest.mock import Mock, patch
from alembic.config import Config
from alembic.migration import MigrationContext
from alembic.operations import Operations
from alembic.script import ScriptDirectory
class CategoryMigrationTests(unittest.TestCase):
def test_graph_upgrade_and_downgrade(self):
backend = Path(__file__).resolve().parents[1]
config = Config()
config.set_main_option("script_location", str(backend / "alembic"))
scripts = ScriptDirectory.from_config(config)
self.assertEqual(scripts.get_heads(), ["c82d19e4a601"])
self.assertEqual(scripts.get_revision("c82d19e4a601").down_revision, "5f8c1c2a9d40")
list(scripts.walk_revisions()) # raises on a broken chain
spec = importlib.util.spec_from_file_location("category_migration", backend / "alembic/versions/c82d19e4a601_category_hierarchy.py")
migration = importlib.util.module_from_spec(spec)
spec.loader.exec_module(migration)
output = io.StringIO()
ops = Operations(MigrationContext.configure(dialect_name="postgresql", opts={"as_sql": True, "output_buffer": output}))
inspector = Mock()
inspector.get_columns.return_value = [{"name": "id"}, {"name": "name"}]
with patch.object(migration, "op", ops), patch.object(migration.sa, "inspect", return_value=inspector):
migration.upgrade()
sql = output.getvalue()
self.assertIn("ADD COLUMN parent_id INTEGER", sql)
self.assertIn("REFERENCES question_categories (id) ON DELETE RESTRICT", sql)
self.assertNotIn("UPDATE", sql) # existing IDs/assignments remain intact
output.seek(0)
output.truncate()
inspector.get_columns.return_value.append({"name": "parent_id"})
migration.upgrade() # fresh create_all already includes parent_id
self.assertEqual(output.getvalue(), "")
migration.downgrade()
self.assertIn("DROP CONSTRAINT fk_question_categories_parent", output.getvalue())
self.assertIn("DROP COLUMN parent_id", output.getvalue())
if __name__ == "__main__":
unittest.main()

View file

@ -0,0 +1,251 @@
"""Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests.
No application startup, external services or AI calls; every test uses a disposable SQLite database.
"""
import os
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
import sys
import unittest
from datetime import datetime, timedelta
from types import ModuleType
from unittest.mock import patch
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.database import Base, get_db
from app.models.user import User
from app.models.question import Question
from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.models.quiz_question_link import QuizQuestionLink
from app.models.attempt import QuizAttempt, AttemptAnswer
from app.models.favorite import Favorite
from app.models.course import Course, CourseEnrollment
from app.services.quiz_builder import category_descendants
from app.utils.auth import get_current_user
from app.routers import questions, question_categories, attempts, mobile
# Extraction is outside this milestone. Stub only its unused service import;
# the real routes, ORM, predicates, creation and authorization run below.
with patch.dict(sys.modules, {"app.services.quiz_service": ModuleType("app.services.quiz_service")}):
from app.routers import quizzes
class BuilderTests(unittest.TestCase):
def setUp(self):
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
with self.engine.connect() as conn:
conn.exec_driver_sql("PRAGMA foreign_keys=ON")
Base.metadata.create_all(self.engine)
self.db = Session(self.engine)
self.owner = User(id=1, name="Owner", email="owner@example.test", hashed_password="unused")
self.peer = User(id=2, name="Peer", email="peer@example.test", hashed_password="unused")
self.mod = User(id=3, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")
self.db.add_all([self.owner, self.peer, self.mod])
self.db.flush()
self.course = Course(id=1, title="Course", user_id=3, status="published")
self.db.add(self.course)
self.db.add_all([QuestionCategory(id=1, name="Root", user_id=3),
QuestionCategory(id=2, name="Child", parent_id=1, user_id=3),
QuestionCategory(id=3, name="Leaf", parent_id=2, user_id=3),
QuestionCategory(id=4, name="Empty", user_id=3)])
self.db.add_all([Quiz(id=1, title="Origin", user_id=3, is_published=1),
Quiz(id=2, title="Course quiz", user_id=3, course_id=1, is_published=1, is_shared=1, allow_review=0)])
self.db.flush()
for qid, category, owner, shared, source in [
(1, 1, 3, 1, 1), (2, 2, 3, 1, 1), (3, 3, 1, 0, None),
(4, 2, 2, 0, None), (5, 2, 3, 1, 2), (6, None, 3, 1, None),
]:
self.db.add(Question(id=qid, question_category_id=category, user_id=owner,
is_shared=shared, source_quiz_id=source, question_text=f"Question {qid}",
question_type="mcq", options=["yes", "no"], correct_answer="yes",
explanation="Full explanation", image_path="q.png", explanation_image_path="answer.png"))
self.db.flush()
self.db.query(Question).filter(Question.id == 2).update({"is_shared": None})
self.db.add_all([QuizQuestionLink(quiz_id=1, question_id=1, position=0),
QuizQuestionLink(quiz_id=1, question_id=2, position=1),
QuizQuestionLink(quiz_id=2, question_id=5, position=0)])
self.db.commit()
self.user = self.owner
app = FastAPI()
for path, router in [("questions", questions.router), ("question-categories", question_categories.router),
("quizzes", quizzes.router), ("attempts", attempts.router), ("mobile", mobile.router)]:
app.include_router(router, prefix=f"/{path}")
app.dependency_overrides[get_db] = lambda: self.db
app.dependency_overrides[get_current_user] = lambda: self.user
self.client = TestClient(app)
def tearDown(self):
self.client.close()
self.db.close()
self.engine.dispose()
def generate(self, **overrides):
return self.client.post("/questions/builder", json={"title": " Test ", "count": 2, **overrides})
def count(self, **params):
response = self.client.get("/questions/builder/count", params=params)
self.assertEqual(response.status_code, 200, response.text)
return response.json()["count"]
def answer(self, qid, correct=False, day=0, completed=True, expired=0, quiz_id=1):
attempt = QuizAttempt(user_id=1, quiz_id=quiz_id, completed_at=datetime(2026, 1, 1) + timedelta(days=day) if completed else None,
expired=expired, total_questions=1, score=int(correct))
self.db.add(attempt)
self.db.flush()
self.db.add(AttemptAnswer(attempt_id=attempt.id, question_id=qid, is_correct=correct, user_answer="yes" if correct else "no"))
self.db.commit()
return attempt
def test_descendants_counts_visibility_and_invalid_categories(self):
self.assertEqual(category_descendants(self.db.query(QuestionCategory).all(), [1, 2]), {1, 2, 3})
self.assertEqual(self.count(category_ids=[1, 2]), 3)
self.assertEqual(self.count(category_ids=[1], is_shared=True), 2)
self.assertEqual(self.count(), 4)
cats = self.client.get("/question-categories/").json()
root = next(c for c in cats if c["id"] == 1)
leaf = next(c for c in cats if c["id"] == 3)
self.assertEqual(root["question_count"], 3)
self.assertEqual([b["id"] for b in leaf["breadcrumbs"]], [1, 2, 3])
self.assertEqual(self.client.get("/questions/builder/count?category_ids=999").status_code, 400)
self.assertEqual(self.generate(category_ids=[999]).status_code, 400)
self.assertEqual(self.client.get("/questions/bank/ids?category_ids=1,bad").status_code, 400)
self.assertEqual(self.client.get("/questions/bank?category_ids=-1").status_code, 400)
self.assertEqual(self.client.get("/questions/export/qti?question_ids=1,4").status_code, 400)
bank = self.client.get("/questions/bank", params={"category_ids": "1,2", "search_mode": "keyword"}).json()
self.assertEqual(bank["total"], 3)
self.assertEqual(set(self.client.get("/questions/bank/ids").json()), {1, 2, 3, 6})
def test_parent_cycle_missing_self_delete_and_saved_history(self):
saved = self.generate(category_ids=[1], count=3).json()["id"]
before = [q.question_id for q in self.db.query(QuizQuestionLink).filter_by(quiz_id=saved).order_by(QuizQuestionLink.position)]
attempt = self.answer(1, correct=True, quiz_id=saved)
self.user = self.mod
for parent in (1, 3, 999):
res = self.client.patch("/question-categories/1", json={"name": "Root", "parent_id": parent})
self.assertEqual(res.status_code, 400, res.text)
self.assertEqual(self.client.delete("/question-categories/1").status_code, 400)
self.assertEqual(self.client.delete("/question-categories/3?move_to=3").status_code, 400)
self.assertEqual(self.client.delete("/question-categories/3?move_to=999").status_code, 400)
res = self.client.patch("/question-categories/2", json={"name": "Moved", "parent_id": 4})
self.assertEqual(res.status_code, 200, res.text)
self.assertEqual([b["id"] for b in res.json()["breadcrumbs"]], [4, 2])
self.assertEqual(self.client.delete("/question-categories/3?move_to=4").status_code, 204)
self.assertEqual(before, [q.question_id for q in self.db.query(QuizQuestionLink).filter_by(quiz_id=saved).order_by(QuizQuestionLink.position)])
self.assertEqual(self.db.get(QuizAttempt, attempt.id).score, 1)
self.assertEqual(self.db.get(Question, 3).question_category_id, 4)
self.assertEqual(self.client.post("/question-categories/", json={"name": "New", "parent_id": 999}).status_code, 400)
def test_sampling_exact_zero_insufficient_stale_and_validation(self):
for payload, status in [({"category_ids": [4]}, 400), ({"count": 5}, 400), ({"expected_count": 99}, 409),
({"count": 0}, 422), ({"count": 201}, 422), ({"mode": "bad"}, 422),
({"title": " "}, 422), ({"title": "x" * 201}, 422), ({"time_limit_minutes": 0}, 422), ({"state": "bad"}, 422)]:
self.assertEqual(self.generate(**payload).status_code, status)
sampled = self.generate(count=2).json()
sampled_ids = [q.id for q in self.db.get(Quiz, sampled["id"]).questions]
self.assertEqual(len(sampled_ids), 2)
self.assertEqual(len(set(sampled_ids)), 2)
self.assertTrue(set(sampled_ids) <= {1, 2, 3, 6})
result = self.generate(count=4, expected_count=4, mode="learning")
self.assertEqual(result.status_code, 200, result.text)
quiz = self.db.get(Quiz, result.json()["id"])
self.assertEqual(quiz.title, "Test")
self.assertEqual(quiz.is_shared, 0)
self.assertEqual(quiz.is_published, 0)
self.assertEqual({q.id for q in quiz.questions}, {1, 2, 3, 6})
self.assertEqual(len(quiz.questions), 4)
self.assertEqual(quiz.questions[0].explanation_image_path, "answer.png")
first = self.client.post(f"/attempts/start?quiz_id={quiz.id}").json()
again = self.client.post(f"/attempts/start?quiz_id={quiz.id}").json()
self.assertEqual(first["id"], again["id"])
self.assertEqual({q["id"] for q in self.client.get(f"/quizzes/{quiz.id}").json()["questions"]}, {1, 2, 3, 6})
def test_latest_incorrect_unused_bookmarks(self):
self.answer(1, False, day=0)
self.answer(1, True, day=1)
self.answer(1, False, day=2, expired=1)
self.answer(2, False, day=0)
self.answer(2, True, day=2, completed=False)
self.answer(3, False, day=1, quiz_id=2) # course answers never count
self.db.add_all([Favorite(user_id=1, question_id=2), Favorite(user_id=1, question_id=5), Favorite(user_id=2, question_id=1)])
self.db.commit()
self.assertEqual(self.count(state="unused"), 2)
self.assertEqual(self.count(state="incorrect"), 1)
self.assertEqual(self.count(state="bookmarked"), 1)
res = self.generate(state="incorrect", count=1).json()
self.assertEqual([q.id for q in self.db.get(Quiz, res["id"]).questions], [2])
self.answer(2, True, day=3)
self.assertEqual(self.count(state="incorrect"), 0)
def test_explicit_ids_atomic_permissions_order_and_category_creator(self):
for ids in ([1, 4], [1, 5], [1, 999]):
before = self.db.query(Quiz).count()
res = self.client.post("/questions/from-bank", json={"title": "X", "question_ids": ids})
self.assertEqual(res.status_code, 400, res.text)
self.assertEqual(self.db.query(Quiz).count(), before)
res = self.client.post("/questions/from-bank", json={"title": "X", "question_ids": [2, 1, 2, 3]})
self.assertEqual(res.status_code, 200, res.text)
self.assertEqual([q.id for q in self.db.get(Quiz, res.json()["id"]).questions], [2, 1, 3])
res = self.client.post("/question-categories/3/create-quiz?title=Manual")
self.assertEqual(res.status_code, 200, res.text) # ordinary owner, no origin quiz
self.assertEqual(res.json()["questions_count"], 1)
def test_shared_private_revocation_and_mobile(self):
private_id = self.generate(category_ids=[3], count=1).json()["id"]
res = self.client.patch(f"/quizzes/{private_id}/share?shared=true")
self.assertEqual(res.status_code, 400)
self.assertEqual(self.client.post("/questions/from-bank", json={"title": "X", "question_ids": [3], "is_shared": True}).status_code, 400)
shared_id = self.generate(is_shared=True, count=2, category_ids=[1]).json()["id"]
self.user = self.peer
self.assertEqual(self.client.get(f"/quizzes/{private_id}").status_code, 403)
self.assertEqual(self.client.post(f"/quizzes/{private_id}/shuffle").status_code, 403)
self.assertEqual(self.client.post(f"/attempts/start?quiz_id={private_id}").status_code, 403)
self.assertEqual(self.client.get(f"/quizzes/{shared_id}").status_code, 200)
self.assertEqual(self.client.get(f"/mobile/quizzes/{shared_id}").status_code, 200)
self.assertNotIn(private_id, [q["id"] for q in self.client.get("/quizzes/").json()])
self.assertEqual(self.client.patch(f"/quizzes/{shared_id}/share?shared=false").status_code, 403)
attempt = self.client.post(f"/attempts/start?quiz_id={shared_id}").json()["id"]
self.user = self.owner
self.assertEqual(self.client.patch(f"/quizzes/{shared_id}/share?shared=false").status_code, 200)
self.user = self.peer
for url in (f"/quizzes/{shared_id}", f"/quizzes/{shared_id}/review", f"/attempts/progress?quiz_id={shared_id}",
f"/attempts/quiz/{shared_id}/in-progress", f"/attempts/{attempt}"):
self.assertEqual(self.client.get(url).status_code, 403, url)
self.assertEqual(self.client.get(f"/mobile/quizzes/{shared_id}").status_code, 404)
self.assertEqual(self.client.get("/attempts/in-progress").json(), [])
self.assertEqual(self.client.post("/attempts/progress", json={"quiz_id": shared_id,
"attempt_id": attempt, "answers": {}, "current_idx": 0, "mode": "timed"}).status_code, 403)
self.assertEqual(self.client.post("/mobile/attempts", json={"quiz_id": shared_id, "answers": []}).status_code, 404)
self.assertNotIn(shared_id, [q["id"] for q in self.client.get("/mobile/sync").json()["quizzes"]])
self.assertEqual(self.client.post(f"/attempts/{attempt}/submit", json={"answers": []}).status_code, 403)
self.assertEqual(self.client.post(f"/attempts/start?quiz_id={shared_id}").status_code, 403)
def test_question_revocation_legacy_publication_and_course_enrollment(self):
saved_id = self.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
self.user = self.peer
self.assertEqual(self.client.get("/quizzes/1").status_code, 200) # published legacy quiz
self.db.get(Question, 1).is_shared = 0
self.db.commit()
self.assertEqual(self.client.get("/quizzes/1").status_code, 403)
self.user = self.owner
self.assertEqual(self.client.get(f"/quizzes/{saved_id}").status_code, 403)
self.user = self.peer
self.assertEqual(self.client.get("/quizzes/2").status_code, 403)
self.assertEqual(self.client.post("/attempts/start?quiz_id=2").status_code, 403)
self.db.add(CourseEnrollment(course_id=1, user_id=2))
self.db.commit()
self.assertEqual(self.client.get("/quizzes/2").status_code, 200)
self.assertEqual(self.client.get("/quizzes/2?study=true").status_code, 403)
self.assertEqual(self.client.get("/quizzes/2/review").status_code, 403)
self.assertEqual(self.client.get("/mobile/quizzes/2").status_code, 404)
self.assertNotIn(2, [q["id"] for q in self.client.get("/quizzes/").json()])
self.user = self.mod
self.assertEqual(self.client.patch("/quizzes/2/share?shared=true").status_code, 400)
self.assertEqual(self.client.get("/quizzes/1").status_code, 200)
if __name__ == "__main__":
unittest.main()

View file

@ -10,6 +10,7 @@ const DashboardPage = lazy(() => import('./pages/DashboardPage'))
const UploadPage = lazy(() => import('./pages/UploadPage'))
const DocumentDetailPage = lazy(() => import('./pages/DocumentDetailPage'))
const QuizPage = lazy(() => import('./pages/QuizPage'))
const CustomQuizPage = lazy(() => import('./pages/CustomQuizPage'))
const QuizzesPage = lazy(() => import('./pages/QuizzesPage'))
const ResultsPage = lazy(() => import('./pages/ResultsPage'))
const AdminPage = lazy(() => import('./pages/AdminPage'))
@ -80,6 +81,7 @@ function AppRoutes() {
<Route element={<AppLayout />}>
<Route path="/" element={<DashboardPage />} />
<Route path="/quizzes" element={<QuizzesPage />} />
<Route path="/quizzes/create" element={<CustomQuizPage />} />
<Route path="/quizzes/:id" element={<QuizPage />} />
<Route path="/results/:id" element={<ResultsPage />} />
<Route path="/documents/:id" element={<DocumentDetailPage />} />

View file

@ -0,0 +1,12 @@
.custom-test { max-width: 960px; margin: auto; }
.custom-test h1 { margin: 16px 0; }
.custom-test fieldset { border: 1px solid var(--border); border-radius: 8px; padding: 16px; min-width: 0; }
.custom-test p { margin: 12px 0; }
.custom-test-categories { display: grid; gap: 10px; max-height: 320px; overflow: auto; margin: 12px 0; }
.custom-test-categories label, .custom-test-share { display: flex; align-items: baseline; gap: 8px; }
.custom-test input[type=checkbox] { width: auto; flex-shrink: 0; }
.custom-test-settings { display: grid; grid-template-columns: repeat(auto-fit, minmax(220px, 1fr)); gap: 16px; margin: 20px 0; }
.custom-test-settings label { display: flex; flex-direction: column; gap: 6px; }
.custom-test-settings input, .custom-test-settings select { width: 100%; padding: 10px; border: 1px solid var(--border); border-radius: 6px; background: var(--input-bg); color: var(--text); }
.custom-test form > button { margin: 8px 8px 0 0; }
.custom-test [role=alert] { color: var(--wrong-fg); }

View file

@ -0,0 +1,109 @@
import { useEffect, useState } from 'react'
import { Link, useNavigate } from 'react-router-dom'
import api from '../api/client'
import { useAuth } from '../context/AuthContext'
import './CustomQuizPage.css'
export default function CustomQuizPage() {
const { user } = useAuth()
const navigate = useNavigate()
const [categories, setCategories] = useState([])
const [categoryIds, setCategoryIds] = useState([])
const [state, setState] = useState('all')
const [shared, setShared] = useState(false)
const [title, setTitle] = useState('My Custom Test')
const [mode, setMode] = useState('learning')
const [time, setTime] = useState('')
const [count, setCount] = useState(20)
const [available, setAvailable] = useState(null)
const [countKey, setCountKey] = useState(null)
const [error, setError] = useState('')
const [countError, setCountError] = useState('')
const [submitting, setSubmitting] = useState(false)
const [refresh, setRefresh] = useState(0)
const filterKey = JSON.stringify([categoryIds, state, shared, refresh])
useEffect(() => {
let active = true
api.get('/question-categories/').then(r => { if (active) setCategories(r.data) })
.catch(() => { if (active) setError('Could not load categories. Reload this page to try again.') })
return () => { active = false }
}, [])
useEffect(() => {
let active = true
setAvailable(null)
setCountError('')
const params = new URLSearchParams({ state, is_shared: String(shared) })
categoryIds.forEach(id => params.append('category_ids', id))
api.get('/questions/builder/count', { params }).then(r => {
if (active) { setAvailable(r.data.count); setCountKey(filterKey) }
}).catch(() => { if (active) setCountError('Could not load available count. Try refreshing.') })
return () => { active = false }
}, [filterKey])
const ready = countKey === filterKey && available !== null
const validCount = Number.isInteger(Number(count)) && Number(count) >= 1 && Number(count) <= 200 && Number(count) <= available
const submit = async e => {
e.preventDefault()
if (!ready || !validCount || submitting) return
setSubmitting(true)
setError('')
try {
const result = await api.post('/questions/builder', {
title: title.trim(), category_ids: categoryIds, state, count: Number(count),
expected_count: available, mode, time_limit_minutes: mode === 'timed' && time ? Number(time) : null,
is_shared: shared,
})
navigate(`/quizzes/${result.data.id}`)
} catch (err) {
const detail = err.response?.data?.detail
setError(typeof detail === 'string' ? detail : 'Could not create test. Check your settings and try again.')
setRefresh(v => v + 1)
} finally { setSubmitting(false) }
}
return (
<div className="custom-test">
<Link to="/quizzes"> Quizzes</Link>
<h1>Create Custom Test</h1>
<p>Choose questions from your bank, {user?.name || 'learner'}. Your test saves a fixed selection.</p>
<form onSubmit={submit} className="card">
<fieldset disabled={submitting}>
<legend>Categories</legend>
<p>Select any combination. Parent categories include all descendants; overlapping selections count once. No selection includes the whole bank.</p>
<div className="custom-test-categories">
{categories.map(cat => (
<label key={cat.id}>
<input type="checkbox" checked={categoryIds.includes(cat.id)} onChange={e => setCategoryIds(ids => e.target.checked ? [...ids, cat.id] : ids.filter(id => id !== cat.id))} />
{(cat.breadcrumbs || []).map(c => c.name).join(' ') || cat.name} ({cat.question_count})
</label>
))}
</div>
<button type="button" className="btn btn-secondary btn-sm" onClick={() => setCategoryIds([])}>Clear categories</button>
</fieldset>
<div className="custom-test-settings">
<label>Title<input required maxLength={200} value={title} onChange={e => setTitle(e.target.value)} /></label>
<label>Question state<select value={state} onChange={e => setState(e.target.value)}>
<option value="all">All</option><option value="unused">Unused</option>
<option value="incorrect">Incorrect</option><option value="bookmarked">Bookmarked</option>
</select></label>
<label>Number of questions<input type="number" required min="1" max="200" step="1" value={count} onChange={e => setCount(e.target.value)} /></label>
<label>Mode<select value={mode} onChange={e => setMode(e.target.value)}>
<option value="learning">Study</option><option value="timed">Exam</option>
</select></label>
{mode === 'timed' && <label>Time limit (minutes, optional)<input type="number" min="1" step="1" value={time} onChange={e => setTime(e.target.value)} /></label>}
</div>
<p>Unused means never answered in a completed, nonexpired bank attempt. Incorrect uses your latest such answer.</p>
<label className="custom-test-share"><input type="checkbox" checked={shared} onChange={e => setShared(e.target.checked)} /> Share with other learners (only shareable questions)</label>
<p role="status" aria-live="polite">{ready ? `${available} questions available` : 'Counting available questions…'}</p>
{ready && available === 0 && <p>No questions match these filters.</p>}
{ready && !validCount && available > 0 && <p>Choose 1{Math.min(200, available)} questions.</p>}
{countError && <p role="alert">{countError}</p>}
<button type="button" className="btn btn-secondary" disabled={submitting} onClick={() => setRefresh(v => v + 1)}>Refresh count</button>
{error && <p role="alert">{error}</p>}
<button className="btn btn-primary" type="submit" disabled={submitting || !ready || !validCount || !title.trim()}>{submitting ? 'Creating…' : 'Create Test'}</button>
</form>
</div>
)
}

View file

@ -0,0 +1,100 @@
import { fireEvent, render, screen, waitFor } from '@testing-library/react'
import userEvent from '@testing-library/user-event'
import { MemoryRouter, Route, Routes } from 'react-router-dom'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import api from '../api/client'
import CustomQuizPage from './CustomQuizPage'
vi.mock('../api/client', () => ({ default: { get: vi.fn(), post: vi.fn() } }))
vi.mock('../context/AuthContext', () => ({ useAuth: () => ({ user: { id: 1, name: 'Learner' } }) }))
const categories = [
{ id: 1, name: 'Pediatrics', question_count: 30, breadcrumbs: [{ id: 1, name: 'Pediatrics' }] },
{ id: 2, name: 'Neonatal', question_count: 10, breadcrumbs: [{ id: 1, name: 'Pediatrics' }, { id: 2, name: 'Neonatal' }] },
]
function setupCount(count = 30) {
api.get.mockImplementation(url => Promise.resolve({ data: url === '/question-categories/' ? categories : { count } }))
}
function renderBuilder() {
render(<MemoryRouter initialEntries={['/quizzes/create']}><Routes>
<Route path="/quizzes/create" element={<CustomQuizPage />} />
<Route path="/quizzes/:id" element={<h1>Saved test</h1>} />
</Routes></MemoryRouter>)
}
beforeEach(() => { vi.clearAllMocks(); setupCount() })
describe('CustomQuizPage', () => {
it('shows exact counts, multi-category controls, private study defaults and saves the selection', async () => {
api.post.mockResolvedValue({ data: { id: 123 } })
renderBuilder()
await screen.findByText('30 questions available')
expect(screen.getByLabelText('Mode')).toHaveValue('learning')
expect(screen.getByLabelText(/Share with/)).not.toBeChecked()
expect(screen.queryByLabelText(/Time limit/)).not.toBeInTheDocument()
await userEvent.click(screen.getByLabelText('Pediatrics (30)'))
await userEvent.click(screen.getByLabelText('Pediatrics Neonatal (10)'))
await userEvent.selectOptions(screen.getByLabelText('Question state'), 'incorrect')
await userEvent.selectOptions(screen.getByLabelText('Mode'), 'timed')
fireEvent.change(screen.getByLabelText(/Time limit/), { target: { value: '15' } })
fireEvent.change(screen.getByLabelText('Number of questions'), { target: { value: '10' } })
await userEvent.click(screen.getByLabelText(/Share with/))
await waitFor(() => expect(screen.getByRole('button', { name: 'Create Test' })).toBeEnabled())
const calls = api.get.mock.calls.filter(([url]) => url === '/questions/builder/count')
const params = calls.at(-1)[1].params
expect(params.getAll('category_ids')).toEqual(['1', '2'])
expect(params.get('state')).toBe('incorrect')
expect(params.get('is_shared')).toBe('true')
await userEvent.click(screen.getByRole('button', { name: 'Create Test' }))
expect(api.post).toHaveBeenCalledWith('/questions/builder', {
title: 'My Custom Test', category_ids: [1, 2], state: 'incorrect', count: 10,
mode: 'timed', time_limit_minutes: 15, expected_count: 30, is_shared: true,
})
await screen.findByRole('heading', { name: 'Saved test' })
})
it('blocks zero/insufficient pools and invalid counts', async () => {
setupCount(0)
renderBuilder()
await screen.findByText('No questions match these filters.')
expect(screen.getByRole('button', { name: 'Create Test' })).toBeDisabled()
setupCount(5)
await userEvent.click(screen.getByRole('button', { name: 'Refresh count' }))
await screen.findByText('5 questions available')
expect(screen.getByRole('button', { name: 'Create Test' })).toBeDisabled()
for (const value of ['0', '201', '1.5']) {
fireEvent.change(screen.getByLabelText('Number of questions'), { target: { value } })
expect(screen.getByRole('button', { name: 'Create Test' })).toBeDisabled()
}
fireEvent.change(screen.getByLabelText('Number of questions'), { target: { value: '5' } })
expect(screen.getByRole('button', { name: 'Create Test' })).toBeEnabled()
})
it('reports count failures and server stale-pool errors without navigating', async () => {
api.get.mockImplementation(url => url === '/question-categories/' ? Promise.resolve({ data: categories }) : Promise.reject(new Error('offline')))
renderBuilder()
await screen.findByRole('alert')
expect(screen.getByRole('button', { name: 'Create Test' })).toBeDisabled()
setupCount(30)
await userEvent.click(screen.getByRole('button', { name: 'Refresh count' }))
await screen.findByText('30 questions available')
api.post.mockRejectedValue({ response: { data: { detail: 'Available count changed. Refresh the count and try again' } } })
await userEvent.click(screen.getByRole('button', { name: 'Create Test' }))
expect(await screen.findByRole('alert')).toHaveTextContent('Available count changed')
expect(screen.queryByRole('heading', { name: 'Saved test' })).not.toBeInTheDocument()
})
it('ignores outdated count responses after filters change', async () => {
let resolveOld
api.get.mockImplementation(url => url === '/question-categories/' ? Promise.resolve({ data: categories }) : new Promise(resolve => { resolveOld = resolve }))
renderBuilder()
await screen.findByLabelText('Pediatrics (30)')
const old = resolveOld
setupCount(4)
await userEvent.selectOptions(screen.getByLabelText('Question state'), 'bookmarked')
await screen.findByText('4 questions available')
old({ data: { count: 100 } })
await waitFor(() => expect(screen.queryByText('100 questions available')).not.toBeInTheDocument())
expect(screen.getByRole('button', { name: 'Create Test' })).toBeDisabled()
})
})

View file

@ -1,5 +1,5 @@
import { useState, useEffect, useRef, lazy, Suspense } from 'react'
import { useNavigate } from 'react-router-dom'
import { useNavigate, Link } from 'react-router-dom'
import { useAuth } from '../context/AuthContext'
import api from '../api/client'
import Dialog from '../components/Dialog'
@ -284,7 +284,7 @@ function QuestionEditModal({ question, categories, onSaved, onClose }) {
<label>Question Category</label>
<select value={form.question_category_id} onChange={e => setForm(f => ({ ...f, question_category_id: e.target.value }))}>
<option value=""> Uncategorized </option>
{categories.map(c => <option key={c.id} value={c.id}>{c.name}</option>)}
{categories.map(c => <option key={c.id} value={c.id}>{(c.breadcrumbs || []).map(b => b.name).join(' ') || c.name}</option>)}
</select>
</div>
<div className="form-group">
@ -398,7 +398,7 @@ function CreateQuestionModal({ categories, onCreated, onClose }) {
<select value={form.question_category_id} onChange={e => setForm(f => ({ ...f, question_category_id: e.target.value }))}
style={{ width: '100%', padding: '10px 14px', border: '1px solid var(--border)', borderRadius: 8, background: 'var(--input-bg)', color: 'var(--text)', fontSize: '0.9rem' }}>
<option value="">None</option>
{categories.map(c => <option key={c.id} value={c.id}>{c.name}</option>)}
{categories.map(c => <option key={c.id} value={c.id}>{(c.breadcrumbs || []).map(b => b.name).join(' ') || c.name}</option>)}
</select>
</div>
@ -527,6 +527,8 @@ export default function QuestionBankPage() {
const [selectedIds, setSelectedIds] = useState(new Set())
const [showCreateQuiz, setShowCreateQuiz] = useState(false)
const [newCatName, setNewCatName] = useState('')
const [catParent, setCatParent] = useState('')
const [editingCategory, setEditingCategory] = useState(null)
const [showCatForm, setShowCatForm] = useState(false)
const [assignCatId, setAssignCatId] = useState('')
const [bulkError, setBulkError] = useState('')
@ -666,9 +668,12 @@ export default function QuestionBankPage() {
const addCategory = async () => {
if (!newCatName.trim()) return
try {
const res = await api.post('/question-categories/', { name: newCatName.trim() })
setCategories(prev => [...prev, res.data])
setNewCatName(''); setShowCatForm(false)
const payload = { name: newCatName.trim(), parent_id: catParent ? Number(catParent) : null, description: editingCategory?.description || null }
if (editingCategory) await api.patch(`/question-categories/${editingCategory.id}`, payload)
else await api.post('/question-categories/', payload)
const res = await api.get('/question-categories/')
setCategories(res.data)
setNewCatName(''); setCatParent(''); setEditingCategory(null); setShowCatForm(false)
} catch (err) { await openAlert(err.response?.data?.detail || 'Failed', { title: 'Error' }) }
}
@ -684,7 +689,7 @@ export default function QuestionBankPage() {
setCategories(prev => prev.filter(c => c.id !== catId))
setFilterCatIds(prev => prev.filter(c => c !== catId))
loadQuestions(searchQuery, 0, filterCatIds.filter(c => c !== catId), showUncategorized, showFavorites)
} catch { }
} catch (err) { await openAlert(err.response?.data?.detail || 'Could not delete category', { title: 'Error' }) }
}
const deleteCategory = (catId) => {
@ -773,6 +778,7 @@ export default function QuestionBankPage() {
return (
<div>
<Dialog {...dialogProps} />
<Link className="btn btn-primary" to="/quizzes/create" style={{ marginBottom: 16 }}>Create Custom Test</Link>
{/* Delete category dialog */}
{deletingCatId && (() => {
const cat = categories.find(c => c.id === deletingCatId)
@ -786,7 +792,7 @@ export default function QuestionBankPage() {
<label>Move {cat.question_count} questions to:</label>
<select value={moveToCatId} onChange={e => setMoveToCatId(e.target.value)}>
<option value="">Leave uncategorized</option>
{others.map(c => <option key={c.id} value={c.id}>{c.name}</option>)}
{others.map(c => <option key={c.id} value={c.id}>{(c.breadcrumbs || []).map(b => b.name).join(' ') || c.name}</option>)}
</select>
</div>
)}
@ -869,17 +875,21 @@ export default function QuestionBankPage() {
<button className="btn btn-secondary btn-sm" onClick={handleQtiExport} disabled={qtiExporting}>
{qtiExporting ? 'Exporting QTI...' : `Export QTI${selectedIds.size > 0 ? ` (${selectedIds.size})` : ''}`}
</button>
{isModerator && <button className="btn btn-secondary btn-sm" onClick={() => setShowCatForm(v => !v)}>+ Category</button>}
{isModerator && <button className="btn btn-secondary btn-sm" onClick={() => { setEditingCategory(null); setNewCatName(''); setCatParent(''); setShowCatForm(v => !v) }}>+ Category</button>}
</div>
</div>
{showCatForm && (
<div style={{ marginTop: 12, display: 'flex', gap: 8 }}>
<input type="text" value={newCatName} onChange={e => setNewCatName(e.target.value)} placeholder="New category name..."
<div style={{ marginTop: 12, display: 'flex', gap: 8, flexWrap: 'wrap' }}>
<input aria-label="Category name" type="text" value={newCatName} onChange={e => setNewCatName(e.target.value)} placeholder="New category name..."
onKeyDown={e => e.key === 'Enter' && addCategory()}
style={{ flex: 1, padding: '7px 12px', border: '1px solid var(--border)', borderRadius: 8, fontSize: '0.875rem', background: 'var(--input-bg)', color: 'var(--text)' }} />
<button className="btn btn-primary btn-sm" onClick={addCategory}>Add</button>
<button className="btn btn-secondary btn-sm" onClick={() => { setShowCatForm(false); setNewCatName('') }}>Cancel</button>
<select aria-label="Parent category" value={catParent} onChange={e => setCatParent(e.target.value)}>
<option value="">No parent (root)</option>
{categories.filter(c => !editingCategory || !(c.breadcrumbs || [{ id: c.id }]).some(b => b.id === editingCategory.id)).map(c => <option key={c.id} value={c.id}>{(c.breadcrumbs || []).map(b => b.name).join(' ') || c.name}</option>)}
</select>
<button className="btn btn-primary btn-sm" onClick={addCategory}>{editingCategory ? 'Save category' : 'Add'}</button>
<button className="btn btn-secondary btn-sm" onClick={() => { setShowCatForm(false); setNewCatName(''); setCatParent(''); setEditingCategory(null) }}>Cancel</button>
</div>
)}
</div>
@ -901,9 +911,10 @@ export default function QuestionBankPage() {
<div key={cat.id} style={{ display: 'flex', alignItems: 'center', gap: 2 }}>
<button className={`btn btn-sm ${isActive ? 'btn-primary' : 'btn-secondary'}`}
onClick={() => { setFilterCatIds(prev => prev.includes(cat.id) ? prev.filter(c => c !== cat.id) : [...prev, cat.id]); setShowUncategorized(false); setShowFavorites(false) }}>
{cat.name} <span style={{ opacity: 0.65 }}>({cat.question_count})</span>
{(cat.breadcrumbs || []).map(c => c.name).join(' ') || cat.name} <span style={{ opacity: 0.65 }}>({cat.question_count})</span>
</button>
{isModerator && <button onClick={() => deleteCategory(cat.id)}
{isModerator && <button className="btn btn-secondary btn-sm" aria-label={`Edit category ${cat.name}`} onClick={() => { setEditingCategory(cat); setNewCatName(cat.name); setCatParent(cat.parent_id || ''); setShowCatForm(true) }}>Edit</button>}
{isModerator && <button aria-label={`Delete category ${cat.name}`} onClick={() => deleteCategory(cat.id)}
style={{ background: 'none', border: 'none', cursor: 'pointer', color: '#cbd5e1', fontSize: '0.72rem', padding: '2px 3px', lineHeight: 1 }}
onMouseEnter={e => e.currentTarget.style.color = '#ef4444'} onMouseLeave={e => e.currentTarget.style.color = '#cbd5e1'}></button>}
</div>
@ -1009,7 +1020,7 @@ export default function QuestionBankPage() {
style={{ padding: '5px 10px', border: '1px solid var(--border)', borderRadius: 6, fontSize: '0.82rem', background: 'var(--input-bg)', color: 'var(--text)' }}>
<option value="">Move to category</option>
<option value="remove"> Remove category</option>
{categories.map(c => <option key={c.id} value={c.id}>{c.name}</option>)}
{categories.map(c => <option key={c.id} value={c.id}>{(c.breadcrumbs || []).map(b => b.name).join(' ') || c.name}</option>)}
</select>
<button className="btn btn-sm btn-primary" onClick={bulkAssignCategory} disabled={!assignCatId}>Apply</button>
<button className="btn btn-sm btn-primary" onClick={() => setShowCreateQuiz(true)}>Create Quiz</button>

View file

@ -99,3 +99,26 @@ describe('QuestionBankPage QTI actions', () => {
expect(screen.getByText('Export unavailable')).toBeInTheDocument()
})
})
describe('QuestionBankPage category hierarchy', () => {
it('edits parent assignment with breadcrumb choices and excludes descendants', async () => {
vi.clearAllMocks()
const cats = [
{ id: 1, name: 'Root', parent_id: null, question_count: 2, breadcrumbs: [{ id: 1, name: 'Root' }] },
{ id: 2, name: 'Child', parent_id: 1, question_count: 2, breadcrumbs: [{ id: 1, name: 'Root' }, { id: 2, name: 'Child' }] },
{ id: 3, name: 'Other', parent_id: null, question_count: 0, breadcrumbs: [{ id: 3, name: 'Other' }] },
]
mockInitialRequests()
const initialGet = api.get.getMockImplementation()
api.get.mockImplementation(url => url === '/question-categories/' ? Promise.resolve({ data: cats }) : initialGet(url))
api.patch = vi.fn().mockResolvedValue({ data: {} })
renderPage()
expect(await screen.findByRole('link', { name: 'Create Custom Test' })).toHaveAttribute('href', '/quizzes/create')
await userEvent.click(await screen.findByRole('button', { name: 'Edit category Root' }))
const parent = screen.getByLabelText('Parent category')
expect([...parent.options].map(o => o.text)).toEqual(['No parent (root)', 'Other'])
await userEvent.selectOptions(parent, '3')
await userEvent.click(screen.getByRole('button', { name: 'Save category' }))
await waitFor(() => expect(api.patch).toHaveBeenCalledWith('/question-categories/1', { name: 'Root', parent_id: 3, description: null }))
})
})

View file

@ -190,6 +190,18 @@ function QuestionStudyModal({ question, query, onClose }) {
function QuizCard({ quiz, isModerator, categories, onDelete, onCategoryChange }) {
const navigate = useNavigate()
const { user } = useAuth()
const [shareError, setShareError] = useState('')
const [shareBusy, setShareBusy] = useState(false)
const toggleShare = async () => {
setShareBusy(true)
setShareError('')
try {
const res = await api.patch(`/quizzes/${quiz.id}/share`, null, { params: { shared: quiz.is_shared !== 1 } })
onCategoryChange(quiz.id, quiz.category_id, res.data.is_published, res.data.is_shared)
} catch (err) { setShareError(err.response?.data?.detail || 'Could not update sharing') }
finally { setShareBusy(false) }
}
const [showCatMenu, setShowCatMenu] = useState(false)
const assignCategory = async (catId) => {
@ -221,7 +233,7 @@ function QuizCard({ quiz, isModerator, categories, onDelete, onCategoryChange })
style={{ background: 'none', border: 'none', cursor: 'pointer', color: '#cbd5e1', fontSize: '0.9rem', padding: '4px 6px', borderRadius: 4, textDecoration: 'none', display: 'flex', alignItems: 'center' }}
onMouseEnter={e => e.currentTarget.style.color = 'var(--primary)'}
onMouseLeave={e => e.currentTarget.style.color = '#cbd5e1'}></Link>
<button title={quiz.is_published === 0 ? 'Hidden from users — click to publish' : 'Visible — click to hide'} onClick={togglePublish}
<button title={quiz.is_published === 0 ? 'Unpublished (sharing may still grant access) — click to publish' : 'Published — click to unpublish'} onClick={togglePublish}
style={{ background: 'none', border: 'none', cursor: 'pointer', color: quiz.is_published === 0 ? '#ef4444' : '#cbd5e1', fontSize: '0.9rem', padding: '4px 6px', borderRadius: 4 }}
onMouseEnter={e => e.currentTarget.style.opacity = '0.7'} onMouseLeave={e => e.currentTarget.style.opacity = '1'}>
{quiz.is_published === 0 ? '🙈' : '👁'}
@ -267,6 +279,11 @@ function QuizCard({ quiz, isModerator, categories, onDelete, onCategoryChange })
{quiz.time_limit_minutes && <span style={{ fontSize: '0.75rem', color: '#94a3b8' }}>{quiz.time_limit_minutes} min</span>}
</div>
</div>
<div onClick={e => e.stopPropagation()} style={{ display: 'flex', gap: 8, flexWrap: 'wrap', alignItems: 'center' }}>
{(isModerator || user?.id === quiz.user_id) && <button className="btn btn-secondary btn-sm" disabled={shareBusy} onClick={toggleShare}>{quiz.is_shared === 1 ? 'Unshare test' : 'Share test'}</button>}
{quiz.is_shared === 1 && <Link to={`/quizzes/${quiz.id}`}>Shared test link</Link>}
{shareError && <p role="alert">{shareError}</p>}
</div>
</div>
)
}
@ -321,9 +338,10 @@ export default function QuizzesPage() {
} catch (err) { console.error(err) }
}
const handleCategoryChange = (quizId, catId, isPublished) => {
const handleCategoryChange = (quizId, catId, isPublished, isShared) => {
setQuizzes(prev => prev.map(q => q.id === quizId ? {
...q,
is_shared: isShared !== undefined ? isShared : q.is_shared,
category_id: catId !== undefined ? catId : q.category_id,
is_published: isPublished !== undefined ? isPublished : q.is_published,
} : q))
@ -386,6 +404,8 @@ export default function QuizzesPage() {
</a>
</div>
<Link className="btn btn-primary" to="/quizzes/create" style={{ marginBottom: 16 }}>Create Custom Test</Link>
{/* Search bar */}
<div className="card" style={{ marginBottom: 16 }}>
<div style={{ display: 'flex', gap: 10, alignItems: 'center', flexWrap: 'wrap' }}>

View file

@ -0,0 +1,27 @@
import { render, screen, waitFor } from '@testing-library/react'
import userEvent from '@testing-library/user-event'
import { MemoryRouter } from 'react-router-dom'
import { expect, it, vi } from 'vitest'
import QuizzesPage from './QuizzesPage'
import api from '../api/client'
vi.mock('../api/client', () => ({ default: { get: vi.fn(), patch: vi.fn() } }))
vi.mock('../context/AuthContext', () => ({ useAuth: () => ({ user: { id: 1, role: 'user' } }) }))
it('offers creation, sharing and revocation for an owner, and displays server errors', async () => {
api.get.mockImplementation(url => Promise.resolve({ data: url === '/quizzes/' ? [
{ id: 10, title: 'My Test', user_id: 1, is_shared: 0, is_published: 0, mode: 'learning', questions_count: 2 },
] : [] }))
api.patch.mockResolvedValueOnce({ data: { is_shared: 1, is_published: 0 } })
render(<MemoryRouter><QuizzesPage /></MemoryRouter>)
expect(await screen.findByRole('link', { name: 'Create Custom Test' })).toHaveAttribute('href', '/quizzes/create')
await userEvent.click(screen.getByRole('button', { name: 'Share test' }))
expect(await screen.findByRole('link', { name: 'Shared test link' })).toHaveAttribute('href', '/quizzes/10')
expect(api.patch).toHaveBeenLastCalledWith('/quizzes/10/share', null, { params: { shared: true } })
api.patch.mockResolvedValueOnce({ data: { is_shared: 0, is_published: 0 } })
await userEvent.click(screen.getByRole('button', { name: 'Unshare test' }))
await waitFor(() => expect(screen.queryByRole('link', { name: 'Shared test link' })).not.toBeInTheDocument())
api.patch.mockRejectedValueOnce({ response: { data: { detail: 'This test contains private or course-only questions' } } })
await userEvent.click(screen.getByRole('button', { name: 'Share test' }))
expect(await screen.findByRole('alert')).toHaveTextContent('private or course-only')
})