cmhub/apps/licensing/tests.py

from datetime import timedelta
from decimal import Decimal
from concurrent.futures import ThreadPoolExecutor
from io import StringIO
from types import SimpleNamespace
from unittest.mock import patch

from django.contrib import admin
from django.core.management import call_command
from django.core.management.base import CommandError
from django.db import IntegrityError, close_old_connections
from django.test import SimpleTestCase, TestCase, TransactionTestCase, override_settings
from django.urls import reverse
from django.utils import timezone
from rest_framework.test import APIClient

from apps.billing.models import PointsLedger
from apps.billing.payment_gateways import PaymentOrderCode, PaymentReceipt
from apps.licensing.models import (
    ClientSubscriptionPolicy,
    ClientSubscriptionPolicyAudit,
    ClientDevice,
    DeviceBindingAudit,
    DeviceCredential,
    DeviceSession,
    LegacyMigrationGrant,
    LicenseEvent,
    LicenseSeat,
    MigrationRequest,
    SoftwareEntitlement,
    SoftwareOrder,
    SoftwarePlan,
)
from apps.licensing.services import (
    LicensingError,
    SoftwareOrderAmountMismatchError,
    SoftwareOrderError,
    SoftwareOrderTransactionMismatchError,
    SubscriptionAuthorizationDecision,
    apply_software_payment,
    assign_license_seat,
    confirm_migration_request,
    create_legacy_migration_grant,
    create_migration_request,
    create_software_order,
    evaluate_device_authorization,
    evaluate_subscription_access,
    get_subscription_mode,
    get_published_client_subscription_policy,
    grant_plan_to_existing_users,
    grant_software_entitlement,
    record_device_heartbeat,
    release_license_seat,
    renew_software_entitlement,
    revoke_software_entitlement,
    register_device,
    software_subscription_status,
    set_client_subscription_policy,
)
from apps.users.models import ApiKey, User, UserWallet


class ExistingUserPlanGrantCommandTests(TestCase):
    def setUp(self):
        self.plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="测试",
            duration_days=30,
            price=Decimal("99.00"),
            device_limit=1,
            grace_days=2,
        )
        self.user_one = User.objects.create_user(
            username="existing-one",
            email="existing-one@example.com",
            password="password",
        )
        self.user_two = User.objects.create_user(
            username="existing-two",
            email="existing-two@example.com",
            password="password",
        )
        self.staff_user = User.objects.create_user(
            username="staff-user",
            email="staff-user@example.com",
            password="password",
            is_staff=True,
        )
        self.inactive_user = User.objects.create_user(
            username="inactive-user",
            email="inactive-user@example.com",
            password="password",
            is_active=False,
        )

    def command_options(self, **overrides):
        options = {
            "plan_id": self.plan.pk,
            "reason": "T-633 存量用户测试套餐过渡",
        }
        options.update(overrides)
        return options

    def test_dry_run_is_read_only_and_reports_scope(self):
        stdout = StringIO()

        call_command("grant_existing_users_plan", stdout=stdout, **self.command_options())

        self.assertIn("mode=DRY-RUN", stdout.getvalue())
        self.assertIn("eligible=2", stdout.getvalue())
        self.assertIn("skipped_existing=0", stdout.getvalue())
        self.assertIn("grant_count=2", stdout.getvalue())
        self.assertFalse(SoftwareEntitlement.objects.exists())
        self.assertFalse(LicenseSeat.objects.exists())
        self.assertFalse(LicenseEvent.objects.exists())

    def test_execute_grants_only_active_nonstaff_users_with_audit_events(self):
        before = timezone.now()

        call_command(
            "grant_existing_users_plan",
            execute=True,
            expected_grant_count=2,
            **self.command_options(),
        )

        entitlements = SoftwareEntitlement.objects.order_by("user_id")
        self.assertEqual(entitlements.count(), 2)
        self.assertSetEqual(
            set(entitlements.values_list("user_id", flat=True)),
            {self.user_one.pk, self.user_two.pk},
        )
        entitlement = entitlements.first()
        self.assertEqual(entitlement.source_plan, self.plan)
        self.assertEqual(entitlement.plan_name, "测试")
        self.assertEqual(entitlement.plan_duration_days, 30)
        self.assertEqual(entitlement.plan_grace_days, 2)
        self.assertGreaterEqual(entitlement.starts_at, before)
        self.assertEqual(
            entitlement.expires_at,
            entitlement.starts_at + timedelta(days=30),
        )
        self.assertEqual(
            entitlement.grace_expires_at,
            entitlement.expires_at + timedelta(days=2),
        )
        self.assertEqual(LicenseSeat.objects.count(), 2)
        self.assertEqual(
            LicenseEvent.objects.filter(
                action=LicenseEvent.Action.GRANTED,
                reason="T-633 存量用户测试套餐过渡",
            ).count(),
            2,
        )

    def test_existing_usable_product_entitlement_is_skipped_and_rerun_is_idempotent(self):
        formal_plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="正式会员",
            duration_days=365,
            price=Decimal("999.00"),
            device_limit=1,
            grace_days=7,
        )
        grant_software_entitlement(
            user=self.user_one,
            plan=formal_plan,
            reason="已有正式权益",
        )

        call_command(
            "grant_existing_users_plan",
            execute=True,
            expected_grant_count=1,
            **self.command_options(),
        )
        result = grant_plan_to_existing_users(
            plan=self.plan,
            reason="重复预演",
        )

        self.assertEqual(SoftwareEntitlement.objects.count(), 2)
        self.assertEqual(result.eligible_count, 2)
        self.assertEqual(result.skipped_existing_count, 2)
        self.assertEqual(result.grant_count, 0)
        self.assertFalse(result.executed)

    def test_execute_requires_matching_preview_count(self):
        with self.assertRaisesRegex(CommandError, "当前实际应授予 2 人"):
            call_command(
                "grant_existing_users_plan",
                execute=True,
                expected_grant_count=1,
                **self.command_options(),
            )

        self.assertFalse(SoftwareEntitlement.objects.exists())
        self.assertFalse(LicenseEvent.objects.exists())

    def test_failure_rolls_back_entire_batch(self):
        original_grant = grant_software_entitlement
        call_count = 0

        def fail_second_grant(**kwargs):
            nonlocal call_count
            call_count += 1
            if call_count == 2:
                raise RuntimeError("simulated grant failure")
            return original_grant(**kwargs)

        with patch(
            "apps.licensing.services.grant_software_entitlement",
            side_effect=fail_second_grant,
        ):
            with self.assertRaisesRegex(RuntimeError, "simulated grant failure"):
                call_command(
                    "grant_existing_users_plan",
                    execute=True,
                    expected_grant_count=2,
                    **self.command_options(),
                )

        self.assertFalse(SoftwareEntitlement.objects.exists())
        self.assertFalse(LicenseSeat.objects.exists())
        self.assertFalse(LicenseEvent.objects.exists())


class SubscriptionModeUnitTests(SimpleTestCase):
    @override_settings(
        CMSHOPEE_SUBSCRIPTION_MODE="",
        CMSHOPEE_SUBSCRIPTION_ENFORCEMENT=True,
    )
    def test_legacy_enforcement_setting_maps_to_enforce(self):
        self.assertEqual(get_subscription_mode(), "enforce")

    @override_settings(CMSHOPEE_SUBSCRIPTION_MODE="shadow")
    @patch("apps.licensing.services.evaluate_account_authorization")
    def test_shadow_access_keeps_real_rejection_for_observation(self, evaluate_mock):
        evaluate_mock.return_value = SubscriptionAuthorizationDecision(
            product_code="cmshopee",
            allowed=False,
            code="subscription_required",
        )

        access = evaluate_subscription_access(
            user=SimpleNamespace(),
            product_code="cmshopee",
        )

        self.assertTrue(access.allowed)
        self.assertEqual(access.access_source, "shadow_fallback")
        self.assertEqual(access.entitlement_status, "required")
        self.assertEqual(access.entitlement_code, "subscription_required")

    @patch("apps.licensing.services.evaluate_account_authorization")
    def test_open_status_has_client_contract_without_database_entitlement(self, evaluate_mock):
        evaluate_mock.return_value = SubscriptionAuthorizationDecision(
            product_code="cmshopee",
            allowed=False,
            code="subscription_required",
        )
        user = SimpleNamespace(
            get_username=lambda: "desktop-user",
            get_full_name=lambda: "",
        )

        data = software_subscription_status(
            user=user,
            product_code="cmshopee",
            mode="open",
        )

        self.assertEqual(data["status"], "active")
        self.assertTrue(data["allowed"])
        self.assertEqual(data["account"]["display_name"], "desktop-user")
        self.assertEqual(data["plan"]["code"], "development-open")
        self.assertEqual(data["expires_at"], data["plan"]["expires_at"])
        self.assertEqual(data["access_source"], "open_mode")
        self.assertEqual(data["entitlement_status"], "required")
        self.assertFalse(data["real_entitlement_allowed"])

    def test_admin_index_only_exposes_membership_models_without_deleting_registration(self):
        hidden_models = (
            ClientDevice,
            DeviceSession,
            DeviceBindingAudit,
            LicenseSeat,
            LegacyMigrationGrant,
            MigrationRequest,
            DeviceCredential,
            SoftwareOrder,
            LicenseEvent,
        )
        user = SimpleNamespace(
            is_active=True,
            is_staff=True,
            is_superuser=True,
            has_perm=lambda *_args, **_kwargs: True,
            has_module_perms=lambda *_args, **_kwargs: True,
        )
        request = SimpleNamespace(user=user)

        for model in hidden_models:
            self.assertIn(model, admin.site._registry)
            self.assertEqual(admin.site._registry[model].get_model_perms(request), {})
        self.assertTrue(admin.site._registry[SoftwarePlan].get_model_perms(request))
        self.assertTrue(admin.site._registry[SoftwareEntitlement].get_model_perms(request))
        self.assertEqual(SoftwarePlan._meta.verbose_name, "会员套餐")
        self.assertEqual(SoftwareEntitlement._meta.verbose_name, "用户会员")

    @patch("apps.licensing.services.evaluate_account_authorization")
    def test_enforce_status_reports_real_grace_period_at_top_level(self, evaluate_mock):
        now = timezone.now()
        entitlement = SimpleNamespace(
            id=7,
            source_plan_id=3,
            plan_name="专业版",
            expires_at=now - timedelta(days=1),
            grace_expires_at=now + timedelta(days=2),
        )
        evaluate_mock.return_value = SubscriptionAuthorizationDecision(
            product_code="cmshopee",
            allowed=True,
            code="",
            entitlement=entitlement,
        )
        user = SimpleNamespace(
            get_username=lambda: "member-user",
            get_full_name=lambda: "会员用户",
        )

        data = software_subscription_status(
            user=user,
            product_code="cmshopee",
            now=now,
            mode="enforce",
        )

        self.assertEqual(data["status"], "grace")
        self.assertEqual(data["entitlement_status"], "grace")
        self.assertTrue(data["real_entitlement_allowed"])
        self.assertEqual(data["plan"]["code"], "plan-3")
        self.assertEqual(data["expires_at"], entitlement.expires_at.isoformat())
        self.assertEqual(
            data["grace_expires_at"],
            entitlement.grace_expires_at.isoformat(),
        )


class DeviceRegistrationApiTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="device-user",
            email="device@example.com",
            password="test-password",
        )
        UserWallet.objects.create(user=self.user, points_balance=20)
        self.api_key, self.raw_api_key = ApiKey.create_for_user(self.user, name="desktop")
        self.client = APIClient()
        self.register_url = reverse("api-client-device-register")
        self.heartbeat_url = reverse("api-client-device-heartbeat")
        self.payload = {
            "product_code": ClientDevice.ProductCode.CMSHOPEE,
            "device_id": "v1:installed-device-abcdef",
            "device_id_version": "v1",
            "installation_public_key": "test-installation-public-key",
            "platform": ClientDevice.Platform.WINDOWS,
            "client_version": "0.1.0",
        }

    def register_device(self):
        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.raw_api_key}")
        return self.client.post(self.register_url, self.payload, format="json")

    def test_register_requires_api_key(self):
        response = self.client.post(self.register_url, self.payload, format="json")

        self.assertEqual(response.status_code, 401)
        self.assertEqual(response.data["error"]["code"], "unauthorized")
        self.assertFalse(ClientDevice.objects.exists())

    def test_register_creates_hashed_device_and_one_time_session_token(self):
        response = self.register_device()

        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.data["device"]["product_code"], "cmshopee")
        self.assertIn("device_session_token", response.data)
        self.assertNotIn("installation_public_key", response.data)
        self.assertNotIn("device_id", response.data)

        device = ClientDevice.objects.get(user=self.user)
        session = DeviceSession.objects.get(device=device)
        raw_token = response.data["device_session_token"]
        self.assertNotEqual(device.device_fingerprint, self.payload["device_id"])
        self.assertNotEqual(device.public_key_fingerprint, self.payload["installation_public_key"])
        self.assertNotEqual(session.token_hash, raw_token)
        self.assertTrue(session.matches_token(raw_token))
        self.assertEqual(
            DeviceBindingAudit.objects.filter(
                device=device,
                action=DeviceBindingAudit.Action.REGISTERED,
            ).count(),
            1,
        )

    def test_repeat_registration_reuses_device_and_rotates_session(self):
        first = self.register_device()
        second = self.register_device()

        self.assertEqual(first.status_code, 201)
        self.assertEqual(second.status_code, 200)
        self.assertEqual(ClientDevice.objects.filter(user=self.user).count(), 1)
        device = ClientDevice.objects.get(user=self.user)
        self.assertEqual(
            DeviceSession.objects.filter(device=device, revoked_at__isnull=True).count(),
            1,
        )
        self.assertEqual(DeviceSession.objects.filter(device=device).count(), 2)
        self.assertNotEqual(
            first.data["device_session_token"],
            second.data["device_session_token"],
        )

    @override_settings(DEVICE_SESSION_TTL_SECONDS=172800)
    def test_heartbeat_uses_device_session_and_activity_is_throttled(self):
        registration = self.register_device()
        token = registration.data["device_session_token"]
        self.client.credentials(HTTP_X_DEVICE_SESSION=token)

        response = self.client.post(self.heartbeat_url, {}, format="json")

        self.assertEqual(response.status_code, 200)
        self.assertFalse(response.data["activity_updated"])
        device = ClientDevice.objects.get(user=self.user)
        session = DeviceSession.objects.get(device=device, revoked_at__isnull=True)
        updated = record_device_heartbeat(
            session,
            now=device.last_seen_at + timedelta(days=1, seconds=1),
        )
        self.assertTrue(updated)
        device.refresh_from_db()
        self.assertEqual(
            DeviceBindingAudit.objects.filter(
                device=device,
                action=DeviceBindingAudit.Action.HEARTBEAT,
            ).count(),
            1,
        )

    def test_invalid_or_revoked_device_session_is_rejected(self):
        self.client.credentials(HTTP_X_DEVICE_SESSION="dvs_cmhub_invalid")
        invalid = self.client.post(self.heartbeat_url, {}, format="json")
        self.assertEqual(invalid.status_code, 401)
        self.assertEqual(invalid.data["error"]["code"], "device_session_invalid")

        registration = self.register_device()
        device = ClientDevice.objects.get(user=self.user)
        device.status = ClientDevice.Status.REVOKED
        device.save(update_fields=("status",))
        self.client.credentials(HTTP_X_DEVICE_SESSION=registration.data["device_session_token"])
        revoked = self.client.post(self.heartbeat_url, {}, format="json")
        self.assertEqual(revoked.status_code, 403)
        self.assertEqual(revoked.data["error"]["code"], "device_revoked")

    def test_existing_balance_api_remains_compatible_without_device_header(self):
        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.raw_api_key}")

        response = self.client.get(reverse("api-balance"))

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data["points_balance"], 20)

    @override_settings(DEVICE_ACTIVITY_UPDATE_SECONDS=60)
    def test_heartbeat_updates_when_device_is_stale(self):
        registration = self.register_device()
        device = ClientDevice.objects.get(user=self.user)
        stale_time = timezone.now() - timedelta(minutes=2)
        ClientDevice.objects.filter(pk=device.pk).update(last_seen_at=stale_time)
        self.client.credentials(HTTP_X_DEVICE_SESSION=registration.data["device_session_token"])

        response = self.client.post(self.heartbeat_url, {}, format="json")

        self.assertEqual(response.status_code, 200)
        self.assertTrue(response.data["activity_updated"])
        device.refresh_from_db()
        self.assertGreater(device.last_seen_at, stale_time)

    @override_settings(DEVICE_ACTIVITY_UPDATE_SECONDS=60)
    def test_repeat_registration_updates_stale_device_activity(self):
        self.register_device()
        device = ClientDevice.objects.get(user=self.user)
        stale_time = timezone.now() - timedelta(minutes=2)
        ClientDevice.objects.filter(pk=device.pk).update(last_seen_at=stale_time)

        response = self.register_device()

        self.assertEqual(response.status_code, 200)
        device.refresh_from_db()
        self.assertGreater(device.last_seen_at, stale_time)


class ClientSubscriptionPolicyServiceTests(TestCase):
    def setUp(self):
        self.operator = User.objects.create_user(
            username="client-policy-operator",
            email="client-policy-operator@example.com",
            password="test-password",
            is_staff=True,
            is_superuser=True,
        )

    @override_settings(
        CMSHOPEE_CLIENT_SUBSCRIPTION_POLICY="enforce",
        CMSHOPEE_CLIENT_SUBSCRIPTION_POLICY_UPDATED_AT="2026-07-28T10:00:00+08:00",
    )
    def test_missing_admin_policy_falls_back_to_environment_configuration(self):
        self.assertEqual(
            get_published_client_subscription_policy(),
            {
                "policy_version": 1,
                "subscription_check_enabled": True,
                "subscription_enforcement_enabled": True,
                "updated_at": "2026-07-28T10:00:00+08:00",
            },
        )

    @override_settings(
        CMSHOPEE_CLIENT_SUBSCRIPTION_POLICY="off",
        CMSHOPEE_CLIENT_SUBSCRIPTION_POLICY_UPDATED_AT="2026-07-28T00:00:00+08:00",
    )
    def test_policy_transition_writes_audit_and_immediately_overrides_environment(self):
        policy = set_client_subscription_policy(
            mode=ClientSubscriptionPolicy.Mode.OBSERVE,
            reason="新版客户端灰度检测",
            actor=self.operator,
        )

        self.assertEqual(policy.pk, 1)
        self.assertEqual(policy.mode, ClientSubscriptionPolicy.Mode.OBSERVE)
        self.assertEqual(
            get_published_client_subscription_policy(),
            {
                "policy_version": 1,
                "subscription_check_enabled": True,
                "subscription_enforcement_enabled": False,
                "updated_at": policy.updated_at.isoformat(),
            },
        )
        first_audit = ClientSubscriptionPolicyAudit.objects.get(policy=policy)
        self.assertEqual(first_audit.previous_mode, "")
        self.assertEqual(first_audit.mode, ClientSubscriptionPolicy.Mode.OBSERVE)
        self.assertEqual(first_audit.reason, "新版客户端灰度检测")
        self.assertEqual(first_audit.changed_by, self.operator)

        updated = set_client_subscription_policy(
            mode=ClientSubscriptionPolicy.Mode.ENFORCE,
            reason="灰度验收完成",
            actor=self.operator,
        )

        self.assertEqual(updated.pk, policy.pk)
        self.assertEqual(updated.mode, ClientSubscriptionPolicy.Mode.ENFORCE)
        self.assertEqual(ClientSubscriptionPolicy.objects.count(), 1)
        self.assertEqual(
            ClientSubscriptionPolicyAudit.objects.filter(policy=policy).count(),
            2,
        )
        latest_audit = ClientSubscriptionPolicyAudit.objects.latest("id")
        self.assertEqual(latest_audit.previous_mode, ClientSubscriptionPolicy.Mode.OBSERVE)
        self.assertEqual(latest_audit.mode, ClientSubscriptionPolicy.Mode.ENFORCE)

    def test_policy_transition_requires_reason_and_rejects_same_mode(self):
        with self.assertRaises(LicensingError) as missing_reason:
            set_client_subscription_policy(
                mode=ClientSubscriptionPolicy.Mode.OBSERVE,
                reason="",
                actor=self.operator,
            )
        self.assertEqual(missing_reason.exception.code, "reason_required")
        self.assertFalse(ClientSubscriptionPolicy.objects.exists())

        set_client_subscription_policy(
            mode=ClientSubscriptionPolicy.Mode.OFF,
            reason="创建默认策略",
            actor=self.operator,
        )
        with self.assertRaises(LicensingError) as unchanged:
            set_client_subscription_policy(
                mode=ClientSubscriptionPolicy.Mode.OFF,
                reason="不应产生伪审计",
                actor=self.operator,
            )
        self.assertEqual(unchanged.exception.code, "policy_unchanged")
        self.assertEqual(ClientSubscriptionPolicyAudit.objects.count(), 1)

    def test_database_singleton_rejects_a_second_primary_key(self):
        ClientSubscriptionPolicy.objects.create(mode=ClientSubscriptionPolicy.Mode.OFF)

        with self.assertRaises(IntegrityError):
            ClientSubscriptionPolicy.objects.bulk_create(
                [
                    ClientSubscriptionPolicy(
                        singleton_id=2,
                        mode=ClientSubscriptionPolicy.Mode.OBSERVE,
                    )
                ]
            )


class SoftwareEntitlementServiceTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="entitlement-user",
            email="entitlement@example.com",
            password="test-password",
        )
        self.operator = User.objects.create_user(
            username="entitlement-operator",
            email="operator@example.com",
            password="test-password",
            is_staff=True,
        )
        self.plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="月度套餐",
            duration_days=30,
            price=Decimal("19.90"),
            device_limit=1,
            grace_days=3,
        )

    def create_device(self, suffix):
        device_id = f"v1:entitlement-device-{suffix}"
        public_key = f"entitlement-public-key-{suffix}"
        return ClientDevice.objects.create(
            user=self.user,
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            device_id_version="v1",
            device_fingerprint=ClientDevice.fingerprint_device_id("v1", device_id),
            public_key_fingerprint=ClientDevice.fingerprint_public_key(public_key),
            platform=ClientDevice.Platform.WINDOWS,
            client_version="0.1.0",
        )

    def grant_entitlement(self, **kwargs):
        return grant_software_entitlement(
            user=self.user,
            plan=self.plan,
            reason="客服人工授予",
            actor=self.operator,
            **kwargs,
        )

    def test_grant_snapshots_plan_creates_fixed_seats_and_audit_event(self):
        entitlement = self.grant_entitlement()
        self.plan.name = "已调整套餐"
        self.plan.price = Decimal("29.90")
        self.plan.device_limit = 3
        self.plan.save()
        entitlement.refresh_from_db()

        self.assertEqual(entitlement.plan_name, "月度套餐")
        self.assertEqual(entitlement.plan_price, Decimal("19.90"))
        self.assertEqual(entitlement.plan_device_limit, 1)
        self.assertEqual(LicenseSeat.objects.filter(entitlement=entitlement).count(), 1)
        event = LicenseEvent.objects.get(entitlement=entitlement)
        self.assertEqual(event.action, LicenseEvent.Action.GRANTED)
        self.assertEqual(event.reason, "客服人工授予")
        self.assertEqual(event.actor, self.operator)

    def test_renew_extends_from_later_of_now_or_existing_expiry(self):
        now = timezone.now()
        entitlement = self.grant_entitlement(starts_at=now - timedelta(days=10))
        previous_expiry = entitlement.expires_at

        renewed = renew_software_entitlement(
            entitlement=entitlement,
            reason="用户续订",
            actor=self.operator,
            now=now,
        )

        self.assertEqual(renewed.expires_at, previous_expiry + timedelta(days=30))
        self.assertEqual(renewed.grace_expires_at, renewed.expires_at + timedelta(days=3))
        self.assertEqual(renewed.status, SoftwareEntitlement.Status.ACTIVE)
        self.assertTrue(
            LicenseEvent.objects.filter(
                entitlement=entitlement,
                action=LicenseEvent.Action.RENEWED,
                reason="用户续订",
            ).exists()
        )

    def test_assign_release_and_revoke_are_audited_and_never_exceed_seat_limit(self):
        entitlement = self.grant_entitlement()
        first_device = self.create_device("one")
        second_device = self.create_device("two")

        seat = assign_license_seat(
            entitlement=entitlement,
            device=first_device,
            reason="首次绑定",
            actor=self.operator,
        )
        repeated = assign_license_seat(
            entitlement=entitlement,
            device=first_device,
            reason="重复绑定",
            actor=self.operator,
        )
        self.assertEqual(seat.pk, repeated.pk)
        with self.assertRaisesRegex(LicensingError, "设备席位已用完"):
            assign_license_seat(
                entitlement=entitlement,
                device=second_device,
                reason="超额绑定",
                actor=self.operator,
            )

        released = release_license_seat(
            seat=seat,
            reason="客服解绑",
            actor=self.operator,
        )
        assigned_again = assign_license_seat(
            entitlement=entitlement,
            device=second_device,
            reason="重新绑定",
            actor=self.operator,
        )
        self.assertIsNone(released.device_id)
        self.assertEqual(assigned_again.device_id, second_device.id)
        self.assertEqual(
            LicenseEvent.objects.filter(
                entitlement=entitlement,
                action=LicenseEvent.Action.SEAT_ASSIGNED,
            ).count(),
            2,
        )
        self.assertTrue(
            LicenseEvent.objects.filter(
                entitlement=entitlement,
                action=LicenseEvent.Action.SEAT_RELEASED,
            ).exists()
        )

        revoked = revoke_software_entitlement(
            entitlement=entitlement,
            reason="退款撤销",
            actor=self.operator,
        )
        self.assertEqual(revoked.status, SoftwareEntitlement.Status.REVOKED)
        with self.assertRaisesRegex(LicensingError, "软件权益当前不可用"):
            assign_license_seat(
                entitlement=entitlement,
                device=first_device,
                reason="撤销后绑定",
                actor=self.operator,
            )

    def test_manual_operations_require_reason(self):
        with self.assertRaisesRegex(LicensingError, "必须填写操作原因"):
            grant_software_entitlement(user=self.user, plan=self.plan, reason="")


class SoftwareOrderServiceTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="software-order-user",
            email="software-order@example.com",
            password="test-password",
        )
        self.plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="月度订阅",
            duration_days=30,
            price=Decimal("19.90"),
            device_limit=1,
            grace_days=3,
        )

    def create_order(self):
        return create_software_order(
            user=self.user,
            plan=self.plan,
            pay_method=SoftwareOrder.PayMethod.WEIXIN,
            payment_order_func=lambda _order: PaymentOrderCode(
                code_url="weixin://software-order-test",
                expires_at=timezone.now() + timedelta(minutes=10),
            ),
        )

    def create_alternate_plan(self):
        return SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="其他月度订阅",
            duration_days=30,
            price=Decimal("29.90"),
            device_limit=1,
            grace_days=2,
        )

    @staticmethod
    def payment_for(order, *, amount=None, transaction_id="wx-software-001"):
        return PaymentReceipt(
            order_no=order.order_no,
            pay_method=SoftwareOrder.PayMethod.WEIXIN,
            amount=amount or order.amount_money,
            transaction_id=transaction_id,
            paid_at=timezone.now(),
        )

    def test_order_snapshots_payment_once_and_never_credits_points(self):
        order = self.create_order()
        self.plan.name = "已修改套餐"
        self.plan.price = Decimal("29.90")
        self.plan.save()

        first = apply_software_payment(self.payment_for(order))
        second = apply_software_payment(self.payment_for(order))
        order.refresh_from_db()

        self.assertTrue(first.applied)
        self.assertFalse(second.applied)
        self.assertEqual(order.status, SoftwareOrder.Status.PAID)
        self.assertEqual(order.plan_name, "月度订阅")
        self.assertEqual(order.amount_money, Decimal("19.90"))
        self.assertEqual(order.entitlement.plan_name, "月度订阅")
        self.assertEqual(order.fulfillment_event.action, LicenseEvent.Action.ORDER_FULFILLED)
        self.assertEqual(
            PointsLedger.objects.filter(user=self.user).count(),
            0,
        )

    def test_duplicate_callback_with_changed_transaction_or_amount_is_rejected(self):
        order = self.create_order()
        with self.assertRaises(SoftwareOrderAmountMismatchError):
            apply_software_payment(self.payment_for(order, amount=Decimal("19.89")))
        self.assertEqual(SoftwareEntitlement.objects.count(), 0)

        apply_software_payment(self.payment_for(order))
        with self.assertRaises(SoftwareOrderTransactionMismatchError):
            apply_software_payment(
                self.payment_for(order, transaction_id="wx-software-other")
            )

    def test_second_paid_order_renews_same_plan_entitlement_once(self):
        first_order = self.create_order()
        first = apply_software_payment(self.payment_for(first_order))
        first_expiry = first.entitlement.expires_at

        second_order = self.create_order()
        second = apply_software_payment(
            self.payment_for(second_order, transaction_id="wx-software-002")
        )

        self.assertEqual(second.entitlement.pk, first.entitlement.pk)
        self.assertEqual(second.entitlement.expires_at, first_expiry + timedelta(days=30))
        self.assertEqual(
            SoftwareEntitlement.objects.filter(user=self.user).count(),
            1,
        )

    def test_expired_other_plan_allows_order_and_payment_grants_new_entitlement(self):
        old_plan = self.create_alternate_plan()
        old_entitlement = grant_software_entitlement(
            user=self.user,
            plan=old_plan,
            reason="准备过期套餐测试",
            starts_at=timezone.now() - timedelta(days=40),
        )

        order = self.create_order()
        old_entitlement.refresh_from_db()
        self.assertEqual(old_entitlement.status, SoftwareEntitlement.Status.ACTIVE)

        result = apply_software_payment(
            self.payment_for(order, transaction_id="wx-expired-plan-change")
        )
        old_entitlement.refresh_from_db()

        self.assertEqual(old_entitlement.status, SoftwareEntitlement.Status.EXPIRED)
        self.assertNotEqual(result.entitlement.pk, old_entitlement.pk)
        self.assertEqual(result.entitlement.source_plan_id, self.plan.id)
        self.assertEqual(result.entitlement.status, SoftwareEntitlement.Status.ACTIVE)

    def test_expired_same_plan_creates_new_entitlement_instead_of_renewing_old_one(self):
        old_entitlement = grant_software_entitlement(
            user=self.user,
            plan=self.plan,
            reason="准备同套餐过期测试",
            starts_at=timezone.now() - timedelta(days=40),
        )

        order = self.create_order()
        result = apply_software_payment(
            self.payment_for(order, transaction_id="wx-expired-same-plan")
        )
        old_entitlement.refresh_from_db()

        self.assertEqual(old_entitlement.status, SoftwareEntitlement.Status.EXPIRED)
        self.assertNotEqual(result.entitlement.pk, old_entitlement.pk)
        self.assertEqual(
            SoftwareEntitlement.objects.filter(user=self.user).count(),
            2,
        )

    def test_other_plan_within_grace_period_still_blocks_order_creation(self):
        old_plan = self.create_alternate_plan()
        grant_software_entitlement(
            user=self.user,
            plan=old_plan,
            reason="准备宽限期套餐测试",
            starts_at=timezone.now() - timedelta(days=31),
        )

        with self.assertRaises(SoftwareOrderError) as context:
            self.create_order()

        self.assertEqual(context.exception.code, "plan_change_not_supported")
        self.assertEqual(SoftwareOrder.objects.filter(user=self.user).count(), 0)

    def test_other_active_plan_created_after_order_still_blocks_payment(self):
        order = self.create_order()
        old_plan = self.create_alternate_plan()
        active_entitlement = grant_software_entitlement(
            user=self.user,
            plan=old_plan,
            reason="模拟下单后套餐变化",
        )

        with self.assertRaises(SoftwareOrderError) as context:
            apply_software_payment(
                self.payment_for(order, transaction_id="wx-late-plan-change")
            )

        self.assertEqual(context.exception.code, "plan_change_not_supported")
        order.refresh_from_db()
        active_entitlement.refresh_from_db()
        self.assertEqual(order.status, SoftwareOrder.Status.PENDING)
        self.assertEqual(active_entitlement.status, SoftwareEntitlement.Status.ACTIVE)

    def test_invalid_payment_does_not_expire_elapsed_entitlement(self):
        old_plan = self.create_alternate_plan()
        old_entitlement = grant_software_entitlement(
            user=self.user,
            plan=old_plan,
            reason="准备支付校验失败测试",
            starts_at=timezone.now() - timedelta(days=40),
        )
        order = self.create_order()

        with self.assertRaises(SoftwareOrderAmountMismatchError):
            apply_software_payment(
                self.payment_for(
                    order,
                    amount=order.amount_money - Decimal("0.01"),
                    transaction_id="wx-invalid-expired-plan",
                )
            )

        old_entitlement.refresh_from_db()
        self.assertEqual(old_entitlement.status, SoftwareEntitlement.Status.ACTIVE)


class SoftwareOrderConcurrencyTests(TransactionTestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="software-order-concurrency",
            email="software-order-concurrency@example.com",
            password="test-password",
        )
        self.plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="并发订阅套餐",
            duration_days=30,
            price=Decimal("19.90"),
            device_limit=1,
        )
        self.order = create_software_order(
            user=self.user,
            plan=self.plan,
            pay_method=SoftwareOrder.PayMethod.WEIXIN,
            payment_order_func=lambda _order: PaymentOrderCode(
                code_url="weixin://software-order-concurrency",
                expires_at=timezone.now() + timedelta(minutes=10),
            ),
        )

    def test_concurrent_same_order_callback_fulfills_once(self):
        def apply_callback():
            close_old_connections()
            try:
                order = SoftwareOrder.objects.get(pk=self.order.pk)
                result = apply_software_payment(
                    PaymentReceipt(
                        order_no=order.order_no,
                        pay_method=SoftwareOrder.PayMethod.WEIXIN,
                        amount=order.amount_money,
                        transaction_id="wx-software-concurrency-001",
                        paid_at=timezone.now(),
                    )
                )
                return result.applied
            finally:
                close_old_connections()

        with ThreadPoolExecutor(max_workers=2) as executor:
            outcomes = list(executor.map(lambda _index: apply_callback(), range(2)))

        self.order.refresh_from_db()
        self.assertEqual(outcomes.count(True), 1)
        self.assertEqual(self.order.status, SoftwareOrder.Status.PAID)
        self.assertEqual(
            LicenseEvent.objects.filter(
                action=LicenseEvent.Action.ORDER_FULFILLED,
                metadata__software_order_no=self.order.order_no,
            ).count(),
            1,
        )


class SoftwareEntitlementAdminTests(TestCase):
    def setUp(self):
        self.operator = User.objects.create_user(
            username="licensing-admin",
            email="licensing-admin@example.com",
            password="test-password",
            is_staff=True,
            is_superuser=True,
        )
        self.user = User.objects.create_user(
            username="licensing-target",
            email="licensing-target@example.com",
            password="test-password",
        )
        self.plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="后台套餐",
            duration_days=30,
            price=Decimal("9.90"),
            device_limit=2,
        )
        self.grant_url = reverse("admin:licensing_softwareentitlement_grant")

    def test_admin_grant_requires_staff_and_reason_then_writes_event(self):
        anonymous = self.client.get(self.grant_url)
        self.assertEqual(anonymous.status_code, 302)

        self.client.force_login(self.operator)
        missing_reason = self.client.post(
            self.grant_url,
            {"user": self.user.pk, "plan": self.plan.pk, "reason": ""},
        )
        self.assertEqual(missing_reason.status_code, 200)
        self.assertFalse(SoftwareEntitlement.objects.exists())

        response = self.client.post(
            self.grant_url,
            {"user": self.user.pk, "plan": self.plan.pk, "reason": "后台补偿"},
        )
        self.assertEqual(response.status_code, 302)
        entitlement = SoftwareEntitlement.objects.get(user=self.user)
        self.assertTrue(
            LicenseEvent.objects.filter(
                entitlement=entitlement,
                action=LicenseEvent.Action.GRANTED,
                reason="后台补偿",
                actor=self.operator,
            ).exists()
        )

    def test_admin_index_only_exposes_membership_models(self):
        hidden_models = (
            ClientDevice,
            DeviceSession,
            DeviceBindingAudit,
            LicenseSeat,
            LegacyMigrationGrant,
            MigrationRequest,
            DeviceCredential,
            SoftwareOrder,
            LicenseEvent,
        )
        visible_models = (
            SoftwarePlan,
            SoftwareEntitlement,
            ClientSubscriptionPolicy,
        )
        request = SimpleNamespace(user=self.operator)

        for model in hidden_models:
            self.assertEqual(admin.site._registry[model].get_model_perms(request), {})
        for model in visible_models:
            self.assertTrue(admin.site._registry[model].get_model_perms(request))

        self.assertTrue(LicenseSeat.objects.model._meta.db_table)
        self.assertTrue(DeviceCredential.objects.model._meta.db_table)

        self.client.force_login(self.operator)
        self.assertEqual(
            self.client.get(reverse("admin:licensing_softwareorder_changelist")).status_code,
            200,
        )
        self.assertEqual(SoftwarePlan._meta.verbose_name, "会员套餐")
        self.assertEqual(SoftwareEntitlement._meta.verbose_name, "用户会员")
        self.assertNotIn(ClientSubscriptionPolicyAudit, admin.site._registry)

    def test_client_subscription_policy_admin_requires_superuser_and_writes_audit(self):
        staff = User.objects.create_user(
            username="licensing-policy-staff",
            email="licensing-policy-staff@example.com",
            password="test-password",
            is_staff=True,
        )
        add_url = reverse("admin:licensing_clientsubscriptionpolicy_add")

        self.client.force_login(staff)
        self.assertEqual(self.client.get(add_url).status_code, 403)

        self.client.force_login(self.operator)
        audit_management = {
            "audits-TOTAL_FORMS": "0",
            "audits-INITIAL_FORMS": "0",
            "audits-MIN_NUM_FORMS": "0",
            "audits-MAX_NUM_FORMS": "0",
        }
        missing_reason = self.client.post(
            add_url,
            {
                "mode": ClientSubscriptionPolicy.Mode.OBSERVE,
                "reason": "",
                **audit_management,
            },
        )
        self.assertEqual(missing_reason.status_code, 200)
        self.assertFalse(ClientSubscriptionPolicy.objects.exists())

        created = self.client.post(
            add_url,
            {
                "mode": ClientSubscriptionPolicy.Mode.OBSERVE,
                "reason": "客户端灰度",
                **audit_management,
            },
        )
        self.assertEqual(created.status_code, 302)
        policy = ClientSubscriptionPolicy.objects.get(pk=1)
        self.assertEqual(policy.mode, ClientSubscriptionPolicy.Mode.OBSERVE)
        self.assertTrue(
            ClientSubscriptionPolicyAudit.objects.filter(
                policy=policy,
                reason="客户端灰度",
                changed_by=self.operator,
            ).exists()
        )

        self.client.force_login(staff)
        change_url = reverse(
            "admin:licensing_clientsubscriptionpolicy_change",
            args=(policy.pk,),
        )
        self.assertEqual(self.client.get(change_url).status_code, 403)

        self.client.force_login(self.operator)
        same_mode = self.client.post(
            change_url,
            {
                "mode": ClientSubscriptionPolicy.Mode.OBSERVE,
                "reason": "不应写入",
                "audits-TOTAL_FORMS": "1",
                "audits-INITIAL_FORMS": "1",
                "audits-MIN_NUM_FORMS": "0",
                "audits-MAX_NUM_FORMS": "0",
                "audits-0-id": ClientSubscriptionPolicyAudit.objects.get(policy=policy).pk,
                "audits-0-policy": policy.pk,
            },
        )
        self.assertEqual(same_mode.status_code, 200)
        self.assertEqual(ClientSubscriptionPolicyAudit.objects.count(), 1)
        self.assertEqual(self.client.get(add_url).status_code, 403)


class LicenseSeatConcurrencyTests(TransactionTestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="seat-concurrency-user",
            email="seat-concurrency@example.com",
            password="test-password",
        )
        self.plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="单席位套餐",
            duration_days=30,
            price=Decimal("9.90"),
            device_limit=1,
        )
        self.entitlement = grant_software_entitlement(
            user=self.user,
            plan=self.plan,
            reason="并发测试授予",
        )
        self.first_device = self.create_device("first")
        self.second_device = self.create_device("second")

    def create_device(self, suffix):
        return ClientDevice.objects.create(
            user=self.user,
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            device_id_version="v1",
            device_fingerprint=ClientDevice.fingerprint_device_id("v1", f"concurrent-{suffix}"),
            public_key_fingerprint=ClientDevice.fingerprint_public_key(f"key-{suffix}"),
            platform=ClientDevice.Platform.WINDOWS,
            client_version="0.1.0",
        )

    def test_concurrent_assignments_do_not_exceed_fixed_seat_limit(self):
        def assign(device_id):
            close_old_connections()
            try:
                entitlement = SoftwareEntitlement.objects.get(pk=self.entitlement.pk)
                device = ClientDevice.objects.get(pk=device_id)
                seat = assign_license_seat(
                    entitlement=entitlement,
                    device=device,
                    reason="并发绑定",
                )
                return ("assigned", seat.device_id)
            except LicensingError as exc:
                return (exc.code, None)
            finally:
                close_old_connections()

        with ThreadPoolExecutor(max_workers=2) as executor:
            outcomes = list(
                executor.map(assign, (self.first_device.pk, self.second_device.pk))
            )

        self.assertEqual(sum(outcome[0] == "assigned" for outcome in outcomes), 1)
        self.assertEqual(sum(outcome[0] == "seat_limit_reached" for outcome in outcomes), 1)
        seat = LicenseSeat.objects.get(entitlement=self.entitlement)
        self.assertIn(seat.device_id, {self.first_device.pk, self.second_device.pk})


class LegacyMigrationFlowTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="migration-user",
            email="migration@example.com",
            password="test-password",
        )
        self.other_user = User.objects.create_user(
            username="migration-other",
            email="migration-other@example.com",
            password="test-password",
        )
        self.plan = SoftwarePlan.objects.create(
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            name="存量迁移套餐",
            duration_days=30,
            price=Decimal("1.00"),
            device_limit=1,
        )
        self.api_key, self.raw_api_key = ApiKey.create_for_user(self.user, name="legacy")
        self.device, self.device_session_token = self.register_device_session()
        self.client = APIClient()

    def register_device_session(self):
        result = register_device(
            user=self.user,
            api_key=self.api_key,
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            device_id_version="v1",
            device_id="migration-device-id",
            public_key="migration-device-public-key",
            platform=ClientDevice.Platform.WINDOWS,
            client_version="0.2.0",
        )
        return result.device, result.session_token

    def grant_migration(self):
        return create_legacy_migration_grant(
            user=self.user,
            plan=self.plan,
            reason="历史付费用户迁移",
            eligibility_snapshot={"legacy_customer_id": "legacy-001"},
        )

    def api_headers(self):
        return {
            "HTTP_AUTHORIZATION": f"Bearer {self.raw_api_key}",
            "HTTP_X_DEVICE_SESSION": self.device_session_token,
        }

    def issue_credential(self):
        grant = self.grant_migration()
        migration_request, raw_token = create_migration_request(
            user=self.user,
            device=self.device,
        )
        _request, credential, created = confirm_migration_request(
            request_id=migration_request.request_id,
            user=self.user,
        )
        self.assertTrue(created)
        return grant.entitlement, credential, raw_token

    def test_request_confirm_poll_and_revoke_flow_keeps_credential_hashed(self):
        self.grant_migration()
        create_response = self.client.post(
            reverse("api-client-migration-request-create"),
            {},
            format="json",
            **self.api_headers(),
        )
        self.assertEqual(create_response.status_code, 201)
        raw_credential = create_response.data["device_credential_token"]
        self.assertTrue(create_response.data["confirmation_url"].endswith(create_response.data["request_id"]))
        migration_request = MigrationRequest.objects.get(
            request_id=create_response.data["request_id"]
        )
        self.assertNotEqual(migration_request.credential_token_hash, raw_credential)

        pending = self.client.get(
            reverse(
                "api-client-migration-request-detail",
                args=(migration_request.request_id,),
            ),
            **self.api_headers(),
        )
        self.assertEqual(pending.status_code, 200)
        self.assertEqual(pending.data["status"], MigrationRequest.Status.PENDING)
        self.assertNotIn("device_credential_token", pending.data)

        self.client.force_login(self.user)
        confirm_url = reverse("portal-migration-confirm", args=(migration_request.request_id,))
        self.assertEqual(self.client.get(confirm_url).status_code, 200)
        self.assertEqual(self.client.post(confirm_url).status_code, 302)
        self.assertEqual(self.client.post(confirm_url).status_code, 302)

        credential = DeviceCredential.objects.get(migration_request=migration_request)
        self.assertNotEqual(credential.token_hash, raw_credential)
        self.assertEqual(credential.token_hash, MigrationRequest.hash_credential_token(raw_credential))
        self.assertEqual(credential.seat.device_id, self.device.id)
        self.assertEqual(DeviceCredential.objects.count(), 1)
        self.assertEqual(
            LicenseEvent.objects.filter(
                action=LicenseEvent.Action.CREDENTIAL_ISSUED,
            ).count(),
            1,
        )

        confirmed = self.client.get(
            reverse(
                "api-client-migration-request-detail",
                args=(migration_request.request_id,),
            ),
            **self.api_headers(),
        )
        self.assertEqual(confirmed.status_code, 200)
        self.assertEqual(confirmed.data["status"], MigrationRequest.Status.CONFIRMED)

        revoke = self.client.post(
            reverse("portal-device-credential-revoke", args=(credential.pk,))
        )
        self.assertEqual(revoke.status_code, 302)
        credential.refresh_from_db()
        credential.seat.refresh_from_db()
        self.assertIsNotNone(credential.revoked_at)
        self.assertIsNone(credential.seat.device_id)
        self.assertTrue(
            LicenseEvent.objects.filter(
                action=LicenseEvent.Action.CREDENTIAL_REVOKED,
                reason="用户自助解绑设备",
            ).exists()
        )

    def test_request_requires_eligible_current_device_and_web_confirmation_same_user(self):
        no_grant = self.client.post(
            reverse("api-client-migration-request-create"),
            {},
            format="json",
            **self.api_headers(),
        )
        self.assertEqual(no_grant.status_code, 403)
        self.assertEqual(no_grant.data["error"]["code"], "migration_not_eligible")

        self.grant_migration()
        missing_device_session = self.client.post(
            reverse("api-client-migration-request-create"),
            {},
            format="json",
            HTTP_AUTHORIZATION=f"Bearer {self.raw_api_key}",
        )
        self.assertEqual(missing_device_session.status_code, 401)
        self.assertEqual(missing_device_session.data["error"]["code"], "device_session_required")

        created = self.client.post(
            reverse("api-client-migration-request-create"),
            {},
            format="json",
            **self.api_headers(),
        )
        migration_request = MigrationRequest.objects.get(request_id=created.data["request_id"])
        self.client.force_login(self.other_user)
        confirmation = self.client.get(
            reverse("portal-migration-confirm", args=(migration_request.request_id,))
        )
        self.assertEqual(confirmation.status_code, 403)
        self.assertFalse(DeviceCredential.objects.exists())

    def test_expired_request_and_duplicate_migration_grant_are_rejected(self):
        self.grant_migration()
        with self.assertRaisesRegex(LicensingError, "已有此产品的迁移资格"):
            self.grant_migration()
        migration_request, _raw_token = create_migration_request(
            user=self.user,
            device=self.device,
            now=timezone.now() - timedelta(minutes=20),
        )
        with self.assertRaisesRegex(LicensingError, "迁移请求已过期"):
            confirm_migration_request(
                request_id=migration_request.request_id,
                user=self.user,
                now=timezone.now(),
            )
        self.assertFalse(DeviceCredential.objects.exists())

    def test_authorization_decision_handles_missing_mismatched_and_expired_credentials(self):
        self.grant_migration()
        migration_request, raw_token = create_migration_request(
            user=self.user,
            device=self.device,
        )
        _request, credential, created = confirm_migration_request(
            request_id=migration_request.request_id,
            user=self.user,
        )
        self.assertTrue(created)
        self.assertEqual(
            evaluate_device_authorization(
                user=self.user,
                product_code="cmshopee",
            ).code,
            "device_not_bound",
        )
        self.assertEqual(
            evaluate_device_authorization(
                user=self.user,
                product_code="cmshopee",
                device=self.device,
            ).code,
            "license_required",
        )
        allowed = evaluate_device_authorization(
            user=self.user,
            product_code="cmshopee",
            device=self.device,
            raw_credential_token=raw_token,
        )
        self.assertTrue(allowed.allowed)
        self.assertEqual(allowed.credential_id, credential.id)

        DeviceCredential.objects.filter(pk=credential.pk).update(
            expires_at=timezone.now() - timedelta(seconds=1)
        )
        expired = evaluate_device_authorization(
            user=self.user,
            product_code="cmshopee",
            device=self.device,
            raw_credential_token=raw_token,
        )
        self.assertTrue(expired.would_reject)
        self.assertEqual(expired.code, "license_expired")

    def test_renewal_extends_active_credential_and_keeps_authorization_valid(self):
        entitlement, credential, raw_token = self.issue_credential()
        previous_credential_expiry = credential.expires_at

        renewed = renew_software_entitlement(
            entitlement=entitlement,
            reason="用户续订",
            now=timezone.now(),
        )

        credential.refresh_from_db()
        self.assertGreater(credential.expires_at, previous_credential_expiry)
        self.assertEqual(credential.expires_at, renewed.grace_expires_at)
        decision = evaluate_device_authorization(
            user=self.user,
            product_code=ClientDevice.ProductCode.CMSHOPEE,
            device=self.device,
            raw_credential_token=raw_token,
        )
        self.assertTrue(decision.allowed)

    def test_revoking_entitlement_revokes_credentials_and_releases_seats(self):
        entitlement, credential, _raw_token = self.issue_credential()

        revoke_software_entitlement(
            entitlement=entitlement,
            reason="运营撤销套餐权益",
        )

        credential.refresh_from_db()
        credential.seat.refresh_from_db()
        self.assertIsNotNone(credential.revoked_at)
        self.assertEqual(credential.revoke_reason, "运营撤销套餐权益")
        self.assertIsNone(credential.seat.device_id)
        self.assertTrue(
            LicenseEvent.objects.filter(
                entitlement=entitlement,
                action=LicenseEvent.Action.CREDENTIAL_REVOKED,
                reason="运营撤销套餐权益",
            ).exists()
        )
        self.assertTrue(
            LicenseEvent.objects.filter(
                entitlement=entitlement,
                action=LicenseEvent.Action.SEAT_RELEASED,
                reason="运营撤销套餐权益",
            ).exists()
        )