from datetime import timedelta from unittest.mock import patch from django.test import TestCase from django.utils import timezone 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 progress.models import UserHearts, UserStats from subscription.models import Plan, Subscription from .models import Choice, Grade, Lesson, Node, NodeProgress, Question, Unit, UserAnswer, UserLessonCompletion class LessonCompleteImpossibleCountTests(TestCase): """رگرسیون: correctCount باید با تعدادِ واقعیِ سوال‌های درس مقایسه بشه، وگرنه کلاینتِ مخرب می‌تونه xp/coins رو متورم کنه.""" def setUp(self): self.client = APIClient() self.user = User.objects.create(device_id='device-complete-1', is_guest=True, name='') self.token = issue_access_token(self.user.id) grade = Grade.objects.create(title='پایه اول', icon='icon', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1) node = Node.objects.create(unit=unit, title='گره ۱', order=1) self.lesson = Lesson.objects.create(node=node, title='درس ۱', intro_text='...') for i in range(3): question = Question.objects.create( lesson=self.lesson, type='multiple_choice', prompt=f'سوال {i}', ) Choice.objects.create(question=question, value='درست', is_correct=True) def _complete(self, correct_count, total_questions): # مثلِ اپ: جواب‌ها اول ثبت می‌شوند؛ امتیاز فقط برای جوابِ ثبت‌شده است. for question in self.lesson.questions.order_by('id')[:max(correct_count, 0)]: self.client.post( f'/api/questions/{question.id}/answer', {'selectedOptionId': str(question.choices.get(is_correct=True).id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) return self.client.post( f'/api/lessons/{self.lesson.id}/complete', {'correctCount': correct_count, 'totalQuestions': total_questions}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) def test_correct_count_exceeding_actual_questions_is_rejected(self): # فقط ۳ سوال واقعی در این درس هست؛ کلاینت ادعای ۱۰۰ تای درست از # ۱۰ تای کل رو می‌کنه. response = self._complete(correct_count=100, total_questions=10) self.assertEqual(response.status_code, 400) self.assertIn('correctCount', response.data) def test_negative_correct_count_is_rejected(self): response = self._complete(correct_count=-1, total_questions=3) self.assertEqual(response.status_code, 400) def test_zero_total_questions_is_rejected(self): response = self._complete(correct_count=1, total_questions=0) self.assertEqual(response.status_code, 400) def test_valid_completion_still_grants_expected_reward(self): # ۳ از ۳ سوالِ واقعیِ همین درس، درست — باید ۳ ستاره و xp کامل بده. response = self._complete(correct_count=3, total_questions=3) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['stars'], 3) self.assertEqual(response.data['xpEarned'], 30) self.assertEqual(response.data['coinsEarned'], 15) class LessonCompleteRatioIgnoresClientTotalQuestionsTests(TestCase): """رگرسیون: ratio باید از actual_question_count محاسبه بشه نه از totalQuestیِ ارسالیِ کلاینت، وگرنه حتی بعدِ سقف‌زدنِ correctCount کلاینت می‌تونه با totalQuestions دروغین ratio رو مصنوعاً ۱۰۰٪ کنه.""" def setUp(self): self.client = APIClient() self.user = User.objects.create(device_id='device-complete-2', is_guest=True, name='') self.token = issue_access_token(self.user.id) grade = Grade.objects.create(title='پایه دوم', icon='icon', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۲', order=1) node = Node.objects.create(unit=unit, title='گره ۲', order=1) self.lesson = Lesson.objects.create(node=node, title='درس ۲', intro_text='...') for i in range(3): question = Question.objects.create( lesson=self.lesson, type='multiple_choice', prompt=f'سوال {i}', ) Choice.objects.create(question=question, value='درست', is_correct=True) def _complete(self, correct_count, total_questions): # مثلِ اپ: جواب‌ها اول ثبت می‌شوند؛ امتیاز فقط برای جوابِ ثبت‌شده است. for question in self.lesson.questions.order_by('id')[:max(correct_count, 0)]: self.client.post( f'/api/questions/{question.id}/answer', {'selectedOptionId': str(question.choices.get(is_correct=True).id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) return self.client.post( f'/api/lessons/{self.lesson.id}/complete', {'correctCount': correct_count, 'totalQuestions': total_questions}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) def test_manipulated_total_questions_does_not_yield_full_stars(self): # درسِ ۳-سوالی؛ کلاینت فقط ۱ درست داشته ولی totalQuestions=1 # می‌فرسته تا وانمود کنه ۱۰۰٪ جواب داده. response = self._complete(correct_count=1, total_questions=1) self.assertEqual(response.status_code, 200) self.assertNotEqual(response.data['stars'], 3) def test_ratio_is_based_on_actual_lesson_question_count(self): # ۱ از ۳ سوالِ واقعی = ratio نزدیکِ ۰.۳۳ → طبقِ آستانه‌های موجود # (>=0.9 → 3, >=0.6 → 2, >0 → 1)، باید دقیقاً ۱ ستاره بده. response = self._complete(correct_count=1, total_questions=1) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['stars'], 1) class QuestionAnswerThrottleTests(TestCase): """محدودیتِ per-user روی QuestionAnswerView (throttle_scope='question_answer')؛ Idempotency-Key فقط retry با همون کلید رو می‌گیره، نه حجمِ درخواست‌های متفاوت.""" def setUp(self): from django.core.cache import cache cache.clear() self.client = APIClient() self.user = User.objects.create(device_id='device-answer-throttle-1', is_guest=True, name='') self.token = issue_access_token(self.user.id) grade = Grade.objects.create(title='پایه اول', icon='icon', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1) self.node = Node.objects.create(unit=unit, title='گره ۱', order=1) self.lesson = Lesson.objects.create(node=self.node, title='درس ۱', intro_text='...') self.question = Question.objects.create( lesson=self.lesson, type='multiple_choice', prompt='سوال ۱', ) self.correct_choice = Choice.objects.create( question=self.question, value='درست', is_correct=True, ) def _answer(self, idempotency_key=None, remote_addr=None): kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'} if idempotency_key: kwargs['HTTP_IDEMPOTENCY_KEY'] = idempotency_key if remote_addr: kwargs['REMOTE_ADDR'] = remote_addr return self.client.post( f'/api/questions/{self.question.id}/answer', {'selectedOptionId': str(self.correct_choice.id)}, format='json', **kwargs, ) def test_requests_within_the_limit_still_succeed(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '3/min'}): for _ in range(3): response = self._answer() self.assertEqual(response.status_code, 200) def test_excessive_requests_from_same_user_return_429(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '3/min'}): for _ in range(3): response = self._answer() self.assertEqual(response.status_code, 200) blocked_response = self._answer() self.assertEqual(blocked_response.status_code, 429) def test_changing_client_ip_does_not_bypass_the_user_limit(self): # ⚠️ توجه: از یک کاربرِ *غیرمهمان* استفاده می‌شه (نه self.user که # مهمان است)، چون طبقِ فیکسِ GuestAwareContentThrottle حالا کلیدِ # bucket برای کاربرهای مهمان عمداً بر اساسِ IP است (نگاه کنید به # QuestionAnswerGuestThrottleBypassTests پایین‌تر) — این تست فقط # رفتارِ per-user (نه IP) برای کاربرهایِ ثبت‌نام‌شده رو تضمین # می‌کنه که باید دقیقاً مثلِ قبل دست‌نخورده بمونه. registered_user = User.objects.create(is_guest=False, name='', phone='+989120000701') registered_token = issue_access_token(registered_user.id) def _answer_as(remote_addr): return self.client.post( f'/api/questions/{self.question.id}/answer', {'selectedOptionId': str(self.correct_choice.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {registered_token}', REMOTE_ADDR=remote_addr, ) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '3/min'}): for i in range(3): response = _answer_as(f'10.2.0.{i + 1}') self.assertEqual(response.status_code, 200) blocked_response = _answer_as('10.2.0.99') self.assertEqual(blocked_response.status_code, 429) def test_throttled_request_creates_no_answer_and_no_hearts_or_progress_change(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '1/min'}): first_response = self._answer() self.assertEqual(first_response.status_code, 200) answer_count_before = UserAnswer.objects.filter(user_id=self.user.id).count() hearts_before = UserHearts.objects.filter(user_id=self.user.id).first() hearts_snapshot = hearts_before.current_hearts if hearts_before else None progress_before = NodeProgress.objects.filter(user_id=self.user.id, node=self.node).first() progress_snapshot = ( {'status': progress_before.status, 'last_activity_at': progress_before.last_activity_at} if progress_before else None ) blocked_response = self._answer() self.assertEqual(blocked_response.status_code, 429) self.assertEqual( UserAnswer.objects.filter(user_id=self.user.id).count(), answer_count_before, ) hearts_after = UserHearts.objects.filter(user_id=self.user.id).first() self.assertEqual( hearts_after.current_hearts if hearts_after else None, hearts_snapshot, ) progress_after = NodeProgress.objects.filter(user_id=self.user.id, node=self.node).first() if progress_snapshot is None: self.assertIsNone(progress_after) else: self.assertEqual(progress_after.status, progress_snapshot['status']) self.assertEqual(progress_after.last_activity_at, progress_snapshot['last_activity_at']) def test_idempotency_key_retry_still_works_within_the_limit(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '10/min'}): first_response = self._answer(idempotency_key='retry-key-1') self.assertEqual(first_response.status_code, 200) second_response = self._answer(idempotency_key='retry-key-1') self.assertEqual(second_response.status_code, 200) self.assertEqual( UserAnswer.objects.filter( user_id=self.user.id, question=self.question, idempotency_key='retry-key-1', ).count(), 1, ) class LessonCompleteThrottleTests(TestCase): """محدودیتِ per-user روی LessonCompleteView (throttle_scope='lesson_complete')؛ Idempotency-Key فقط retry با همون کلید رو می‌گیره، نه حجمِ درخواست‌های متفاوت.""" def setUp(self): from django.core.cache import cache cache.clear() self.client = APIClient() self.user = User.objects.create(device_id='device-complete-throttle-1', is_guest=True, name='') self.token = issue_access_token(self.user.id) grade = Grade.objects.create(title='پایه اول', icon='icon', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1) self.node = Node.objects.create(unit=unit, title='گره ۱', order=1) self.lesson = Lesson.objects.create(node=self.node, title='درس ۱', intro_text='...') for i in range(3): Question.objects.create( lesson=self.lesson, type='multiple_choice', prompt=f'سوال {i}', ) def _complete(self, idempotency_key=None, remote_addr=None): kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'} if idempotency_key: kwargs['HTTP_IDEMPOTENCY_KEY'] = idempotency_key if remote_addr: kwargs['REMOTE_ADDR'] = remote_addr return self.client.post( f'/api/lessons/{self.lesson.id}/complete', {'correctCount': 1, 'totalQuestions': 3}, format='json', **kwargs, ) def test_requests_within_the_limit_still_succeed(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'lesson_complete': '3/min'}): for _ in range(3): response = self._complete() self.assertEqual(response.status_code, 200) def test_excessive_requests_from_same_user_return_429(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'lesson_complete': '3/min'}): for _ in range(3): response = self._complete() self.assertEqual(response.status_code, 200) blocked_response = self._complete() self.assertEqual(blocked_response.status_code, 429) def test_changing_client_ip_does_not_bypass_the_user_limit(self): # ⚠️ توجه: از یک کاربرِ *غیرمهمان* استفاده می‌شه (نه self.user که # مهمان است)، چون طبقِ فیکسِ GuestAwareContentThrottle حالا کلیدِ # bucket برای کاربرهای مهمان عمداً بر اساسِ IP است (نگاه کنید به # LessonCompleteGuestThrottleBypassTests پایین‌تر) — این تست فقط # رفتارِ per-user (نه IP) برای کاربرهایِ ثبت‌نام‌شده رو تضمین # می‌کنه که باید دقیقاً مثلِ قبل دست‌نخورده بمونه. registered_user = User.objects.create(is_guest=False, name='', phone='+989120000702') registered_token = issue_access_token(registered_user.id) def _complete_as(remote_addr): return self.client.post( f'/api/lessons/{self.lesson.id}/complete', {'correctCount': 1, 'totalQuestions': 3}, format='json', HTTP_AUTHORIZATION=f'Bearer {registered_token}', REMOTE_ADDR=remote_addr, ) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'lesson_complete': '3/min'}): for i in range(3): response = _complete_as(f'10.3.0.{i + 1}') self.assertEqual(response.status_code, 200) blocked_response = _complete_as('10.3.0.99') self.assertEqual(blocked_response.status_code, 429) def test_throttled_request_creates_no_completion_or_other_side_effects(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'lesson_complete': '1/min'}): first_response = self._complete() self.assertEqual(first_response.status_code, 200) completion_count_before = UserLessonCompletion.objects.filter(user_id=self.user.id).count() stats_before = UserStats.objects.filter(user_id=self.user.id).first() stats_snapshot = ( {'xp': stats_before.xp, 'coins': stats_before.coins, 'progress_percent': stats_before.progress_percent} if stats_before else None ) progress_before = NodeProgress.objects.filter(user_id=self.user.id, node=self.node).first() progress_snapshot = ( {'status': progress_before.status, 'last_activity_at': progress_before.last_activity_at} if progress_before else None ) blocked_response = self._complete() self.assertEqual(blocked_response.status_code, 429) self.assertEqual( UserLessonCompletion.objects.filter(user_id=self.user.id).count(), completion_count_before, ) stats_after = UserStats.objects.filter(user_id=self.user.id).first() if stats_snapshot is None: self.assertIsNone(stats_after) else: self.assertEqual(stats_after.xp, stats_snapshot['xp']) self.assertEqual(stats_after.coins, stats_snapshot['coins']) self.assertEqual(stats_after.progress_percent, stats_snapshot['progress_percent']) progress_after = NodeProgress.objects.filter(user_id=self.user.id, node=self.node).first() if progress_snapshot is None: self.assertIsNone(progress_after) else: self.assertEqual(progress_after.status, progress_snapshot['status']) self.assertEqual(progress_after.last_activity_at, progress_snapshot['last_activity_at']) class QuestionAnswerGuestThrottleBypassTests(TestCase): """اثباتِ فیکسِ throttle: چرخشِ حساب مهمان (هر deviceId یه pk تازه) نباید bucket تازه‌ی question_answer بده — GuestAwareContentThrottle برای مهمان‌ها کلید رو بر اساسِ IP می‌سازه، نه user.pk.""" def setUp(self): from django.core.cache import cache cache.clear() self.client = APIClient() grade = Grade.objects.create(title='پایه اول', icon='icon', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1) node = Node.objects.create(unit=unit, title='گره ۱', order=1) self.lesson = Lesson.objects.create(node=node, title='درس ۱', intro_text='...') self.question = Question.objects.create( lesson=self.lesson, type='multiple_choice', prompt='سوال ۱', ) self.correct_choice = Choice.objects.create( question=self.question, value='درست', is_correct=True, ) def _answer(self, token): return self.client.post( f'/api/questions/{self.question.id}/answer', {'selectedOptionId': str(self.correct_choice.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', ) def test_rotating_guest_accounts_share_the_same_throttle_bucket(self): # سه حساب مهمانِ متفاوت (شبیه‌سازیِ چرخشِ حساب برای دورزدنِ throttle)، # همه از همون کلاینت/IP — باید همه یک bucket مشترک رو مصرف کنند. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '2/min'}): guest1 = User.objects.create(device_id='device-answer-guest-rot-1', is_guest=True, name='') response1 = self._answer(issue_access_token(guest1.id)) self.assertEqual(response1.status_code, 200) guest2 = User.objects.create(device_id='device-answer-guest-rot-2', is_guest=True, name='') response2 = self._answer(issue_access_token(guest2.id)) self.assertEqual(response2.status_code, 200) # سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون # bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk. guest3 = User.objects.create(device_id='device-answer-guest-rot-3', is_guest=True, name='') response3 = self._answer(issue_access_token(guest3.id)) self.assertEqual(response3.status_code, 429) def test_registered_users_keep_independent_per_user_buckets(self): # عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ # قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '1/min'}): user_a = User.objects.create(is_guest=False, name='', phone='+989120000711') response_a = self._answer(issue_access_token(user_a.id)) self.assertEqual(response_a.status_code, 200) user_b = User.objects.create(is_guest=False, name='', phone='+989120000712') response_b = self._answer(issue_access_token(user_b.id)) self.assertEqual(response_b.status_code, 200) def test_blocked_guest_request_creates_no_answer_or_hearts_change(self): # درخواستِ بلاک‌شده (از حسابِ مهمانِ چرخشی) نباید UserAnswer بسازه # یا هارت/progress رو تغییر بده. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'question_answer': '1/min'}): guest1 = User.objects.create(device_id='device-answer-guest-rot-4', is_guest=True, name='') first_response = self._answer(issue_access_token(guest1.id)) self.assertEqual(first_response.status_code, 200) guest2 = User.objects.create(device_id='device-answer-guest-rot-5', is_guest=True, name='') blocked_response = self._answer(issue_access_token(guest2.id)) self.assertEqual(blocked_response.status_code, 429) self.assertEqual(UserAnswer.objects.filter(user_id=guest2.id).count(), 0) self.assertFalse(UserHearts.objects.filter(user_id=guest2.id).exists()) class LessonCompleteGuestThrottleBypassTests(TestCase): """اثباتِ فیکسِ throttle: چرخشِ حساب مهمان نباید bucket تازه‌ی lesson_complete بده — GuestAwareContentThrottle برای مهمان‌ها کلید رو بر اساسِ IP می‌سازه، نه user.pk.""" def setUp(self): from django.core.cache import cache cache.clear() self.client = APIClient() grade = Grade.objects.create(title='پایه اول', icon='icon', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1) self.node = Node.objects.create(unit=unit, title='گره ۱', order=1) self.lesson = Lesson.objects.create(node=self.node, title='درس ۱', intro_text='...') for i in range(3): Question.objects.create( lesson=self.lesson, type='multiple_choice', prompt=f'سوال {i}', ) def _complete(self, token): return self.client.post( f'/api/lessons/{self.lesson.id}/complete', {'correctCount': 1, 'totalQuestions': 3}, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', ) def test_rotating_guest_accounts_share_the_same_throttle_bucket(self): # سه حساب مهمانِ متفاوت (شبیه‌سازیِ چرخشِ حساب برای دورزدنِ throttle)، # همه از همون کلاینت/IP — باید همه یک bucket مشترک رو مصرف کنند. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'lesson_complete': '2/hour'}): guest1 = User.objects.create(device_id='device-complete-guest-rot-1', is_guest=True, name='') response1 = self._complete(issue_access_token(guest1.id)) self.assertEqual(response1.status_code, 200) guest2 = User.objects.create(device_id='device-complete-guest-rot-2', is_guest=True, name='') response2 = self._complete(issue_access_token(guest2.id)) self.assertEqual(response2.status_code, 200) # سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون # bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk. guest3 = User.objects.create(device_id='device-complete-guest-rot-3', is_guest=True, name='') response3 = self._complete(issue_access_token(guest3.id)) self.assertEqual(response3.status_code, 429) def test_registered_users_keep_independent_per_user_buckets(self): # عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ # قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'lesson_complete': '1/hour'}): user_a = User.objects.create(is_guest=False, name='', phone='+989120000721') response_a = self._complete(issue_access_token(user_a.id)) self.assertEqual(response_a.status_code, 200) user_b = User.objects.create(is_guest=False, name='', phone='+989120000722') response_b = self._complete(issue_access_token(user_b.id)) self.assertEqual(response_b.status_code, 200) def test_blocked_guest_request_creates_no_completion_or_reward_side_effects(self): # درخواستِ بلاک‌شده (از حسابِ مهمانِ چرخشی) نباید UserLessonCompletion # بسازه یا UserStats/NodeProgress رو تغییر بده. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'lesson_complete': '1/hour'}): guest1 = User.objects.create(device_id='device-complete-guest-rot-4', is_guest=True, name='') first_response = self._complete(issue_access_token(guest1.id)) self.assertEqual(first_response.status_code, 200) guest2 = User.objects.create(device_id='device-complete-guest-rot-5', is_guest=True, name='') blocked_response = self._complete(issue_access_token(guest2.id)) self.assertEqual(blocked_response.status_code, 429) self.assertEqual(UserLessonCompletion.objects.filter(user_id=guest2.id).count(), 0) self.assertFalse(UserStats.objects.filter(user_id=guest2.id).exists()) self.assertFalse(NodeProgress.objects.filter(user_id=guest2.id, node=self.node).exists()) class QuestionAnswerMascotMoodTests(TestCase): """پاسخِ /answer باید mascotMood بده: درست → happy، غلط → encourage (هم‌خوان با RobotMood در MathApp_frontend).""" def setUp(self): from django.core.cache import cache cache.clear() self.client = APIClient() self.user = User.objects.create(device_id='device-mascot-1', is_guest=True, name='') self.token = issue_access_token(self.user.id) grade = Grade.objects.create(title='پایه اول', icon='i', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1) node = Node.objects.create(unit=unit, title='گره ۱', order=1) lesson = Lesson.objects.create(node=node, title='درس ۱', intro_text='...') self.question = Question.objects.create(lesson=lesson, type='multiple_choice', prompt='۱+۱؟') self.correct = Choice.objects.create(question=self.question, value='۲', is_correct=True) self.wrong = Choice.objects.create(question=self.question, value='۳', is_correct=False) def _answer(self, choice_id): return self.client.post( f'/api/questions/{self.question.id}/answer', {'selectedOptionId': str(choice_id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) def test_correct_answer_mood_happy(self): r = self._answer(self.correct.id) self.assertEqual(r.status_code, 200) self.assertEqual(r.data['mascotMood'], 'happy') def test_wrong_answer_mood_encourage(self): r = self._answer(self.wrong.id) self.assertEqual(r.status_code, 200) self.assertEqual(r.data['mascotMood'], 'encourage') def test_idempotent_replay_keeps_mood(self): k = 'mascot-idem-1' first = self.client.post( f'/api/questions/{self.question.id}/answer', {'selectedOptionId': str(self.wrong.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', HTTP_IDEMPOTENCY_KEY=k, ) replay = self.client.post( f'/api/questions/{self.question.id}/answer', {'selectedOptionId': str(self.wrong.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', HTTP_IDEMPOTENCY_KEY=k, ) self.assertEqual(first.data['mascotMood'], 'encourage') self.assertEqual(replay.data['mascotMood'], 'encourage') # ───────────────────────────────────────────────────────────────────────────── # مقیاس‌پذیریِ HomeView: رفعِ N+1، کشِ برنامه‌ی درسی، ۴۰۴ گریدِ ناموجود # ───────────────────────────────────────────────────────────────────────────── from django.core.cache import cache as _django_cache from django.test.utils import CaptureQueriesContext from django.db import connection class HomeViewScalabilityTests(TestCase): def setUp(self): _django_cache.clear() self.client = APIClient() self.user = User.objects.create(device_id='dev-home-scale', is_guest=True, name='') self.token = issue_access_token(self.user.id) def _grade_with_nodes(self, n_nodes): grade = Grade.objects.create(title=f'g{n_nodes}', icon='i', color='#fff') unit = Unit.objects.create(grade=grade, title='u', order=1) for i in range(n_nodes): node = Node.objects.create(unit=unit, title=f'n{i}', order=i) Lesson.objects.create(node=node, title=f'l{i}', intro_text='.') return grade def _home(self, grade_id): return self.client.get( f'/api/home?gradeId={grade_id}', HTTP_AUTHORIZATION=f'Bearer {self.token}') def test_query_count_is_flat_regardless_of_node_count(self): g_small = self._grade_with_nodes(2) g_big = self._grade_with_nodes(20) _django_cache.clear() with CaptureQueriesContext(connection) as q_small: self.assertEqual(self._home(g_small.id).status_code, 200) _django_cache.clear() with CaptureQueriesContext(connection) as q_big: self.assertEqual(self._home(g_big.id).status_code, 200) # قبلِ رفعِ N+1، q_big چند برابرِ q_small بود (یک COUNT به‌ازای هر node). self.assertEqual(len(q_small), len(q_big)) def test_second_call_hits_cache_not_curriculum_tables(self): g = self._grade_with_nodes(5) self.assertEqual(self._home(g.id).status_code, 200) # پر کردنِ کش with CaptureQueriesContext(connection) as q: self.assertEqual(self._home(g.id).status_code, 200) sql = ' '.join(e['sql'] for e in q).lower() self.assertNotIn('content_unit', sql) # content_nodeprogress دیتای کاربر است (جزیره‌های تمام‌شده)، نه برنامه‌ی درسی. self.assertNotRegex(sql, r'content_node(?!progress)') # فقط کوئریِ per-user باید بمونه self.assertIn('userlessoncompletion', sql.replace('_', '')) def test_content_change_invalidates_tree_cache(self): g = self._grade_with_nodes(3) self._home(g.id) unit = g.units.first() Node.objects.create(unit=unit, title='new-node', order=99) # post_save → bump version r = self._home(g.id) node_titles = [n['title'] for u in r.data['units'] for n in u['nodes']] self.assertIn('new-node', node_titles) def test_missing_grade_returns_404(self): r = self._home(999999) self.assertEqual(r.status_code, 404) def test_grade_with_no_units_returns_empty_not_404(self): g = Grade.objects.create(title='empty', icon='i', color='#fff') r = self._home(g.id) self.assertEqual(r.status_code, 200) self.assertEqual(r.data['units'], []) class GradesListCacheTests(TestCase): def setUp(self): _django_cache.clear() self.client = APIClient() def test_list_is_cached_and_invalidated_on_new_grade(self): Grade.objects.create(title='اول', icon='i', color='#111') r1 = self.client.get('/api/grades') self.assertEqual(len(r1.data['grades']), 1) with CaptureQueriesContext(connection) as q: self.client.get('/api/grades') self.assertEqual(len([e for e in q if 'content_grade' in e['sql'].lower()]), 0) Grade.objects.create(title='دوم', icon='i', color='#222') # post_save → bump r3 = self.client.get('/api/grades') self.assertEqual(len(r3.data['grades']), 2) class UnlimitedHeartsSubscriptionTests(TestCase): """رگرسیون: هر مشترکِ فعال (فارغ از پلن) نباید با جوابِ غلط هارت از دست بده.""" def setUp(self): self.client = APIClient() self.user = User.objects.create(device_id='dev-unlimited-hearts', is_guest=True, name='') self.token = issue_access_token(self.user.id) grade = Grade.objects.create(title='پایه اول', icon='icon', color='#fff') unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1) node = Node.objects.create(unit=unit, title='گره ۱', order=1) lesson = Lesson.objects.create(node=node, title='درس ۱', intro_text='...') self.question = Question.objects.create(lesson=lesson, type='multiple_choice', prompt='سوال ۱') Choice.objects.create(question=self.question, value='درست', is_correct=True) self.wrong_choice = Choice.objects.create(question=self.question, value='غلط', is_correct=False) def _answer_wrong(self): return self.client.post( f'/api/questions/{self.question.id}/answer', {'selectedOptionId': str(self.wrong_choice.id)}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) def test_wrong_answer_costs_a_heart_without_subscription(self): UserHearts.objects.create(user_id=self.user.id, current_hearts=5, max_hearts=5) response = self._answer_wrong() self.assertEqual(response.status_code, 200) self.assertEqual(response.data['heartsRemaining'], 4) def test_wrong_answer_does_not_cost_a_heart_with_active_subscription(self): UserHearts.objects.create(user_id=self.user.id, current_hearts=5, max_hearts=5) plan = Plan.objects.create(title='ماهانه', price=100000) Subscription.objects.create( user_id=self.user.id, plan=plan, status='active', expires_at=timezone.now() + timedelta(days=30), ) response = self._answer_wrong() self.assertEqual(response.status_code, 200) self.assertEqual(response.data['heartsRemaining'], 5) self.assertEqual(UserHearts.objects.get(user_id=self.user.id).current_hearts, 5) def test_expired_subscription_still_costs_a_heart(self): UserHearts.objects.create(user_id=self.user.id, current_hearts=5, max_hearts=5) plan = Plan.objects.create(title='ماهانه', price=100000) Subscription.objects.create( user_id=self.user.id, plan=plan, status='active', expires_at=timezone.now() - timedelta(days=1), ) response = self._answer_wrong() self.assertEqual(response.status_code, 200) self.assertEqual(response.data['heartsRemaining'], 4) class ReadThrottleTests(TestCase): def setUp(self): _django_cache.clear() self.client = APIClient() self.user = User.objects.create(device_id='dev-read-throttle', is_guest=True, name='') self.token = issue_access_token(self.user.id) self.grade = Grade.objects.create(title='g', icon='i', color='#fff') def test_read_endpoint_throttles_after_limit(self): # rate پیش‌فرض '120/min' — بعد از ۱۲۰ درخواستِ موفق، ۱۲۱اُم باید 429 بشه. codes = set() for _ in range(122): codes.add(self.client.get( f'/api/home?gradeId={self.grade.id}', HTTP_AUTHORIZATION=f'Bearer {self.token}').status_code) self.assertIn(429, codes) class HomeGradeFallbackTests(TestCase): """ /api/home باید در نبودِ ?grade= از پایه‌ی ذخیره‌شده‌ی خودِ کاربر استفاده کند. پیش‌تر این‌طور نبود و نتیجه‌اش در اپ این می‌شد: کاربر در صفحه‌ی آشنایی پایه را انتخاب می‌کرد و بعد صفحه‌ی خانه خالی می‌آمد. """ def setUp(self): _django_cache.clear() self.client = APIClient() self.user = User.objects.create(device_id='dev-home-fallback', is_guest=True, name='') self.token = issue_access_token(self.user.id) self.grade = Grade.objects.create(title='پایه نهم', icon='grade-9', color='#FF9800') unit = Unit.objects.create(grade=self.grade, title='فصل ۱', order=1) node = Node.objects.create(unit=unit, title='درس ۱', order=1) Lesson.objects.create(node=node, title='توان صحیح', intro_text='.') def tearDown(self): _django_cache.clear() def _home(self, query=''): return self.client.get(f'/api/home{query}', HTTP_AUTHORIZATION=f'Bearer {self.token}') def test_uses_the_saved_grade_when_no_param_is_given(self): self.user.grade_id = self.grade.id self.user.save(update_fields=['grade_id']) response = self._home() self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data['units']), 1) self.assertEqual(response.data['units'][0]['title'], 'فصل ۱') def test_explicit_param_wins_over_the_saved_grade(self): other = Grade.objects.create(title='پایه دهم', icon='grade-10', color='#E91E63') unit = Unit.objects.create(grade=other, title='فصل مثلثات', order=1) node = Node.objects.create(unit=unit, title='درس ۱', order=1) Lesson.objects.create(node=node, title='سینوس', intro_text='.') self.user.grade_id = self.grade.id self.user.save(update_fields=['grade_id']) response = self._home(f'?gradeId={other.id}') self.assertEqual(response.data['units'][0]['title'], 'فصل مثلثات') def test_user_without_a_grade_gets_an_empty_map_not_an_error(self): response = self._home() self.assertEqual(response.status_code, 200) self.assertEqual(response.data['units'], []) def test_unknown_explicit_grade_is_still_404(self): self.assertEqual(self._home('?gradeId=999999').status_code, 404) def test_a_deleted_saved_grade_degrades_to_an_empty_map(self): # پایه‌ای که کاربر ذخیره کرده حذف شده — این خطای کاربر نیست. self.user.grade_id = 999999 self.user.save(update_fields=['grade_id']) response = self._home() self.assertEqual(response.status_code, 200) self.assertEqual(response.data['units'], []) class NodeProgressUnitParamTests(TestCase): def test_non_numeric_unit_is_a_400_not_a_500(self): from rest_framework.test import APIClient from accounts.authentication import issue_access_token from accounts.models import User user = User.objects.create(device_id='unit-param', is_guest=True) response = APIClient().get('/api/progress/nodes?unit=abc', HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}') self.assertEqual(response.status_code, 400)