""" تست‌های لایه‌ی 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) # «علوم تجربی سوم» در KNOWN_COLLECTIONS هست؛ collectionِ ناشناخته اصلاً جستجو نمی‌شود. @override_settings(**{**RAG_ON, 'RAG_GRADE_TITLE_OVERRIDES': {1: 'سوم'}}) 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() @override_settings(INTERNAL_SERVICE_KEY='test-internal-key') 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='test-internal-key', ) 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) # پیام برای هر دو ثبت می‌شود، ولی جوابِ کش‌شده به API نرسیده و از سهمیه کم نمی‌شود # (۲۰۲۶-۱۰-۰۲: فقط سوالی پول دارد که به API برسد). self.assertEqual(Message.objects.filter(conversation__user_id=self.user.id, role='user').count(), 2) self.assertEqual(r1.data['remainingQuota'], r2.data['remainingQuota']) self.assertEqual(Message.objects.filter(role='user').order_by('id').last().charge, 'none') 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 VisionJsonRepairTests(TestCase): """ مدلِ vision در متنِ ریاضی LaTeX می‌نویسد و بک‌اسلشِ تنها JSON را می‌شکند، پس صفحه‌های پرفرمول — باارزش‌ترین‌ها — بی‌سروصدا دور ریخته می‌شدند. """ def _load(self, raw): import json from rag.services.pdf import repair_json_escapes return json.loads(repair_json_escapes(raw))['p'] def test_latex_commands_survive_instead_of_killing_the_page(self): self.assertEqual(self._load(r'{"p":"\frac{2}{\sqrt{3}}"}'), r'\frac{2}{\sqrt{3}}') def test_latex_starting_with_a_valid_escape_letter_is_not_mangled(self): # دامِ اصلی: \f و \t و \b و \n از نظرِ JSON معتبرند، پس بدونِ تفکیک # \frac به formfeed+«rac» و \theta به tab+«heta» تبدیل می‌شد. self.assertEqual(self._load(r'{"p":"\times \theta \beta \nabla"}'), r'\times \theta \beta \nabla') def test_a_real_newline_is_still_a_newline(self): self.assertEqual(self._load('{"p":"گام ۱\\n۲. ادامه"}'), 'گام ۱\n۲. ادامه') self.assertEqual(self._load('{"p":"اول\\nحالت دوم"}'), 'اول\nحالت دوم') def test_escaped_quotes_still_work(self): self.assertEqual(self._load(r'{"p":"او گفت \"سلام\""}'), 'او گفت "سلام"') class IngestBookRefreshTests(TestCase): """ بازخوانیِ صفحه‌های معیوب (شکل/جدولِ نوشته‌نشده). ingestِ دوباره **جایگزین نمی‌کند**: هر ردیف uuid4ِ تازه می‌گیرد، پس بدونِ پاک‌سازی نسخه‌ی تکراری کنارِ قدیمی می‌نشیند. این کلاس همان را می‌بندد. """ def setUp(self): reset_clients_cache() self.calls = [] def tearDown(self): reset_clients_cache() def _problem(self, text): return {'type': 'نوع', 'keywords': [], 'generalSolution': 'حل', 'exampleFromImage': {'problem': text, 'solution': 'ج'}} def _run(self, text, **kwargs): from rag.services import ingest as ingest_service def fake_images(path, first_page=1, dpi=200, last_page=None): self.calls.append(first_page) return [f'image-p{first_page}'] with patch('rag.services.pdf.extract_page_images', side_effect=fake_images), \ patch('rag.services.pdf.extract_problem_types', return_value={'grade': 'نهم', 'chapter': 'فصل ۱: مجموعه‌ها', 'problemTypes': [self._problem(text)]}), \ patch('rag.services.ingest.ensure_collection'), \ patch('rag.services.ingest.upsert_points'), \ patch('rag.services.ingest.delete_points') as delete_mock, \ patch('rag.services.ingest.embed_text', return_value=[0.01] * 1536): result = ingest_service.ingest_book(subject='ریاضی', grade='نهم', **kwargs) return result, delete_mock def test_refreshing_a_page_replaces_its_chunks_instead_of_duplicating_them(self): from rag.models import LessonChunk self._run('با توجه به شکلِ زیر', chapters=[1], max_pages=1) old = LessonChunk.objects.get() old_point_id = old.qdrant_point_id result, delete_mock = self._run( 'مثلثی با اضلاع ۳ و ۴ و ۵', chapters=[1], max_pages=1, refresh_pages={9}, ) # یک ردیف، نه دو تا — و متنش تازه است. chunk = LessonChunk.objects.get() self.assertEqual(chunk.problem, 'مثلثی با اضلاع ۳ و ۴ و ۵') self.assertEqual(result['chunks_replaced'], 1) # نقطه‌ی قدیمیِ Qdrant هم واقعاً پاک شده، نه فقط ردیفِ دیتابیس. delete_mock.assert_called_once_with('ریاضی نهم', [str(old_point_id)]) self.assertNotEqual(chunk.qdrant_point_id, old_point_id) def test_a_page_marked_for_refresh_is_not_skipped_by_resume(self): self._run('اولی', chapters=[1], max_pages=1) self.calls.clear() # بدونِ کم کردنِ صفحه‌های refresh از مجموعه‌ی «انجام‌شده»، منطقِ # ازسرگیری ردش می‌کرد و بازخوانی بی‌سروصدا هیچ کاری نمی‌کرد. self._run('دومی', chapters=[1], max_pages=1, refresh_pages={9}) self.assertIn(9, self.calls) def test_a_failed_clear_leaves_the_old_data_alone(self): from rag.models import LessonChunk from rag.services import ingest as ingest_service self._run('اصلی', chapters=[1], max_pages=1) 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.ingest.ensure_collection'), \ patch('rag.services.ingest.delete_points', side_effect=RuntimeError('qdrant down')): result = ingest_service.ingest_book( subject='ریاضی', grade='نهم', chapters=[1], max_pages=1, refresh_pages={9}, ) # پاک‌سازی نشد، پس صفحه اصلاً خوانده نشد و دادهٔ قبلی سرِ جایش است — # وگرنه خواندنِ دوباره نسخه‌ی تکراری می‌ساخت. self.assertEqual(LessonChunk.objects.get().problem, 'اصلی') self.assertEqual(result['failed_pages'], [9]) self.assertEqual(result['chunks_replaced'], 0) def test_refresh_is_limited_to_the_chapters_of_this_run(self): from rag.management.commands.rag_ingest_book import Command from rag.services.textbooks import TEXTBOOKS book = TEXTBOOKS['ریاضی نهم'] chapter_one = book.chapter(1) first, last = book.pdf_pages(chapter_one) opts = {'refresh_pages': f'{first},{last},{last + 5}', 'refresh': None} chosen = Command()._resolve_refresh(opts, 'ریاضی نهم', book, [chapter_one]) # صفحه‌ی بیرونِ فصلِ انتخاب‌شده کنار گذاشته می‌شود، وگرنه یک اجرای # محدود به یک فصل می‌توانست صفحه‌ی فصلِ دیگر را پاک کند. self.assertEqual(chosen, {first, last}) 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') class FigureGuardTests(TestCase): """ محافظی که تصمیم می‌گیرد کدام مسئله بدونِ دیدنِ صفحه قابلِ قضاوت نیست. نسخه‌ی اولش یک regexِ تخت بود و فاجعه‌بار گشاد: «روبه‌رو» را می‌گرفت که در هندسه یعنی «زاویه‌ی مقابل»، و «زیر را» را که در «کسرهای زیر را گویا کنید» می‌آید. اندازه‌گیری روی نمونه‌ی ۵۰تایی نشان داد **۸۴٪** از چیزهایی که رد کرده بود واقعاً قابلِ قضاوت بودند — یعنی ~۷۸۴ مسئله‌ی سالم بی‌صدا از چرخه‌ی ساختِ سوال بیرون مانده بود، بی‌هیچ خطایی در هیچ لاگی. """ def _guard(self, text): from rag.management.commands.rag_verify_solutions import needs_figure return needs_figure(text) def test_opposite_angles_are_geometry_not_a_figure_reference(self): self.assertFalse(self._guard('آیا در هر لوزی زاویه‌های روبه‌رو با هم برابر است؟')) def test_data_that_follows_the_word_below_is_not_a_dangling_reference(self): self.assertFalse(self._guard('کسرهای زیر را گویا کنید: ۲/√۳')) self.assertFalse(self._guard('دستگاه زیر را حل کنید: ۲x - ۳y = ۷')) def test_shaped_like_is_not_a_reference_to_a_drawing(self): self.assertFalse(self._guard('ظرفی به شکل مخروطی با شعاع ۴ و ارتفاع ۱۲ داریم')) def test_a_reference_followed_by_its_description_is_self_contained(self): # همان چیزی که قاعده‌ی self-contained در پرامپتِ ingest تولید می‌کند. self.assertFalse(self._guard( 'با توجه به شکل مقابل که شامل مجموعه A با اعضای {a,1} است، ...')) def test_a_bare_reference_with_no_description_is_still_caught(self): self.assertTrue(self._guard('با توجه به شکل، مختصات هر نقطه را بنویسید.')) self.assertTrue(self._guard('مساحت شکل روبه‌رو چقدر است؟')) def test_colouring_tasks_can_never_be_judged_from_text(self): self.assertTrue(self._guard('شکل زیر را رنگ‌آمیزی کنید'))