import logging from dataclasses import dataclass from django.db import transaction from django.db.models import Prefetch from django.utils.translation import gettext as _ from django.utils.functional import cached_property from labs.models import ( User as DBUser, Exam as DBExam, ExamVariant as DBExamVariant, ExamAttempt as DBExamAttempt, UserExamRedemption, ) from labs.exams.exceptions import ( ExamRedeemError, ExamStartError, ) from labs.exams.attempt import ExamAttempt, get_in_progress_exam_attempt from labs.billing.exam import redeem_exam_with_credits from labs.billing.exceptions import InsufficientCreditsError logger = logging.getLogger(__name__) @dataclass class CanStartResult: eligible: bool = False requires_redemption: bool = False message: str = "" class Exam: @staticmethod def get_frontend_exam_list(user: DBUser) -> list[dict]: qs = ( DBExam.objects.filter(published=True) .only("id", "code", "credits", "employee_only") .prefetch_related( Prefetch( "variants", queryset=DBExamVariant.objects.only( "exam", *DBExamVariant.TO_FRONTEND_FIELDS ), ) ) ) if not (user.is_authenticated and user.is_pingcap_employee): qs = qs.filter(employee_only=False) return [ { "id": e.id, "code": e.code, "credits": e.credits, "employee_only": e.employee_only, "variants": [v.to_frontend_detail() for v in e.variants.all()], } for e in qs ] def __init__(self, db_exam: DBExam, user: DBUser): self.db_exam = db_exam self.user = user def get_frontend_detail(self) -> dict: result = self.can_start() data = self.db_exam.to_frontend_detail() data.update( { "can_start": result.eligible, "requires_redemption": result.requires_redemption, "start_suggestion": result.message, } ) if self.latest_attempt: data["latest_attempt"] = self.latest_attempt.get_frontend_detail() else: data["latest_attempt"] = None return data @cached_property def latest_attempt(self) -> ExamAttempt | None: db_exam_attempt = ( DBExamAttempt.objects.filter( user=self.user, exam=self.db_exam, ) .order_by("-created_at") .first() ) if db_exam_attempt: return ExamAttempt(db_exam_attempt) return None def _evaluate_start_rules( self, *, lock_redemption: bool = False, ) -> tuple[CanStartResult, UserExamRedemption | None]: """ Evaluate whether the user can start this exam. Args: lock_redemption: If True, lock the redemption record for transactional use. Returns: (CanStartResult, redemption): The eligibility result and the locked redemption record (only when lock_redemption=True and redemption exists). """ result = CanStartResult( eligible=False, requires_redemption=False, message="", ) exam_costs_credits = self.db_exam.credits > 0 # 1. Must be logged in if not self.user.is_authenticated: result.requires_redemption = exam_costs_credits result.message = _("You must be logged in to start an exam.") return result, None # 2. Cannot retake a passed exam if DBExamAttempt.objects.filter( user_id=self.user.id, exam_id=self.db_exam.id, is_passed=True, ).exists(): result.message = _("You have already passed this exam.") return result, None # 3. PingCAP employees bypass employee_only and credit checks if self.user.is_pingcap_employee: active_attempt = get_in_progress_exam_attempt(self.user) if active_attempt: result.message = _( "You already have an active exam in progress, please finish it first." ) return result, None result.eligible = True return result, None # 4. Non-employees cannot start employee-only exams if self.db_exam.employee_only: result.message = _("This exam is for PingCAP employees only.") return result, None active_attempt = get_in_progress_exam_attempt(self.user) # 5. Calculate requires_redemption for paid exams (before active exam check) # This allows user to know if redemption is needed even when another exam is in progress if exam_costs_credits: if active_attempt and active_attempt.db_exam == self.db_exam: result.requires_redemption = False else: has_unused_redemption = UserExamRedemption.objects.filter( user_id=self.user.id, exam_id=self.db_exam.id, exam_attempt__isnull=True, ).exists() result.requires_redemption = not has_unused_redemption # 6. Cannot start if another exam is in progress if active_attempt: result.message = _( "You already have an active exam in progress, please finish it first." ) return result, None # 7. Free exams are eligible if not exam_costs_credits: result.eligible = True return result, None # 8. Paid exams: check redemption status if result.requires_redemption: result.message = _("Redeem with credits to start this exam.") return result, None # 9. Has unused redemption, eligible to start result.eligible = True if lock_redemption: redemption = ( UserExamRedemption.objects.filter( user_id=self.user.id, exam_id=self.db_exam.id, exam_attempt__isnull=True, ) .select_for_update() .order_by("created_at", "id") .only("id") .first() ) if redemption is None: result.eligible = False result.requires_redemption = True result.message = _("Redeem with credits to start this exam.") return result, None return result, redemption return result, None def can_start(self) -> CanStartResult: result, _ = self._evaluate_start_rules() return result def redeem(self) -> None: result = self.can_start() if not result.requires_redemption: if result.eligible: raise ExamRedeemError( _("You can start this exam now without redeeming.") ) raise ExamRedeemError(result.message) try: redeem_exam_with_credits(user=self.user, exam=self.db_exam) except InsufficientCreditsError: raise ExamRedeemError( _("You don't have enough credits to redeem this exam.") ) def start_exam(self, variant: DBExamVariant) -> ExamAttempt: assert self.db_exam.id == variant.exam_id if not self.user.is_authenticated: raise ExamStartError(_("You must be logged in to start an exam.")) try: with transaction.atomic(): # Lock the user row to guarantee: at most one in-progress attempt per user. DBUser.objects.select_for_update().only("id").get(id=self.user.id) result, redemption = self._evaluate_start_rules(lock_redemption=True) if not result.eligible: raise ExamStartError(result.message) exam_attempt = ExamAttempt.create( user=self.user, exam=self.db_exam, variant=variant ) if redemption is not None: redemption.exam_attempt = exam_attempt.db_exam_attempt redemption.save(update_fields=["exam_attempt"]) return exam_attempt except ExamStartError: raise except Exception: logger.exception( "Failed to start exam " f"user_id={self.user.id} exam_id={self.db_exam.id} variant_id={variant.id}" ) raise ExamStartError( _("Failed to start exam, please refresh the page and try again.") )