fix: allow expired members to buy new plans

This commit is contained in:
QiuSW
2026-07-23 17:42:08 +08:00
parent e12659faa0
commit 976b55c374
6 changed files with 207 additions and 21 deletions
+112
View File
@@ -32,6 +32,7 @@ from apps.licensing.models import (
from apps.licensing.services import (
LicensingError,
SoftwareOrderAmountMismatchError,
SoftwareOrderError,
SoftwareOrderTransactionMismatchError,
SubscriptionAuthorizationDecision,
apply_software_payment,
@@ -675,6 +676,16 @@ class SoftwareOrderServiceTests(TestCase):
),
)
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(
@@ -736,6 +747,107 @@ class SoftwareOrderServiceTests(TestCase):
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):