"""
سقفِ جواب‌های هم‌زمانِ دستیار (ai_tutor/concurrency.py): رگبارِ سوال نباید همه‌ی نخ‌های worker را بگیرد.
هم‌زمانیِ واقعیِ چند نخ با مدلِ واقعی آزموده نشده؛ این‌جا جا را با گرفتنِ دستی پر می‌کنیم.
"""
from unittest import mock

from django.core.cache import cache
from django.core.signals import request_finished
from django.db import close_old_connections
from django.test import TestCase, override_settings

from accounts.authentication import issue_access_token
from accounts.models import User
from ai_tutor import concurrency
from ai_tutor.models import Message
from content.models import Grade


@override_settings(AI_MAX_CONCURRENT_PER_WORKER=1)
class AskConcurrencyCapTests(TestCase):
    def setUp(self):
        cache.clear()
        grade = Grade.objects.create(title='پایه نهم', icon='', color='')
        self.user = User.objects.create(phone='09120007801', grade_id=grade.id)
        self.assertEqual(concurrency.in_flight(), 0)

    def tearDown(self):
        self.assertEqual(concurrency.in_flight(), 0, 'هر تست باید جا را پس بدهد')

    def _ask(self, **extra):
        return self.client.post(
            '/api/ai/ask', {'userQuestion': 'توان چیست؟', **extra}, content_type='application/json',
            HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}',
        )

    def _rag(self):
        return (
            mock.patch('rag.services.config.rag_is_configured', return_value=True),
            mock.patch('rag.services.retrieval.retrieve_context',
                       return_value=([{'title': 'فصل', 'summary': 'متن'}], {'gradeTitle': 'نهم'})),
            mock.patch('rag.services.generation.generate_answer', return_value=('جواب', [], {})),
            mock.patch('rag.services.generation.stream_answer', side_effect=lambda *a, **k: iter(['جو', 'اب'])),
        )

    def test_full_means_503_with_retry_after_and_no_quota_used(self):
        self.assertTrue(concurrency.try_acquire())  # یک سوالِ دیگر در حالِ جواب است
        try:
            with self.assertLogs('ai_tutor.views', level='WARNING'):
                response = self._ask()
        finally:
            concurrency.release()
        self.assertEqual(response.status_code, 503)
        self.assertEqual(response.json()['detail'], 'دستیار الان خیلی شلوغه؛ چند ثانیه‌ی دیگه دوباره بپرس.')
        self.assertEqual(response['Retry-After'], '10')
        self.assertFalse(Message.objects.exists())  # نه پیامی ساخته شد نه سهمیه‌ای رزرو شد

    def test_json_answer_gives_the_slot_back(self):
        a, b, c, d = self._rag()
        with a, b, c, d:
            self.assertEqual(self._ask().status_code, 200)
            self.assertEqual(concurrency.in_flight(), 0)
            self.assertEqual(self._ask().status_code, 200)  # سقفِ ۱: دومی هم جا گرفت

    def test_a_failed_answer_gives_the_slot_back(self):
        a, b, _, d = self._rag()
        failing = mock.patch('rag.services.generation.generate_answer', side_effect=RuntimeError('provider'))
        with a, b, failing, d, self.assertLogs('ai_tutor.views', level='ERROR'):
            self.assertEqual(self._ask().status_code, 503)

    def test_stream_holds_the_slot_until_it_is_read(self):
        a, b, c, d = self._rag()
        with a, b, c, d:
            response = self._ask(stream=True)
            self.assertEqual(concurrency.in_flight(), 1)  # هنوز خوانده نشده
            body = b''.join(response.streaming_content).decode()  # کلاینتِ تست پایانش close() می‌زند
        self.assertIn('event: done', body)

    def test_a_stream_that_is_never_read_still_gives_the_slot_back(self):
        a, b, c, d = self._rag()
        with a, b, c, d:
            response = self._ask(stream=True)
            # کاربر پیش از اولین تکه رفت. سیگنالِ request_finished در TestCase اتصالِ تستی را می‌بندد؛ کلاینتِ
            # تست هم برای پاسخِ استریم همین‌طور جدایش می‌کند.
            request_finished.disconnect(close_old_connections)
            try:
                response.close()
            finally:
                request_finished.connect(close_old_connections)

    def test_release_without_acquire_is_a_loud_bug(self):
        with self.assertRaises(RuntimeError):
            concurrency.release()