Files
MathApp/progress/tests_live_gamification.py
Zareian 701d8badcc make streaks and the leaderboard live, and stop three AI chats from freezing the API
Found in a deeper pass over what the frontend's flow actually touches:

- Nothing in the app ever wrote StreakRecord: "days in a row" was 0 forever
  for every real user. Answers and lesson/island completions now record the
  day (Tehran time); a broken chain shows 0. A streak error never fails the
  answer itself.
- /api/leaderboard only changed when someone ran rebuild_leaderboard by
  hand. It is now computed live (60s cache); weekly counts island XP too.
- gunicorn ran 3 sync workers with a 30s timeout, so every "ask Pi" stream
  pinned a whole worker (~10s, GapGPT buffers the answer) and three at once
  queued the entire API. gunicorn.conf.py (read automatically from the
  working directory) switches to gthread, 8 threads, 120s timeout.
- Every /api/ response now says Cache-Control: private, no-store, so turning
  on caching in the ArvanCloud panel can never serve one student another's data.
- /api/user/summary named no active plan ever ("subscription app not
  implemented yet" had long stopped being true).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-24 06:39:50 +03:30

132 lines
6.0 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
استریک و جدولِ رتبه‌بندی از فعالیتِ واقعی زنده می‌شوند.
تا ۲۰۲۶-۰۹-۲۴ هیچ‌جای اپ 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):
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}').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'], 'طلایی')