from django.core.cache import cache from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient from rest_framework.throttling import ScopedRateThrottle from unittest.mock import patch from accounts.authentication import issue_access_token from accounts.models import User from payment.models import Order from .models import DiscountCode, Offer class ValidateDiscountPermissionTests(TestCase): """ 🔒 اثبات فیکسِ امنیتی: ValidateDiscountView قبلاً permission_classes نداشت (یعنی دیفالتِ پروژه AllowAny اعمال می‌شد) و یک کلاینتِ بدونِ احرازهویت می‌تونست بی‌نهایت کد رو probe کنه و valid/type/value رو ببینه — درحالی‌که RedeemDiscountView (اندپوینتِ واقعیِ مصرف) از قبل IsAuthenticated داره. این تست هم رفتارِ ۴۰۱ برای anonymous، هم رفتارِ صحیحِ اعتبارسنجی برای کاربرِ لاگین‌کرده رو تضمین می‌کنه. """ def setUp(self): self.client = APIClient() self.user = User.objects.create(phone='09000000038', name='') self.token = issue_access_token(self.user.id) self.discount = DiscountCode.objects.create( code='SUMMER10', type='percent', value=10, max_uses=5, expires_at=None, ) def test_unauthenticated_request_is_rejected(self): response = self.client.get('/api/discount/validate', {'code': self.discount.code}) self.assertEqual(response.status_code, 401) def test_authenticated_request_validates_usable_code(self): response = self.client.get( '/api/discount/validate', {'code': self.discount.code}, HTTP_AUTHORIZATION=f'Bearer {self.token}', ) self.assertEqual(response.status_code, 200) self.assertTrue(response.data['valid']) self.assertEqual(response.data['type'], 'percent') self.assertEqual(response.data['value'], 10) def test_authenticated_request_reports_unusable_code_as_invalid(self): expired = DiscountCode.objects.create( code='EXPIRED5', type='fixed', value=5000, max_uses=5, expires_at=timezone.now() - timezone.timedelta(days=1), ) response = self.client.get( '/api/discount/validate', {'code': expired.code}, HTTP_AUTHORIZATION=f'Bearer {self.token}', ) self.assertEqual(response.status_code, 200) self.assertFalse(response.data['valid']) class RedeemOfferCodeThrottleTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ throttle_scope='offer_redeem' روی RedeemOfferCodeView: قبلاً این endpoint هیچ throttle ای نداشت، پس یک کاربرِ احرازهویت‌شده می‌تونست بی‌نهایت کدِ آفر رو probe کنه (چون کدهای نامعتبر فقط یک ValidationError ساده‌ی ۴۰۰ برمی‌گردوندن، نه ۴۲۹). این تست‌ها هم رفتارِ throttle (per-user، نه IP) و هم عدمِ رگرسیون در رفتارِ موفقِ redemption رو تضمین می‌کنن. """ def setUp(self): cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000039', name='') self.token = issue_access_token(self.user.id) def _redeem(self, code): return self.client.post( '/api/discount/offers/redeem', {'code': code}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}', ) def test_unauthenticated_request_is_rejected(self): # رفتارِ IsAuthenticated موجود دست‌نخورده مونده؛ throttle جایگزینِ # authentication نیست. response = self.client.post( '/api/discount/offers/redeem', {'code': 'ANYCODE'}, format='json', ) self.assertEqual(response.status_code, 401) def test_invalid_code_attempts_are_throttled_after_configured_limit(self): # این دقیقاً همون سناریوی probing هست: کدهای نامعتبرِ پشتِ‌سرِهم. # قبل از این فیکس، همه‌شون ۴۰۰ می‌گرفتن و هیچ‌وقت بلاک نمی‌شدن. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '3/hour'}): for i in range(3): response = self._redeem(f'INVALID-{i}') self.assertEqual(response.status_code, 400) blocked_response = self._redeem('INVALID-EXTRA') self.assertEqual(blocked_response.status_code, 429) def test_throttle_is_scoped_per_user_not_shared_globally(self): # کاربرِ غیرمهمان (نه یک حسابِ مهمانِ دوم، که دیگه bucket جدا نداره). other_user = User.objects.create(name='', phone='+989120000701') other_token = issue_access_token(other_user.id) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '1/hour'}): first_user_response = self._redeem('INVALID-A') self.assertEqual(first_user_response.status_code, 400) first_user_blocked = self._redeem('INVALID-B') self.assertEqual(first_user_blocked.status_code, 429) # کاربرِ دومِ متفاوت هنوز باید بتونه تلاش کنه — throttle # per-user است، نه یک شمارنده‌ی سراسری روی endpoint. second_user_response = self.client.post( '/api/discount/offers/redeem', {'code': 'INVALID-C'}, format='json', HTTP_AUTHORIZATION=f'Bearer {other_token}', ) self.assertEqual(second_user_response.status_code, 400) def test_successful_redemption_still_works_within_the_limit(self): # عدمِ رگرسیون در رفتارِ اصلیِ redemption: با یک کدِ معتبر و در # محدوده‌ی مجاز، endpoint باید دقیقاً مثلِ قبل کار کنه. offer = Offer.objects.create( name='Test Offer', kind='free_full', code='VALIDCODE1', is_active=True, ) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '10/hour'}): response = self._redeem(offer.code) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['offerName'], offer.name) self.assertEqual(response.data['kind'], offer.kind) def test_blocked_request_does_not_call_redeem_offer_code(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '1/hour'}): first_response = self._redeem('INVALID-X') self.assertEqual(first_response.status_code, 400) with patch('discount.views.redeem_offer_code') as mock_redeem: blocked_response = self._redeem('INVALID-Y') self.assertEqual(blocked_response.status_code, 429) mock_redeem.assert_not_called() class RedeemOfferCodePerUserThrottleTests(TestCase): """هر کاربرِ ثبت‌نام‌شده سطلِ throttleِ جدای خودش را دارد؛ مصرفِ حسابِ دیگر به او نمی‌رسد.""" def setUp(self): cache.clear() self.client = APIClient() def _redeem(self, code, token): return self.client.post( '/api/discount/offers/redeem', {'code': code}, 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, {'offer_redeem': '1/hour'}): user_a = User.objects.create(name='', phone='+989120000801') response_a = self._redeem('INVALID-REG-A', issue_access_token(user_a.id)) self.assertEqual(response_a.status_code, 400) user_b = User.objects.create(name='', phone='+989120000802') response_b = self._redeem('INVALID-REG-B', issue_access_token(user_b.id)) self.assertEqual(response_b.status_code, 400) class ValidateDiscountThrottleTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ throttle_scope='discount_validate' روی ValidateDiscountView: قبلاً این endpoint هیچ throttle ای نداشت، در حالی که type/value کدهای معتبر رو هم برمی‌گردوند — یعنی یک کاربرِ احرازهویت‌شده می‌تونست بی‌نهایت کدهای تخفیف رو probe کنه. این تست‌ها هم رفتارِ throttle (per-user، نه IP)، هم عدمِ رگرسیون در پاسخِ validation و رفتارِ authentication رو تضمین می‌کنن. """ def setUp(self): cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000040', name='') self.token = issue_access_token(self.user.id) self.discount = DiscountCode.objects.create( code='THROTTLE10', type='percent', value=10, max_uses=5, expires_at=None, ) def _validate(self, code, token=None): headers = {} if token: headers['HTTP_AUTHORIZATION'] = f'Bearer {token}' return self.client.get('/api/discount/validate', {'code': code}, **headers) def test_unauthenticated_request_is_still_rejected(self): # رفتارِ IsAuthenticated موجود دست‌نخورده مونده؛ throttle جایگزینِ # authentication نیست. response = self._validate('ANYCODE') self.assertEqual(response.status_code, 401) def test_repeated_code_probing_is_throttled_after_configured_limit(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '3/hour'}): for i in range(3): response = self._validate(f'PROBE-{i}', token=self.token) self.assertEqual(response.status_code, 200) blocked_response = self._validate('PROBE-EXTRA', token=self.token) self.assertEqual(blocked_response.status_code, 429) def test_throttle_is_scoped_per_user_not_shared_globally(self): # کاربرِ غیرمهمان (نه یک حسابِ مهمانِ دوم، که دیگه bucket جدا نداره). other_user = User.objects.create(name='', phone='+989120000731') other_token = issue_access_token(other_user.id) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '1/hour'}): first_user_response = self._validate('PROBE-A', token=self.token) self.assertEqual(first_user_response.status_code, 200) first_user_blocked = self._validate('PROBE-B', token=self.token) self.assertEqual(first_user_blocked.status_code, 429) # کاربرِ دومِ متفاوت هنوز باید بتونه چک کنه — throttle # per-user است، نه یک شمارنده‌ی سراسری روی endpoint. second_user_response = self._validate('PROBE-C', token=other_token) self.assertEqual(second_user_response.status_code, 200) def test_valid_code_response_unchanged_within_the_limit(self): # عدمِ رگرسیون: پاسخِ اعتبارسنجیِ کدِ معتبر دقیقاً مثلِ قبل باید کار کنه. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '10/hour'}): response = self._validate(self.discount.code, token=self.token) self.assertEqual(response.status_code, 200) self.assertTrue(response.data['valid']) self.assertEqual(response.data['type'], 'percent') self.assertEqual(response.data['value'], 10) def test_invalid_code_response_unchanged_within_the_limit(self): # عدمِ رگرسیون: پاسخ برای کدِ نامعتبر باید همچنان ۲۰۰ با valid=False # باشه (نه ۴۲۹)، تا وقتی که به سقفِ throttle نرسیده باشیم. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '10/hour'}): response = self._validate('NOT-A-REAL-CODE', token=self.token) self.assertEqual(response.status_code, 200) self.assertFalse(response.data['valid']) class RedeemDiscountThrottleTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ throttle_scope='discount_redeem' روی RedeemDiscountView: این endpoint همون oracle ای که ValidateDiscountView و RedeemOfferCodeView رو throttle کرد داره (۴۰۴ برای کدِ ناموجود، ۴۰۰ برای کدِ نامعتبر)، با این تفاوت که حدسِ درست یک کدِ واقعی رو مصرف می‌کنه، نه فقط اطلاعات لو می‌ده. این تست‌ها هم رفتارِ throttle (per-user، نه IP)، هم عدمِ رگرسیون در ردیمِ موفق رو تضمین می‌کنن. """ def setUp(self): cache.clear() self.client = APIClient() self.user = User.objects.create(phone='09000000041', name='') self.token = issue_access_token(self.user.id) self.order = Order.objects.create( user_id=self.user.id, amount=10000, idempotency_key='order-redeem-throttle-1', ) def _redeem(self, code, order_id=None, token=None): headers = {} if token: headers['HTTP_AUTHORIZATION'] = f'Bearer {token}' return self.client.post( '/api/discount/redeem', {'code': code, 'orderId': order_id or self.order.id}, format='json', **headers, ) def test_unauthenticated_request_is_still_rejected(self): # رفتارِ IsAuthenticated موجود دست‌نخورده مونده؛ throttle جایگزینِ # authentication نیست. response = self._redeem('ANYCODE') self.assertEqual(response.status_code, 401) def test_repeated_code_probing_is_throttled_after_configured_limit(self): # سناریوی brute-force: کدهای ناموجودِ پشتِ‌سرِهم روی همون سفارش. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '3/hour'}): for i in range(3): response = self._redeem(f'PROBE-{i}', token=self.token) self.assertEqual(response.status_code, 404) blocked_response = self._redeem('PROBE-EXTRA', token=self.token) self.assertEqual(blocked_response.status_code, 429) def test_throttle_is_scoped_per_user_not_shared_globally(self): # کاربرِ غیرمهمان (نه یک حسابِ مهمانِ دوم، که دیگه bucket جدا نداره). other_user = User.objects.create(name='', phone='+989120000301') other_token = issue_access_token(other_user.id) other_order = Order.objects.create( user_id=other_user.id, amount=5000, idempotency_key='order-redeem-throttle-2', ) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '1/hour'}): first_user_response = self._redeem('PROBE-A', token=self.token) self.assertEqual(first_user_response.status_code, 404) first_user_blocked = self._redeem('PROBE-B', token=self.token) self.assertEqual(first_user_blocked.status_code, 429) # کاربرِ دومِ متفاوت هنوز باید بتونه تلاش کنه — throttle # per-user است، نه یک شمارنده‌ی سراسری روی endpoint. second_user_response = self._redeem('PROBE-C', order_id=other_order.id, token=other_token) self.assertEqual(second_user_response.status_code, 404) def test_successful_redemption_still_works_within_the_limit(self): # عدمِ رگرسیون: رفتارِ اصلیِ redemption باید دقیقاً مثلِ قبل کار کنه. discount = DiscountCode.objects.create( code='REDEEMTHROTTLE10', type='percent', value=10, max_uses=5, expires_at=None, ) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '10/hour'}): response = self._redeem(discount.code, token=self.token) self.assertEqual(response.status_code, 200) self.assertTrue(response.data['success']) self.assertEqual(response.data['discountAmount'], 1000) self.order.refresh_from_db() self.assertEqual(self.order.amount, 9000) def test_blocked_request_does_not_mutate_order(self): # درخواستِ بلاک‌شده نباید هیچ Redemption ای بسازه یا مبلغِ سفارش # رو تغییر بده — throttle باید *قبل* از رسیدن به منطقِ ویو رد کنه. discount = DiscountCode.objects.create( code='REDEEMTHROTTLEBLOCK', type='fixed', value=1000, max_uses=5, expires_at=None, ) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '1/hour'}): first_response = self._redeem('PROBE-X', token=self.token) self.assertEqual(first_response.status_code, 404) blocked_response = self._redeem(discount.code, token=self.token) self.assertEqual(blocked_response.status_code, 429) self.order.refresh_from_db() self.assertEqual(self.order.amount, 10000) class RedeemDiscountPerUserThrottleTests(TestCase): """هر کاربرِ ثبت‌نام‌شده سطلِ throttleِ جدای خودش را دارد؛ مصرفِ حسابِ دیگر به او نمی‌رسد.""" def setUp(self): cache.clear() self.client = APIClient() def _redeem(self, code, order_id, token): return self.client.post( '/api/discount/redeem', {'code': code, 'orderId': order_id}, 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, {'discount_redeem': '1/hour'}): user_a = User.objects.create(name='', phone='+989120000401') order_a = Order.objects.create( user_id=user_a.id, amount=10000, idempotency_key='order-redeem-reg-a', ) response_a = self._redeem('PROBE-REG-A', order_a.id, issue_access_token(user_a.id)) self.assertEqual(response_a.status_code, 404) user_b = User.objects.create(name='', phone='+989120000402') order_b = Order.objects.create( user_id=user_b.id, amount=10000, idempotency_key='order-redeem-reg-b', ) response_b = self._redeem('PROBE-REG-B', order_b.id, issue_access_token(user_b.id)) self.assertEqual(response_b.status_code, 404) class ValidateDiscountPerUserThrottleTests(TestCase): """هر کاربرِ ثبت‌نام‌شده سطلِ throttleِ جدای خودش را دارد؛ مصرفِ حسابِ دیگر به او نمی‌رسد.""" def setUp(self): cache.clear() self.client = APIClient() self.discount = DiscountCode.objects.create( code='GUESTROTVALIDATE10', type='percent', value=10, max_uses=5, expires_at=None, ) def _validate(self, code, token): return self.client.get( '/api/discount/validate', {'code': code}, HTTP_AUTHORIZATION=f'Bearer {token}', ) def test_registered_users_keep_independent_per_user_buckets(self): # عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ # قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '1/hour'}): user_a = User.objects.create(name='', phone='+989120000741') response_a = self._validate(self.discount.code, issue_access_token(user_a.id)) self.assertEqual(response_a.status_code, 200) user_b = User.objects.create(name='', phone='+989120000742') response_b = self._validate(self.discount.code, issue_access_token(user_b.id)) self.assertEqual(response_b.status_code, 200)