feat: Sprint 2 — test taking + results
Backend:
- Models: TestAttempt, AttemptAnswer (migration 002)
- POST /api/attempts: start attempt, shuffle questions/answers,
hide is_correct, expose is_multiple for UI hints
- POST /api/attempts/{id}/submit: save answers, calculate score,
strict matching (selected == correct), return full result
- GET /api/attempts/{id}/result: fetch saved result
- Register attempts router in main.py
Frontend:
- api/attempts.ts: types + API functions
- TestTake page: one question at a time, progress bar, timer
with auto-submit, back navigation controlled by test setting,
radio/checkbox based on is_multiple
- AttemptResult page: score, pass/fail, per-question breakdown
with correct/selected/missed answer highlighting
- App.tsx: add /tests/:testId/take and /attempts/:id/result routes
- TestDetail: add "Пройти тест" button
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,223 @@
|
||||
import random
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.database import get_db
|
||||
from app.models.attempt import AttemptAnswer, TestAttempt
|
||||
from app.models.test import Question, Test
|
||||
from app.schemas.attempt import (
|
||||
AttemptResult,
|
||||
AttemptStart,
|
||||
AttemptStarted,
|
||||
AttemptSubmitDto,
|
||||
AnswerForTest,
|
||||
AnswerResult,
|
||||
QuestionForTest,
|
||||
QuestionResult,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/attempts", tags=["attempts"])
|
||||
|
||||
|
||||
@router.post("", response_model=AttemptStarted, status_code=201)
|
||||
async def start_attempt(data: AttemptStart, db: AsyncSession = Depends(get_db)):
|
||||
"""Начать попытку прохождения теста. Возвращает вопросы в случайном порядке без правильных ответов."""
|
||||
|
||||
result = await db.execute(
|
||||
select(Test)
|
||||
.options(selectinload(Test.questions).selectinload(Question.answers))
|
||||
.where(Test.id == data.test_id, Test.is_active == True)
|
||||
)
|
||||
test = result.scalar_one_or_none()
|
||||
if not test:
|
||||
raise HTTPException(status_code=404, detail="Тест не найден")
|
||||
|
||||
attempt = TestAttempt(test_id=test.id, status="in_progress")
|
||||
db.add(attempt)
|
||||
await db.commit()
|
||||
await db.refresh(attempt)
|
||||
|
||||
# Перемешиваем вопросы и ответы случайно
|
||||
questions_shuffled = list(test.questions)
|
||||
random.shuffle(questions_shuffled)
|
||||
|
||||
questions_for_test = []
|
||||
for q in questions_shuffled:
|
||||
correct_count = sum(1 for a in q.answers if a.is_correct)
|
||||
answers_shuffled = list(q.answers)
|
||||
random.shuffle(answers_shuffled)
|
||||
questions_for_test.append(
|
||||
QuestionForTest(
|
||||
id=q.id,
|
||||
text=q.text,
|
||||
is_multiple=correct_count > 1,
|
||||
answers=[AnswerForTest(id=a.id, text=a.text) for a in answers_shuffled],
|
||||
)
|
||||
)
|
||||
|
||||
return AttemptStarted(
|
||||
id=attempt.id,
|
||||
test_id=test.id,
|
||||
test_title=test.title,
|
||||
test_description=test.description,
|
||||
started_at=attempt.started_at,
|
||||
time_limit=test.time_limit,
|
||||
allow_navigation_back=test.allow_navigation_back,
|
||||
questions=questions_for_test,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{attempt_id}/submit", response_model=AttemptResult)
|
||||
async def submit_attempt(
|
||||
attempt_id: int, data: AttemptSubmitDto, db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""Завершить попытку: сохранить ответы, подсчитать результат."""
|
||||
|
||||
result = await db.execute(
|
||||
select(TestAttempt)
|
||||
.where(TestAttempt.id == attempt_id, TestAttempt.status == "in_progress")
|
||||
)
|
||||
attempt = result.scalar_one_or_none()
|
||||
if not attempt:
|
||||
raise HTTPException(status_code=404, detail="Попытка не найдена или уже завершена")
|
||||
|
||||
# Загружаем тест с вопросами и ответами
|
||||
test_result = await db.execute(
|
||||
select(Test)
|
||||
.options(selectinload(Test.questions).selectinload(Question.answers))
|
||||
.where(Test.id == attempt.test_id)
|
||||
)
|
||||
test = test_result.scalar_one()
|
||||
|
||||
# Сохраняем выбранные ответы
|
||||
submitted = {qa.question_id: set(qa.answer_ids) for qa in data.answers}
|
||||
|
||||
for qa in data.answers:
|
||||
for answer_id in qa.answer_ids:
|
||||
db.add(AttemptAnswer(
|
||||
attempt_id=attempt.id,
|
||||
question_id=qa.question_id,
|
||||
answer_id=answer_id,
|
||||
))
|
||||
|
||||
# Подсчёт результата
|
||||
correct_count = 0
|
||||
question_results = []
|
||||
|
||||
for question in test.questions:
|
||||
correct_ids = {a.id for a in question.answers if a.is_correct}
|
||||
selected_ids = submitted.get(question.id, set())
|
||||
is_correct = selected_ids == correct_ids
|
||||
|
||||
if is_correct:
|
||||
correct_count += 1
|
||||
|
||||
question_results.append(
|
||||
QuestionResult(
|
||||
id=question.id,
|
||||
text=question.text,
|
||||
is_answered_correctly=is_correct,
|
||||
answers=[
|
||||
AnswerResult(
|
||||
id=a.id,
|
||||
text=a.text,
|
||||
is_correct=a.is_correct,
|
||||
is_selected=a.id in selected_ids,
|
||||
)
|
||||
for a in question.answers
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
total = len(test.questions)
|
||||
score = round((correct_count / total) * 100, 1) if total > 0 else 0.0
|
||||
passed = score >= test.passing_score
|
||||
finished_at = datetime.now(timezone.utc)
|
||||
|
||||
attempt.finished_at = finished_at
|
||||
attempt.score = score
|
||||
attempt.passed = passed
|
||||
attempt.correct_count = correct_count
|
||||
attempt.total_count = total
|
||||
attempt.status = "finished"
|
||||
|
||||
await db.commit()
|
||||
|
||||
return AttemptResult(
|
||||
id=attempt.id,
|
||||
test_id=test.id,
|
||||
test_title=test.title,
|
||||
started_at=attempt.started_at,
|
||||
finished_at=finished_at,
|
||||
score=score,
|
||||
passed=passed,
|
||||
passing_score=test.passing_score,
|
||||
correct_count=correct_count,
|
||||
total_count=total,
|
||||
questions=question_results,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{attempt_id}/result", response_model=AttemptResult)
|
||||
async def get_result(attempt_id: int, db: AsyncSession = Depends(get_db)):
|
||||
"""Получить результат завершённой попытки."""
|
||||
|
||||
result = await db.execute(
|
||||
select(TestAttempt)
|
||||
.options(selectinload(TestAttempt.answers).selectinload(AttemptAnswer.answer))
|
||||
.where(TestAttempt.id == attempt_id, TestAttempt.status == "finished")
|
||||
)
|
||||
attempt = result.scalar_one_or_none()
|
||||
if not attempt:
|
||||
raise HTTPException(status_code=404, detail="Результат не найден")
|
||||
|
||||
test_result = await db.execute(
|
||||
select(Test)
|
||||
.options(selectinload(Test.questions).selectinload(Question.answers))
|
||||
.where(Test.id == attempt.test_id)
|
||||
)
|
||||
test = test_result.scalar_one()
|
||||
|
||||
# Восстанавливаем выбранные ответы из БД
|
||||
selected_by_question: dict[int, set[int]] = {}
|
||||
for aa in attempt.answers:
|
||||
selected_by_question.setdefault(aa.question_id, set()).add(aa.answer_id)
|
||||
|
||||
question_results = []
|
||||
for question in test.questions:
|
||||
correct_ids = {a.id for a in question.answers if a.is_correct}
|
||||
selected_ids = selected_by_question.get(question.id, set())
|
||||
question_results.append(
|
||||
QuestionResult(
|
||||
id=question.id,
|
||||
text=question.text,
|
||||
is_answered_correctly=selected_ids == correct_ids,
|
||||
answers=[
|
||||
AnswerResult(
|
||||
id=a.id,
|
||||
text=a.text,
|
||||
is_correct=a.is_correct,
|
||||
is_selected=a.id in selected_ids,
|
||||
)
|
||||
for a in question.answers
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
return AttemptResult(
|
||||
id=attempt.id,
|
||||
test_id=test.id,
|
||||
test_title=test.title,
|
||||
started_at=attempt.started_at,
|
||||
finished_at=attempt.finished_at,
|
||||
score=attempt.score,
|
||||
passed=attempt.passed,
|
||||
passing_score=test.passing_score,
|
||||
correct_count=attempt.correct_count,
|
||||
total_count=attempt.total_count,
|
||||
questions=question_results,
|
||||
)
|
||||
+2
-1
@@ -1,7 +1,7 @@
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.api import tests
|
||||
from app.api import attempts, tests
|
||||
|
||||
app = FastAPI(
|
||||
title="QA Test App API",
|
||||
@@ -19,6 +19,7 @@ app.add_middleware(
|
||||
)
|
||||
|
||||
app.include_router(tests.router)
|
||||
app.include_router(attempts.router)
|
||||
|
||||
|
||||
@app.get("/api/health")
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
from sqlalchemy.sql import func
|
||||
|
||||
from app.database import Base
|
||||
|
||||
|
||||
class TestAttempt(Base):
|
||||
__tablename__ = "test_attempts"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
test_id: Mapped[int] = mapped_column(Integer, ForeignKey("tests.id"), nullable=False)
|
||||
started_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now()
|
||||
)
|
||||
finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
||||
score: Mapped[float | None] = mapped_column(Float, nullable=True) # процент 0–100
|
||||
passed: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
|
||||
correct_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
total_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
# in_progress | finished
|
||||
status: Mapped[str] = mapped_column(String(20), default="in_progress", nullable=False)
|
||||
|
||||
test: Mapped["Test"] = relationship("Test") # type: ignore[name-defined]
|
||||
answers: Mapped[list["AttemptAnswer"]] = relationship(
|
||||
"AttemptAnswer", back_populates="attempt", cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
|
||||
class AttemptAnswer(Base):
|
||||
"""Одна запись = один выбранный вариант ответа сотрудника."""
|
||||
|
||||
__tablename__ = "attempt_answers"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
attempt_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("test_attempts.id"), nullable=False
|
||||
)
|
||||
question_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("questions.id"), nullable=False
|
||||
)
|
||||
answer_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("answers.id"), nullable=False
|
||||
)
|
||||
|
||||
attempt: Mapped["TestAttempt"] = relationship("TestAttempt", back_populates="answers")
|
||||
answer: Mapped["Answer"] = relationship("Answer") # type: ignore[name-defined]
|
||||
@@ -0,0 +1,89 @@
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
# ── Начало попытки ──────────────────────────────────────────
|
||||
|
||||
class AttemptStart(BaseModel):
|
||||
test_id: int
|
||||
|
||||
|
||||
class AnswerForTest(BaseModel):
|
||||
"""Вариант ответа без поля is_correct — не раскрываем правильные ответы."""
|
||||
id: int
|
||||
text: str
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
class QuestionForTest(BaseModel):
|
||||
id: int
|
||||
text: str
|
||||
is_multiple: bool # True = несколько правильных ответов → показываем чекбоксы
|
||||
answers: list[AnswerForTest] # перемешаны случайно
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
class AttemptStarted(BaseModel):
|
||||
"""Возвращается после старта попытки."""
|
||||
id: int
|
||||
test_id: int
|
||||
test_title: str
|
||||
test_description: Optional[str]
|
||||
started_at: datetime
|
||||
time_limit: Optional[int] # минуты, из теста
|
||||
allow_navigation_back: bool # из теста
|
||||
questions: list[QuestionForTest] # перемешаны случайно
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
# ── Сдача попытки ────────────────────────────────────────────
|
||||
|
||||
class QuestionAnswer(BaseModel):
|
||||
"""Ответы сотрудника на один вопрос."""
|
||||
question_id: int
|
||||
answer_ids: list[int] # выбранные варианты (может быть пустым)
|
||||
|
||||
|
||||
class AttemptSubmitDto(BaseModel):
|
||||
answers: list[QuestionAnswer]
|
||||
|
||||
|
||||
# ── Результат ────────────────────────────────────────────────
|
||||
|
||||
class AnswerResult(BaseModel):
|
||||
id: int
|
||||
text: str
|
||||
is_correct: bool # правильный ли ответ вообще
|
||||
is_selected: bool # выбрал ли его сотрудник
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
class QuestionResult(BaseModel):
|
||||
id: int
|
||||
text: str
|
||||
is_answered_correctly: bool # вся комбинация ответов верна
|
||||
answers: list[AnswerResult]
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
class AttemptResult(BaseModel):
|
||||
id: int
|
||||
test_id: int
|
||||
test_title: str
|
||||
started_at: datetime
|
||||
finished_at: datetime
|
||||
score: float # процент правильных ответов
|
||||
passed: bool # преодолён ли порог зачёта
|
||||
passing_score: int # порог из теста
|
||||
correct_count: int
|
||||
total_count: int
|
||||
questions: list[QuestionResult]
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
Reference in New Issue
Block a user