from unittest.mock import patch from django.core.cache import cache from django.test import TestCase, override_settings from rest_framework.test import APIClient from rest_framework.throttling import ScopedRateThrottle from accounts.authentication import issue_access_token from accounts.models import User from .models import AskIdempotency, Conversation, Message from .views import generate_tutor_answer class AskIdempotencyFollowUpsTests(TestCase): """ 🔒 تستِ رگرسیونِ فیکس: _build_ask_response قبلاً برای بازسازیِ suggestedFollowUps دوباره generate_tutor_answer رو صدا می‌زد — که با یک AI provider واقعی یعنی یک فراخوانیِ واقعیِ دیگه به‌ازای هر replay. حالا suggested_follow_ups در همون درخواستِ اولِ موفق ذخیره می‌شه و replay بدونِ هیچ فراخوانیِ دیگری به generate_tutor_answer همون مقدار رو برمی‌گردونه. """ def setUp(self): # ⚠️ افزوده شده: بعد از اضافه‌شدنِ ScopedRateThrottle (scope='ai_ask') # به AskAIView، cache باید بینِ تست‌ها پاک بشه؛ وگرنه شمارشگرِ throttle # (که در محیطِ تست روی LocMemCache نگه‌داری می‌شه و به‌صورتِ خودکار # بینِ تست‌ها پاک نمی‌شه) می‌تونه از یک تستِ قبلی باقی بمونه. cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000021', name='') self.token = issue_access_token(self.user.id) def _ask(self, idempotency_key): return self.client.post( '/api/ai/ask', {'userQuestion': 'یک به‌علاوه یک چند می‌شود؟'}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', HTTP_IDEMPOTENCY_KEY=idempotency_key, ) def test_initial_request_stores_generated_follow_ups(self): response = self._ask('ask-key-1') self.assertEqual(response.status_code, 200) record = AskIdempotency.objects.get(user_id=self.user.id, idempotency_key='ask-key-1') self.assertEqual(record.suggested_follow_ups, response.data['suggestedFollowUps']) self.assertTrue(record.suggested_follow_ups) def test_idempotent_retry_returns_same_follow_ups_without_calling_generator_again(self): first_response = self._ask('ask-key-2') self.assertEqual(first_response.status_code, 200) original_follow_ups = first_response.data['suggestedFollowUps'] # ⚠️ در مسیرِ replay، generate_tutor_answer نباید اصلاً صدا زده # بشه؛ اگه صدا زده بشه، این patch باعثِ خطا می‌شه و تست fail # می‌کنه. with patch( 'ai_tutor.views.generate_tutor_answer', side_effect=AssertionError('generate_tutor_answer must not be called on idempotent replay'), ): retry_response = self._ask('ask-key-2') self.assertEqual(retry_response.status_code, 200) self.assertEqual(retry_response.data['suggestedFollowUps'], original_follow_ups) self.assertEqual(retry_response.json()['answer'], first_response.json()['answer']) self.assertEqual(retry_response.data['messageId'], first_response.data['messageId']) class AskAIGenerationFailureTests(TestCase): """ 🔒 تستِ رگرسیونِ فیکسِ اصلیِ این تسک: جداکردنِ فراخوانیِ generate_tutor_answer از قفلِ ردیفِ کاربر نباید رفتارِ صحتیِ سیستم رو خراب کنه. اگه generate_tutor_answer خطا بده (مثلاً چون یک AI provider واقعی موقتاً در دسترس نیست)، نباید: - هیچ Message یتیمی (پیامِ کاربر بدونِ پاسخ) در دیتابیس بمونه؛ - اسلاتِ کوتای روزانه برای همیشه مصرف‌شده حساب بشه. """ def setUp(self): # ⚠️ افزوده شده: نگاه کنید به توضیحِ مشابه در AskIdempotencyFollowUpsTests.setUp cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000022', name='') self.token = issue_access_token(self.user.id) def _ask(self, idempotency_key=None): headers = {} if idempotency_key: headers['HTTP_IDEMPOTENCY_KEY'] = idempotency_key return self.client.post( '/api/ai/ask', {'userQuestion': 'دو به‌علاوه دو چند می‌شود؟'}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', **headers, ) def test_generation_failure_leaves_no_orphan_message_and_returns_503(self): with patch( 'ai_tutor.views.generate_tutor_answer', side_effect=ConnectionError('provider unreachable'), ): response = self._ask() self.assertEqual(response.status_code, 503) self.assertEqual(Message.objects.filter(conversation__user_id=self.user.id).count(), 0) self.assertEqual(Conversation.objects.filter(user_id=self.user.id).count(), 0) def test_generation_failure_does_not_consume_quota_slot(self): # یک سوالِ ناموفق می‌فرستیم (که نباید کوتا مصرف کنه)، بعد دقیقاً # به اندازه‌ی سقفِ روزانه سوالِ موفق پشتِ سرِ هم می‌فرستیم؛ همه باید ۲۰۰ # بگیرن، یعنی سوالِ ناموفقِ اول اسلاتی از کوتا مصرف نکرده. from .views import daily_quota DAILY_AI_QUOTA = daily_quota(self.user.id) with patch( 'ai_tutor.views.generate_tutor_answer', side_effect=ConnectionError('provider unreachable'), ): failed_response = self._ask() self.assertEqual(failed_response.status_code, 503) for _ in range(DAILY_AI_QUOTA): ok_response = self._ask() self.assertEqual(ok_response.status_code, 200) # این سوالِ اضافه باید رد بشه چون سقف (نه یکی کمتر) واقعاً پر شده. throttled_response = self._ask() self.assertEqual(throttled_response.status_code, 429) def test_generation_failure_on_existing_conversation_keeps_conversation(self): # وقتی conversationId از قبل وجود داره (متعلق به این کاربر)، در # صورتِ شکستِ generate_tutor_answer فقط user_messageِ همین # درخواست باید حذف بشه، نه خودِ conversation (که ممکنه پیام‌های # دیگه‌ای هم داشته باشه). conversation = Conversation.objects.create(user_id=self.user.id) Message.objects.create(conversation=conversation, role='user', text='سوالِ قدیمی') Message.objects.create(conversation=conversation, role='assistant', text='پاسخِ قدیمی') with patch( 'ai_tutor.views.generate_tutor_answer', side_effect=ConnectionError('provider unreachable'), ): response = self.client.post( '/api/ai/ask', {'userQuestion': 'یک سوالِ جدید', 'conversationId': str(conversation.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) self.assertEqual(response.status_code, 503) self.assertTrue(Conversation.objects.filter(pk=conversation.id).exists()) # فقط دو پیامِ قدیمی باید باقی مونده باشن؛ پیامِ کاربرِ رزروشده‌ی # همین درخواستِ ناموفق باید پاک شده باشه. self.assertEqual(conversation.messages.count(), 2) class AskAIConcurrentIdempotencyKeyTests(TestCase): """ 🔒⚛️ تستِ رگرسیونِ فیکسِ این تسک برای حالتِ خیلی نادرِ رقابتِ هم‌زمانِ دو درخواست با دقیقاً همون Idempotency-Key، بعد از اینکه فراخوانیِ AI از زیرِ قفلِ ردیفِ کاربر بیرون اومده. این تست به‌جای threadهای واقعی (که روی SQLite ناپایدارن)، این رقابت رو به‌صورتِ قطعی/تکرارپذیر شبیه‌سازی می‌کنه: داخلِ side_effectِ generate_tutor_answerِ درخواستِ «اول»، یک درخواستِ HTTP کاملاً جداگانه (با همون کلید) فرستاده می‌شه که کاملاً پردازش می‌شه و AskIdempotency رو زودتر می‌سازه (برنده)؛ درخواستِ «اول» بعد از برگشتن از generate_tutor_answer باید در فازِ نهایی به IntegrityError بخوره، رزروِ خودش رو پاک کنه، و نتیجه‌ی همون برنده رو برگردونه. """ def setUp(self): # ⚠️ افزوده شده: نگاه کنید به توضیحِ مشابه در AskIdempotencyFollowUpsTests.setUp cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000023', name='') self.token = issue_access_token(self.user.id) def test_losing_request_cleans_up_and_returns_winners_result(self): key = 'race-key-1' inner_client = APIClient() # ⚠️ چون generate_tutor_answer برای کلِ طولِ این with-block پچ # شده، درخواستِ داخلیِ زیر هم به همین side_effect برمی‌خوره؛ برای # جلوگیری از بازگشتِ بی‌نهایت، فقط اولین فراخوانی (درخواستِ # بیرونی) درخواستِ داخلی رو می‌سازه؛ خودِ فراخوانیِ generate_tutor_ # answerِ درخواستِ داخلی (فراخوانیِ دوم) صرفاً یک پاسخِ قطعی # برمی‌گردونه. call_count = {'n': 0} def side_effect(user_question, grade_id, lesson_id, context=None): call_count['n'] += 1 if call_count['n'] == 1: # این «درخواستِ داخلی» با همون idempotency key، در طولِ # فراخوانیِ AIِ درخواستِ بیرونی (که قفلِ ردیفِ کاربر رو # دیگه نگه نداره) کاملاً اجرا و کامل می‌شه — یعنی برنده‌ی # رقابت می‌شه. resp = inner_client.post( '/api/ai/ask', {'userQuestion': user_question}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', HTTP_IDEMPOTENCY_KEY=key, ) assert resp.status_code == 200, resp.content side_effect.inner_response = resp return 'پاسخِ واقعی برای سوال', ['ادامه بده'] return 'پاسخِ دستیار برای درخواستِ داخلی', ['ادامه بده'] with patch('ai_tutor.views.generate_tutor_answer', side_effect=side_effect): outer_response = self.client.post( '/api/ai/ask', {'userQuestion': 'سوالِ رقابتی'}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', HTTP_IDEMPOTENCY_KEY=key, ) inner_response = side_effect.inner_response self.assertEqual(outer_response.status_code, 200) # هر دو پاسخ باید دقیقاً به همون AskIdempotency (نتیجه‌ی برنده) # اشاره کنن. self.assertEqual(outer_response.data['messageId'], inner_response.data['messageId']) self.assertEqual(outer_response.json()['answer'], inner_response.json()['answer']) # فقط یک رکوردِ AskIdempotency برای این کلید باید وجود داشته باشه. self.assertEqual( AskIdempotency.objects.filter(user_id=self.user.id, idempotency_key=key).count(), 1 ) # فقط یک پیامِ کاربر و یک پیامِ دستیار باید نهایتاً باقی مونده # باشن؛ پیامِ رزروشده‌ی درخواستِ بازنده باید پاک شده باشه (یعنی # کوتای روزانه فقط یک بار مصرف شده، نه دوبار). self.assertEqual( Message.objects.filter(conversation__user_id=self.user.id, role='user').count(), 1 ) self.assertEqual( Message.objects.filter(conversation__user_id=self.user.id, role='assistant').count(), 1 ) class AskAIConversationApiContractTests(TestCase): """ 🔒 تستِ رگرسیونِ این تسک: اضافه‌شدنِ مرزِ context به generate_tutor_answer نباید هیچ تغییری در قراردادِ API endpoint (POST /api/ai/ask) بده — نه برای شروعِ مکالمه‌ی جدید (بدونِ conversationId)، نه برای ادامه‌ی مکالمه‌ی موجود (با conversationId)، و نه در چک‌های مالکیت/احراز هویت. """ def setUp(self): # ⚠️ افزوده شده: نگاه کنید به توضیحِ مشابه در AskIdempotencyFollowUpsTests.setUp cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000024', name='') self.token = issue_access_token(self.user.id) self.other_user = User.objects.create(phone='09000000025', name='') self.other_token = issue_access_token(self.other_user.id) def test_no_conversation_id_creates_new_conversation(self): self.assertEqual(Conversation.objects.filter(user_id=self.user.id).count(), 0) response = self.client.post( '/api/ai/ask', {'userQuestion': 'سلام'}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) self.assertEqual(response.status_code, 200) conversations = Conversation.objects.filter(user_id=self.user.id) self.assertEqual(conversations.count(), 1) self.assertEqual(response.data['conversationId'], str(conversations.first().id)) def test_existing_conversation_id_continues_same_conversation(self): conversation = Conversation.objects.create(user_id=self.user.id) Message.objects.create(conversation=conversation, role='user', text='سوالِ قبلی') Message.objects.create(conversation=conversation, role='assistant', text='پاسخِ قبلی') response = self.client.post( '/api/ai/ask', {'userQuestion': 'سوالِ دوم', 'conversationId': str(conversation.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['conversationId'], str(conversation.id)) # هنوز فقط همون یک conversation باید وجود داشته باشه؛ چیزِ جدیدی # ساخته نشده. self.assertEqual(Conversation.objects.filter(user_id=self.user.id).count(), 1) self.assertEqual(conversation.messages.count(), 4) def test_conversation_id_belonging_to_another_user_is_rejected(self): other_conversation = Conversation.objects.create(user_id=self.other_user.id) response = self.client.post( '/api/ai/ask', {'userQuestion': 'سوال', 'conversationId': str(other_conversation.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) self.assertEqual(response.status_code, 404) self.assertEqual(other_conversation.messages.count(), 0) class GenerationBoundaryContextTests(TestCase): """ 🏗️ تستِ رگرسیونِ اصلیِ این تسک: مرزِ تولیدِ پاسخ (generate_tutor_answer(user_question, grade_id, lesson_id, context)) باید: - همچنان gradeId/lessonId رو بگیره (metadataِ لازم برای RAG/Qdrantِ آینده)، دقیقاً همون‌هایی که در بدنه‌ی API درخواست اومدن؛ - بتونه یک context اپلیکیشن‌-ساخته (رشته/شیء/لیستِ پیام) هم کنارِ اون‌ها بگیره، بدونِ اینکه بدونه اون context چطور retrieve شده؛ و AskAIView (که هنوز لایه‌ی RAG/Qdrant رو پیاده نمی‌کنه) نباید خودش تاریخچه‌ی خامِ مکالمه رو بسازه و به این مرز بفرسته. """ def setUp(self): # ⚠️ افزوده شده: نگاه کنید به توضیحِ مشابه در AskIdempotencyFollowUpsTests.setUp cache.clear() def test_generate_tutor_answer_accepts_grade_lesson_and_application_provided_context(self): # مرز باید grade_id/lesson_id رو (به‌عنوانِ metadataِ RAGِ آینده) و # context رو (چه رشته باشه، چه شیء/دیکشنری، چه لیستِ پیام‌های # retrieve-شده) به‌عنوانِ آرگومان‌هایی مستقل کنارِ سوالِ کاربر قبول # کنه. answer_text, follow_ups = generate_tutor_answer( 'مشتق تابع sin(x) چیست؟', 7, '42', context={'retrieved_snippets': ['قاعده‌ی مشتق‌گیریِ توابعِ مثلثاتی']}, ) self.assertIn('مشتق تابع sin(x) چیست؟', answer_text) self.assertTrue(follow_ups) # بدونِ context هم (رفتارِ فعلی/پیش‌فرض) باید عادی کار کنه؛ context # باید کاملاً اختیاری بمونه. answer_text_no_context, _ = generate_tutor_answer('یک سوال بدون context', None, None) self.assertIn('یک سوال بدون context', answer_text_no_context) def test_ask_ai_view_passes_grade_and_lesson_id_from_request_to_generation_boundary(self): # gradeId/lessonIdِ همون درخواستِ API باید دقیقاً همون‌طور که # فرستاده شدن به مرزِ تولید برسن — این metadataی لازم برای # RAG/Qdrantِ آینده‌ست، نه تاریخچه. client = APIClient() user = User.objects.create(phone='09000000026', name='') token = issue_access_token(user.id) with patch( 'ai_tutor.views.generate_tutor_answer', wraps=generate_tutor_answer, ) as mocked_generate: response = client.post( '/api/ai/ask', { 'userQuestion': 'مشتق چیست؟', 'gradeId': 10, 'lessonId': 'lesson-55', }, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', ) self.assertEqual(response.status_code, 200) mocked_generate.assert_called_once_with('مشتق چیست؟', 10, 'lesson-55', None) def test_ask_ai_view_does_not_send_conversation_history_to_generation_boundary(self): # این تستِ کلیدیِ این تسکه: وقتی AskAIView مرزِ تولید رو صدا می‌زنه، # نباید هیچ تاریخچه‌ی خامِ Message/Conversation (لیستی از پیام‌های # قبلی، یا خودِ QuerySet پیام‌ها) بهش پاس بده — چون RAG هنوز اینجا # پیاده نشده، context باید None بمونه (فقط gradeId/lessonId/سوال # مجازن). client = APIClient() user = User.objects.create(phone='09000000027', name='') token = issue_access_token(user.id) conversation = Conversation.objects.create(user_id=user.id) Message.objects.create(conversation=conversation, role='user', text='سوالِ قدیمیِ اول') Message.objects.create(conversation=conversation, role='assistant', text='پاسخِ قدیمیِ اول') Message.objects.create(conversation=conversation, role='user', text='سوالِ قدیمیِ دوم') Message.objects.create(conversation=conversation, role='assistant', text='پاسخِ قدیمیِ دوم') with patch( 'ai_tutor.views.generate_tutor_answer', wraps=generate_tutor_answer, ) as mocked_generate: response = client.post( '/api/ai/ask', {'userQuestion': 'سوالِ جدید', 'conversationId': str(conversation.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', ) self.assertEqual(response.status_code, 200) mocked_generate.assert_called_once() call_args, call_kwargs = mocked_generate.call_args all_args = list(call_args) + list(call_kwargs.values()) # آرگومانِ اول باید فقط همین سوالِ تازه باشه، نه چیزِ دیگه‌ای. self.assertEqual(call_args[0], 'سوالِ جدید') # هیچ آرگومانی نباید حاویِ متنِ پیام‌های قدیمی یا خودِ آبجکت‌های # Message/Conversation باشه — یعنی هیچ تاریخچه‌ی خامی به مرزِ تولید # پاس داده نشده. for arg in all_args: self.assertNotIsInstance(arg, (Message, Conversation)) if isinstance(arg, str): self.assertNotIn('قدیمی', arg) # context (چهارمین آرگومان) در این تسک باید None باشه (RAG/Qdrant # هنوز پیاده نشده)؛ gradeId/lessonId هم چون در این درخواست فرستاده # نشدن، None هستن. context_arg = call_args[3] if len(call_args) > 3 else call_kwargs.get('context') self.assertIsNone(context_arg) class AskAIViewThrottleTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ ScopedRateThrottle (scope='ai_ask') روی AskAIView: قبلاً این endpoint هیچ throttle مخصوصی نداشت (فقط DAILY_AI_QUOTA که سقفِ روزانه است، نه محدودیتِ burst کوتاه‌مدت). این تست ثابت می‌کند که درخواست‌های مجاز داخلِ نرخ همچنان کار می‌کنند و درخواستِ بیش از حد با 429 رد می‌شود — بدون اینکه به کوتای روزانه یا idempotency دست بزند. """ def setUp(self): self.client = APIClient() cache.clear() self.user = User.objects.create(phone='09000000028', name='') self.token = issue_access_token(self.user.id) def _ask(self, idempotency_key): return self.client.post( '/api/ai/ask', {'userQuestion': 'یک به‌علاوه یک چند می‌شود؟'}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', HTTP_IDEMPOTENCY_KEY=idempotency_key, ) # سقفِ روزانه (پیش‌فرض ۲) نباید پیش از throttle برسد؛ این تست فقط throttle را می‌سنجد. @override_settings(AI_DAILY_QUOTA_FREE=10) def test_requests_within_rate_succeed_and_excess_is_throttled(self): # هر درخواست idempotency-key متفاوت دارد تا هر بار واقعاً از کوتای # روزانه مصرف کند (نه یک replay ساده)، و صرفاً throttle سنجیده شود. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'ai_ask': '3/hour'}): for i in range(3): response = self._ask(f'ai-ask-throttle-key-{i}') self.assertEqual(response.status_code, 200) throttled_response = self._ask('ai-ask-throttle-key-3') self.assertEqual(throttled_response.status_code, 429) class AskAIViewPerUserThrottleTests(TestCase): """هر کاربرِ ثبت‌نام‌شده سطلِ throttleِ جدای خودش را دارد؛ مصرفِ حسابِ دیگر به او نمی‌رسد.""" def setUp(self): cache.clear() self.client = APIClient() def _ask(self, token, idempotency_key): return self.client.post( '/api/ai/ask', {'userQuestion': 'یک به‌علاوه یک چند می‌شود؟'}, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', HTTP_IDEMPOTENCY_KEY=idempotency_key, ) def test_registered_users_keep_independent_per_user_buckets(self): # عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ # قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن — throttle # برای اونا هنوز بر اساسِ user.pk است، نه IP. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'ai_ask': '1/hour'}): user_a = User.objects.create(name='', phone='+989120000101') response_a = self._ask(issue_access_token(user_a.id), 'user-a-key-1') self.assertEqual(response_a.status_code, 200) user_b = User.objects.create(name='', phone='+989120000102') response_b = self._ask(issue_access_token(user_b.id), 'user-b-key-1') self.assertEqual(response_b.status_code, 200) class StubNeverReachesProductionTests(TestCase): """بی‌کلیدِ RAG و بیرون از DEBUG، دانش‌آموز ۵۰۳ می‌گیرد، نه متنِ «پاسخ نمونه (stub)».""" @override_settings(DEBUG=False, OPENAI_API_KEY='', QDRANT_API_KEY='', RAG_FORCE_MOCK=False) def test_missing_keys_are_a_503_and_cost_nothing(self): cache.clear() user = User.objects.create(phone='09000000029', name='') response = APIClient().post( '/api/ai/ask', {'userQuestion': '۲+۲ چند است؟'}, format='json', HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}', ) self.assertEqual(response.status_code, 503) self.assertFalse(Message.objects.filter(conversation__user_id=user.id).exists()) class GlobalDailyCapTests(TestCase): """سقفِ روزانه‌ی کلِ سامانه برای سوال‌هایی که به مدل می‌رسند (AI_GLOBAL_DAILY_MAX).""" @override_settings(AI_GLOBAL_DAILY_MAX=2) def test_the_whole_system_stops_at_the_daily_cap(self): cache.clear() codes = [] for i in range(3): user = User.objects.create(phone=f'0900{i:07d}', name='') codes.append(APIClient().post( '/api/ai/ask', {'userQuestion': f'۲+{i} چند است؟'}, format='json', HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}', REMOTE_ADDR=f'10.0.0.{i}', ).status_code) self.assertEqual(codes, [200, 200, 503])