MathApp/rag/tests.py

"""
تست‌های لایه‌ی RAG. همه‌ی فراخوانی‌های شبکه (OpenAI/GapGPT و Qdrant) mock
می‌شن؛ این تست‌ها بدونِ کلید و بدونِ اینترنت هم سبز می‌مونن.
"""
from types import SimpleNamespace
from unittest.mock import patch

from django.core.cache import cache
from django.core.management import call_command
from django.test import TestCase, override_settings
from rest_framework.test import APIClient

from accounts.authentication import issue_access_token
from accounts.models import User

from .models import LessonChunk
from .services.clients import reset_clients_cache

RAG_ON = dict(
    OPENAI_API_KEY='test-openai',
    QDRANT_API_KEY='test-qdrant',
    RAG_FORCE_MOCK=False,
    RAG_GRADE_TITLE_OVERRIDES={1: 'ششم دبستان'},
)


def _fake_openai(embedding=None, chat_reply='پاسخِ آزمایشیِ مدل.'):
    dim = 1536
    vec = embedding or [0.01] * dim

    class _Embeddings:
        def create(self, **kwargs):
            return SimpleNamespace(data=[SimpleNamespace(embedding=list(vec))])

    class _Completions:
        def create(self, **kwargs):
            return SimpleNamespace(
                choices=[SimpleNamespace(message=SimpleNamespace(content=chat_reply))]
            )

    return SimpleNamespace(
        embeddings=_Embeddings(),
        chat=SimpleNamespace(completions=_Completions()),
    )


class _FakeQdrant:
    def __init__(self, points=None):
        self._collections = set()
        self._points = points or []
        self.upserted = []

    def get_collections(self):
        return SimpleNamespace(collections=[SimpleNamespace(name=n) for n in self._collections])

    def create_collection(self, collection_name, **kwargs):
        self._collections.add(collection_name)

    def upsert(self, collection_name, points, **kwargs):
        self._collections.add(collection_name)
        self.upserted.extend(points)

    def query_points(self, collection_name, query, limit=4, query_filter=None):
        pts = [SimpleNamespace(score=0.9, payload=p) for p in self._points][:limit]
        return SimpleNamespace(points=pts)


@override_settings(**RAG_ON)
class AskAIRagPathTests(TestCase):
    def setUp(self):
        cache.clear()
        reset_clients_cache()
        self.client = APIClient()
        self.user = User.objects.create(device_id='dev-rag-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)

    def _ask(self, body):
        return self.client.post(
            '/api/ai/ask', body, format='json',
            HTTP_AUTHORIZATION=f'Bearer {self.token}',
        )

    def test_ask_with_retrieval_returns_answer_and_used_chunks(self):
        qdrant = _FakeQdrant(points=[
            {'summary': 'چرخه‌ی آب شاملِ تبخیر، تراکم و بارش است.', 'lesson_title': 'آب', 'lesson_number': 3},
            {'summary': 'خورشید آب را گرم می‌کند و بخار می‌شود.', 'lesson_title': 'آب', 'lesson_number': 3},
        ])
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply='آب تبخیر می‌شود.')), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            resp = self._ask({
                'userQuestion': 'چرخه‌ی آب چیست؟',
                'gradeId': 1,
                'subject': 'علوم تجربی',
                'lessonNumber': 3,
            })
        self.assertEqual(resp.status_code, 200, resp.content)
        self.assertEqual(resp.data['answer'], 'آب تبخیر می‌شود.')
        self.assertEqual(len(resp.data['usedChunks']), 2)
        self.assertEqual(resp.data['usedChunks'][0]['title'], 'آب')

    def test_ask_without_subject_skips_retrieval_but_still_answers(self):
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply='پاسخِ عمومی.')), \
             patch('rag.services.clients.get_qdrant_client', return_value=_FakeQdrant()):
            resp = self._ask({'userQuestion': 'یک سوالِ کلی'})
        self.assertEqual(resp.status_code, 200, resp.content)
        self.assertEqual(resp.data['answer'], 'پاسخِ عمومی.')
        self.assertEqual(resp.data['usedChunks'], [])

    def test_provider_failure_becomes_503_and_releases_slot(self):
        from ai_tutor.models import Conversation, Message

        def boom(*a, **k):
            raise ConnectionError('qdrant down')

        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai()), \
             patch('rag.services.retrieval.search', side_effect=boom):
            resp = self._ask({
                'userQuestion': 'سوال', 'gradeId': 1, 'subject': 'علوم تجربی', 'lessonNumber': 3,
            })
        self.assertEqual(resp.status_code, 503)
        self.assertEqual(Conversation.objects.filter(user_id=self.user.id).count(), 0)
        self.assertEqual(Message.objects.filter(conversation__user_id=self.user.id).count(), 0)


@override_settings(**RAG_ON)
class IngestTextTests(TestCase):
    def setUp(self):
        cache.clear()
        reset_clients_cache()
        self.client = APIClient()

    def test_ingest_text_creates_chunks(self):
        qdrant = _FakeQdrant()
        reply = '{"summary": "خلاصه‌ی این تکه."}'
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply=reply)), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            resp = self.client.post(
                '/api/rag/ingest/text',
                {
                    'lessonId': 12, 'gradeId': 1, 'subject': 'علوم تجربی',
                    'lessonNumber': 3, 'title': 'آب',
                    'text': 'الف' * 900,
                },
                format='json',
                HTTP_X_INTERNAL_KEY='dev-internal-service-key-change-me',
            )
        self.assertEqual(resp.status_code, 200, resp.content)
        self.assertGreater(resp.data['chunksCreated'], 0)
        self.assertEqual(LessonChunk.objects.filter(lesson_id=12).count(), resp.data['chunksCreated'])
        self.assertTrue(qdrant.upserted)
        self.assertEqual(resp.data['collection'], 'علوم تجربی ششم دبستان'.lower())

    def test_ingest_requires_internal_key(self):
        resp = self.client.post(
            '/api/rag/ingest/text',
            {'lessonId': 1, 'subject': 'علوم تجربی', 'lessonNumber': 1, 'text': 'x'},
            format='json',
        )
        # بدونِ X-Internal-Key و بدونِ توکن: DRF بسته به authenticator یا
        # 401 می‌ده یا 403 — هر دو یعنی «رد شد».
        self.assertIn(resp.status_code, (401, 403))


class IngestPayloadContractTests(TestCase):
    """
    قراردادِ بینِ چیزی که ingest می‌نویسه و چیزی که retrieval دنبالش می‌گرده.

    این تست‌ها عمداً ریزبینانه‌ان چون شکستنِ این قرارداد هیچ خطایی تولید نمی‌کنه:
    نقطه‌ها سالم در Qdrant می‌شینن، ولی فیلترِ جستجو هیچ‌وقت بهشون نمی‌خوره و
    محتوا بی‌صدا نامرئی می‌مونه.
    """

    def test_math_payload_key_matches_the_key_retrieval_filters_on(self):
        from rag.services.ingest import _payload
        from rag.services.known_collections import filter_key_for_subject

        chunk = LessonChunk(
            lesson_id=1, grade_id=1, subject='ریاضی', lesson_number=2, chapter=2,
            chapter_title='فصل ۲: کسر', order=1, title='کسر',
            original_text='x', summary='x', collection='ریاضی ششم',
        )
        payload = _payload(chunk)

        # کلیدی که ریاضی روش فیلتر می‌شه باید دقیقاً در payload باشه، با همون مقدار.
        self.assertEqual(filter_key_for_subject('ریاضی'), 'chapter')
        self.assertIn('chapter', payload)
        self.assertEqual(payload['chapter'], 'فصل ۲: کسر')

    def test_text_subject_payload_keeps_lesson_number_filter(self):
        from rag.services.ingest import _payload
        from rag.services.known_collections import filter_key_for_subject

        chunk = LessonChunk(
            lesson_id=1, grade_id=1, subject='علوم تجربی', lesson_number=3, chapter=None,
            order=1, title='آب', original_text='x', summary='x', collection='علوم تجربی سوم',
        )
        payload = _payload(chunk)

        self.assertEqual(filter_key_for_subject('علوم تجربی'), 'lesson_number')
        self.assertEqual(payload['lesson_number'], 3)
        # فصل نداره، پس نباید کلیدِ خالیِ chapter بنویسه.
        self.assertNotIn('chapter', payload)

    def test_structured_fields_survive_into_payload(self):
        """
        استخراجِ vision گرونه؛ اگه ساختارش فقط به متنِ صاف تبدیل بشه، برای
        ساختِ سؤال‌های مسیرِ درس باید دوباره همون هزینه داده بشه.
        """
        from rag.services.ingest import _payload

        chunk = LessonChunk(
            lesson_id=1, grade_id=1, subject='ریاضی', lesson_number=2, chapter=2,
            chapter_title='فصل ۲: کسر', grade_title='ششم دبستان',
            problem_type='مقایسه کسرها', keywords=['کسر', 'مقایسه'],
            general_solution='مخرج مشترک بگیرید.',
            problem='کدام کسر بزرگ‌تر است؟', solution='۳/۴ بزرگ‌تر است.',
            order=1, title='کسر', original_text='x', summary='x', collection='ریاضی ششم',
        )
        payload = _payload(chunk)

        self.assertEqual(payload['type'], 'مقایسه کسرها')
        self.assertEqual(payload['keywords'], ['کسر', 'مقایسه'])
        self.assertEqual(payload['generalSolution'], 'مخرج مشترک بگیرید.')
        self.assertEqual(payload['problem'], 'کدام کسر بزرگ‌تر است؟')
        self.assertEqual(payload['solution'], '۳/۴ بزرگ‌تر است.')
        self.assertEqual(payload['grade'], 'ششم دبستان')

    def test_payload_text_reads_what_ingest_writes(self):
        """پلِ نهایی: متنی که retrieval از payload درمی‌آره نباید خالی باشه."""
        from rag.services.ingest import _payload
        from rag.services.retrieval import _payload_text

        chunk = LessonChunk(
            lesson_id=1, grade_id=1, subject='ریاضی', lesson_number=2, chapter=2,
            chapter_title='فصل ۲: کسر', general_solution='مخرج مشترک بگیرید.',
            order=1, title='کسر', original_text='متنِ اصلی', summary='خلاصه‌ی تکه',
            collection='ریاضی ششم',
        )
        self.assertTrue(_payload_text(_payload(chunk)))


class RagConfigTests(TestCase):
    def test_rag_disabled_without_keys(self):
        from rag.services.config import rag_is_configured

        with override_settings(OPENAI_API_KEY='', QDRANT_API_KEY=''):
            self.assertFalse(rag_is_configured())
        with override_settings(OPENAI_API_KEY='a', QDRANT_API_KEY='b', RAG_FORCE_MOCK=True):
            self.assertFalse(rag_is_configured())
        with override_settings(OPENAI_API_KEY='a', QDRANT_API_KEY='b', RAG_FORCE_MOCK=False):
            self.assertTrue(rag_is_configured())


def _parse_sse(streaming_content):
    """bytes iterator → [(event, data_str), ...]"""
    raw = b''.join(streaming_content).decode('utf-8')
    events = []
    for block in raw.split('\n\n'):
        block = block.strip()
        if not block:
            continue
        ev, data = None, None
        for line in block.splitlines():
            if line.startswith('event: '):
                ev = line[7:]
            elif line.startswith('data: '):
                data = line[6:]
        events.append((ev, data))
    return events


@override_settings(**RAG_ON)
class AskAIStreamTests(TestCase):
    def setUp(self):
        cache.clear()
        reset_clients_cache()
        self.client = APIClient()
        self.user = User.objects.create(device_id='dev-stream-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)

    def _ask_stream(self, body):
        return self.client.post(
            '/api/ai/ask', {**body, 'stream': True}, format='json',
            HTTP_AUTHORIZATION=f'Bearer {self.token}',
        )

    def test_stream_emits_chunks_then_done_and_persists(self):
        from ai_tutor.models import Message

        chunks_out, meta = [{'title': 'کسر', 'summary': 'خلاصه‌ی کسر'}], {'collection': 'ریاضی ششم'}
        with patch('rag.services.retrieval.retrieve_context', return_value=(chunks_out, meta)), \
             patch('rag.services.generation.stream_answer', return_value=iter(['برای ', 'جمع ', 'کسرها...'])):
            resp = self._ask_stream({
                'userQuestion': 'جمع کسر؟', 'subject': 'ریاضی', 'grade': 'ششم', 'lessonNumber': 2,
            })
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp['content-type'], 'text/event-stream')
        events = _parse_sse(resp.streaming_content)
        kinds = [e for e, _ in events]
        self.assertEqual(kinds.count('chunk'), 3)
        self.assertEqual(kinds[-1], 'done')
        import json as _json
        done = _json.loads(events[-1][1])
        self.assertIn('messageId', done)
        self.assertEqual(done['usedChunks'][0]['title'], 'کسر')
        # پیامِ کاربر و دستیار هر دو ثبت شده و کوتا مصرف شده
        self.assertEqual(Message.objects.filter(conversation__user_id=self.user.id, role='assistant').count(), 1)
        msg = Message.objects.get(conversation__user_id=self.user.id, role='assistant')
        self.assertEqual(msg.text, 'برای جمع کسرها...')

    def test_stream_provider_failure_emits_error_and_releases_slot(self):
        from ai_tutor.models import Conversation, Message

        def boom(*a, **k):
            raise ConnectionError('provider down')

        with patch('rag.services.retrieval.retrieve_context', side_effect=boom):
            resp = self._ask_stream({'userQuestion': 'x', 'subject': 'ریاضی', 'grade': 'ششم', 'lessonNumber': 2})
        events = _parse_sse(resp.streaming_content)
        self.assertEqual(events[-1][0], 'error')
        self.assertEqual(Conversation.objects.filter(user_id=self.user.id).count(), 0)
        self.assertEqual(Message.objects.filter(conversation__user_id=self.user.id).count(), 0)


@override_settings(**RAG_ON, RAG_ANSWER_CACHE_TTL=300)
class AskAIAnswerCacheTests(TestCase):
    def setUp(self):
        cache.clear()
        reset_clients_cache()
        self.client = APIClient()
        self.user = User.objects.create(device_id='dev-cache-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)

    def _ask(self, body):
        return self.client.post(
            '/api/ai/ask', body, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}',
        )

    def test_identical_question_second_time_skips_generation_but_still_persists(self):
        from ai_tutor.models import Message

        qdrant = _FakeQdrant(points=[{'summary': 'خلاصه', 'lesson_title': 'کسر', 'lesson_number': 2}])
        body = {'userQuestion': 'سوال یکسان', 'subject': 'ریاضی', 'grade': 'ششم', 'lessonNumber': 2}

        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply='پاسخ ۱')) as oai, \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            r1 = self._ask(body)
            calls_after_first = oai.call_count
            r2 = self._ask(body)

        self.assertEqual(r1.status_code, 200)
        self.assertEqual(r2.status_code, 200)
        self.assertEqual(r1.data['answer'], r2.data['answer'])
        # درخواستِ دوم نباید کلاینتِ OpenAI رو دوباره صدا بزنه (کش)
        self.assertEqual(oai.call_count, calls_after_first)
        # ولی پیام و کوتا برای هر دو درخواست ثبت/مصرف شده
        self.assertEqual(Message.objects.filter(conversation__user_id=self.user.id, role='user').count(), 2)
        self.assertGreater(r1.data['remainingQuota'], r2.data['remainingQuota'])


class MathPayloadShapeTests(TestCase):
    """
    نقاطِ ریاضیِ ما باید هم‌شکلِ نقاطِ موجودِ تیمِ دیگر باشند. مهم‌ترینش نبودِ
    `summary` است: با آن، retrieval به‌جای راه‌حل، متنِ صافِ «Type: …» را
    به‌عنوان context به مدل می‌دهد و کیفیت بی‌سروصدا افت می‌کند.
    """

    def _math_chunk(self, **over):
        data = dict(
            lesson_id=None, subject='ریاضی', lesson_number=4, chapter=4,
            chapter_title='فصل ۴: توان و ریشه', grade_title='نهم', order=1,
            title='فصل ۴: توان و ریشه', original_text='Type: توان صحیح',
            summary='Type: توان صحیح', collection='ریاضی نهم', source_page=67,
            problem_type='توان صحیح', keywords=['توان'],
            general_solution='پایه را در خودش ضرب کن.',
            problem='۲ به توان ۳ چند است؟', solution='۸',
        )
        data.update(over)
        return LessonChunk(**data)

    def test_math_payload_has_no_summary_or_lesson_title(self):
        from rag.services.ingest import _payload

        payload = _payload(self._math_chunk())
        self.assertNotIn('summary', payload)
        self.assertNotIn('lesson_title', payload)
        self.assertEqual(payload['chapter'], 'فصل ۴: توان و ریشه')
        self.assertEqual(payload['grade'], 'نهم')
        self.assertEqual(payload['source_page'], 67)

    def test_retrieval_reads_solution_not_the_flattened_text(self):
        from rag.services.ingest import _payload
        from rag.services.retrieval import _payload_text

        text = _payload_text(_payload(self._math_chunk()))
        self.assertIn('پایه را در خودش ضرب کن.', text)
        self.assertNotIn('Type:', text)

    def test_text_subject_still_gets_summary(self):
        from rag.services.ingest import _payload

        payload = _payload(self._math_chunk(
            problem_type='', keywords=[], general_solution='', problem='', solution='',
            subject='علوم تجربی', summary='خلاصه‌ی درس', title='زنگ علوم',
        ))
        self.assertEqual(payload['summary'], 'خلاصه‌ی درس')
        self.assertEqual(payload['lesson_title'], 'زنگ علوم')


class TextbookLayoutTests(TestCase):
    def test_every_chapter_title_is_registered_for_retrieval(self):
        """عنوانی که ingest می‌نویسد باید همانی باشد که retrieval می‌سازد."""
        from rag.services.known_collections import resolve_math_chapter
        from rag.services.textbooks import TEXTBOOKS

        for book in TEXTBOOKS.values():
            for chapter in book.chapters:
                self.assertEqual(
                    resolve_math_chapter(book.subject, book.grade, chapter.number),
                    chapter.title,
                    f'{book.key} فصل {chapter.number}',
                )

    def test_page_ranges_are_contiguous_for_books_we_ingest_ourselves(self):
        """
        فقط کتاب‌هایی که فایلِ PDF دارند. پایه‌های ششم و هفتم را تیمِ دیگر
        ingest کرده و نقاطشان شماره‌ی صفحه ندارند، پس بازه‌ی صفحه برایشان صفر
        است و معنایی ندارد.
        """
        from rag.services.textbooks import INGESTABLE

        self.assertTrue(INGESTABLE, 'هیچ کتابِ قابلِ ingest ثبت نشده')
        for book in INGESTABLE.values():
            previous = None
            for chapter in book.chapters:
                self.assertLess(0, chapter.first_page, book.key)
                self.assertLessEqual(chapter.first_page, chapter.last_page, book.key)
                if previous is not None:
                    # بازه‌ها نباید همپوشانی یا حفره داشته باشند، وگرنه صفحه‌ای
                    # دو بار ingest می‌شود یا اصلاً دیده نمی‌شود.
                    self.assertEqual(chapter.first_page, previous.last_page + 1, book.key)
                previous = chapter

    def test_pdf_pages_apply_the_offset(self):
        from rag.services.textbooks import MATH_9

        first, last = MATH_9.pdf_pages(MATH_9.chapter(1))
        self.assertEqual((first, last), (9, 25))


class GradeAwarePromptTests(TestCase):
    def test_secondary_grade_is_not_addressed_as_an_elementary_pupil(self):
        from rag.services.generation import system_prompt

        self.assertIn('نهم', system_prompt('نهم'))
        self.assertNotIn('دبستان', system_prompt('نهم'))
        self.assertIn('دبستان', system_prompt('سوم'))


@override_settings(**RAG_ON)
class AnswerAnnotationTests(TestCase):
    """پاسخ باید با متادیتای تولید ثبت شود، وگرنه لایک/دیسلایک بعداً قابلِ تحلیل نیست."""

    def setUp(self):
        cache.clear()
        reset_clients_cache()
        self.user = User.objects.create(phone='09120000777', is_guest=False, grade_id=1)
        self.client = APIClient()
        self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}')

    def tearDown(self):
        cache.clear()
        reset_clients_cache()

    def test_assistant_message_records_model_prompt_version_and_retrieval(self):
        from ai_tutor.models import Message
        from rag.services.generation import PROMPT_VERSION

        qdrant = _FakeQdrant(points=[{'summary': 'خلاصه', 'lesson_title': 'کسر', 'lesson_number': 2}])
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply='پاسخ')), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            response = self.client.post('/api/ai/ask', {
                'userQuestion': 'کسر چیست؟', 'subject': 'ریاضی',
                'grade': 'ششم', 'lessonNumber': 2,
            }, format='json')

        self.assertEqual(response.status_code, 200)
        message = Message.objects.get(pk=response.data['messageId'])
        self.assertEqual(message.prompt_version, PROMPT_VERSION)
        self.assertTrue(message.model_name)
        self.assertEqual(message.grade_title, 'ششم')
        self.assertEqual(message.retrieval_collection, 'ریاضی ششم')
        self.assertEqual(message.chunk_count, 1)
        self.assertFalse(message.from_cache)

    def test_second_identical_question_is_marked_as_cached(self):
        from ai_tutor.models import Message

        qdrant = _FakeQdrant(points=[{'summary': 'خلاصه', 'lesson_title': 'کسر', 'lesson_number': 2}])
        body = {'userQuestion': 'همان سوال', 'subject': 'ریاضی', 'grade': 'ششم', 'lessonNumber': 2}
        with override_settings(RAG_ANSWER_CACHE_TTL=600), \
             patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply='پاسخ')), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            self.client.post('/api/ai/ask', body, format='json')
            second = self.client.post('/api/ai/ask', body, format='json')

        message = Message.objects.get(pk=second.data['messageId'])
        self.assertTrue(message.from_cache)
        # متادیتای تولید از cache هم می‌آید، پس نباید خالی شود.
        self.assertTrue(message.prompt_version)

    def test_feedback_stores_optional_reason(self):
        from ai_tutor.models import MessageFeedback

        qdrant = _FakeQdrant(points=[{'summary': 'خلاصه', 'lesson_number': 2}])
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai()), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            ask = self.client.post('/api/ai/ask', {
                'userQuestion': 'سوال', 'subject': 'ریاضی', 'grade': 'ششم', 'lessonNumber': 2,
            }, format='json')

        response = self.client.post(
            f"/api/ai/messages/{ask.data['messageId']}/feedback",
            {'isHelpful': False, 'reason': 'جواب ربطی به سوال نداشت'}, format='json',
        )
        self.assertEqual(response.status_code, 200)
        feedback = MessageFeedback.objects.get(message_id=ask.data['messageId'])
        self.assertFalse(feedback.is_helpful)
        self.assertEqual(feedback.reason, 'جواب ربطی به سوال نداشت')


@override_settings(**RAG_ON)
class IngestBookResumeTests(TestCase):
    """
    ازسرگیریِ ingestِ کتاب. مهم‌ترین حالت، صفحه‌ی بی‌مسئله است: بدونِ نشانه‌ی
    IngestedPage هر اجرای دوباره یک فراخوانیِ vision دیگر برایش خرج می‌کند.
    """

    def setUp(self):
        reset_clients_cache()
        self.calls = []

    def tearDown(self):
        reset_clients_cache()

    def _run(self, pages_content, **kwargs):
        """pages_content: {pdf_page: [problemType, ...]}؛ صفحه‌ی غایب یعنی بی‌مسئله."""
        from rag.services import ingest as ingest_service

        def fake_images(path, first_page=1, dpi=200, last_page=None):
            return [f'image-p{first_page}']

        def fake_extract(image, grade, chapter_name):
            page = int(str(image).split('p')[-1])
            self.calls.append(page)
            return {'grade': grade, 'chapter': chapter_name,
                    'problemTypes': pages_content.get(page, [])}

        with patch('rag.services.pdf.extract_page_images', side_effect=fake_images), \
             patch('rag.services.pdf.extract_problem_types', side_effect=fake_extract), \
             patch('rag.services.ingest.ensure_collection'), \
             patch('rag.services.ingest.upsert_points'), \
             patch('rag.services.ingest.embed_text', return_value=[0.01] * 1536):
            return ingest_service.ingest_book(subject='ریاضی', grade='نهم', **kwargs)

    def test_pages_without_problems_are_not_billed_twice(self):
        problem = {'type': 'مجموعه', 'keywords': ['مجموعه'], 'generalSolution': 'حل',
                   'exampleFromImage': {'problem': 'سوال', 'solution': 'جواب'}}

        first = self._run({10: [problem]}, chapters=[1], max_pages=3)
        self.assertEqual(first['pages_processed'], 3)
        self.assertEqual(first['chunks_created'], 1)
        self.assertEqual(self.calls, [9, 10, 11])

        # اجرای دوم: هر سه صفحه — شاملِ ۹ و ۱۱ که chunk نساختند — رد می‌شوند.
        self.calls.clear()
        second = self._run({10: [problem]}, chapters=[1], max_pages=3)
        self.assertEqual(self.calls, [12, 13, 14])
        self.assertEqual(second['pages_skipped'], 3)

    def test_a_failing_page_is_retried_on_the_next_run(self):
        from rag.services import ingest as ingest_service

        def fake_images(path, first_page=1, dpi=200, last_page=None):
            return [f'image-p{first_page}']

        def boom(image, grade, chapter_name):
            raise RuntimeError('provider down')

        with patch('rag.services.pdf.extract_page_images', side_effect=fake_images), \
             patch('rag.services.pdf.extract_problem_types', side_effect=boom), \
             patch('rag.services.ingest.ensure_collection'), \
             patch('rag.services.ingest.upsert_points'):
            result = ingest_service.ingest_book(
                subject='ریاضی', grade='نهم', chapters=[1], max_pages=2,
            )

        self.assertEqual(result['status'], 'partial')
        self.assertEqual(result['failed_pages'], [9, 10])
        self.assertEqual(result['pages_processed'], 0)

        # هیچ نشانه‌ای زده نشده، پس اجرای بعدی دوباره سراغشان می‌رود.
        self.calls.clear()
        self._run({}, chapters=[1], max_pages=2)
        self.assertEqual(self.calls, [9, 10])

    def test_chapter_title_comes_from_our_catalog_not_from_the_model(self):
        from rag.models import LessonChunk

        # مدل عنوانِ فصلِ متفاوتی برمی‌گرداند؛ باید نادیده گرفته شود، وگرنه
        # فیلترِ retrieval روی رشته‌ی دقیق دیگر نمی‌خورد.
        problem = {'type': 'مجموعه', 'keywords': [], 'generalSolution': 'حل',
                   'exampleFromImage': {'problem': 'س', 'solution': 'ج'}}
        with patch('rag.services.pdf.extract_page_images',
                   side_effect=lambda path, first_page=1, dpi=200, last_page=None: [f'image-p{first_page}']), \
             patch('rag.services.pdf.extract_problem_types',
                   return_value={'grade': 'چیزِ دیگر', 'chapter': 'فصلِ اشتباه',
                                 'problemTypes': [problem]}), \
             patch('rag.services.ingest.ensure_collection'), \
             patch('rag.services.ingest.upsert_points'), \
             patch('rag.services.ingest.embed_text', return_value=[0.01] * 1536):
            from rag.services import ingest as ingest_service
            ingest_service.ingest_book(subject='ریاضی', grade='نهم', chapters=[1], max_pages=1)

        chunk = LessonChunk.objects.get()
        self.assertEqual(chunk.chapter_title, 'فصل ۱: مجموعه‌ها')
        self.assertEqual(chunk.grade_title, 'نهم')


class LessonContextResolutionTests(TestCase):
    """
    اپ فقط lessonId دارد؛ نباید مجبور باشد نامِ collectionِ Qdrant را بداند.
    """

    def setUp(self):
        from content.models import Grade, Lesson, Node, Unit

        grade = Grade.objects.create(title='پایه نهم', icon='grade-9', color='#FF9800')
        unit = Unit.objects.create(grade=grade, title='فصل ۴: توان و ریشه', order=4)
        node = Node.objects.create(unit=unit, title='توان صحیح', order=1)
        self.lesson = Lesson.objects.create(node=node, title='محاسبه توان منفی', intro_text='.')

        other_unit = Unit.objects.create(grade=grade, title='یک عنوانِ ناشناخته', order=9)
        other_node = Node.objects.create(unit=other_unit, title='درس', order=1)
        self.orphan_lesson = Lesson.objects.create(node=other_node, title='x', intro_text='.')

    def test_derives_subject_grade_and_chapter(self):
        from rag.services.lesson_context import resolve_from_lesson

        self.assertEqual(
            resolve_from_lesson(self.lesson.id),
            {'subject': 'ریاضی', 'gradeTitle': 'نهم', 'chapter': 4},
        )

    def test_unknown_chapter_title_still_yields_the_grade(self):
        from rag.services.lesson_context import resolve_from_lesson

        self.assertEqual(resolve_from_lesson(self.orphan_lesson.id), {'gradeTitle': 'نهم'})

    def test_bad_input_is_not_an_error(self):
        from rag.services.lesson_context import resolve_from_lesson

        for value in (None, '', 'abc', 999999):
            self.assertEqual(resolve_from_lesson(value), {})

    def test_grade_prefix_variants_normalize_to_the_collection_form(self):
        from rag.services.lesson_context import normalize_grade_title

        self.assertEqual(normalize_grade_title('پایه نهم'), 'نهم')
        self.assertEqual(normalize_grade_title('پایهٔ دهم'), 'دهم')
        self.assertEqual(normalize_grade_title('ششم'), 'ششم')


@override_settings(**RAG_ON)
class AskDerivesContextFromLessonTests(TestCase):
    def setUp(self):
        from content.models import Grade, Lesson, Node, Unit

        cache.clear()
        reset_clients_cache()
        self.user = User.objects.create(device_id='dev-ctx-1', is_guest=True)
        self.client = APIClient()
        self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}')

        grade = Grade.objects.create(title='پایه نهم', icon='g', color='#fff')
        unit = Unit.objects.create(grade=grade, title='فصل ۴: توان و ریشه', order=4)
        node = Node.objects.create(unit=unit, title='توان صحیح', order=1)
        self.lesson = Lesson.objects.create(node=node, title='توان منفی', intro_text='.')

    def tearDown(self):
        cache.clear()
        reset_clients_cache()

    def test_lesson_id_alone_is_enough_to_hit_the_right_collection(self):
        from ai_tutor.models import Message

        qdrant = _FakeQdrant(points=[{'generalSolution': 'معکوسِ پایه را بگیر.',
                                      'chapter': 'فصل ۴: توان و ریشه'}])
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply='پاسخ')), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            response = self.client.post('/api/ai/ask', {
                'userQuestion': 'چرا توان منفی معکوس می‌شود؟',
                'lessonId': str(self.lesson.id),
            }, format='json')

        self.assertEqual(response.status_code, 200)
        message = Message.objects.get(pk=response.data['messageId'])
        # بدونِ اینکه اپ subject/grade/lessonNumber فرستاده باشد:
        self.assertEqual(message.retrieval_collection, 'ریاضی نهم')
        self.assertEqual(message.retrieval_filter_value, 'فصل ۴: توان و ریشه')
        self.assertEqual(message.grade_title, 'نهم')

    def test_explicit_values_still_win_over_the_derived_ones(self):
        from ai_tutor.models import Message

        qdrant = _FakeQdrant(points=[{'summary': 'خلاصه', 'lesson_number': 2}])
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai()), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            response = self.client.post('/api/ai/ask', {
                'userQuestion': 'سوال', 'lessonId': str(self.lesson.id),
                'subject': 'علوم تجربی', 'grade': 'سوم', 'lessonNumber': 2,
            }, format='json')

        message = Message.objects.get(pk=response.data['messageId'])
        self.assertEqual(message.retrieval_collection, 'علوم تجربی سوم')


class NonIngestableBookTests(TestCase):
    def test_a_book_without_a_pdf_is_rejected_by_ingest_not_silently_skipped(self):
        from rag.services.ingest import ingest_book

        # پایه‌ی ششم فقط ساختار دارد (دادهٔ واقعی‌اش روی Qdrantِ تیمِ دیگر است).
        with self.assertRaises(FileNotFoundError):
            ingest_book(subject='ریاضی', grade='ششم', chapters=[1], max_pages=1)


@override_settings(**RAG_ON)
class FollowUpStaysGroundedTests(TestCase):
    """
    پیامِ دومِ مکالمه هم باید به کتاب مستند بماند.

    اپ در ادامه فقط `conversationId` می‌فرستد. چون Conversation فیلدی برای
    عنوانِ پایه ندارد، grade_title خالی می‌شد و retrieval بی‌هیچ خطایی رد
    می‌شد — یعنی فقط جوابِ اول از کتاب می‌آمد.
    """

    def setUp(self):
        from content.models import Grade, Lesson, Node, Unit

        cache.clear()
        reset_clients_cache()
        self.user = User.objects.create(device_id='dev-follow-1', is_guest=True)
        self.client = APIClient()
        self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}')

        grade = Grade.objects.create(title='پایه نهم', icon='g', color='#fff')
        unit = Unit.objects.create(grade=grade, title='فصل ۴: توان و ریشه', order=4)
        node = Node.objects.create(unit=unit, title='توان صحیح', order=1)
        self.lesson = Lesson.objects.create(node=node, title='توان منفی', intro_text='.')

    def tearDown(self):
        cache.clear()
        reset_clients_cache()

    def test_second_message_of_a_conversation_still_hits_the_collection(self):
        from ai_tutor.models import Message

        qdrant = _FakeQdrant(points=[{'generalSolution': 'معکوسِ پایه.',
                                      'chapter': 'فصل ۴: توان و ریشه'}])
        with patch('rag.services.clients.get_openai_client', return_value=_fake_openai(chat_reply='پاسخ')), \
             patch('rag.services.clients.get_qdrant_client', return_value=qdrant):
            first = self.client.post('/api/ai/ask', {
                'userQuestion': 'سوالِ اول', 'lessonId': str(self.lesson.id),
            }, format='json')
            # پیامِ دوم: فقط conversationId، همان‌طور که اپ می‌فرستد.
            second = self.client.post('/api/ai/ask', {
                'userQuestion': 'سوالِ دوم، متفاوت',
                'conversationId': str(first.data['conversationId']),
            }, format='json')

        self.assertEqual(second.status_code, 200, second.data)
        message = Message.objects.get(pk=second.data['messageId'])
        self.assertEqual(message.retrieval_collection, 'ریاضی نهم')
        self.assertEqual(message.retrieval_filter_value, 'فصل ۴: توان و ریشه')
        self.assertEqual(message.chunk_count, 1)


class ReviewFixRegressionTests(TestCase):
    """نگهبانِ باگ‌هایی که مرورِ کد پیدا کرد؛ هرکدام قبلاً بی‌سروصدا خراب می‌کردند."""

    def test_a_chunk_with_only_a_problem_type_still_counts_as_structured(self):
        from rag.services.ingest import _payload

        chunk = LessonChunk(
            lesson_id=None, subject='ریاضی', lesson_number=1, chapter=1,
            chapter_title='فصل ۱: مجموعه‌ها', grade_title='نهم', order=1,
            title='فصل ۱: مجموعه‌ها', original_text='Type: مجموعه', summary='Type: مجموعه',
            collection='ریاضی نهم', problem_type='مجموعه', keywords=['مجموعه'],
            general_solution='', problem='', solution='',
        )
        payload = _payload(chunk)
        self.assertNotIn('summary', payload)
        self.assertEqual(payload['type'], 'مجموعه')

    def test_cache_key_separates_grades(self):
        from rag.services.answer_cache import make_key

        self.assertNotEqual(
            make_key(None, None, 'کسر چیست؟', 'نهم'),
            make_key(None, None, 'کسر چیست؟', 'سوم'),
        )

    def test_grade_id_path_builds_the_same_collection_as_the_lesson_path(self):
        from content.models import Grade
        from rag.services.config import build_collection_name, resolve_grade_title

        grade = Grade.objects.create(title='پایه نهم', icon='g', color='#fff')
        # «پایه نهم» نباید به «ریاضی پایه نهم» تبدیل شود.
        self.assertEqual(resolve_grade_title(grade.id), 'نهم')
        self.assertEqual(
            build_collection_name('ریاضی', resolve_grade_title(grade.id)), 'ریاضی نهم',
        )

    def test_feedback_reason_survives_a_second_vote_without_one(self):
        from ai_tutor.models import Conversation, Message, MessageFeedback

        user = User.objects.create(device_id='dev-reason', is_guest=True)
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}')
        conversation = Conversation.objects.create(user_id=user.id)
        message = Message.objects.create(conversation=conversation, role='assistant', text='پاسخ')

        client.post(f'/api/ai/messages/{message.id}/feedback',
                    {'isHelpful': False, 'reason': 'ربطی نداشت'}, format='json')
        client.post(f'/api/ai/messages/{message.id}/feedback',
                    {'isHelpful': True}, format='json')

        feedback = MessageFeedback.objects.get(message=message)
        self.assertTrue(feedback.is_helpful)
        self.assertEqual(feedback.reason, 'ربطی نداشت')

    def test_a_failed_upsert_rolls_back_so_the_page_is_retried(self):
        from rag.services import ingest as ingest_service

        def boom(*args, **kwargs):
            raise RuntimeError('qdrant down')

        rows = [{'order': 1, 'source_page': 42, 'original_text': 'x', 'summary': 'x',
                 'structured': {'chapter_title': 'فصل ۱: مجموعه‌ها', 'grade_title': 'نهم',
                                'problem_type': 'مجموعه'}}]
        with patch('rag.services.ingest.ensure_collection'), \
             patch('rag.services.ingest.embed_text', return_value=[0.01] * 1536), \
             patch('rag.services.ingest.upsert_points', side_effect=boom):
            with self.assertRaises(RuntimeError):
                ingest_service._persist_and_upsert(
                    rows, lesson_id=None, grade_id=None, subject='ریاضی',
                    lesson_number=1, chapter=1, title='فصل ۱: مجموعه‌ها',
                    collection='ریاضی نهم',
                )

        # هیچ ردیفی نباید بماند، وگرنه _done_pages آن صفحه را انجام‌شده می‌بیند
        # و chunkها برای همیشه در DB می‌مانند بدونِ اینکه به Qdrant برسند.
        self.assertEqual(LessonChunk.objects.filter(source_page=42).count(), 0)


class IngestUpsertRollbackTests(TestCase):
    """
    وقتی نوشتن در Qdrant شکست می‌خورد، هیچ ردِ نیمه‌کاره‌ای نباید بماند.

    چرا این تست هست: upsert با wait=True ممکن است واقعاً نوشته باشد و فقط
    جوابش تایم‌اوت شده باشد. تا امروز در آن حالت فقط ردیف‌های DB پس گرفته
    می‌شدند و نقطه‌ها در Qdrant می‌ماندند؛ و چون هر اجرای دوباره شناسه‌ی تازه
    می‌سازد، هر retry یک نسخه‌ی تکراری اضافه می‌کرد. روی صفحه‌ی ۱۲۵ کتابِ
    ریاضی نهم همین اتفاق افتاد: سه نسخه از یک مسئله و دو نقطه‌ی بی‌صاحب.
    """

    @override_settings(**RAG_ON)
    def test_failed_upsert_removes_both_the_rows_and_the_points(self):
        from .services.ingest import _persist_and_upsert

        with patch('rag.services.ingest.ensure_collection'), \
             patch('rag.services.ingest.embed_text', return_value=[0.0] * 8), \
             patch('rag.services.ingest.upsert_points', side_effect=TimeoutError('boom')), \
             patch('rag.services.ingest.delete_points') as delete_mock:
            with self.assertRaises(TimeoutError):
                _persist_and_upsert(
                    [{'order': 1, 'original_text': 'x', 'summary': 'خلاصه', 'source_page': 125}],
                    lesson_id=1, grade_id=1, subject='ریاضی', lesson_number=1,
                    chapter=7, title='عبارت‌های گویا', collection='ریاضی نهم',
                )

        self.assertEqual(LessonChunk.objects.filter(collection='ریاضی نهم').count(), 0)
        delete_mock.assert_called_once()
        collection, point_ids = delete_mock.call_args[0]
        self.assertEqual(collection, 'ریاضی نهم')
        self.assertEqual(len(point_ids), 1, 'شناسه‌ی همان نقطه باید برای پاک شدن برود.')

    @override_settings(**RAG_ON)
    def test_a_failure_while_cleaning_up_does_not_hide_the_original_error(self):
        """پاک‌سازیِ ناموفق نباید خطای اصلی را بپوشاند؛ صفحه باید ناتمام بماند."""
        from .services.ingest import _persist_and_upsert

        with patch('rag.services.ingest.ensure_collection'), \
             patch('rag.services.ingest.embed_text', return_value=[0.0] * 8), \
             patch('rag.services.ingest.upsert_points', side_effect=TimeoutError('boom')), \
             patch('rag.services.ingest.delete_points', side_effect=RuntimeError('qdrant down')):
            with self.assertRaises(TimeoutError):
                _persist_and_upsert(
                    [{'order': 1, 'original_text': 'x', 'summary': 'خلاصه', 'source_page': 9}],
                    lesson_id=1, grade_id=1, subject='ریاضی', lesson_number=1,
                    chapter=1, title='مجموعه‌ها', collection='ریاضی نهم',
                )

        self.assertEqual(LessonChunk.objects.filter(collection='ریاضی نهم').count(), 0)


class VerifySolutionsLimitTests(TestCase):
    """
    دو خطای ۴۲۹ هست و نباید یکسان با آن‌ها رفتار کرد.

    فشارِ لحظه‌ایِ موازی‌کاری گذراست و عقب‌نشینی حلش می‌کند؛ ولی سقفِ مصرفِ
    حساب ساعت‌ها باز نمی‌شود. یک اجرا ۱۱۹ بار دوباره تلاش کرد، ۶۵ دقیقه طول
    کشید، هیچ chunkی سنجیده نشد، و مهلتِ باز شدن را هم جلوتر برد.
    """

    def setUp(self):
        self.chunk = LessonChunk.objects.create(
            subject='ریاضی', lesson_number=1, order=1, title='توان',
            original_text='x', summary='x', collection='ریاضی نهم',
            problem='حاصل ۲^۳ چند است؟', solution='۸',
        )

    @override_settings(**RAG_ON)
    def test_account_limit_stops_at_once_instead_of_retrying(self):
        from django.core.management.base import CommandError

        error = Exception("Error code: 429 - {'code': 'initial_usage_limit_exceeded'}")
        client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(
            create=lambda **kw: (_ for _ in ()).throw(error))))

        with patch('rag.services.clients.get_openai_client', return_value=client), \
             patch('rag.management.commands.rag_verify_solutions.time.sleep') as slept:
            with self.assertRaises(CommandError) as caught:
                call_command('rag_verify_solutions', '--collection', 'ریاضی نهم')

        self.assertIn('سقفِ مصرف', str(caught.exception))
        slept.assert_not_called()   # حتی یک بار هم نباید صبر کرده باشد

    @override_settings(**RAG_ON)
    def test_an_exhausted_quota_also_stops_at_once(self):
        """
        ۴۰۳ با متنِ «insufficient user quota» — با فاصله، نه زیرخط.

        محافظِ اول دنبالِ `insufficient_quota` می‌گشت و این را نمی‌گرفت؛ یک
        اجرا ۷۷۰ فراخوانیِ محکوم‌به‌شکست زد به‌جای اینکه سرِ اولی بایستد.
        """
        from django.core.management.base import CommandError

        error = Exception("Error code: 403 - {'message': 'insufficient user quota, "
                          "remaining quota: $-0.1'}")
        client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(
            create=lambda **kw: (_ for _ in ()).throw(error))))

        with patch('rag.services.clients.get_openai_client', return_value=client), \
             patch('rag.management.commands.rag_verify_solutions.time.sleep') as slept:
            with self.assertRaises(CommandError):
                call_command('rag_verify_solutions', '--collection', 'ریاضی نهم')
        slept.assert_not_called()

    @override_settings(**RAG_ON)
    def test_a_transient_429_is_still_retried(self):
        calls = {'n': 0}

        def flaky(**kwargs):
            calls['n'] += 1
            if calls['n'] == 1:
                raise Exception('Error code: 429 - rate limit, slow down')
            return SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(
                content='{"verdict": "agrees", "answer": "۸", "why": ""}'))])

        client = SimpleNamespace(chat=SimpleNamespace(
            completions=SimpleNamespace(create=flaky)))
        with patch('rag.services.clients.get_openai_client', return_value=client), \
             patch('rag.management.commands.rag_verify_solutions.time.sleep'):
            call_command('rag_verify_solutions', '--collection', 'ریاضی نهم')

        self.chunk.refresh_from_db()
        self.assertEqual(self.chunk.solution_verdict, LessonChunk.VERDICT_AGREES)
        self.assertEqual(calls['n'], 2, 'باید دقیقاً یک بار دوباره تلاش می‌کرد.')

    @override_settings(**RAG_ON)
    def test_figure_dependent_problems_are_labelled_without_any_call(self):
        """هزینه‌ی صفر برای چیزی که جوابش از متن درنمی‌آید."""
        self.chunk.problem = 'مساحت شکلِ زیر را حساب کنید.'
        self.chunk.save(update_fields=['problem'])

        def boom(**kwargs):
            raise AssertionError('نباید هیچ فراخوانی‌ای می‌شد.')

        client = SimpleNamespace(chat=SimpleNamespace(
            completions=SimpleNamespace(create=boom)))
        with patch('rag.services.clients.get_openai_client', return_value=client):
            call_command('rag_verify_solutions', '--collection', 'ریاضی نهم')

        self.chunk.refresh_from_db()
        self.assertEqual(self.chunk.solution_verdict, LessonChunk.VERDICT_UNJUDGEABLE)
        self.assertEqual(self.chunk.checked_by, 'pre-filter')