diff --git a/backend/alembic/versions/c82d19e4a601_category_hierarchy.py b/backend/alembic/versions/c82d19e4a601_category_hierarchy.py
new file mode 100644
index 0000000..a93a23d
--- /dev/null
+++ b/backend/alembic/versions/c82d19e4a601_category_hierarchy.py
@@ -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")
diff --git a/backend/app/models/question_category.py b/backend/app/models/question_category.py
index f7ec9b6..f451d75 100644
--- a/backend/app/models/question_category.py
+++ b/backend/app/models/question_category.py
@@ -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)
diff --git a/backend/app/routers/attempts.py b/backend/app/routers/attempts.py
index 07b9f77..e46e646 100644
--- a/backend/app/routers/attempts.py
+++ b/backend/app/routers/attempts.py
@@ -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)
diff --git a/backend/app/routers/mobile.py b/backend/app/routers/mobile.py
index 154563f..d9a94ac 100644
--- a/backend/app/routers/mobile.py
+++ b/backend/app/routers/mobile.py
@@ -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
diff --git a/backend/app/routers/question_categories.py b/backend/app/routers/question_categories.py
index 70c2ba1..2c7d2be 100644
--- a/backend/app/routers/question_categories.py
+++ b/backend/app/routers/question_categories.py
@@ -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]))
diff --git a/backend/app/routers/questions.py b/backend/app/routers/questions.py
index 2a9d2d8..290cad6 100644
--- a/backend/app/routers/questions.py
+++ b/backend/app/routers/questions.py
@@ -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:
diff --git a/backend/app/routers/quizzes.py b/backend/app/routers/quizzes.py
index cb80e50..0542c77 100644
--- a/backend/app/routers/quizzes.py
+++ b/backend/app/routers/quizzes.py
@@ -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)
diff --git a/backend/app/services/quiz_builder.py b/backend/app/services/quiz_builder.py
new file mode 100644
index 0000000..9b480d6
--- /dev/null
+++ b/backend/app/services/quiz_builder.py
@@ -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))
diff --git a/backend/app/utils/quiz_access.py b/backend/app/utils/quiz_access.py
new file mode 100644
index 0000000..164fa6f
--- /dev/null
+++ b/backend/app/utils/quiz_access.py
@@ -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}
diff --git a/backend/tests/test_category_migration.py b/backend/tests/test_category_migration.py
new file mode 100644
index 0000000..a6608c7
--- /dev/null
+++ b/backend/tests/test_category_migration.py
@@ -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()
diff --git a/backend/tests/test_quiz_builder.py b/backend/tests/test_quiz_builder.py
new file mode 100644
index 0000000..8b559e4
--- /dev/null
+++ b/backend/tests/test_quiz_builder.py
@@ -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()
diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx
index f383979..ee4d2b1 100644
--- a/frontend/src/App.jsx
+++ b/frontend/src/App.jsx
@@ -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() {
}>
} />
} />
+ } />
} />
} />
} />
diff --git a/frontend/src/pages/CustomQuizPage.css b/frontend/src/pages/CustomQuizPage.css
new file mode 100644
index 0000000..e50d731
--- /dev/null
+++ b/frontend/src/pages/CustomQuizPage.css
@@ -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); }
diff --git a/frontend/src/pages/CustomQuizPage.jsx b/frontend/src/pages/CustomQuizPage.jsx
new file mode 100644
index 0000000..4504171
--- /dev/null
+++ b/frontend/src/pages/CustomQuizPage.jsx
@@ -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 (
+
+
← Quizzes
+
Create Custom Test
+
Choose questions from your bank, {user?.name || 'learner'}. Your test saves a fixed selection.
+
+
@@ -398,7 +398,7 @@ function CreateQuestionModal({ categories, onCreated, onClose }) {
@@ -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 (
+ Create Custom Test
{/* Delete category dialog */}
{deletingCatId && (() => {
const cat = categories.find(c => c.id === deletingCatId)
@@ -786,7 +792,7 @@ export default function QuestionBankPage() {
)}
@@ -869,17 +875,21 @@ export default function QuestionBankPage() {
- {isModerator && }
+ {isModerator && }
{showCatForm && (
-
@@ -901,9 +911,10 @@ export default function QuestionBankPage() {
- {isModerator &&
@@ -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)' }}>
- {categories.map(c => )}
+ {categories.map(c => )}
diff --git a/frontend/src/pages/QuestionBankPage.test.jsx b/frontend/src/pages/QuestionBankPage.test.jsx
index 47b40b7..0bb6f57 100644
--- a/frontend/src/pages/QuestionBankPage.test.jsx
+++ b/frontend/src/pages/QuestionBankPage.test.jsx
@@ -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 }))
+ })
+})
diff --git a/frontend/src/pages/QuizzesPage.jsx b/frontend/src/pages/QuizzesPage.jsx
index bf8aa53..8dba28e 100644
--- a/frontend/src/pages/QuizzesPage.jsx
+++ b/frontend/src/pages/QuizzesPage.jsx
@@ -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'}>✏️
-