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(device_id='device-ask-1', is_guest=True, 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.data['answer'], first_response.data['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(device_id='device-ask-fail-1', is_guest=True, 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(device_id='device-ask-race-1', is_guest=True, 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.data['answer'], inner_response.data['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(device_id='device-contract-1', is_guest=True, name='') self.token = issue_access_token(self.user.id) self.other_user = User.objects.create(device_id='device-contract-2', is_guest=True, 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(device_id='device-boundary-grade-1', is_guest=True, 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(device_id='device-boundary-1', is_guest=True, 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(device_id='device-ai-throttle-1', is_guest=True, 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_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 AskAIViewGuestThrottleBypassTests(TestCase): """ 🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='ai_ask' برای هر کاربرِ احرازهویت‌شده (مهمان یا نه) user.pk بود. چون GuestSessionView با هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک bucket throttle تازه برای AskAIView بگیره. حالا GuestAwareAIThrottle برای کاربرهای مهمان کلید رو بر اساسِ IP (همون get_ident که guest_session هم استفاده می‌کنه) می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده. رفتارِ throttle برای کاربرِ غیرمهمان (per-user، مثلِ قبل) دست‌نخورده مونده. """ 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_rotating_guest_accounts_share_the_same_throttle_bucket(self): # سه حساب مهمانِ متفاوت (شبیه‌سازیِ چرخشِ حساب برای دورزدنِ throttle)، # همه از همون کلاینت/IP — باید همه یک bucket مشترک رو مصرف کنند. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'ai_ask': '2/hour'}): guest1 = User.objects.create(device_id='device-guest-rot-1', is_guest=True, name='') response1 = self._ask(issue_access_token(guest1.id), 'guest-rot-key-1') self.assertEqual(response1.status_code, 200) guest2 = User.objects.create(device_id='device-guest-rot-2', is_guest=True, name='') response2 = self._ask(issue_access_token(guest2.id), 'guest-rot-key-2') self.assertEqual(response2.status_code, 200) # سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون # bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk. guest3 = User.objects.create(device_id='device-guest-rot-3', is_guest=True, name='') response3 = self._ask(issue_access_token(guest3.id), 'guest-rot-key-3') self.assertEqual(response3.status_code, 429) 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(is_guest=False, 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(is_guest=False, 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(device_id='device-no-keys', is_guest=True, 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())