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(phone='09000000042', 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(phone='09000000043', 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(phone='09000000044', 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(phone='09000000045', 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(phone='09000000046', 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(phone='09000000047', 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):
        # 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):
        # کلیدِ throttle شناسه‌ی کاربر است، نه IP؛ تغییرِ IP سطلِ تازه نمی‌دهد.
        registered_user = User.objects.create(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 PurchasePerUserThrottleTests(TestCase):
    """هر کاربرِ ثبت‌نام‌شده سطلِ throttleِ جدای خودش را دارد؛ مصرفِ حسابِ دیگر به او نمی‌رسد."""

    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_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(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(name='', phone='+989120000602')
            response_b = self._purchase('key-reg-b', issue_access_token(user_b.id))

        self.assertEqual(response_b.status_code, 200)


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

    def setUp(self):
        from django.core.cache import cache
        cache.clear()
        # migrationِ 0005 پلنِ یک‌ساله را می‌سازد؛ این تست‌ها شکلِ یک ردیفِ
        # مشخص را می‌سنجند و باید از جدولِ خالی شروع کنند.
        Plan.objects.all().delete()

    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'], '')


class YearlyPlanSeedTests(TestCase):
    """migrationهای 0005 و 0007: سرورِ بی‌پلن پلنِ سالانه و ماهانه با قیمتِ ۱۴۰۵-۰۷ می‌گیرد."""

    def test_the_yearly_plan_exists_after_migrations(self):
        plan = Plan.objects.get(title='اشتراک یک‌ساله')
        self.assertEqual(plan.duration_days, 365)
        self.assertEqual((plan.price, plan.ai_daily_quota), (1_790_000, 10))
        self.assertEqual(plan.currency, 'IRT')
        self.assertTrue(plan.features)

    def test_a_monthly_plan_is_created_next_to_it(self):
        plan = Plan.objects.get(duration_days=30)
        self.assertEqual((plan.price, plan.ai_daily_quota), (179_000, 10))

    def test_repricing_leaves_an_admin_edited_yearly_plan_alone(self):
        import importlib
        from django.apps import apps as global_apps
        reprice = importlib.import_module('subscription.migrations.0007_plans_priced_by_ai_cost')
        Plan.objects.filter(title='اشتراک یک‌ساله').update(price=450_000, ai_daily_quota=9)
        reprice.apply_prices(global_apps, None)
        importlib.import_module('subscription.migrations.0008_paid_quota_ten_a_day').apply(global_apps, None)
        plan = Plan.objects.get(title='اشتراک یک‌ساله')
        self.assertEqual((plan.price, plan.ai_daily_quota), (450_000, 9))
        self.assertEqual(Plan.objects.filter(duration_days=30).count(), 1)

    def test_only_question_packs_are_on_sale(self):
        from django.core.cache import cache
        cache.clear()
        data = self.client.get('/api/subscription/plans').json()
        self.assertEqual({p['kind'] for p in data['plans']}, {'pack'})
        self.assertEqual([p['questionCredits'] for p in data['plans']], [20, 60, 150])
        self.assertEqual([p['heartsDays'] for p in data['plans']], [7, 30, 60])
        self.assertEqual(data['freeAiDailyQuota'], 3)
        # پلن‌های اشتراکی پاک نشده‌اند؛ فقط فروخته نمی‌شوند.
        self.assertFalse(Plan.objects.get(title='اشتراک یک‌ساله').on_sale)

    def test_it_never_overwrites_existing_plans(self):
        import importlib
        from django.apps import apps as global_apps
        seed = importlib.import_module('subscription.migrations.0005_seed_yearly_plan')
        Plan.objects.all().delete()
        Plan.objects.create(title='پلنِ ادمین', price=1000)

        seed.create_yearly_plan(global_apps, None)

        self.assertEqual(list(Plan.objects.values_list('title', flat=True)), ['پلنِ ادمین'])


class PlanCacheInvalidationTests(TestCase):
    """
    ذخیره‌ی پلن در ادمین باید فهرستِ پلن‌ها را تازه کند. تا ۲۰۲۶-۱۰-۰۳ گیرنده‌ی سیگنال کلیدِ
    قدیمیِ 'v1' را پاک می‌کرد و فهرست تا یک ساعت قیمتِ قبلی را نشان می‌داد، در حالی که سفارش با
    قیمتِ تازه ساخته می‌شد.
    """

    def test_admin_price_change_shows_up_immediately(self):
        from django.core.cache import cache
        cache.clear()
        plan = Plan.objects.create(title='بسته‌ی آزمایشی', price=45000, kind='pack', question_credits=7)
        client = APIClient()

        def price():
            plans = client.get('/api/subscription/plans').json()['plans']
            return next(p['price'] for p in plans if p['id'] == str(plan.id))

        self.assertEqual(price(), 45000)
        plan.price = 99000
        plan.save()
        self.assertEqual(price(), 99000)