""" استریک و جدولِ رتبه‌بندی از فعالیتِ واقعی زنده می‌شوند. تا ۲۰۲۶-۰۹-۲۴ هیچ‌جای اپ 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'], 'طلایی') 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'])