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

feat: add admin client policy control

Browse Source
This commit is contained in:
QiuSW
2026-07-28 15:42:23 +08:00
parent 9544aaf4cc
commit 2171968780
17 changed files with 558 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/licensing/tests.py
+189 -1
View File
@@ -8,7 +8,7 @@ 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.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
@@ -17,6 +17,8 @@ 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,
@@ -44,6 +46,7 @@ from apps.licensing.services import (
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,
@@ -52,6 +55,7 @@ from apps.licensing.services import (
revoke_software_entitlement,
register_device,
software_subscription_status,
set_client_subscription_policy,
)
from apps.users.models import ApiKey, User, UserWallet
@@ -495,6 +499,114 @@ class DeviceRegistrationApiTests(TestCase):
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(
@@ -973,6 +1085,7 @@ class SoftwareEntitlementAdminTests(TestCase):
visible_models = (
SoftwarePlan,
SoftwareEntitlement,
ClientSubscriptionPolicy,
)
request = SimpleNamespace(user=self.operator)
@@ -991,6 +1104,81 @@ class SoftwareEntitlementAdminTests(TestCase):
)
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):