Zareian
645baf846b
- /api/payment/orders/{id}/pay requires the short-lived signature that paymentUrl carries;
without it (or with another order's) it is the same 404 as an unknown order
- Session.refresh_token stores the SHA-256 of the token; migration accounts.0007 hashes the
existing rows so tokens clients already hold keep working
- /api/leaderboard requires a token (guests have one); names of children were scrapable
- RAG mock only for `manage.py test`, not any command with an argument named "test"
- tools/loadtest: user factory, load generator, fake GapGPT+Qdrant, with method notes
- security-before-production.md lists only open items; closed ones become a guards table
with the tests that hold them. Comments describe current behaviour, not history
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
199 lines
8.6 KiB
Python
199 lines
8.6 KiB
Python
"""
|
||
استریک و جدولِ رتبه‌بندی از فعالیتِ واقعی زنده می‌شوند.
|
||
|
||
تا ۲۰۲۶-۰۹-۲۴ هیچ‌جای اپ StreakRecord نمی‌نوشت («روز پیاپی» برای هر کاربرِ
|
||
واقعی صفر می‌ماند) و جدولِ رتبه‌بندی فقط با اجرای دستیِ rebuild_leaderboard
|
||
عوض می‌شد.
|
||
"""
|
||
from datetime import date
|
||
from unittest import mock
|
||
|
||
from django.core.cache import cache
|
||
from django.test import TestCase
|
||
|
||
from accounts.authentication import issue_access_token
|
||
from accounts.models import User
|
||
from content.models import Choice, Grade, Lesson, Node, Question, Unit
|
||
from progress.models import StreakRecord
|
||
from progress.services import effective_current_streak, record_activity
|
||
|
||
|
||
class StreakTests(TestCase):
|
||
def test_consecutive_days_grow_the_streak(self):
|
||
record_activity(1, today=date(2026, 9, 20))
|
||
record_activity(1, today=date(2026, 9, 21))
|
||
streak = record_activity(1, today=date(2026, 9, 22))
|
||
|
||
self.assertEqual(streak.current_streak, 3)
|
||
self.assertEqual(streak.longest_streak, 3)
|
||
|
||
def test_same_day_counts_once(self):
|
||
record_activity(1, today=date(2026, 9, 20))
|
||
streak = record_activity(1, today=date(2026, 9, 20))
|
||
|
||
self.assertEqual(streak.current_streak, 1)
|
||
self.assertEqual(streak.active_days.count(), 1)
|
||
|
||
def test_a_missed_day_restarts_it_but_keeps_the_longest(self):
|
||
for day in (20, 21, 22):
|
||
record_activity(1, today=date(2026, 9, day))
|
||
streak = record_activity(1, today=date(2026, 9, 25))
|
||
|
||
self.assertEqual(streak.current_streak, 1)
|
||
self.assertEqual(streak.longest_streak, 3)
|
||
|
||
def test_shown_streak_is_zero_once_the_chain_breaks(self):
|
||
"""عددِ ذخیره‌شده فقط سرِ فعالیت عوض می‌شود؛ نمایش باید شکستن را ببیند."""
|
||
streak = record_activity(1, today=date(2026, 9, 20))
|
||
|
||
self.assertEqual(effective_current_streak(streak, today=date(2026, 9, 21)), 1)
|
||
self.assertEqual(effective_current_streak(streak, today=date(2026, 9, 23)), 0)
|
||
self.assertEqual(effective_current_streak(None), 0)
|
||
|
||
|
||
class ActivityFeedsStreakAndLeaderboardTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
grade = Grade.objects.create(title='پایه نهم', icon='', color='')
|
||
unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1)
|
||
self.node = Node.objects.create(unit=unit, title='جزیره', order=1)
|
||
for i in range(4):
|
||
lesson = Lesson.objects.create(node=self.node, title=f'درس {i}')
|
||
q = Question.objects.create(lesson=lesson, type='multiple_choice', prompt=f'س {i}')
|
||
self.correct = Choice.objects.create(question=q, value='الف', is_correct=True)
|
||
Choice.objects.create(question=q, value='ب', is_correct=False)
|
||
self.question = q
|
||
self.user = User.objects.create(phone='09120000031', name='سارا', grade_id=grade.id)
|
||
self.auth = {'HTTP_AUTHORIZATION': f'Bearer {issue_access_token(self.user.id)}'}
|
||
|
||
def test_answering_a_question_starts_the_streak(self):
|
||
self.client.post(
|
||
f'/api/questions/{self.question.pk}/answer',
|
||
{'selectedOptionId': self.correct.pk}, content_type='application/json', **self.auth,
|
||
)
|
||
|
||
summary = self.client.get('/api/user/summary', **self.auth).json()
|
||
self.assertEqual(summary['streak']['current'], 1)
|
||
calendar = self.client.get('/api/user/streak/calendar', **self.auth).json()
|
||
self.assertEqual(len(calendar['activeDays']), 1)
|
||
|
||
def test_a_finished_island_shows_up_on_the_leaderboard(self):
|
||
for question in Question.objects.filter(lesson__node=self.node).order_by('id')[:3]:
|
||
self.client.post(
|
||
f'/api/questions/{question.pk}/answer',
|
||
{'selectedOptionId': question.choices.get(is_correct=True).pk},
|
||
content_type='application/json', **self.auth,
|
||
)
|
||
self.client.post(
|
||
f'/api/nodes/{self.node.pk}/quiz/complete',
|
||
{'correctCount': 3, 'totalQuestions': 4},
|
||
content_type='application/json', **self.auth,
|
||
)
|
||
|
||
for period in ('weekly', 'all_time'):
|
||
entries = self.client.get(f'/api/leaderboard?period={period}', **self.auth).json()['entries']
|
||
self.assertEqual(entries[0]['userId'], str(self.user.id), period)
|
||
self.assertEqual(entries[0]['name'], 'سارا')
|
||
self.assertEqual(entries[0]['xpInPeriod'], 30, period)
|
||
|
||
def test_streak_failure_never_breaks_the_answer(self):
|
||
with mock.patch('progress.services.record_activity', side_effect=RuntimeError('boom')):
|
||
response = self.client.post(
|
||
f'/api/questions/{self.question.pk}/answer',
|
||
{'selectedOptionId': self.correct.pk}, content_type='application/json', **self.auth,
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertFalse(StreakRecord.objects.filter(user_id=self.user.id).exists())
|
||
|
||
|
||
class ApiResponsesAreNeverCachedTests(TestCase):
|
||
def test_api_responses_say_no_store(self):
|
||
response = self.client.get('/api/grades')
|
||
self.assertIn('no-store', response['Cache-Control'])
|
||
|
||
def test_a_view_that_sets_its_own_cache_control_keeps_it(self):
|
||
from django.http import HttpResponse
|
||
from config.middleware import APINoStoreMiddleware
|
||
|
||
def view(request):
|
||
response = HttpResponse()
|
||
response['Cache-Control'] = 'no-cache'
|
||
return response
|
||
|
||
request = mock.Mock(path='/api/ai/ask')
|
||
self.assertEqual(APINoStoreMiddleware(view)(request)['Cache-Control'], 'no-cache')
|
||
|
||
|
||
class SummaryShowsTheActivePlanTests(TestCase):
|
||
def test_active_subscription_is_named(self):
|
||
from subscription.models import Plan, Subscription
|
||
user = User.objects.create(phone='09120000032')
|
||
auth = {'HTTP_AUTHORIZATION': f'Bearer {issue_access_token(user.id)}'}
|
||
self.assertIsNone(self.client.get('/api/user/summary', **auth).json()['subscription'])
|
||
|
||
plan = Plan.objects.create(title='طلایی', price=1000)
|
||
Subscription.objects.create(user_id=user.id, plan=plan, status='active')
|
||
|
||
self.assertEqual(self.client.get('/api/user/summary', **auth).json()['subscription'], 'طلایی')
|
||
|
||
|
||
class HeartsLikeDuolingoTests(TestCase):
|
||
"""هر ۵ ساعت یک قلب؛ ۲۴ ساعت بعد از اولین قلبِ ازدست‌رفته، همه پر."""
|
||
|
||
def _hearts_after_losing(self, n):
|
||
from progress.models import UserHearts
|
||
h = UserHearts.objects.create(user_id=1)
|
||
for _ in range(n):
|
||
h.lose_heart()
|
||
h.save()
|
||
return h
|
||
|
||
def _at(self, hours):
|
||
from django.utils import timezone
|
||
from datetime import timedelta
|
||
return mock.patch('django.utils.timezone.now', return_value=timezone.now() + timedelta(hours=hours))
|
||
|
||
def test_one_heart_comes_back_every_five_hours(self):
|
||
h = self._hearts_after_losing(5)
|
||
with self._at(4.9):
|
||
h.apply_regen()
|
||
self.assertEqual(h.current_hearts, 0)
|
||
with self._at(5.1):
|
||
h.apply_regen()
|
||
self.assertEqual(h.current_hearts, 1)
|
||
with self._at(15.1):
|
||
h.apply_regen()
|
||
self.assertEqual(h.current_hearts, 3)
|
||
|
||
def test_everything_refills_24_hours_after_the_first_loss(self):
|
||
h = self._hearts_after_losing(5)
|
||
with self._at(24.1):
|
||
h.apply_regen()
|
||
self.assertEqual(h.current_hearts, 5)
|
||
self.assertIsNone(h.next_refill_at)
|
||
self.assertIsNone(h.full_refill_at)
|
||
|
||
def test_an_old_24_hour_timer_is_shortened_to_five(self):
|
||
from datetime import timedelta
|
||
from django.utils import timezone
|
||
from progress.models import UserHearts
|
||
h = UserHearts.objects.create(
|
||
user_id=2, current_hearts=2, next_refill_at=timezone.now() + timedelta(hours=23),
|
||
)
|
||
h.apply_regen()
|
||
self.assertLessEqual(h.next_refill_at, timezone.now() + timedelta(hours=5, seconds=1))
|
||
self.assertIsNotNone(h.full_refill_at)
|
||
|
||
def test_api_reports_both_timers(self):
|
||
user = User.objects.create(phone='09120000033')
|
||
from progress.models import UserHearts
|
||
h = UserHearts.objects.create(user_id=user.id)
|
||
h.lose_heart()
|
||
h.save()
|
||
body = self.client.get(
|
||
'/api/user/hearts', HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}',
|
||
).json()
|
||
self.assertEqual(body['currentHearts'], 4)
|
||
self.assertIsNotNone(body['nextRefillAt'])
|
||
self.assertIsNotNone(body['fullRefillAt'])
|