MathApp/subscription/tests.py

from unittest.mock import patch

from django.conf import settings
from django.db.models import F
from django.test import TestCase, TransactionTestCase, override_settings
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 payment.models import Order
from payment.services import IdempotencyKeyConflict, create_order

from .models import Plan, Subscription


class PurchaseIdempotencyRetryTests(TestCase):
    """Regression test: retrying an already-paid purchase must not regress
    an active subscription back to pending."""

    def setUp(self):
        self.client = APIClient()
        self.user = User.objects.create(device_id='device-purchase-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)
        self.plan = Plan.objects.create(
            title='ماهانه', price=100000, currency='IRT', duration_days=30,
        )

    def _purchase(self, idempotency_key):
        return self.client.post(
            '/api/subscription/purchase',
            {'planId': str(self.plan.id)},
            format='json',
            HTTP_AUTHORIZATION=f'Bearer {self.token}',
            HTTP_IDEMPOTENCY_KEY=idempotency_key,
        )

    def test_retry_of_paid_purchase_does_not_regress_active_subscription_to_pending(self):
        subscription = Subscription.objects.create(
            user_id=self.user.id, plan=self.plan, status='active',
        )
        order = Order.objects.create(
            user_id=self.user.id,
            subscription_id=subscription.id,
            amount=self.plan.price,
            currency=self.plan.currency,
            status='paid',
            provider='gateway',
            idempotency_key='retry-key-1',
        )
        subscription.order_id = order.id
        subscription.save(update_fields=['order_id'])

        response = self._purchase('retry-key-1')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data['orderId'], str(order.id))
        subscription.refresh_from_db()
        self.assertEqual(subscription.status, 'active')
        self.assertEqual(Order.objects.filter(idempotency_key='retry-key-1').count(), 1)

    def test_genuinely_new_purchase_still_starts_pending(self):
        response = self._purchase('brand-new-key-1')

        self.assertEqual(response.status_code, 200)
        subscription = Subscription.objects.get(user_id=self.user.id)
        self.assertEqual(subscription.status, 'pending')
        self.assertEqual(str(subscription.order_id), response.data['orderId'])


class PurchaseExternalCallNotUnderLockTests(TransactionTestCase):
    """Regression test: create_order must not run inside PurchaseView's
    transaction.atomic block, since it will eventually call an external
    payment gateway. Uses TransactionTestCase (not TestCase) because
    TestCase wraps every test in its own outer atomic block, which would
    make connection.in_atomic_block always True and hide this bug."""

    def setUp(self):
        self.client = APIClient()
        self.user = User.objects.create(device_id='device-purchase-lock-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)
        self.plan = Plan.objects.create(
            title='ماهانه', price=100000, currency='IRT', duration_days=30,
        )

    def test_create_order_is_called_without_an_open_db_transaction(self):
        from django.db import connection

        with patch('subscription.views.create_order', wraps=create_order) as mocked_create_order:
            def _assert_no_open_transaction(*args, **kwargs):
                # Fails here if the view still calls create_order under its own lock.
                assert not connection.in_atomic_block, (
                    'create_order must not be called while a DB transaction from '
                    'PurchaseView is open'
                )
                return create_order(*args, **kwargs)

            mocked_create_order.side_effect = _assert_no_open_transaction

            response = self.client.post(
                '/api/subscription/purchase',
                {'planId': str(self.plan.id)},
                format='json',
                HTTP_AUTHORIZATION=f'Bearer {self.token}',
            )

        self.assertEqual(response.status_code, 200)
        mocked_create_order.assert_called_once()


class PurchaseOrderCreationFailureCleanupTests(TestCase):
    """Regression test: even though create_order runs outside the view's
    lock, a failed order must still clean up the subscription correctly
    (delete if freshly created, restore prior state if it already existed)."""

    def setUp(self):
        self.client = APIClient()
        self.user = User.objects.create(device_id='device-purchase-cleanup-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)
        self.plan = Plan.objects.create(
            title='ماهانه', price=100000, currency='IRT', duration_days=30,
        )

    def _purchase(self):
        return self.client.post(
            '/api/subscription/purchase',
            {'planId': str(self.plan.id)},
            format='json',
            HTTP_AUTHORIZATION=f'Bearer {self.token}',
        )

    def test_failed_order_creation_deletes_freshly_created_subscription(self):
        self.assertFalse(Subscription.objects.filter(user_id=self.user.id).exists())

        with patch(
            'subscription.views.create_order',
            side_effect=IdempotencyKeyConflict('تست شکستِ create_order'),
        ):
            response = self._purchase()

        self.assertEqual(response.status_code, 409)
        self.assertFalse(Subscription.objects.filter(user_id=self.user.id).exists())

    def test_failed_order_creation_restores_existing_subscription_exactly(self):
        annual_plan = Plan.objects.create(
            title='سالانه', price=900000, currency='IRT', duration_days=365,
        )
        subscription = Subscription.objects.create(
            user_id=self.user.id, plan=annual_plan, status='active',
        )
        subscription.order_id = 4242
        subscription.save(update_fields=['order_id'])

        with patch(
            'subscription.views.create_order',
            side_effect=IdempotencyKeyConflict('تست شکستِ create_order'),
        ):
            response = self._purchase()

        self.assertEqual(response.status_code, 409)
        subscription.refresh_from_db()
        self.assertEqual(subscription.status, 'active')
        self.assertEqual(subscription.plan_id, annual_plan.id)
        self.assertEqual(subscription.order_id, 4242)


class PurchaseFailureCleanupConcurrencyRegressionTests(TestCase):
    """Regression test: cleanup after a failed create_order must not
    erase or overwrite a valid state written by a concurrent request in
    between phase 1 and this cleanup."""

    def setUp(self):
        self.client = APIClient()
        self.user = User.objects.create(device_id='device-purchase-race-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)
        self.plan = Plan.objects.create(
            title='ماهانه', price=100000, currency='IRT', duration_days=30,
        )
        self.other_plan = Plan.objects.create(
            title='سالانه', price=900000, currency='IRT', duration_days=365,
        )

    def _purchase(self):
        return self.client.post(
            '/api/subscription/purchase',
            {'planId': str(self.plan.id)},
            format='json',
            HTTP_AUTHORIZATION=f'Bearer {self.token}',
        )

    def test_failure_delete_path_does_not_erase_concurrent_success_on_new_subscription(self):
        self.assertFalse(Subscription.objects.filter(user_id=self.user.id).exists())

        def _fail_after_concurrent_success(*args, **kwargs):
            # Simulates a concurrent request succeeding between phase 1 and cleanup.
            concurrent_order = Order.objects.create(
                user_id=self.user.id,
                amount=self.other_plan.price,
                currency=self.other_plan.currency,
                status='paid',
                provider='gateway',
                idempotency_key='concurrent-key-1',
            )
            Subscription.objects.filter(user_id=self.user.id).update(
                status='active', plan_id=self.other_plan.id, order_id=concurrent_order.id,
            )
            raise IdempotencyKeyConflict('تست شکستِ create_order بعد از موفقیتِ هم‌زمان')

        with patch('subscription.views.create_order', side_effect=_fail_after_concurrent_success):
            response = self._purchase()

        self.assertEqual(response.status_code, 409)
        subscription = Subscription.objects.get(user_id=self.user.id)
        self.assertEqual(subscription.status, 'active')
        self.assertEqual(subscription.plan_id, self.other_plan.id)

    def test_failure_restore_path_does_not_overwrite_concurrent_success_on_existing_subscription(self):
        original_order_id = 4242
        subscription = Subscription.objects.create(
            user_id=self.user.id, plan=self.other_plan, status='active',
        )
        subscription.order_id = original_order_id
        subscription.save(update_fields=['order_id'])

        def _fail_after_concurrent_success(*args, **kwargs):
            # Phase 1 already set status to 'pending'; a concurrent request
            # now moves the subscription to a new active state.
            concurrent_order = Order.objects.create(
                user_id=self.user.id,
                amount=self.plan.price,
                currency=self.plan.currency,
                status='paid',
                provider='gateway',
                idempotency_key='concurrent-key-2',
            )
            Subscription.objects.filter(user_id=self.user.id).update(
                status='active', plan_id=self.plan.id, order_id=concurrent_order.id,
            )
            raise IdempotencyKeyConflict('تست شکستِ create_order بعد از موفقیتِ هم‌زمان')

        with patch('subscription.views.create_order', side_effect=_fail_after_concurrent_success):
            response = self._purchase()

        self.assertEqual(response.status_code, 409)
        subscription.refresh_from_db()
        self.assertEqual(subscription.status, 'active')
        self.assertEqual(subscription.plan_id, self.plan.id)
        self.assertNotEqual(subscription.order_id, original_order_id)


class PurchaseIdenticalIntermediateStateRaceRegressionTests(TestCase):
    """Regression test: two concurrent purchases for the same plan can
    reach an identical intermediate state (pending, same plan_id,
    order_id=None), so cleanup can't distinguish them by field comparison
    alone. Simulates a concurrent request B writing the same phase-1
    state while request A is failing."""

    def setUp(self):
        self.client = APIClient()
        self.user = User.objects.create(device_id='device-purchase-race-identical-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)
        self.plan = Plan.objects.create(
            title='ماهانه', price=100000, currency='IRT', duration_days=30,
        )

    def _purchase(self, idempotency_key):
        return self.client.post(
            '/api/subscription/purchase',
            {'planId': str(self.plan.id)},
            format='json',
            HTTP_AUTHORIZATION=f'Bearer {self.token}',
            HTTP_IDEMPOTENCY_KEY=idempotency_key,
        )

    def test_failed_request_does_not_delete_concurrent_requests_identical_pending_state(self):
        def _simulate_concurrent_identical_purchase_then_fail(*args, **kwargs):
            # Request B: concurrently repeats phase 1 (update_or_create + version
            # bump) for the same user/plan, reaching the same intermediate state.
            Subscription.objects.update_or_create(
                user_id=self.user.id, defaults={'plan': self.plan, 'status': 'pending'},
            )
            b_subscription = Subscription.objects.get(user_id=self.user.id)
            Subscription.objects.filter(pk=b_subscription.pk).update(version=F('version') + 1)
            # Request A: its create_order fails.
            raise IdempotencyKeyConflict('create_order failure for request A')

        with patch(
            'subscription.views.create_order',
            side_effect=_simulate_concurrent_identical_purchase_then_fail,
        ):
            response = self._purchase('key-A')

        self.assertEqual(response.status_code, 409)
        # Subscription now holds B's state and must survive A's failure cleanup
        # even though A and B wrote identical status/plan_id/order_id.
        subscription = Subscription.objects.get(user_id=self.user.id)
        self.assertEqual(subscription.status, 'pending')
        self.assertEqual(subscription.plan_id, self.plan.id)
        self.assertIsNone(subscription.order_id)
        self.assertEqual(subscription.version, 2)


class PurchaseThrottleTests(TestCase):
    """
    🔒 اثبات محدودیتِ جدیدِ per-user روی PurchaseView (throttle_scope='purchase'):
    تعدادِ زیادِ درخواستِ purchase با Idempotency-Key های *متفاوت* از یک
    اکانتِ احرازهویت‌شده باید بعد از حدِ مجاز با ۴۲۹ رد بشه — چون محافظتِ
    idempotency فقط جلوی reuse یک کلیدِ یکسان رو می‌گیره، نه حجمِ بالای
    تلاش‌های *متفاوت*. ScopedRateThrottle این‌جا با pk کاربر (نه IP) کلید
    می‌خوره، پس تغییرِ IP از همون اکانت این محدودیت رو دور نمی‌زنه.
    """

    def setUp(self):
        from django.core.cache import cache
        cache.clear()
        self.client = APIClient()
        self.user = User.objects.create(device_id='device-purchase-throttle-1', is_guest=True, name='')
        self.token = issue_access_token(self.user.id)
        self.plan = Plan.objects.create(
            title='ماهانه', price=100000, currency='IRT', duration_days=30,
        )

    def _purchase(self, idempotency_key, remote_addr=None):
        kwargs = {}
        if remote_addr:
            kwargs['REMOTE_ADDR'] = remote_addr
        return self.client.post(
            '/api/subscription/purchase',
            {'planId': str(self.plan.id)},
            format='json',
            HTTP_AUTHORIZATION=f'Bearer {self.token}',
            HTTP_IDEMPOTENCY_KEY=idempotency_key,
            **kwargs,
        )

    @override_settings(
        REST_FRAMEWORK={
            **settings.REST_FRAMEWORK,
            'DEFAULT_THROTTLE_RATES': {
                **settings.REST_FRAMEWORK['DEFAULT_THROTTLE_RATES'],
                'purchase': '3/hour',
            },
        }
    )
    def test_requests_within_the_limit_still_succeed(self):
        # ⚠️ مثلِ GuestSessionThrottleTests: override_settings به‌تنهایی روی
        # ScopedRateThrottle.THROTTLE_RATES (که یک‌بار موقعِ import گرفته
        # شده) اثر نداره؛ برای اینکه نرخِ واقعاً محدودشده اعمال بشه، مستقیماً
        # دیکشنریِ زنده‌ی THROTTLE_RATES رو patch می‌کنیم.
        with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'purchase': '3/hour'}):
            for i in range(3):
                response = self._purchase(f'key-within-limit-{i}')
                self.assertEqual(response.status_code, 200)

    def test_excessive_purchase_attempts_from_same_user_are_blocked(self):
        with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'purchase': '3/hour'}):
            for i in range(3):
                response = self._purchase(f'key-block-{i}')
                self.assertEqual(response.status_code, 200)

            blocked_response = self._purchase('key-block-extra')
        self.assertEqual(blocked_response.status_code, 429)

    def test_changing_client_ip_does_not_bypass_the_user_limit(self):
        # کاربرِ غیرمهمان لازمه، نه self.user، چون کلیدِ bucketِ مهمان‌ها
        # عمداً بر اساسِ IP است (نگاه کنید به PurchaseGuestThrottleBypassTests)
        # و این تست رفتارِ per-user برای ثبت‌نام‌شده‌ها را می‌سنجد.
        registered_user = User.objects.create(is_guest=False, name='', phone='+989120000501')
        registered_token = issue_access_token(registered_user.id)

        def _purchase_as(idempotency_key, remote_addr):
            return self.client.post(
                '/api/subscription/purchase',
                {'planId': str(self.plan.id)},
                format='json',
                HTTP_AUTHORIZATION=f'Bearer {registered_token}',
                HTTP_IDEMPOTENCY_KEY=idempotency_key,
                REMOTE_ADDR=remote_addr,
            )

        with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'purchase': '3/hour'}):
            for i in range(3):
                response = _purchase_as(f'key-ip-{i}', f'10.1.0.{i + 1}')
                self.assertEqual(response.status_code, 200)

            blocked_response = _purchase_as('key-ip-extra', '10.1.0.99')
        self.assertEqual(blocked_response.status_code, 429)

    def test_throttled_request_creates_no_order_and_does_not_mutate_subscription(self):
        with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'purchase': '1/hour'}):
            first_response = self._purchase('key-mutate-1')
            self.assertEqual(first_response.status_code, 200)

            subscription = Subscription.objects.get(user_id=self.user.id)
            snapshot = {
                'status': subscription.status,
                'plan_id': subscription.plan_id,
                'order_id': subscription.order_id,
                'version': subscription.version,
            }
            order_count_before = Order.objects.filter(user_id=self.user.id).count()

            with patch('subscription.views.create_order') as mock_create_order:
                blocked_response = self._purchase('key-mutate-2')
                self.assertEqual(blocked_response.status_code, 429)
                mock_create_order.assert_not_called()

        self.assertEqual(Order.objects.filter(user_id=self.user.id).count(), order_count_before)
        subscription.refresh_from_db()
        self.assertEqual(subscription.status, snapshot['status'])
        self.assertEqual(subscription.plan_id, snapshot['plan_id'])
        self.assertEqual(subscription.order_id, snapshot['order_id'])
        self.assertEqual(subscription.version, snapshot['version'])


class PurchaseGuestThrottleBypassTests(TestCase):
    """
    🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='purchase' برای هر
    کاربرِ احرازهویت‌شده (مهمان یا نه) user.pk بود. چون GuestSessionView
    با هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم
    می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک bucket throttle
    تازه برای سقفِ ۱۰/ساعتِ purchase بگیره. این endpoint حساسِ مالیه: هر
    درخواستِ پذیرفته‌شده Subscription می‌سازه/آپدیت می‌کنه و create_order()
    رو صدا می‌زنه. حالا GuestAwarePurchaseThrottle برای کاربرهای مهمان
    کلید رو بر اساسِ IP (همون get_ident که guest_session هم استفاده
    می‌کنه) می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده. رفتارِ
    throttle برای کاربرِ غیرمهمان (per-user، مثلِ قبل) دست‌نخورده مونده.
    """

    def setUp(self):
        from django.core.cache import cache
        cache.clear()
        self.client = APIClient()
        self.plan = Plan.objects.create(
            title='ماهانه', price=100000, currency='IRT', duration_days=30,
        )

    def _purchase(self, idempotency_key, token):
        return self.client.post(
            '/api/subscription/purchase',
            {'planId': str(self.plan.id)},
            format='json',
            HTTP_AUTHORIZATION=f'Bearer {token}',
            HTTP_IDEMPOTENCY_KEY=idempotency_key,
        )

    def test_rotating_guest_accounts_share_the_same_throttle_bucket(self):
        # سه حساب مهمانِ متفاوت (شبیه‌سازیِ چرخشِ حساب برای دورزدنِ throttle)،
        # همه از همون کلاینت/IP — باید همه یک bucket مشترک رو مصرف کنند.
        with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'purchase': '2/hour'}):
            guest1 = User.objects.create(device_id='device-purchase-guest-rot-1', is_guest=True, name='')
            response1 = self._purchase('key-guest-rot-1', issue_access_token(guest1.id))
            self.assertEqual(response1.status_code, 200)

            guest2 = User.objects.create(device_id='device-purchase-guest-rot-2', is_guest=True, name='')
            response2 = self._purchase('key-guest-rot-2', issue_access_token(guest2.id))
            self.assertEqual(response2.status_code, 200)

            # سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
            # bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk — یعنی
            # سقفِ ۲/ساعت هنوز اعمال می‌شه، صرف‌نظر از تعدادِ حساب‌های
            # مهمانِ ساخته‌شده.
            guest3 = User.objects.create(device_id='device-purchase-guest-rot-3', is_guest=True, name='')
            response3 = self._purchase('key-guest-rot-3', issue_access_token(guest3.id))

        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, {'purchase': '1/hour'}):
            guest1 = User.objects.create(device_id='device-purchase-guest-rot-4', is_guest=True, name='')
            response1 = self._purchase('key-guest-rot-4', issue_access_token(guest1.id))
            self.assertEqual(response1.status_code, 200)

            guest2 = User.objects.create(device_id='device-purchase-guest-rot-5', is_guest=True, name='')
            blocked = self._purchase('key-guest-rot-5', issue_access_token(guest2.id))
            self.assertEqual(blocked.status_code, 429)

            guest3 = User.objects.create(device_id='device-purchase-guest-rot-6', is_guest=True, name='')
            still_blocked = self._purchase('key-guest-rot-6', issue_access_token(guest3.id))

        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, {'purchase': '1/hour'}):
            user_a = User.objects.create(is_guest=False, name='', phone='+989120000601')
            response_a = self._purchase('key-reg-a', issue_access_token(user_a.id))
            self.assertEqual(response_a.status_code, 200)

            user_b = User.objects.create(is_guest=False, name='', phone='+989120000602')
            response_b = self._purchase('key-reg-b', issue_access_token(user_b.id))

        self.assertEqual(response_b.status_code, 200)

    def test_blocked_guest_request_does_not_mutate_subscription_or_call_purchase_logic(self):
        # درخواستِ بلاک‌شده (از حسابِ مهمانِ چرخشی) نباید Subscription/Order
        # بسازه یا create_order() رو صدا بزنه — throttle باید *قبل* از
        # رسیدن به منطقِ purchase رد کنه، حتی با Idempotency-Key معتبر.
        with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'purchase': '1/hour'}):
            guest1 = User.objects.create(device_id='device-purchase-guest-rot-7', is_guest=True, name='')
            first_response = self._purchase('key-guest-rot-7', issue_access_token(guest1.id))
            self.assertEqual(first_response.status_code, 200)

            # حسابِ مهمانِ تازه‌ی دوم باید بلاک بشه (bucket مشترکِ IP از
            # قبل مصرف شده) و نباید Subscription/Order برای آن بسازد.
            guest2 = User.objects.create(device_id='device-purchase-guest-rot-8', is_guest=True, name='')
            with patch('subscription.views.create_order') as mock_create_order:
                blocked_response = self._purchase('key-guest-rot-8', issue_access_token(guest2.id))
                self.assertEqual(blocked_response.status_code, 429)
                mock_create_order.assert_not_called()

        self.assertFalse(Subscription.objects.filter(user_id=guest2.id).exists())
        self.assertEqual(Order.objects.filter(user_id=guest2.id).count(), 0)


class PlanCardContractTests(TestCase):
    """
    کارتِ اشتراک در فرانت فهرستِ امکانات و یک برچسب نشان می‌دهد؛ تا امروز هر
    دو در کدِ اپ هاردکد بودند، یعنی عوض کردنِ یک بند به دیپلویِ تازه نیاز داشت.
    """

    def setUp(self):
        from django.core.cache import cache
        cache.clear()

    def test_plans_carry_their_feature_list_and_badge(self):
        Plan.objects.create(title='طلایی', price=49000, popular=True,
                            badge='محبوب‌ترین',
                            features=['درس‌های نامحدود', 'بدون تبلیغ'])
        row = self.client.get('/api/subscription/plans').json()['plans'][0]
        self.assertEqual(row['badge'], 'محبوب‌ترین')
        self.assertEqual(row['features'], ['درس‌های نامحدود', 'بدون تبلیغ'])

    def test_a_plan_without_them_still_serializes(self):
        Plan.objects.create(title='رایگان', price=0)
        row = self.client.get('/api/subscription/plans').json()['plans'][0]
        self.assertEqual(row['features'], [])
        self.assertEqual(row['badge'], '')