code split: api/routes/tests

backend/app/api/routes/tests/__init__.py

@@ -0,0 +1,35 @@
+from fastapi import APIRouter
+
+from .attempts import (
+    save_answers,
+    start_test,
+    submit_test,
+)
+from .history import my_history
+from .results import get_result
+from .schemas import AnswerSubmit, BulkAnswerSubmit
+
+from .catalog import router as catalog_router
+from .attempts import router as attempts_router
+from .results import router as results_router
+from .leaderboard import router as leaderboard_router
+from .history import router as history_router
+
+router = APIRouter(prefix="/tests", tags=["Tests"])
+
+router.include_router(catalog_router)
+router.include_router(attempts_router)
+router.include_router(results_router)
+router.include_router(leaderboard_router)
+router.include_router(history_router)
+
+# Re-exports for tests/test_attempt_persistence.py
+__all__ = [
+    "AnswerSubmit",
+    "BulkAnswerSubmit",
+    "get_result",
+    "my_history",
+    "save_answers",
+    "start_test",
+    "submit_test",
+]

backend/app/api/routes/{tests.py → tests/attempts.py}

@@ -1,183 +1,27 @@
 from datetime import datetime, timezone
-from typing import List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException
-from fastapi_cache.decorator import cache
-from pydantic import BaseModel
-from sqlalchemy import func, select
-from sqlalchemy.orm import Session, aliased, joinedload
+from sqlalchemy.orm import Session
 
-from app.api.deps import get_current_user, require_aspirant
+from app.api.deps import require_aspirant
 from app.core.database import get_db
-from app.models.models import (
-    PracticeAttemptCounter,
-    Question,
-    Test,
-    TestAttempt,
-    TestStatus,
-    UserAnswer,
-)
+from app.models.models import Question, Test, TestAttempt, TestStatus, UserAnswer
 from app.services.cloudinary_service import (
     optimize_delivery_image_url,
     optimize_delivery_image_urls,
 )
-from app.services.scoring import evaluate_answer
-
-router = APIRouter(prefix="/tests", tags=["Tests"])
-
-MAX_REATTEMPTS = 5  # max reattempts per test (6 total including first)
-
-
-# ── Schemas ───────────────────────────────────────────────────────
-
-
-class AnswerSubmit(BaseModel):
-    question_id: int
-    selected_answer: Optional[str] = None
-    time_spent_seconds: int = 0
-
-
-class BulkAnswerSubmit(BaseModel):
-    answers: List[AnswerSubmit]
-
-
-class ViolationUpdate(BaseModel):
-    tab_violations: Optional[int] = None
-    fullscreen_violations: Optional[int] = None
-
-
-def _submitted_attempts_query(db: Session, user_id: int, test_id: int):
-    return db.query(TestAttempt).filter(
-        TestAttempt.user_id == user_id,
-        TestAttempt.test_id == test_id,
-        TestAttempt.status == TestStatus.submitted,
-    )
-
-
-def _first_submitted_attempt(
-    db: Session, user_id: int, test_id: int
-) -> Optional[TestAttempt]:
-    return (
-        _submitted_attempts_query(db, user_id, test_id)
-        .order_by(TestAttempt.id.asc())
-        .first()
-    )
-
-
-def _practice_counter(
-    db: Session, user_id: int, test_id: int
-) -> Optional[PracticeAttemptCounter]:
-    return (
-        db.query(PracticeAttemptCounter)
-        .filter(
-            PracticeAttemptCounter.user_id == user_id,
-            PracticeAttemptCounter.test_id == test_id,
-        )
-        .first()
-    )
-
-
-def _attempt_progress(
-    db: Session,
-    user_id: int,
-    test_id: int,
-    submitted_count: Optional[int] = None,
-) -> dict:
-    if submitted_count is None:
-        submitted_count = _submitted_attempts_query(db, user_id, test_id).count()
-
-    counter = _practice_counter(db, user_id, test_id)
-    stored_practice_count = counter.count if counter else 0
-    legacy_practice_count = max(submitted_count - 1, 0)
-    practice_count = max(stored_practice_count, legacy_practice_count)
-
-    return {
-        "submitted_count": submitted_count,
-        "practice_count": practice_count,
-        "total_used": (1 if submitted_count > 0 else 0) + practice_count,
-    }
-
-
-def _set_practice_count(
-    db: Session, user_id: int, test_id: int, count: int
-) -> PracticeAttemptCounter:
-    counter = _practice_counter(db, user_id, test_id)
-    if not counter:
-        counter = PracticeAttemptCounter(user_id=user_id, test_id=test_id, count=0)
-        db.add(counter)
-
-    counter.count = max(counter.count or 0, count)
-    counter.updated_at = datetime.now(timezone.utc)
-    return counter
-
-
-def _has_previous_submission(
-    db: Session, user_id: int, test_id: int, attempt_id: int
-) -> bool:
-    return (
-        db.query(TestAttempt.id)
-        .filter(
-            TestAttempt.user_id == user_id,
-            TestAttempt.test_id == test_id,
-            TestAttempt.status == TestStatus.submitted,
-            TestAttempt.id != attempt_id,
-        )
-        .first()
-        is not None
-    )
-
-
-# ── Tests ─────────────────────────────────────────────────────────
-
-
-@router.get("")
-def list_tests(db: Session = Depends(get_db), _=Depends(require_aspirant)):
-    results = (
-        db.query(Test, func.count(Question.id).label("question_count"))
-        .outerjoin(Question, Test.id == Question.test_id)
-        .filter(Test.is_published.is_(True))
-        .group_by(Test.id)
-        .order_by(Test.created_at.desc())
-        .all()
-    )
-    return [
-        {
-            "id": test.id,
-            "title": test.title,
-            "description": test.description,
-            "duration_minutes": test.duration_minutes,
-            "total_marks": test.total_marks,
-            "question_count": question_count,
-            "series_id": test.series_id,
-            "created_at": test.created_at,
-            "category": test.category,
-            "series_name": test.series_name,
-            "test_type": test.test_type,
-            "subject": test.subject,
-        }
-        for test, question_count in results
-    ]
-
-
-@router.get("/{test_id}")
-def get_test(test_id: int, db: Session = Depends(get_db), _=Depends(require_aspirant)):
-    test = db.query(Test).filter(Test.id == test_id).first()
-    if not test:
-        raise HTTPException(status_code=404, detail="Test not found")
-    question_count = db.query(Question).filter(Question.test_id == test_id).count()
-    return {
-        "id": test.id,
-        "title": test.title,
-        "description": test.description,
-        "duration_minutes": test.duration_minutes,
-        "total_marks": test.total_marks,
-        "question_count": question_count,
-        "series_id": test.series_id,
-        "created_at": test.created_at,
-    }
-
+from app.api.routes.tests.helpers import (
+    MAX_REATTEMPTS,
+    _attempt_out,
+    _attempt_progress,
+    _has_previous_submission,
+    _set_practice_count,
+    _submitted_attempts_query,
+)
+from app.api.routes.tests.schemas import BulkAnswerSubmit, ViolationUpdate
+from app.api.routes.tests.result_builder import _evaluate_submission, _result_payload
 
-# ── Attempt ───────────────────────────────────────────────────────
+router = APIRouter()
 
 
 def _enforce_reattempt_limit(progress: dict) -> None:
@@ -256,20 +100,6 @@ def start_test(
     }
 
 
-def _attempt_out(a: TestAttempt) -> dict:
-    return {
-        "id": a.id,
-        "test_id": a.test_id,
-        "status": a.status,
-        "started_at": a.started_at,
-        "submitted_at": a.submitted_at,
-        "score": a.score,
-        "total_marks": a.total_marks,
-        "tab_violations": a.tab_violations,
-        "fullscreen_violations": a.fullscreen_violations,
-    }
-
-
 @router.get("/{test_id}/attempt/{attempt_id}/questions")
 def get_questions(
     test_id: int,
@@ -388,154 +218,6 @@ def save_answers(
     return {"message": "Saved"}
 
 
-def _percentage(
-    score: Optional[float], total_marks: Optional[float], digits: int = 2
-) -> float:
-    return round((score or 0) / total_marks * 100, digits) if total_marks else 0
-
-
-def _base_answer_detail(
-    q: Question, selected_answer, is_correct, marks_awarded, time_spent_seconds
-):
-    return {
-        "question_id": q.id,
-        "question_text": q.question_text,
-        "question_type": q.question_type,
-        "question_image_url": optimize_delivery_image_url(q.question_image_url),
-        "options": q.options,
-        "option_images": optimize_delivery_image_urls(q.option_images),
-        "correct_answer": q.correct_answer,
-        "selected_answer": selected_answer,
-        "is_correct": is_correct,
-        "marks_awarded": marks_awarded,
-        "marks": q.marks,
-        "negative_marks": q.negative_marks,
-        "time_spent_seconds": time_spent_seconds,
-        "topper_answer": None,
-        "topper_time_seconds": 0,
-    }
-
-
-def _evaluate_submission(test: Test, answers: List[AnswerSubmit]):
-    submitted_by_question = {ans.question_id: ans for ans in answers}
-    earned = 0.0
-    details = []
-
-    # Test.questions relationship declares order_by="Question.order_index",
-    # so the lazy-load already returns questions sorted by order_index.
-    for q in test.questions:
-        submitted = submitted_by_question.get(q.id)
-        selected = submitted.selected_answer if submitted else None
-        if selected is not None and selected.strip() == "":
-            selected = None
-        is_correct, marks = evaluate_answer(q, selected)
-        earned += marks
-        details.append(
-            _base_answer_detail(
-                q,
-                selected,
-                is_correct,
-                marks,
-                submitted.time_spent_seconds if submitted else 0,
-            )
-        )
-
-    total_marks = sum(q.marks for q in test.questions)
-    return max(0.0, round(earned, 2)), total_marks, details
-
-
-def _attempt_answer_details(test: Test, answers: list[UserAnswer]):
-    answers_map = {a.question_id: a for a in answers}
-    details = []
-    for q in sorted(test.questions, key=lambda x: x.order_index):
-        ua = answers_map.get(q.id)
-        details.append(
-            _base_answer_detail(
-                q,
-                ua.selected_answer if ua else None,
-                ua.is_correct if ua else None,
-                ua.marks_awarded if ua else 0,
-                ua.time_spent_seconds if ua else 0,
-            )
-        )
-    return details
-
-
-def _result_payload(
-    *,
-    attempt_id,
-    attempt_number: int,
-    attempts_remaining: int,
-    counts_for_leaderboard: bool,
-    persisted: bool,
-    test: Test,
-    score: float,
-    total_marks: float,
-    submitted_at: Optional[datetime],
-    tab_violations: int,
-    answer_details: list[dict],
-    current_user,
-    db: Session,
-    client_result_id: Optional[str] = None,
-) -> dict:
-    first_attempts = _get_first_attempts(test.id, db)
-
-    topper = first_attempts[0] if first_attempts else None
-    topper_data = None
-    topper_answers_map = {}
-    if topper:
-        topper_answers_map = {ua.question_id: ua for ua in topper.answers}
-        topper_data = {
-            "user_id": topper.user_id,
-            "full_name": topper.user.full_name,
-            "score": topper.score,
-            "total_marks": topper.total_marks,
-            "percentage": _percentage(topper.score, topper.total_marks),
-        }
-
-    correct = 0
-    incorrect = 0
-    skipped = 0
-    for detail in answer_details:
-        topper_ua = topper_answers_map.get(detail["question_id"])
-        detail["topper_answer"] = topper_ua.selected_answer if topper_ua else None
-        detail["topper_time_seconds"] = topper_ua.time_spent_seconds if topper_ua else 0
-        if detail["is_correct"] is True:
-            correct += 1
-        elif detail["is_correct"] is False:
-            incorrect += 1
-        else:
-            skipped += 1
-    rank = next(
-        (i + 1 for i, a in enumerate(first_attempts) if a.user_id == current_user.id),
-        None,
-    )
-
-    return {
-        "attempt_id": attempt_id,
-        "client_result_id": client_result_id,
-        "attempt_number": attempt_number,
-        "attempts_remaining": attempts_remaining,
-        "max_attempts": MAX_REATTEMPTS + 1,
-        "counts_for_leaderboard": counts_for_leaderboard,
-        "persisted": persisted,
-        "test_id": test.id,
-        "test_title": test.title,
-        "score": score,
-        "total_marks": total_marks,
-        "percentage": _percentage(score, total_marks),
-        "correct": correct,
-        "incorrect": incorrect,
-        "skipped": skipped,
-        "submitted_at": submitted_at,
-        "tab_violations": tab_violations,
-        "rank": rank,
-        "total_participants": len(first_attempts),
-        "topper": topper_data,
-        "answers": answer_details,
-    }
-
-
 @router.post("/{test_id}/attempt/{attempt_id}/submit")
 def submit_test(
     test_id: int,
@@ -623,256 +305,3 @@ def submit_test(
     db.commit()
 
     return {"id": client_result_id, "persisted": False, "result": result}
-
-
-# ── Result ────────────────────────────────────────────────────────
-
-
-@router.get("/attempt/{attempt_id}/result")
-def get_result(
-    attempt_id: int,
-    db: Session = Depends(get_db),
-    current_user=Depends(get_current_user),
-):
-    attempt = (
-        db.query(TestAttempt)
-        .filter(TestAttempt.id == attempt_id, TestAttempt.user_id == current_user.id)
-        .first()
-    )
-    if not attempt or attempt.status != TestStatus.submitted:
-        raise HTTPException(status_code=404, detail="Result not found")
-
-    test = attempt.test
-    all_user_attempts = (
-        _submitted_attempts_query(db, current_user.id, test.id)
-        .order_by(TestAttempt.id.asc())
-        .all()
-    )
-    attempt_number = next(
-        (i + 1 for i, a in enumerate(all_user_attempts) if a.id == attempt_id), 1
-    )
-    is_first = attempt_number == 1
-
-    progress = _attempt_progress(db, current_user.id, test.id, len(all_user_attempts))
-    total_marks = attempt.total_marks or sum(q.marks for q in test.questions)
-
-    return _result_payload(
-        attempt_id=attempt_id,
-        client_result_id=None,
-        attempt_number=attempt_number,
-        attempts_remaining=max(0, MAX_REATTEMPTS + 1 - progress["total_used"]),
-        counts_for_leaderboard=is_first,
-        persisted=True,
-        test=test,
-        score=attempt.score or 0,
-        total_marks=total_marks,
-        submitted_at=attempt.submitted_at,
-        tab_violations=attempt.tab_violations,
-        answer_details=_attempt_answer_details(test, attempt.answers),
-        current_user=current_user,
-        db=db,
-    )
-
-
-def _get_first_attempts(test_id: int, db: Session) -> list[TestAttempt]:
-    if db.get_bind().dialect.name == "postgresql":
-        return _first_attempts_pg(test_id, db)
-    return _first_attempts_generic(test_id, db)
-
-
-def _first_attempts_pg(test_id: int, db: Session) -> list[TestAttempt]:
-    """Fetch first attempts using PostgreSQL DISTINCT ON."""
-    first_attempts_subquery = (
-        db.query(TestAttempt)
-        .filter(
-            TestAttempt.test_id == test_id,
-            TestAttempt.status == TestStatus.submitted,
-        )
-        .distinct(TestAttempt.user_id)
-        .order_by(TestAttempt.user_id, TestAttempt.id.asc())
-        .subquery()
-    )
-
-    FirstAttempt = aliased(TestAttempt, first_attempts_subquery)
-
-    return (
-        db.query(FirstAttempt)
-        .options(joinedload(FirstAttempt.user))
-        .order_by(
-            FirstAttempt.score.desc(),
-            FirstAttempt.id.asc(),
-        )
-        .all()
-    )
-
-
-def _first_attempts_generic(test_id: int, db: Session) -> list[TestAttempt]:
-    """Fetch first attempts using ROW_NUMBER() for non-PostgreSQL dialects."""
-    first_attempt_ids_subquery = (
-        db.query(
-            TestAttempt.id.label("attempt_id"),
-            func.row_number()
-            .over(
-                partition_by=TestAttempt.user_id,
-                order_by=TestAttempt.id.asc(),
-            )
-            .label("attempt_rank"),
-        )
-        .filter(
-            TestAttempt.test_id == test_id,
-            TestAttempt.status == TestStatus.submitted,
-        )
-        .subquery()
-    )
-
-    return (
-        db.query(TestAttempt)
-        .options(joinedload(TestAttempt.user))
-        .join(
-            first_attempt_ids_subquery,
-            TestAttempt.id == first_attempt_ids_subquery.c.attempt_id,
-        )
-        .filter(first_attempt_ids_subquery.c.attempt_rank == 1)
-        .order_by(
-            TestAttempt.score.desc(),
-            TestAttempt.id.asc(),
-        )
-        .all()
-    )
-
-
-# ── Leaderboard ───────────────────────────────────────────────────
-
-
-def _leaderboard_cache_key_builder(
-    func, namespace="", *, request=None, response=None, args=None, kwargs=None
-):
-    args = args or ()
-    kwargs = kwargs or {}
-
-    test_id = kwargs.get("test_id")
-    current_user = kwargs.get("current_user")
-
-    if test_id is None and request is not None:
-        test_id = request.path_params.get("test_id")
-    if test_id is None and args:
-        test_id = args[0]
-    if current_user is None and len(args) >= 3:
-        current_user = args[2]
-
-    user_id = getattr(current_user, "id", "unknown")
-    return (
-        f"{namespace}:{func.__module__}:{func.__name__}:test:{test_id}:user:{user_id}"
-    )
-
-
-@router.get("/{test_id}/leaderboard")
-@cache(expire=60, key_builder=_leaderboard_cache_key_builder)
-def get_leaderboard(
-    test_id: int, db: Session = Depends(get_db), current_user=Depends(require_aspirant)
-):
-    test = db.query(Test).filter(Test.id == test_id).first()
-    if not test:
-        raise HTTPException(status_code=404, detail="Test not found")
-
-    first_attempts = _get_first_attempts(test_id, db)
-
-    leaderboard = []
-    current_user_rank = None
-    for rank, attempt in enumerate(first_attempts, 1):
-        score = attempt.score or 0.0
-        pct = round(score / attempt.total_marks * 100, 1) if attempt.total_marks else 0
-        if attempt.user_id == current_user.id:
-            current_user_rank = rank
-        leaderboard.append(
-            {
-                "rank": rank,
-                "user_id": attempt.user_id,
-                "full_name": attempt.user.full_name,
-                "score": attempt.score,
-                "total_marks": attempt.total_marks,
-                "percentage": pct,
-                "submitted_at": attempt.submitted_at,
-                "tab_violations": attempt.tab_violations,
-                "is_current_user": attempt.user_id == current_user.id,
-            }
-        )
-
-    return {
-        "test_id": test_id,
-        "test_title": test.title,
-        "total_participants": len(leaderboard),
-        "current_user_rank": current_user_rank,
-        "leaderboard": leaderboard,
-    }
-
-
-# ── My attempts ───────────────────────────────────────────────────
-
-
-@router.get("/{test_id}/my-attempts")
-def my_attempts(
-    test_id: int, db: Session = Depends(get_db), current_user=Depends(get_current_user)
-):
-    first_attempt = _first_submitted_attempt(db, current_user.id, test_id)
-    if not first_attempt:
-        return []
-
-    submitted_count = _submitted_attempts_query(db, current_user.id, test_id).count()
-    progress = _attempt_progress(db, current_user.id, test_id, submitted_count)
-
-    return [
-        {
-            "attempt_id": first_attempt.id,
-            "attempt_number": 1,
-            "counts_for_leaderboard": True,
-            "is_first": True,
-            "status": first_attempt.status,
-            "score": first_attempt.score,
-            "total_marks": first_attempt.total_marks,
-            "percentage": _percentage(
-                first_attempt.score, first_attempt.total_marks, 1
-            ),
-            "started_at": first_attempt.started_at,
-            "submitted_at": first_attempt.submitted_at,
-            "attempts_remaining": max(0, MAX_REATTEMPTS + 1 - progress["total_used"]),
-        }
-    ]
-
-
-@router.get("/my/history")
-def my_history(db: Session = Depends(get_db), current_user=Depends(get_current_user)):
-    attempts = (
-        db.query(TestAttempt)
-        .filter(
-            TestAttempt.user_id == current_user.id,
-            TestAttempt.status == TestStatus.submitted,
-        )
-        .order_by(TestAttempt.id.asc())
-        .all()
-    )
-
-    first_by_test: dict[int, TestAttempt] = {}
-    for attempt in attempts:
-        first_by_test.setdefault(attempt.test_id, attempt)
-
-    first_attempts = sorted(
-        first_by_test.values(),
-        key=lambda a: a.started_at or a.submitted_at or datetime.min,
-        reverse=True,
-    )
-
-    return [
-        {
-            "id": a.id,
-            "test_id": a.test_id,
-            "status": a.status,
-            "started_at": a.started_at,
-            "submitted_at": a.submitted_at,
-            "score": a.score,
-            "total_marks": a.total_marks,
-            "tab_violations": a.tab_violations,
-            "fullscreen_violations": a.fullscreen_violations,
-        }
-        for a in first_attempts
-    ]

backend/app/api/routes/tests/catalog.py

@@ -0,0 +1,56 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.api.deps import require_aspirant
+from app.core.database import get_db
+from app.models.models import Question, Test
+
+router = APIRouter()
+
+
+@router.get("/")
+def list_tests(db: Session = Depends(get_db), _=Depends(require_aspirant)):
+    results = (
+        db.query(Test, func.count(Question.id).label("question_count"))
+        .outerjoin(Question, Test.id == Question.test_id)
+        .filter(Test.is_published.is_(True))
+        .group_by(Test.id)
+        .order_by(Test.created_at.desc())
+        .all()
+    )
+    return [
+        {
+            "id": test.id,
+            "title": test.title,
+            "description": test.description,
+            "duration_minutes": test.duration_minutes,
+            "total_marks": test.total_marks,
+            "question_count": question_count,
+            "series_id": test.series_id,
+            "created_at": test.created_at,
+            "category": test.category,
+            "series_name": test.series_name,
+            "test_type": test.test_type,
+            "subject": test.subject,
+        }
+        for test, question_count in results
+    ]
+
+
+@router.get("/{test_id}")
+def get_test(test_id: int, db: Session = Depends(get_db), _=Depends(require_aspirant)):
+    test = db.query(Test).filter(Test.id == test_id).first()
+    if not test:
+        raise HTTPException(status_code=404, detail="Test not found")
+    question_count = db.query(Question).filter(Question.test_id == test_id).count()
+    return {
+        "id": test.id,
+        "title": test.title,
+        "description": test.description,
+        "duration_minutes": test.duration_minutes,
+        "total_marks": test.total_marks,
+        "question_count": question_count,
+        "series_id": test.series_id,
+        "created_at": test.created_at,
+    }

backend/app/api/routes/tests/helpers.py

@@ -0,0 +1,181 @@
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session, aliased, joinedload
+
+from app.models.models import (
+    PracticeAttemptCounter,
+    TestAttempt,
+    TestStatus,
+)
+
+MAX_REATTEMPTS = 5  # max reattempts per test (6 total including first)
+
+
+def _submitted_attempts_query(db: Session, user_id: int, test_id: int):
+    return db.query(TestAttempt).filter(
+        TestAttempt.user_id == user_id,
+        TestAttempt.test_id == test_id,
+        TestAttempt.status == TestStatus.submitted,
+    )
+
+
+def _first_submitted_attempt(
+    db: Session, user_id: int, test_id: int
+) -> Optional[TestAttempt]:
+    return (
+        _submitted_attempts_query(db, user_id, test_id)
+        .order_by(TestAttempt.id.asc())
+        .first()
+    )
+
+
+def _practice_counter(
+    db: Session, user_id: int, test_id: int
+) -> Optional[PracticeAttemptCounter]:
+    return (
+        db.query(PracticeAttemptCounter)
+        .filter(
+            PracticeAttemptCounter.user_id == user_id,
+            PracticeAttemptCounter.test_id == test_id,
+        )
+        .first()
+    )
+
+
+def _attempt_progress(
+    db: Session,
+    user_id: int,
+    test_id: int,
+    submitted_count: Optional[int] = None,
+) -> dict:
+    if submitted_count is None:
+        submitted_count = _submitted_attempts_query(db, user_id, test_id).count()
+
+    counter = _practice_counter(db, user_id, test_id)
+    stored_practice_count = counter.count if counter else 0
+    legacy_practice_count = max(submitted_count - 1, 0)
+    practice_count = max(stored_practice_count, legacy_practice_count)
+
+    return {
+        "submitted_count": submitted_count,
+        "practice_count": practice_count,
+        "total_used": (1 if submitted_count > 0 else 0) + practice_count,
+    }
+
+
+def _set_practice_count(
+    db: Session, user_id: int, test_id: int, count: int
+) -> PracticeAttemptCounter:
+    counter = _practice_counter(db, user_id, test_id)
+    if not counter:
+        counter = PracticeAttemptCounter(user_id=user_id, test_id=test_id, count=0)
+        db.add(counter)
+
+    counter.count = max(counter.count or 0, count)
+    counter.updated_at = datetime.now(timezone.utc)
+    return counter
+
+
+def _has_previous_submission(
+    db: Session, user_id: int, test_id: int, attempt_id: int
+) -> bool:
+    return (
+        db.query(TestAttempt.id)
+        .filter(
+            TestAttempt.user_id == user_id,
+            TestAttempt.test_id == test_id,
+            TestAttempt.status == TestStatus.submitted,
+            TestAttempt.id != attempt_id,
+        )
+        .first()
+        is not None
+    )
+
+
+def _attempt_out(a: TestAttempt) -> dict:
+    return {
+        "id": a.id,
+        "test_id": a.test_id,
+        "status": a.status,
+        "started_at": a.started_at,
+        "submitted_at": a.submitted_at,
+        "score": a.score,
+        "total_marks": a.total_marks,
+        "tab_violations": a.tab_violations,
+        "fullscreen_violations": a.fullscreen_violations,
+    }
+
+
+def _percentage(
+    score: Optional[float], total_marks: Optional[float], digits: int = 2
+) -> float:
+    return round((score or 0) / total_marks * 100, digits) if total_marks else 0
+
+
+def _get_first_attempts(test_id: int, db: Session) -> list[TestAttempt]:
+    if db.get_bind().dialect.name == "postgresql":
+        return _first_attempts_pg(test_id, db)
+    return _first_attempts_generic(test_id, db)
+
+
+def _first_attempts_pg(test_id: int, db: Session) -> list[TestAttempt]:
+    """Fetch first attempts using PostgreSQL DISTINCT ON."""
+    first_attempts_subquery = (
+        db.query(TestAttempt)
+        .filter(
+            TestAttempt.test_id == test_id,
+            TestAttempt.status == TestStatus.submitted,
+        )
+        .distinct(TestAttempt.user_id)
+        .order_by(TestAttempt.user_id, TestAttempt.id.asc())
+        .subquery()
+    )
+
+    FirstAttempt = aliased(TestAttempt, first_attempts_subquery)
+
+    return (
+        db.query(FirstAttempt)
+        .options(joinedload(FirstAttempt.user))
+        .order_by(
+            FirstAttempt.score.desc(),
+            FirstAttempt.id.asc(),
+        )
+        .all()
+    )
+
+
+def _first_attempts_generic(test_id: int, db: Session) -> list[TestAttempt]:
+    """Fetch first attempts using ROW_NUMBER() for non-PostgreSQL dialects."""
+    first_attempt_ids_subquery = (
+        db.query(
+            TestAttempt.id.label("attempt_id"),
+            func.row_number()
+            .over(
+                partition_by=TestAttempt.user_id,
+                order_by=TestAttempt.id.asc(),
+            )
+            .label("attempt_rank"),
+        )
+        .filter(
+            TestAttempt.test_id == test_id,
+            TestAttempt.status == TestStatus.submitted,
+        )
+        .subquery()
+    )
+
+    return (
+        db.query(TestAttempt)
+        .options(joinedload(TestAttempt.user))
+        .join(
+            first_attempt_ids_subquery,
+            TestAttempt.id == first_attempt_ids_subquery.c.attempt_id,
+        )
+        .filter(first_attempt_ids_subquery.c.attempt_rank == 1)
+        .order_by(
+            TestAttempt.score.desc(),
+            TestAttempt.id.asc(),
+        )
+        .all()
+    )

backend/app/api/routes/tests/history.py

@@ -0,0 +1,84 @@
+from datetime import datetime
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_current_user
+from app.core.database import get_db
+from app.models.models import TestAttempt, TestStatus
+from app.api.routes.tests.helpers import (
+    MAX_REATTEMPTS,
+    _attempt_progress,
+    _first_submitted_attempt,
+    _percentage,
+    _submitted_attempts_query,
+)
+
+router = APIRouter()
+
+
+@router.get("/{test_id}/my-attempts")
+def my_attempts(
+    test_id: int, db: Session = Depends(get_db), current_user=Depends(get_current_user)
+):
+    first_attempt = _first_submitted_attempt(db, current_user.id, test_id)
+    if not first_attempt:
+        return []
+
+    submitted_count = _submitted_attempts_query(db, current_user.id, test_id).count()
+    progress = _attempt_progress(db, current_user.id, test_id, submitted_count)
+
+    return [
+        {
+            "attempt_id": first_attempt.id,
+            "attempt_number": 1,
+            "counts_for_leaderboard": True,
+            "is_first": True,
+            "status": first_attempt.status,
+            "score": first_attempt.score,
+            "total_marks": first_attempt.total_marks,
+            "percentage": _percentage(
+                first_attempt.score, first_attempt.total_marks, 1
+            ),
+            "started_at": first_attempt.started_at,
+            "submitted_at": first_attempt.submitted_at,
+            "attempts_remaining": max(0, MAX_REATTEMPTS + 1 - progress["total_used"]),
+        }
+    ]
+
+
+@router.get("/my/history")
+def my_history(db: Session = Depends(get_db), current_user=Depends(get_current_user)):
+    attempts = (
+        db.query(TestAttempt)
+        .filter(
+            TestAttempt.user_id == current_user.id,
+            TestAttempt.status == TestStatus.submitted,
+        )
+        .order_by(TestAttempt.id.asc())
+        .all()
+    )
+
+    first_by_test: dict[int, TestAttempt] = {}
+    for attempt in attempts:
+        first_by_test.setdefault(attempt.test_id, attempt)
+
+    first_attempts = sorted(
+        first_by_test.values(),
+        key=lambda a: a.started_at or a.submitted_at or datetime.min,
+        reverse=True,
+    )
+
+    return [
+        {
+            "id": a.id,
+            "test_id": a.test_id,
+            "status": a.status,
+            "started_at": a.started_at,
+            "submitted_at": a.submitted_at,
+            "score": a.score,
+            "total_marks": a.total_marks,
+            "tab_violations": a.tab_violations,
+            "fullscreen_violations": a.fullscreen_violations,
+        }
+        for a in first_attempts
+    ]

backend/app/api/routes/tests/leaderboard.py

@@ -0,0 +1,73 @@
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi_cache.decorator import cache
+from sqlalchemy.orm import Session
+
+from app.api.deps import require_aspirant
+from app.core.database import get_db
+from app.models.models import Test
+from app.api.routes.tests.helpers import _get_first_attempts
+
+router = APIRouter()
+
+
+def _leaderboard_cache_key_builder(
+    func, namespace="", *, request=None, response=None, args=None, kwargs=None
+):
+    args = args or ()
+    kwargs = kwargs or {}
+
+    test_id = kwargs.get("test_id")
+    current_user = kwargs.get("current_user")
+
+    if test_id is None and request is not None:
+        test_id = request.path_params.get("test_id")
+    if test_id is None and args:
+        test_id = args[0]
+    if current_user is None and len(args) >= 3:
+        current_user = args[2]
+
+    user_id = getattr(current_user, "id", "unknown")
+    return (
+        f"{namespace}:{func.__module__}:{func.__name__}:test:{test_id}:user:{user_id}"
+    )
+
+
+@router.get("/{test_id}/leaderboard")
+@cache(expire=60, key_builder=_leaderboard_cache_key_builder)
+def get_leaderboard(
+    test_id: int, db: Session = Depends(get_db), current_user=Depends(require_aspirant)
+):
+    test = db.query(Test).filter(Test.id == test_id).first()
+    if not test:
+        raise HTTPException(status_code=404, detail="Test not found")
+
+    first_attempts = _get_first_attempts(test_id, db)
+
+    leaderboard = []
+    current_user_rank = None
+    for rank, attempt in enumerate(first_attempts, 1):
+        score = attempt.score or 0.0
+        pct = round(score / attempt.total_marks * 100, 1) if attempt.total_marks else 0
+        if attempt.user_id == current_user.id:
+            current_user_rank = rank
+        leaderboard.append(
+            {
+                "rank": rank,
+                "user_id": attempt.user_id,
+                "full_name": attempt.user.full_name,
+                "score": attempt.score,
+                "total_marks": attempt.total_marks,
+                "percentage": pct,
+                "submitted_at": attempt.submitted_at,
+                "tab_violations": attempt.tab_violations,
+                "is_current_user": attempt.user_id == current_user.id,
+            }
+        )
+
+    return {
+        "test_id": test_id,
+        "test_title": test.title,
+        "total_participants": len(leaderboard),
+        "current_user_rank": current_user_rank,
+        "leaderboard": leaderboard,
+    }

backend/app/api/routes/tests/result_builder.py

@@ -0,0 +1,159 @@
+from datetime import datetime
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.models import Question, Test, UserAnswer
+from app.services.cloudinary_service import (
+    optimize_delivery_image_url,
+    optimize_delivery_image_urls,
+)
+from app.services.scoring import evaluate_answer
+from app.api.routes.tests.helpers import (
+    MAX_REATTEMPTS,
+    _get_first_attempts,
+    _percentage,
+)
+from app.api.routes.tests.schemas import AnswerSubmit
+
+
+def _base_answer_detail(
+    q: Question, selected_answer, is_correct, marks_awarded, time_spent_seconds
+):
+    return {
+        "question_id": q.id,
+        "question_text": q.question_text,
+        "question_type": q.question_type,
+        "question_image_url": optimize_delivery_image_url(q.question_image_url),
+        "options": q.options,
+        "option_images": optimize_delivery_image_urls(q.option_images),
+        "correct_answer": q.correct_answer,
+        "selected_answer": selected_answer,
+        "is_correct": is_correct,
+        "marks_awarded": marks_awarded,
+        "marks": q.marks,
+        "negative_marks": q.negative_marks,
+        "time_spent_seconds": time_spent_seconds,
+        "topper_answer": None,
+        "topper_time_seconds": 0,
+    }
+
+
+def _evaluate_submission(test: Test, answers: List[AnswerSubmit]):
+    submitted_by_question = {ans.question_id: ans for ans in answers}
+    earned = 0.0
+    details = []
+
+    # Test.questions relationship declares order_by="Question.order_index",
+    # so the lazy-load already returns questions sorted by order_index.
+    for q in test.questions:
+        submitted = submitted_by_question.get(q.id)
+        selected = submitted.selected_answer if submitted else None
+        if selected is not None and selected.strip() == "":
+            selected = None
+        is_correct, marks = evaluate_answer(q, selected)
+        earned += marks
+        details.append(
+            _base_answer_detail(
+                q,
+                selected,
+                is_correct,
+                marks,
+                submitted.time_spent_seconds if submitted else 0,
+            )
+        )
+
+    total_marks = sum(q.marks for q in test.questions)
+    return max(0.0, round(earned, 2)), total_marks, details
+
+
+def _attempt_answer_details(test: Test, answers: list[UserAnswer]):
+    answers_map = {a.question_id: a for a in answers}
+    details = []
+    for q in sorted(test.questions, key=lambda x: x.order_index):
+        ua = answers_map.get(q.id)
+        details.append(
+            _base_answer_detail(
+                q,
+                ua.selected_answer if ua else None,
+                ua.is_correct if ua else None,
+                ua.marks_awarded if ua else 0,
+                ua.time_spent_seconds if ua else 0,
+            )
+        )
+    return details
+
+
+def _result_payload(
+    *,
+    attempt_id,
+    attempt_number: int,
+    attempts_remaining: int,
+    counts_for_leaderboard: bool,
+    persisted: bool,
+    test: Test,
+    score: float,
+    total_marks: float,
+    submitted_at: Optional[datetime],
+    tab_violations: int,
+    answer_details: list[dict],
+    current_user,
+    db: Session,
+    client_result_id: Optional[str] = None,
+) -> dict:
+    first_attempts = _get_first_attempts(test.id, db)
+
+    topper = first_attempts[0] if first_attempts else None
+    topper_data = None
+    topper_answers_map = {}
+    if topper:
+        topper_answers_map = {ua.question_id: ua for ua in topper.answers}
+        topper_data = {
+            "user_id": topper.user_id,
+            "full_name": topper.user.full_name,
+            "score": topper.score,
+            "total_marks": topper.total_marks,
+            "percentage": _percentage(topper.score, topper.total_marks),
+        }
+
+    correct = 0
+    incorrect = 0
+    skipped = 0
+    for detail in answer_details:
+        topper_ua = topper_answers_map.get(detail["question_id"])
+        detail["topper_answer"] = topper_ua.selected_answer if topper_ua else None
+        detail["topper_time_seconds"] = topper_ua.time_spent_seconds if topper_ua else 0
+        if detail["is_correct"] is True:
+            correct += 1
+        elif detail["is_correct"] is False:
+            incorrect += 1
+        else:
+            skipped += 1
+    rank = next(
+        (i + 1 for i, a in enumerate(first_attempts) if a.user_id == current_user.id),
+        None,
+    )
+
+    return {
+        "attempt_id": attempt_id,
+        "client_result_id": client_result_id,
+        "attempt_number": attempt_number,
+        "attempts_remaining": attempts_remaining,
+        "max_attempts": MAX_REATTEMPTS + 1,
+        "counts_for_leaderboard": counts_for_leaderboard,
+        "persisted": persisted,
+        "test_id": test.id,
+        "test_title": test.title,
+        "score": score,
+        "total_marks": total_marks,
+        "percentage": _percentage(score, total_marks),
+        "correct": correct,
+        "incorrect": incorrect,
+        "skipped": skipped,
+        "submitted_at": submitted_at,
+        "tab_violations": tab_violations,
+        "rank": rank,
+        "total_participants": len(first_attempts),
+        "topper": topper_data,
+        "answers": answer_details,
+    }

backend/app/api/routes/tests/results.py

@@ -0,0 +1,60 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_current_user
+from app.core.database import get_db
+from app.models.models import TestAttempt, TestStatus
+from app.api.routes.tests.helpers import (
+    MAX_REATTEMPTS,
+    _attempt_progress,
+    _submitted_attempts_query,
+)
+from app.api.routes.tests.result_builder import _attempt_answer_details, _result_payload
+
+router = APIRouter()
+
+
+@router.get("/attempt/{attempt_id}/result")
+def get_result(
+    attempt_id: int,
+    db: Session = Depends(get_db),
+    current_user=Depends(get_current_user),
+):
+    attempt = (
+        db.query(TestAttempt)
+        .filter(TestAttempt.id == attempt_id, TestAttempt.user_id == current_user.id)
+        .first()
+    )
+    if not attempt or attempt.status != TestStatus.submitted:
+        raise HTTPException(status_code=404, detail="Result not found")
+
+    test = attempt.test
+    all_user_attempts = (
+        _submitted_attempts_query(db, current_user.id, test.id)
+        .order_by(TestAttempt.id.asc())
+        .all()
+    )
+    attempt_number = next(
+        (i + 1 for i, a in enumerate(all_user_attempts) if a.id == attempt_id), 1
+    )
+    is_first = attempt_number == 1
+
+    progress = _attempt_progress(db, current_user.id, test.id, len(all_user_attempts))
+    total_marks = attempt.total_marks or sum(q.marks for q in test.questions)
+
+    return _result_payload(
+        attempt_id=attempt_id,
+        client_result_id=None,
+        attempt_number=attempt_number,
+        attempts_remaining=max(0, MAX_REATTEMPTS + 1 - progress["total_used"]),
+        counts_for_leaderboard=is_first,
+        persisted=True,
+        test=test,
+        score=attempt.score or 0,
+        total_marks=total_marks,
+        submitted_at=attempt.submitted_at,
+        tab_violations=attempt.tab_violations,
+        answer_details=_attempt_answer_details(test, attempt.answers),
+        current_user=current_user,
+        db=db,
+    )

backend/app/api/routes/tests/schemas.py

@@ -0,0 +1,18 @@
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class AnswerSubmit(BaseModel):
+    question_id: int
+    selected_answer: Optional[str] = None
+    time_spent_seconds: int = 0
+
+
+class BulkAnswerSubmit(BaseModel):
+    answers: List[AnswerSubmit]
+
+
+class ViolationUpdate(BaseModel):
+    tab_violations: Optional[int] = None
+    fullscreen_violations: Optional[int] = None