# 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(phone='09000000060', 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(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(phone='09000000061', 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(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(phone='09000000062', 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(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 SupportCreatePerUserThrottleTests(TestCase): """هر کاربرِ ثبت‌نام‌شده سطلِ throttleِ جدای خودش را دارد؛ مصرفِ حسابِ دیگر به او نمی‌رسد.""" 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_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(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(name='', phone='+989120001002') response_b = self._create_bug_report(issue_access_token(user_b.id), 'b') self.assertEqual(response_b.status_code, 200) class TicketListTests(TestCase): """GET /api/support/tickets: فقط تیکت‌های خودِ کاربر، تازه‌ترین اول، با شمارِ پاسخِ پشتیبانی.""" def setUp(self): cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000063', name='') self.other = User.objects.create(phone='09000000064', 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())