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 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 ( 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, 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, ) 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") 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.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 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, ) 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, "用户会员") 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() )