import threading

from django.db import close_old_connections
from django.test import TransactionTestCase
from django.utils import timezone

from labs.exams.exam import Exam as DomainExam
from labs.models import Exam, ExamAttempt, ExamVariant, User


class ExamAttemptSingleInProgressConcurrencyTests(TransactionTestCase):
    def _start_exam_in_thread(
        self,
        *,
        user_id: int,
        exam_id: int,
        variant_id: int,
        barrier: threading.Barrier,
        results: list[dict],
        index: int,
    ) -> None:
        close_old_connections()
        try:
            user = User.objects.get(id=user_id)
            exam = Exam.objects.get(id=exam_id)
            variant = ExamVariant.objects.get(id=variant_id)
            barrier.wait()
            attempt = DomainExam(exam, user).start_exam(variant)
            results[index] = {"ok": True, "attempt_id": attempt.db_exam_attempt.id}
        except Exception as e:
            results[index] = {"ok": False, "error": str(e)}
        finally:
            close_old_connections()

    def test_concurrent_start_same_exam_creates_single_attempt(self):
        user = User.objects.create_user(email="test-concurrency@test.com")
        exam = Exam.objects.create(code="exam_concurrency_1", credits=0)
        variant = ExamVariant.objects.create(
            exam=exam,
            language_code="en",
            title="Exam Concurrency 1 (EN)",
            duration_seconds=3600,
            pass_score=0,
            total_score=0,
        )

        barrier = threading.Barrier(2)
        results: list[dict] = [{}, {}]
        threads = [
            threading.Thread(
                target=self._start_exam_in_thread,
                kwargs={
                    "user_id": user.id,
                    "exam_id": exam.id,
                    "variant_id": variant.id,
                    "barrier": barrier,
                    "results": results,
                    "index": i,
                },
            )
            for i in range(2)
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=10)
            self.assertFalse(t.is_alive())

        now = timezone.now()
        self.assertEqual(
            ExamAttempt.objects.filter(
                user=user,
                status=ExamAttempt.Status.IN_PROGRESS,
                submitted_at__isnull=True,
                expires_at__gte=now,
            ).count(),
            1,
        )
        ok = [r for r in results if r.get("ok") is True]
        err = [r for r in results if r.get("ok") is False]
        self.assertEqual(len(ok), 1)
        self.assertEqual(len(err), 1)
        self.assertEqual(
            err[0].get("error"),
            "You already have an active exam in progress, please finish it first.",
        )

    def test_concurrent_start_cross_exam_creates_single_attempt(self):
        user = User.objects.create_user(email="test-concurrency-cross@test.com")

        exam_a = Exam.objects.create(code="exam_concurrency_a", credits=0)
        variant_a = ExamVariant.objects.create(
            exam=exam_a,
            language_code="en",
            title="Exam Concurrency A (EN)",
            duration_seconds=3600,
            pass_score=0,
            total_score=0,
        )
        exam_b = Exam.objects.create(code="exam_concurrency_b", credits=0)
        variant_b = ExamVariant.objects.create(
            exam=exam_b,
            language_code="en",
            title="Exam Concurrency B (EN)",
            duration_seconds=3600,
            pass_score=0,
            total_score=0,
        )

        barrier = threading.Barrier(2)
        results: list[dict] = [{}, {}]
        threads = [
            threading.Thread(
                target=self._start_exam_in_thread,
                kwargs={
                    "user_id": user.id,
                    "exam_id": exam_a.id,
                    "variant_id": variant_a.id,
                    "barrier": barrier,
                    "results": results,
                    "index": 0,
                },
            ),
            threading.Thread(
                target=self._start_exam_in_thread,
                kwargs={
                    "user_id": user.id,
                    "exam_id": exam_b.id,
                    "variant_id": variant_b.id,
                    "barrier": barrier,
                    "results": results,
                    "index": 1,
                },
            ),
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=10)
            self.assertFalse(t.is_alive())

        now = timezone.now()
        self.assertEqual(
            ExamAttempt.objects.filter(
                user=user,
                status=ExamAttempt.Status.IN_PROGRESS,
                submitted_at__isnull=True,
                expires_at__gte=now,
            ).count(),
            1,
        )
        ok = [r for r in results if r.get("ok") is True]
        err = [r for r in results if r.get("ok") is False]
        self.assertEqual(len(ok), 1)
        self.assertEqual(len(err), 1)
        self.assertEqual(
            err[0].get("error"),
            "You already have an active exam in progress, please finish it first.",
        )