import logging from datetime import timedelta from django.db import transaction from django.db.models import F from django.utils import timezone from django.utils.translation import gettext as _ from labs.models import ( User as DBUser, Exam as DBExam, ExamVariant as DBExamVariant, ExamSection as DBExamSection, ExamAttempt as DBExamAttempt, ExamAttemptSection as DBExamAttemptSection, ) from labs.exams.exceptions import ( ExamAttemptSubmitSectionError, ExamAttemptNotInProgressError, ExamAttemptExpiredError, ) from labs.exams.section import get_section_handler, ExamSectionHandler logger = logging.getLogger(__name__) class ExamAttempt: def __init__(self, db_exam_attempt: DBExamAttempt): self.db_exam_attempt: DBExamAttempt = db_exam_attempt self.db_exam_variant: DBExamVariant = db_exam_attempt.variant self.db_exam: DBExam = db_exam_attempt.exam @classmethod def create( cls, user: DBUser, exam: DBExam, variant: DBExamVariant ) -> "ExamAttempt": now = timezone.now() expires_at = now + timedelta(seconds=variant.duration_seconds) with transaction.atomic(): db_exam_attempt = DBExamAttempt.objects.create( user=user, exam=exam, variant=variant, language_code=user.language_code, status=DBExamAttempt.Status.IN_PROGRESS, started_at=now, expires_at=expires_at, ) sections = DBExamSection.objects.filter(variant=variant).order_by( "display_order" ) attempt_sections = [] for i, section in enumerate(sections): attempt_sections.append( DBExamAttemptSection( attempt=db_exam_attempt, section=section, display_order=section.display_order, started_at=now if i == 0 else None, status=DBExamAttemptSection.Status.AVAILABLE if i == 0 else DBExamAttemptSection.Status.LOCKED, ) ) DBExamAttemptSection.objects.bulk_create(attempt_sections) return cls(db_exam_attempt) def ensure_in_progress(self) -> None: if self.db_exam_attempt.status != DBExamAttempt.Status.IN_PROGRESS: raise ExamAttemptNotInProgressError() if self.db_exam_attempt.expires_at < timezone.now(): raise ExamAttemptExpiredError() def get_frontend_detail(self) -> dict: if self.db_exam_attempt.status == DBExamAttempt.Status.IN_PROGRESS: now = timezone.now() remaining_seconds = max( 0, int((self.db_exam_attempt.expires_at - now).total_seconds()) ) else: remaining_seconds = 0 return { "id": self.db_exam_attempt.id, "exam_id": self.db_exam_attempt.exam_id, "variant_id": self.db_exam_attempt.variant_id, "language_code": self.db_exam_attempt.language_code, "started_at": self.db_exam_attempt.started_at, "submitted_at": self.db_exam_attempt.submitted_at, "expires_at": self.db_exam_attempt.expires_at, "score": self.db_exam_attempt.score, "is_passed": self.db_exam_attempt.is_passed, "status": self.db_exam_attempt.status, "result_detail": self.db_exam_attempt.result_detail, "remaining_seconds": remaining_seconds, } def get_frontend_sections(self) -> list[dict]: return list( DBExamAttemptSection.objects.select_related("section") .filter(attempt=self.db_exam_attempt) .order_by("display_order") .annotate( section_title=F("section__title"), section_type=F("section__section_type"), ) .values( "section_id", "section_title", "section_type", "display_order", "status" ) ) def get_section_handler(self, section: DBExamAttemptSection) -> ExamSectionHandler: return get_section_handler(section) def submit_section(self, section_handler: ExamSectionHandler) -> None: with transaction.atomic(): current = ( DBExamAttemptSection.objects.select_related("section") .select_for_update() .get( id=section_handler.db_exam_attempt_section.id, ) ) if current.status != DBExamAttemptSection.Status.AVAILABLE: raise ExamAttemptSubmitSectionError( _("This section is not available to be submitted.") ) section_handler = self.get_section_handler(current) ack_allowed, denial_reason = section_handler.can_submit() if not ack_allowed: raise ExamAttemptSubmitSectionError(denial_reason) current.status = DBExamAttemptSection.Status.COMPLETED current.completed_at = timezone.now() current.save() next = ( DBExamAttemptSection.objects.filter( attempt=self.db_exam_attempt, status=DBExamAttemptSection.Status.LOCKED, display_order__gt=current.display_order, ) .order_by("display_order") .first() ) if next: next.status = DBExamAttemptSection.Status.AVAILABLE next.started_at = timezone.now() next.save() else: self.submit_exam() def submit_exam(self) -> None: attempt_id = self.db_exam_attempt.id with transaction.atomic(): db_attempt = ( DBExamAttempt.objects.select_for_update() .select_related("variant", "exam") .get(id=attempt_id) ) if db_attempt.submitted_at is not None: return now = timezone.now() if db_attempt.status == DBExamAttempt.Status.IN_PROGRESS: expires_at = db_attempt.expires_at if expires_at is not None and now > expires_at: db_attempt.status = DBExamAttempt.Status.EXPIRED else: db_attempt.status = DBExamAttempt.Status.COMPLETED db_attempt.submitted_at = now score = 0 for db_section in DBExamAttemptSection.objects.select_related( "section" ).filter(attempt_id=attempt_id): section_handler = self.get_section_handler(db_section) section_score, _ = section_handler.calculate_score() score += section_score db_attempt.score = score db_attempt.is_passed = score >= db_attempt.variant.pass_score db_attempt.save( update_fields=["status", "submitted_at", "score", "is_passed", "updated_at"] ) self.db_exam_attempt = db_attempt self.db_exam_variant = db_attempt.variant self.db_exam = db_attempt.exam if db_attempt.is_passed: self._issue_certificate_if_configured() def _issue_certificate_if_configured(self) -> None: """Issue certificate if the exam has a certificate configured.""" cert_short_name = self.db_exam.certificate_short_name if not cert_short_name: return try: from labs.certificate.issue import issue_certificate issuance = issue_certificate( user=self.db_exam_attempt.user, cert_short_name=cert_short_name, ) # Save certificate info to result_detail on success self.db_exam_attempt.result_detail["certificate"] = { "cert_number": issuance.cert_number, "cert_name": issuance.cert_type.name, "cert_short_name": cert_short_name, "issued_at": issuance.issued_at.isoformat(), } self.db_exam_attempt.save(update_fields=["result_detail"]) logger.info( f"Certificate {cert_short_name} issued for user " f"{self.db_exam_attempt.user.email} after passing exam {self.db_exam.code}" ) except Exception as e: logger.error( f"Failed to issue certificate {cert_short_name} for user " f"{self.db_exam_attempt.user.email}: {e}" ) def get_in_progress_exam_attempt(user: DBUser) -> ExamAttempt | None: now = timezone.now() db_exam_attempt = ( DBExamAttempt.objects.filter( user=user, status=DBExamAttempt.Status.IN_PROGRESS, submitted_at__isnull=True, expires_at__gte=now, ) .select_related("exam") .order_by("-created_at") .first() ) if not db_exam_attempt: return None return ExamAttempt(db_exam_attempt)