Zareian
5f3aae0756
خطاهای این پروژه خاموش‌اند: نامِ اشتباهِ collection یا فیلترِ نخورده نتیجه‌ی خالی می‌دهد نه استثنا، و دستیار فقط بدتر جواب می‌دهد. این تست‌ها نگهبانِ همان‌جاهایند. TextbookLayoutTests بررسی می‌کند رشته‌ی هر فصل بینِ ingest و retrieval یکی باشد و بازه‌ی صفحه‌ها پیوسته بماند. IngestBookResumeTests مطمئن می‌شود صفحه‌ی بی‌مسئله دوبار پول نمی‌گیرد، صفحه‌ی خطادار در اجرای بعد دوباره سراغش می‌روند، و upsertِ شکست‌خورده rollback می‌شود. LessonContextResolutionTests نگهبانِ نرمال‌سازیِ «پایه نهم» → «ریاضی نهم» است و اینکه مسیرِ gradeId و مسیرِ lessonId به یک collection برسند. GradeAwarePromptTests جلوی برگشتنِ لحنِ دبستانی برای پایه‌های بالا را می‌گیرد. FollowUpStaysGroundedTests پیامِ دومِ یک گفتگو را هم مستند نگه می‌دارد. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
782 lines
39 KiB
Python
782 lines
39 KiB
Python
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.objects.create(
|
||
lesson=self.lesson, type='multiple_choice', prompt=f'سوال {i}',
|
||
)
|
||
|
||
def _complete(self, correct_count, total_questions):
|
||
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.objects.create(
|
||
lesson=self.lesson, type='multiple_choice', prompt=f'سوال {i}',
|
||
)
|
||
|
||
def _complete(self, correct_count, total_questions):
|
||
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?grade={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)
|
||
self.assertNotIn('content_node', sql)
|
||
# فقط کوئریِ 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?grade={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'?grade={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('?grade=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'], [])
|