Files
MathApp/support/tests.py
Zareian ab5e53730a Audit fixes 2026-10-03: seed keeps admin edits, plan cache, refresh limit, payment and tutor guards
- seed_content no longer undoes admin work: a SeedRecord ledger (content/seed_sync.py)
  stops deleted/renamed/re-worded content from being recreated and keeps admin-edited
  lesson text; new rows in the seed files still arrive (migration content.0010)
- plan list cache is invalidated on Plan save (receiver used a stale 'v1' key)
- token refresh per-IP cap 30/hour -> 600/hour (TOKEN_REFRESH_IP_RATE)
- a failed gateway result only changes a pending order (forged NOK no longer
  overwrites refunded/canceled)
- tutor gets the correct answer only after the student has answered
- support reply notification moved to a TicketMessage post_save signal
- global daily cap on paid AI questions (AI_GLOBAL_DAILY_MAX, default 3000)
- Bazaar verify uses only the order's SKU; bazaar orders require a SKU
- /api/health warns on the committed signing key; deploy.sh saves local commits;
  400 for bad ?unit= / ?period=; account delete keeps the referrer's Conversion
- earlier today: no stub answers outside DEBUG, ?grade= removed, logged fallbacks

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-03 16:05:27 +03:30

425 lines
21 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.
# Create your tests here.
from unittest.mock import patch
from django.core.cache import cache
from django.test import TestCase
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 .models import BugReport, Ticket, TicketMessage
class BugReportCreateThrottleTests(TestCase):
"""
🔒 اثبات محدودیتِ جدیدِ per-user روی BugReportCreateView
(throttle_scope='support_create'): هیچ idempotency ای روی این
endpoint نیست، پس بدونِ throttle یک کاربرِ احرازهویت‌شده می‌تونست
تعدادِ نامحدودی BugReport با درخواست‌های پرتکرار بسازه.
ScopedRateThrottle با pk کاربر کلید می‌خوره، نه IP.
"""
def setUp(self):
cache.clear()
self.client = APIClient()
self.user = User.objects.create(device_id='device-bugreport-throttle-1', is_guest=True, name='')
self.token = issue_access_token(self.user.id)
def _create(self, i=0, remote_addr=None):
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'}
if remote_addr:
kwargs['REMOTE_ADDR'] = remote_addr
return self.client.post(
'/api/support/bug-reports',
{'title': f'باگ {i}', 'description': f'توضیحات {i}'},
format='json',
**kwargs,
)
def test_requests_within_the_limit_still_succeed(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = self._create(i)
self.assertEqual(response.status_code, 200)
def test_excessive_requests_from_same_user_return_429(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = self._create(i)
self.assertEqual(response.status_code, 200)
blocked_response = self._create('extra')
self.assertEqual(blocked_response.status_code, 429)
def test_changing_client_ip_does_not_bypass_the_user_limit(self):
# کاربرِ غیرمهمان (نه self.user که مهمانه)، چون کلیدِ throttleِ
# مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه.
registered_user = User.objects.create(is_guest=False, name='', phone='+989120000901')
registered_token = issue_access_token(registered_user.id)
def _create_as(i, remote_addr):
return self.client.post(
'/api/support/bug-reports',
{'title': f'باگ {i}', 'description': f'توضیحات {i}'},
format='json',
HTTP_AUTHORIZATION=f'Bearer {registered_token}',
REMOTE_ADDR=remote_addr,
)
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = _create_as(i, f'10.4.0.{i + 1}')
self.assertEqual(response.status_code, 200)
blocked_response = _create_as('extra', '10.4.0.99')
self.assertEqual(blocked_response.status_code, 429)
def test_throttled_request_creates_no_additional_bug_report(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
first_response = self._create()
self.assertEqual(first_response.status_code, 200)
count_before = BugReport.objects.filter(user_id=self.user.id).count()
blocked_response = self._create('blocked')
self.assertEqual(blocked_response.status_code, 429)
self.assertEqual(BugReport.objects.filter(user_id=self.user.id).count(), count_before)
class TicketCreateThrottleTests(TestCase):
"""
🔒 اثبات محدودیتِ جدیدِ per-user روی TicketCreateView
(throttle_scope='support_create'، مشترک با BugReport/TicketMessage).
"""
def setUp(self):
cache.clear()
self.client = APIClient()
self.user = User.objects.create(device_id='device-ticket-throttle-1', is_guest=True, name='')
self.token = issue_access_token(self.user.id)
def _create(self, i=0, remote_addr=None):
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'}
if remote_addr:
kwargs['REMOTE_ADDR'] = remote_addr
return self.client.post(
'/api/support/tickets',
{'subject': f'موضوع {i}', 'category': 'billing'},
format='json',
**kwargs,
)
def test_requests_within_the_limit_still_succeed(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = self._create(i)
self.assertEqual(response.status_code, 200)
def test_excessive_requests_from_same_user_return_429(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = self._create(i)
self.assertEqual(response.status_code, 200)
blocked_response = self._create('extra')
self.assertEqual(blocked_response.status_code, 429)
def test_changing_client_ip_does_not_bypass_the_user_limit(self):
# کاربرِ غیرمهمان (نه self.user که مهمانه)، چون کلیدِ throttleِ
# مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه.
registered_user = User.objects.create(is_guest=False, name='', phone='+989120000902')
registered_token = issue_access_token(registered_user.id)
def _create_as(i, remote_addr):
return self.client.post(
'/api/support/tickets',
{'subject': f'موضوع {i}', 'category': 'billing'},
format='json',
HTTP_AUTHORIZATION=f'Bearer {registered_token}',
REMOTE_ADDR=remote_addr,
)
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = _create_as(i, f'10.5.0.{i + 1}')
self.assertEqual(response.status_code, 200)
blocked_response = _create_as('extra', '10.5.0.99')
self.assertEqual(blocked_response.status_code, 429)
def test_throttled_request_creates_no_additional_ticket(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
first_response = self._create()
self.assertEqual(first_response.status_code, 200)
count_before = Ticket.objects.filter(user_id=self.user.id).count()
blocked_response = self._create('blocked')
self.assertEqual(blocked_response.status_code, 429)
self.assertEqual(Ticket.objects.filter(user_id=self.user.id).count(), count_before)
class TicketMessageCreateThrottleTests(TestCase):
"""
🔒 اثبات محدودیتِ جدیدِ per-user روی TicketMessageCreateView
(throttle_scope='support_create'، مشترک با BugReport/Ticket).
"""
def setUp(self):
cache.clear()
self.client = APIClient()
self.user = User.objects.create(device_id='device-ticketmsg-throttle-1', is_guest=True, name='')
self.token = issue_access_token(self.user.id)
self.ticket = Ticket.objects.create(user_id=self.user.id, subject='موضوع', category='billing')
def _create(self, i=0, remote_addr=None):
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'}
if remote_addr:
kwargs['REMOTE_ADDR'] = remote_addr
return self.client.post(
f'/api/support/tickets/{self.ticket.id}/messages',
{'text': f'پیام {i}'},
format='json',
**kwargs,
)
def test_requests_within_the_limit_still_succeed(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = self._create(i)
self.assertEqual(response.status_code, 200)
def test_excessive_requests_from_same_user_return_429(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = self._create(i)
self.assertEqual(response.status_code, 200)
blocked_response = self._create('extra')
self.assertEqual(blocked_response.status_code, 429)
def test_changing_client_ip_does_not_bypass_the_user_limit(self):
# کاربرِ غیرمهمان (نه self.user که مهمانه)، چون کلیدِ throttleِ
# مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه.
registered_user = User.objects.create(is_guest=False, name='', phone='+989120000903')
registered_token = issue_access_token(registered_user.id)
registered_ticket = Ticket.objects.create(
user_id=registered_user.id, subject='موضوع', category='billing',
)
def _create_as(i, remote_addr):
return self.client.post(
f'/api/support/tickets/{registered_ticket.id}/messages',
{'text': f'پیام {i}'},
format='json',
HTTP_AUTHORIZATION=f'Bearer {registered_token}',
REMOTE_ADDR=remote_addr,
)
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
for i in range(3):
response = _create_as(i, f'10.6.0.{i + 1}')
self.assertEqual(response.status_code, 200)
blocked_response = _create_as('extra', '10.6.0.99')
self.assertEqual(blocked_response.status_code, 429)
def test_throttled_request_creates_no_additional_message(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
first_response = self._create()
self.assertEqual(first_response.status_code, 200)
count_before = TicketMessage.objects.filter(ticket=self.ticket).count()
blocked_response = self._create('blocked')
self.assertEqual(blocked_response.status_code, 429)
self.assertEqual(TicketMessage.objects.filter(ticket=self.ticket).count(), count_before)
class SupportCreateGuestThrottleBypassTests(TestCase):
"""
🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='support_create'
(مشترک بینِ BugReportCreateView، TicketCreateView و
TicketMessageCreateView) برای هر کاربرِ احرازهویت‌شده (مهمان یا نه)
user.pk بود. چون GuestSessionView با هر deviceId جدید یک کاربرِ
مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم می‌تونست با ساختنِ پی‌درپیِ
حساب‌های مهمان، هر بار یک bucket throttle تازه برای این scopeِ
مشترک بگیره و سقفِ ۳/ساعتِ ساختِ رکوردهای پشتیبانی (BugReport/
Ticket/TicketMessage) رو چند برابر کنه. حالا GuestAwareSupportThrottle
برای کاربرهای مهمان کلید رو بر اساسِ IP (همون get_ident که
guest_session هم استفاده می‌کنه) می‌سازه، پس چرخشِ حساب مهمان دیگه
bucket تازه نمی‌ده — این هم روی bucket مشترکِ بینِ هر سه Endpoint و
هم روی رفتارِ per-user برای کاربرهای غیرمهمان تست می‌شه.
"""
def setUp(self):
cache.clear()
self.client = APIClient()
def _create_bug_report(self, token, i=0, remote_addr=None):
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {token}'}
if remote_addr:
kwargs['REMOTE_ADDR'] = remote_addr
return self.client.post(
'/api/support/bug-reports',
{'title': f'باگ {i}', 'description': f'توضیحات {i}'},
format='json',
**kwargs,
)
def _create_ticket(self, token, i=0):
return self.client.post(
'/api/support/tickets',
{'subject': f'موضوع {i}', 'category': 'billing'},
format='json',
HTTP_AUTHORIZATION=f'Bearer {token}',
)
def test_rotating_guest_accounts_share_the_same_throttle_bucket(self):
# سه حساب مهمانِ متفاوت (شبیه‌سازیِ چرخشِ حساب برای دورزدنِ throttle)،
# همه از همون کلاینت/IP — باید همه یک bucket مشترک رو مصرف کنند،
# حتی وقتی از دو Endpointِ متفاوتِ همون scope استفاده می‌کنن.
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '2/hour'}):
guest1 = User.objects.create(device_id='device-support-guest-rot-1', is_guest=True, name='')
response1 = self._create_bug_report(issue_access_token(guest1.id), 1)
self.assertEqual(response1.status_code, 200)
guest2 = User.objects.create(device_id='device-support-guest-rot-2', is_guest=True, name='')
response2 = self._create_ticket(issue_access_token(guest2.id), 2)
self.assertEqual(response2.status_code, 200)
# سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
# bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk — یعنی
# سقفِ ۲/ساعت هنوز اعمال می‌شه، صرف‌نظر از تعدادِ حساب‌های
# مهمانِ ساخته‌شده یا این‌که کدوم Endpoint صدا زده شده.
guest3 = User.objects.create(device_id='device-support-guest-rot-3', is_guest=True, name='')
response3 = self._create_bug_report(issue_access_token(guest3.id), 3)
self.assertEqual(response3.status_code, 429)
def test_rotated_guest_account_cannot_get_a_fresh_bucket_after_limit(self):
# حتی وقتی حسابِ مهمانِ جدید به سقف رسیده، یک حسابِ مهمانِ دیگر
# (چهارمین pk) هم نباید bucket تازه‌ای بگیره — سقف هنوز به IP
# مشترک متصل است.
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
guest1 = User.objects.create(device_id='device-support-guest-rot-4', is_guest=True, name='')
response1 = self._create_bug_report(issue_access_token(guest1.id), 4)
self.assertEqual(response1.status_code, 200)
guest2 = User.objects.create(device_id='device-support-guest-rot-5', is_guest=True, name='')
blocked = self._create_bug_report(issue_access_token(guest2.id), 5)
self.assertEqual(blocked.status_code, 429)
guest3 = User.objects.create(device_id='device-support-guest-rot-6', is_guest=True, name='')
still_blocked = self._create_bug_report(issue_access_token(guest3.id), 6)
self.assertEqual(still_blocked.status_code, 429)
def test_registered_users_keep_independent_per_user_buckets(self):
# عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ
# قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن — throttle
# برای اونا هنوز بر اساسِ user.pk است، نه IP.
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
user_a = User.objects.create(is_guest=False, name='', phone='+989120001001')
response_a = self._create_bug_report(issue_access_token(user_a.id), 'a')
self.assertEqual(response_a.status_code, 200)
user_b = User.objects.create(is_guest=False, name='', phone='+989120001002')
response_b = self._create_bug_report(issue_access_token(user_b.id), 'b')
self.assertEqual(response_b.status_code, 200)
def test_blocked_guest_request_does_not_create_support_record(self):
# درخواستِ بلاک‌شده (از حسابِ مهمانِ چرخشی) نباید هیچ BugReport یا
# Ticket جدیدی بسازه — throttle باید *قبل* از رسیدن به منطقِ
# ساختِ رکورد رد کنه.
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
guest1 = User.objects.create(device_id='device-support-guest-rot-7', is_guest=True, name='')
first_response = self._create_bug_report(issue_access_token(guest1.id), 7)
self.assertEqual(first_response.status_code, 200)
# حسابِ مهمانِ تازه‌ی دوم باید بلاک بشه (bucket مشترکِ IP از
# قبل مصرف شده) و نباید BugReport یا Ticket جدیدی بسازد.
guest2 = User.objects.create(device_id='device-support-guest-rot-8', is_guest=True, name='')
bug_count_before = BugReport.objects.count()
ticket_count_before = Ticket.objects.count()
blocked_bug_response = self._create_bug_report(issue_access_token(guest2.id), 8)
self.assertEqual(blocked_bug_response.status_code, 429)
blocked_ticket_response = self._create_ticket(issue_access_token(guest2.id), 8)
self.assertEqual(blocked_ticket_response.status_code, 429)
self.assertEqual(BugReport.objects.count(), bug_count_before)
self.assertEqual(Ticket.objects.count(), ticket_count_before)
class TicketListTests(TestCase):
"""GET /api/support/tickets: فقط تیکت‌های خودِ کاربر، تازه‌ترین اول، با شمارِ پاسخِ پشتیبانی."""
def setUp(self):
cache.clear()
self.client = APIClient()
self.user = User.objects.create(device_id='device-ticket-list-1', is_guest=True, name='')
self.other = User.objects.create(device_id='device-ticket-list-2', is_guest=True, name='')
self.auth = {'HTTP_AUTHORIZATION': f'Bearer {issue_access_token(self.user.id)}'}
def test_lists_only_own_tickets_with_reply_count(self):
old = Ticket.objects.create(user_id=self.user.id, subject='قدیمی', category='learning')
TicketMessage.objects.create(ticket=old, sender='user', text='سلام')
TicketMessage.objects.create(ticket=old, sender='support', text='جواب')
new = Ticket.objects.create(user_id=self.user.id, subject='تازه', category='payment')
Ticket.objects.create(user_id=self.other.id, subject='مالِ دیگری', category='other')
response = self.client.get('/api/support/tickets', **self.auth)
self.assertEqual(response.status_code, 200)
tickets = response.json()['tickets']
self.assertEqual([t['ticketId'] for t in tickets], [str(new.pk), str(old.pk)])
self.assertEqual(tickets[1]['supportReplies'], 1)
self.assertEqual(tickets[0]['supportReplies'], 0)
def test_requires_authentication(self):
self.assertIn(self.client.get('/api/support/tickets').status_code, (401, 403))
def test_listing_does_not_spend_the_create_quota(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
for _ in range(3):
self.assertEqual(self.client.get('/api/support/tickets', **self.auth).status_code, 200)
response = self.client.post(
'/api/support/tickets', {'subject': 'س', 'category': 'other'}, format='json', **self.auth,
)
self.assertEqual(response.status_code, 200)
class TicketReplyNotificationTests(TestCase):
"""پاسخی که پشتیبان از صفحه‌ی خودِ تیکت (inline) می‌دهد هم اعلان می‌سازد (تا ۲۰۲۶-۱۰-۰۳ نمی‌ساخت)."""
def test_inline_reply_on_the_ticket_page_notifies_the_user(self):
from django.contrib.auth.models import User as Staff
from django.test import Client
from notifications.models import Notification
staff = Staff.objects.create_superuser('support-staff', 's@example.com', 'pw')
ticket = Ticket.objects.create(user_id=77, subject='s', category='c')
client = Client()
client.force_login(staff)
client.post(f'/admin/support/ticket/{ticket.id}/change/', {
'user_id': 77, 'subject': 's', 'category': 'c', 'status': 'open', 'priority': 'normal',
'messages-TOTAL_FORMS': '1', 'messages-INITIAL_FORMS': '0',
'messages-MIN_NUM_FORMS': '0', 'messages-MAX_NUM_FORMS': '1000',
'messages-0-sender': 'support', 'messages-0-text': 'پاسخ', 'messages-0-ticket': str(ticket.id),
})
self.assertEqual(ticket.messages.count(), 1)
self.assertEqual(Notification.objects.filter(user_id=77, type='ticket_replied').count(), 1)
def test_a_users_own_message_does_not_notify_them(self):
from notifications.models import Notification
ticket = Ticket.objects.create(user_id=78, subject='s', category='c')
TicketMessage.objects.create(ticket=ticket, sender='user', text='سلام')
self.assertFalse(Notification.objects.filter(user_id=78).exists())