chengma/cmhub
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add software subscription orders

Browse Source
This commit is contained in:
QiuSW
2026-07-21 11:52:49 +08:00
parent 341864dc70
commit bc9d1aa0f7
25 changed files with 1179 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/licensing/tests.py
+215
View File
@@ -8,6 +8,8 @@ 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,
@@ -18,14 +20,19 @@ from apps.licensing.models import (
LicenseSeat,
MigrationRequest,
SoftwareEntitlement,
SoftwareOrder,
SoftwarePlan,
)
from apps.licensing.services import (
LicensingError,
SoftwareOrderAmountMismatchError,
SoftwareOrderTransactionMismatchError,
apply_software_payment,
assign_license_seat,
confirm_migration_request,
create_legacy_migration_grant,
create_migration_request,
create_software_order,
evaluate_device_authorization,
grant_software_entitlement,
record_device_heartbeat,
@@ -345,6 +352,152 @@ class SoftwareEntitlementServiceTests(TestCase):
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),
),
)
@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,
)
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(
@@ -506,6 +659,19 @@ class LegacyMigrationFlowTests(TestCase):
"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(
@@ -673,3 +839,52 @@ class LegacyMigrationFlowTests(TestCase):
)
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()
)