"""
اعتبارِ بسته زیرِ درخواستِ هم‌زمان (بازبینیِ ۲۰۲۶-۱۰-۰۲).

پیش از این فقط ادعا بود که قفلِ ردیفِ کاربر و `transaction_mode: IMMEDIATE`ِ SQLite جلوی
خرجِ دوباره‌ی یک اعتبار را می‌گیرند. این‌جا دو thread واقعی، با دو اتصالِ جدا به دیتابیسِ فایلیِ
تست، هم‌زمان می‌پرسند. جوابِ دستیار ساختگی است و ۰٫۴ ثانیه طول می‌کشد تا دو درخواست واقعاً روی
هم بیفتند. فقط روی SQLite سنجیده شده؛ بعد از مهاجرت به Postgres دوباره باید سبز شود.
"""
import threading
import time
from unittest.mock import patch

from django.core.cache import cache
from django.db import connection
from django.test import TransactionTestCase, override_settings
from rest_framework.test import APIClient

from accounts.authentication import issue_access_token
from accounts.models import User
from subscription.models import QuestionCredit
from subscription.services import credit_balance

from .models import Message


def _slow_answer(*args, **kwargs):
    time.sleep(0.4)
    return 'جواب', ['بیشتر توضیح بده']


@override_settings(AI_DAILY_QUOTA_FREE=0)
class ConcurrentCreditTests(TransactionTestCase):
    def setUp(self):
        cache.clear()
        self.user = User.objects.create(phone='09120007201', is_guest=False)

    def tearDown(self):
        cache.clear()

    def race(self, *headers):
        codes = []
        barrier = threading.Barrier(len(headers))

        def ask(extra):
            try:
                client = APIClient()
                client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}')
                barrier.wait()
                codes.append(client.post('/api/ai/ask', {'userQuestion': '۲ + ۲؟'}, format='json', **extra).status_code)
            finally:
                connection.close()

        with patch('rag.services.config.rag_is_configured', return_value=False), \
                patch('ai_tutor.views.generate_tutor_answer', side_effect=_slow_answer):
            threads = [threading.Thread(target=ask, args=(h,)) for h in headers]
            for t in threads:
                t.start()
            for t in threads:
                t.join()
        return sorted(codes)

    def test_two_questions_at_once_with_one_credit_get_one_answer(self):
        QuestionCredit.objects.create(user_id=self.user.id, delta=1, reason='grant')
        self.assertEqual(self.race({}, {}), [200, 429])
        self.assertEqual(credit_balance(self.user.id), 0)
        self.assertEqual(Message.objects.filter(role='assistant').count(), 1)

    def test_the_same_idempotency_key_twice_at_once_costs_one_credit(self):
        QuestionCredit.objects.create(user_id=self.user.id, delta=2, reason='grant')
        key = {'HTTP_IDEMPOTENCY_KEY': 'same-key'}
        self.assertEqual(self.race(key, key), [200, 200])
        self.assertEqual(credit_balance(self.user.id), 1)
        self.assertEqual(Message.objects.filter(role='user').count(), 1)