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

feat: add staged subscription rollout mode

Browse Source
This commit is contained in:
QiuSW
2026-07-22 15:13:28 +08:00
parent e3b90ccea5
commit 2604a59d67
17 changed files with 551 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/tests.py
+117 -13
View File
@@ -63,7 +63,12 @@ from apps.billing.services import RechargePayment
from apps.moderation.models import SensitiveWord
from apps.moderation.providers.keyword import reset_keyword_matcher_cache
from apps.portal.models import DownloadRelease
from apps.licensing.models import ClientDevice, SoftwareOrder, SoftwarePlan
from apps.licensing.models import (
ClientDevice,
SoftwareEntitlement,
SoftwareOrder,
SoftwarePlan,
)
from apps.licensing.services import (
create_software_order,
grant_software_entitlement,
@@ -1437,6 +1442,82 @@ class GenerateApiTests(TestCase):
starts_at=starts_at,
)
@override_settings(
CMSHOPEE_SUBSCRIPTION_MODE="open",
PUBLIC_BASE_URL="https://cm.example.test",
)
def test_cmshopee_open_mode_returns_compatible_access_without_writing_entitlement(self):
response = self.client.get(
"/api/v1/cmshopee/subscription/status",
**self.auth_header(),
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["product_code"], "cmshopee")
self.assertEqual(data["status"], "active")
self.assertTrue(data["allowed"])
self.assertIsNone(data["code"])
self.assertEqual(data["account"]["username"], self.user.username)
self.assertEqual(data["plan"]["code"], "development-open")
self.assertEqual(data["plan"]["display_name"], "开发测试长期会员")
self.assertEqual(data["expires_at"], data["plan"]["expires_at"])
self.assertIsNone(data["grace_expires_at"])
self.assertEqual(data["manage_url"], "https://cm.example.test/subscription")
self.assertEqual(data["access_source"], "open_mode")
self.assertEqual(data["entitlement_status"], "required")
self.assertTrue(data["notice_id"])
self.assertFalse(SoftwareEntitlement.objects.filter(user=self.user).exists())
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="shadow")
def test_cmshopee_shadow_mode_allows_missing_entitlement_and_reports_real_status(self):
response = self.client.get(
"/api/v1/cmshopee/subscription/status",
**self.auth_header(),
)
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["status"], "active")
self.assertTrue(data["allowed"])
self.assertEqual(data["access_source"], "shadow_fallback")
self.assertEqual(data["entitlement_status"], "required")
self.assertEqual(data["plan"]["code"], "shadow-fallback")
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="open")
def test_cmshopee_open_mode_submit_does_not_create_entitlement(self):
response = self.post_with_provider(
"/api/v1/cmshopee/generate/title",
{"prompt": "开发测试标题", "model": self.title_alias},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(self.provider.text_calls), 1)
self.assertFalse(SoftwareEntitlement.objects.filter(user=self.user).exists())
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="shadow")
def test_cmshopee_shadow_submit_logs_real_entitlement_status(self):
with self.assertLogs("cmhub.licensing.authorization", level="INFO") as captured:
response = self.post_with_provider(
"/api/v1/cmshopee/generate/title",
{"prompt": "影子模式标题", "model": self.title_alias},
)
self.assertEqual(response.status_code, 200)
events = [
getattr(record, "cmshopee_authorization", None)
for record in captured.records
if getattr(record, "cmshopee_authorization", None)
]
self.assertEqual(len(events), 1)
self.assertEqual(events[0]["subscription_mode"], "shadow")
self.assertEqual(events[0]["access_source"], "shadow_fallback")
self.assertEqual(events[0]["entitlement_status"], "required")
self.assertTrue(events[0]["would_reject"])
self.assertEqual(events[0]["would_reject_code"], "subscription_required")
self.assertTrue(events[0]["final_allowed"])
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="enforce")
def test_cmshopee_subscription_status_is_required_without_entitlement(self):
response = self.client.get(
"/api/v1/cmshopee/subscription/status",
@@ -1444,17 +1525,29 @@ class GenerateApiTests(TestCase):
)
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.json(),
{
"product_code": "cmshopee",
"status": "required",
"allowed": False,
"code": "subscription_required",
"plan": None,
},
data = response.json()
self.assertEqual(data["status"], "required")
self.assertFalse(data["allowed"])
self.assertEqual(data["code"], "subscription_required")
self.assertIsNone(data["plan"])
self.assertEqual(data["entitlement_status"], "required")
self.assertEqual(data["access_source"], "entitlement")
@override_settings(
CMSHOPEE_SUBSCRIPTION_MODE="",
CMSHOPEE_SUBSCRIPTION_ENFORCEMENT=True,
)
def test_cmshopee_legacy_enforcement_setting_remains_supported(self):
response = self.client.get(
"/api/v1/cmshopee/subscription/status",
**self.auth_header(),
)
self.assertEqual(response.status_code, 200)
self.assertFalse(response.json()["allowed"])
self.assertEqual(response.json()["code"], "subscription_required")
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="enforce")
def test_cmshopee_subscription_status_is_active_without_device_session(self):
entitlement = self.create_cmshopee_entitlement()
@@ -1468,8 +1561,18 @@ class GenerateApiTests(TestCase):
self.assertEqual(data["status"], "active")
self.assertTrue(data["allowed"])
self.assertIsNone(data["code"])
self.assertEqual(data["plan"]["display_name"], entitlement.plan_name)
self.assertEqual(data["plan"]["name"], entitlement.plan_name)
self.assertEqual(data["plan"]["code"], f"plan-{entitlement.source_plan_id}")
self.assertEqual(data["expires_at"], entitlement.expires_at.isoformat())
self.assertEqual(
data["grace_expires_at"],
entitlement.grace_expires_at.isoformat(),
)
self.assertEqual(data["access_source"], "entitlement")
self.assertEqual(data["entitlement_status"], "active")
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="enforce")
def test_cmshopee_subscription_status_is_expired_after_grace_period(self):
self.create_cmshopee_entitlement(
starts_at=timezone.now() - timedelta(days=40),
@@ -1485,8 +1588,9 @@ class GenerateApiTests(TestCase):
self.assertEqual(data["status"], "expired")
self.assertFalse(data["allowed"])
self.assertEqual(data["code"], "subscription_expired")
self.assertEqual(data["entitlement_status"], "expired")
@override_settings(CMSHOPEE_SUBSCRIPTION_ENFORCEMENT=True)
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="enforce")
def test_cmshopee_enforcement_rejects_without_subscription(self):
response = self.post_with_provider(
"/api/v1/cmshopee/generate/title",
@@ -1498,7 +1602,7 @@ class GenerateApiTests(TestCase):
self.assertEqual(self.provider.text_calls, [])
self.assert_generation_not_charged()
@override_settings(CMSHOPEE_SUBSCRIPTION_ENFORCEMENT=True)
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="enforce")
def test_cmshopee_account_subscription_allows_multiple_device_contexts(self):
self.create_cmshopee_entitlement()
@@ -1516,7 +1620,7 @@ class GenerateApiTests(TestCase):
self.assertEqual(second_response.status_code, 200)
self.assertEqual(len(self.provider.text_calls), 2)
@override_settings(CMSHOPEE_SUBSCRIPTION_ENFORCEMENT=True)
@override_settings(CMSHOPEE_SUBSCRIPTION_MODE="enforce")
def test_generic_generation_remains_available_without_subscription(self):
response = self.post_with_provider(
"/api/v1/generate/title",
apps/api/views.py
+15 -11
View File
@@ -1,7 +1,6 @@
import logging
from django.http import HttpResponse
from django.conf import settings
from django.utils import timezone
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt
@@ -79,7 +78,7 @@ from apps.licensing.services import (
create_migration_request,
expire_software_order,
query_and_apply_software_payment,
evaluate_account_authorization,
evaluate_subscription_access,
software_subscription_status,
record_device_heartbeat,
register_device,
@@ -391,7 +390,7 @@ class GenerateImageTaskDetailView(ExternalApiView):
return Response(task_detail_response(task), status=status.HTTP_200_OK)
class CmshopeeShadowAuthorizationMixin:
class CmshopeeSubscriptionAuthorizationMixin:
product_code = "cmshopee"
def optional_client_device(self, request):
@@ -403,7 +402,7 @@ class CmshopeeShadowAuthorizationMixin:
request.headers.get("X-Device-Session", "").strip()
)
device = None
decision = evaluate_account_authorization(
decision = evaluate_subscription_access(
user=request.user,
product_code=self.product_code,
)
@@ -412,9 +411,14 @@ class CmshopeeShadowAuthorizationMixin:
"product_code": self.product_code,
"user_id": request.user.id,
"client_device_id": getattr(device, "id", None),
"would_reject": decision.would_reject,
"would_reject_code": decision.code,
"enforcement": settings.CMSHOPEE_SUBSCRIPTION_ENFORCEMENT,
"would_reject": bool(decision.entitlement_code),
"would_reject_code": decision.entitlement_code,
"final_allowed": decision.allowed,
"final_code": decision.code,
"subscription_mode": decision.mode,
"access_source": decision.access_source,
"entitlement_status": decision.entitlement_status,
"enforcement": decision.mode == "enforce",
}
authorization_logger.info(
"%s %s",
@@ -422,7 +426,7 @@ class CmshopeeShadowAuthorizationMixin:
event,
extra={"cmshopee_authorization": event},
)
if settings.CMSHOPEE_SUBSCRIPTION_ENFORCEMENT and decision.would_reject:
if decision.would_reject:
message = (
"当前账号没有有效的软件订阅"
if decision.code == "subscription_required"
@@ -433,16 +437,16 @@ class CmshopeeShadowAuthorizationMixin:
return device
class CmshopeeGenerateTitleView(CmshopeeShadowAuthorizationMixin, GenerateTitleView):
class CmshopeeGenerateTitleView(CmshopeeSubscriptionAuthorizationMixin, GenerateTitleView):
pass
class CmshopeeAnalyzeImagesView(CmshopeeShadowAuthorizationMixin, AnalyzeImagesView):
class CmshopeeAnalyzeImagesView(CmshopeeSubscriptionAuthorizationMixin, AnalyzeImagesView):
pass
class CmshopeeGenerateImageTaskSubmitView(
CmshopeeShadowAuthorizationMixin,
CmshopeeSubscriptionAuthorizationMixin,
GenerateImageTaskSubmitView,
):
pass
apps/licensing/admin.py
+18 -14
View File
@@ -28,6 +28,13 @@ from .services import (
)
class HiddenFromAdminIndexMixin:
"""Keep registered models addressable without adding menu noise."""
def get_model_perms(self, request):
return {}
def _masked_fingerprint(value: str) -> str:
if not value:
return ""
@@ -35,7 +42,7 @@ def _masked_fingerprint(value: str) -> str:
@admin.register(ClientDevice)
class ClientDeviceAdmin(admin.ModelAdmin):
class ClientDeviceAdmin(HiddenFromAdminIndexMixin, admin.ModelAdmin):
list_display = (
"user",
"product_code",
@@ -81,7 +88,7 @@ class ClientDeviceAdmin(admin.ModelAdmin):
@admin.register(DeviceSession)
class DeviceSessionAdmin(admin.ModelAdmin):
class DeviceSessionAdmin(HiddenFromAdminIndexMixin, admin.ModelAdmin):
list_display = ("device", "expires_at", "last_used_at", "revoked_at", "created_at")
list_filter = ("revoked_at", "expires_at")
search_fields = ("device__user__username", "device__user__email")
@@ -106,7 +113,7 @@ class DeviceSessionAdmin(admin.ModelAdmin):
@admin.register(DeviceBindingAudit)
class DeviceBindingAuditAdmin(admin.ModelAdmin):
class DeviceBindingAuditAdmin(HiddenFromAdminIndexMixin, admin.ModelAdmin):
list_display = ("user", "device", "api_key", "action", "reason", "created_at")
list_filter = ("action", "created_at")
search_fields = ("user__username", "user__email", "api_key__key_prefix")
@@ -334,15 +341,12 @@ class ReadOnlyLicenseAdmin(admin.ModelAdmin):
return False
class HiddenLegacyLicenseAdmin(ReadOnlyLicenseAdmin):
"""Keep legacy records addressable without advertising them in admin."""
def get_model_perms(self, request):
return {}
class HiddenLicenseAdmin(HiddenFromAdminIndexMixin, ReadOnlyLicenseAdmin):
pass
@admin.register(LicenseSeat)
class LicenseSeatAdmin(HiddenLegacyLicenseAdmin):
class LicenseSeatAdmin(HiddenLicenseAdmin):
list_display = ("entitlement", "seat_number", "device", "bound_at", "released_at")
list_filter = ("entitlement__product_code", "bound_at", "released_at")
search_fields = (
@@ -354,7 +358,7 @@ class LicenseSeatAdmin(HiddenLegacyLicenseAdmin):
@admin.register(LicenseEvent)
class LicenseEventAdmin(ReadOnlyLicenseAdmin):
class LicenseEventAdmin(HiddenLicenseAdmin):
list_display = ("entitlement", "action", "seat", "device", "actor", "reason", "created_at")
list_filter = ("action", "entitlement__product_code", "created_at")
search_fields = (
@@ -370,7 +374,7 @@ class LegacyMigrationGrantForm(EntitlementGrantForm):
@admin.register(LegacyMigrationGrant)
class LegacyMigrationGrantAdmin(HiddenLegacyLicenseAdmin):
class LegacyMigrationGrantAdmin(HiddenLicenseAdmin):
list_display = ("user", "product_code", "entitlement", "status", "actor", "created_at")
list_filter = ("product_code", "status", "created_at")
search_fields = ("user__username", "user__email", "reason")
@@ -417,7 +421,7 @@ class LegacyMigrationGrantAdmin(HiddenLegacyLicenseAdmin):
@admin.register(MigrationRequest)
class MigrationRequestAdmin(HiddenLegacyLicenseAdmin):
class MigrationRequestAdmin(HiddenLicenseAdmin):
list_display = ("request_id", "user", "device", "migration_grant", "status", "expires_at", "confirmed_at")
list_filter = ("status", "device__product_code", "expires_at")
search_fields = ("=request_id", "user__username", "user__email")
@@ -425,7 +429,7 @@ class MigrationRequestAdmin(HiddenLegacyLicenseAdmin):
@admin.register(DeviceCredential)
class DeviceCredentialAdmin(HiddenLegacyLicenseAdmin):
class DeviceCredentialAdmin(HiddenLicenseAdmin):
list_display = ("token_prefix", "user", "product_code", "device", "entitlement", "expires_at", "revoked_at")
list_filter = ("product_code", "revoked_at", "expires_at")
search_fields = ("token_prefix", "user__username", "user__email")
@@ -433,7 +437,7 @@ class DeviceCredentialAdmin(HiddenLegacyLicenseAdmin):
@admin.register(SoftwareOrder)
class SoftwareOrderAdmin(ReadOnlyLicenseAdmin):
class SoftwareOrderAdmin(HiddenLicenseAdmin):
list_display = (
"created_at",
"order_no",
apps/licensing/migrations/0005_alter_softwareentitlement_options_and_more.py
+21
View File
@@ -0,0 +1,21 @@
# Generated by Django 5.2.15 on 2026-07-22 07:04
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('licensing', '0004_alter_licenseevent_action_softwareorder'),
]
operations = [
migrations.AlterModelOptions(
name='softwareentitlement',
options={'ordering': ('-expires_at', '-id'), 'verbose_name': '用户会员', 'verbose_name_plural': '用户会员'},
),
migrations.AlterModelOptions(
name='softwareplan',
options={'ordering': ('product_code', 'name', 'id'), 'verbose_name': '会员套餐', 'verbose_name_plural': '会员套餐'},
),
]
apps/licensing/models.py
+4 -4
View File
@@ -108,8 +108,8 @@ class SoftwarePlan(models.Model):
class Meta:
db_table = "software_plan"
verbose_name = "软件套餐"
verbose_name_plural = "软件套餐"
verbose_name = "会员套餐"
verbose_name_plural = "会员套餐"
ordering = ("product_code", "name", "id")
constraints = [
models.CheckConstraint(
@@ -178,8 +178,8 @@ class SoftwareEntitlement(models.Model):
class Meta:
db_table = "software_entitlement"
verbose_name = "软件权益"
verbose_name_plural = "软件权益"
verbose_name = "用户会员"
verbose_name_plural = "用户会员"
ordering = ("-expires_at", "-id")
constraints = [
models.CheckConstraint(
apps/licensing/services.py
+139 -17
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import secrets
from dataclasses import dataclass
from datetime import timedelta
from datetime import datetime, timedelta, timezone as datetime_timezone
from decimal import Decimal
from decimal import InvalidOperation
@@ -98,6 +98,22 @@ class SubscriptionAuthorizationDecision:
return not self.allowed
@dataclass(frozen=True)
class SubscriptionAccessDecision:
product_code: str
mode: str
allowed: bool
code: str
access_source: str
entitlement_status: str
entitlement_code: str
entitlement: SoftwareEntitlement | None = None
@property
def would_reject(self) -> bool:
return not self.allowed
@dataclass(frozen=True)
class DeviceRegistrationResult:
device: ClientDevice
@@ -970,27 +986,133 @@ def evaluate_account_authorization(*, user, product_code: str, now=None):
)
def software_subscription_status(*, user, product_code: str, now=None) -> dict:
decision = evaluate_account_authorization(
def _real_entitlement_status(decision, now) -> str:
if decision.allowed:
if decision.entitlement and decision.entitlement.expires_at <= now:
return "grace"
return "active"
if decision.code == "subscription_expired":
return "expired"
return "required"
def get_subscription_mode() -> str:
mode = str(getattr(settings, "CMSHOPEE_SUBSCRIPTION_MODE", "") or "").strip().lower()
if mode in {"open", "shadow", "enforce"}:
return mode
return (
"enforce"
if getattr(settings, "CMSHOPEE_SUBSCRIPTION_ENFORCEMENT", False)
else "open"
)
def evaluate_subscription_access(*, user, product_code: str, now=None, mode=None):
now = now or timezone.now()
mode = mode or get_subscription_mode()
real_decision = evaluate_account_authorization(
user=user,
product_code=product_code,
now=now,
)
entitlement = decision.entitlement
return {
"product_code": product_code,
"status": (
"active"
if decision.allowed
else ("expired" if decision.code == "subscription_expired" else "required")
),
"allowed": decision.allowed,
"code": decision.code or None,
"plan": {
entitlement_status = _real_entitlement_status(real_decision, now)
if mode == "open":
return SubscriptionAccessDecision(
product_code=product_code,
mode=mode,
allowed=True,
code="",
access_source="open_mode",
entitlement_status=entitlement_status,
entitlement_code=real_decision.code,
)
if mode == "shadow" and not real_decision.allowed:
return SubscriptionAccessDecision(
product_code=product_code,
mode=mode,
allowed=True,
code="",
access_source="shadow_fallback",
entitlement_status=entitlement_status,
entitlement_code=real_decision.code,
)
return SubscriptionAccessDecision(
product_code=product_code,
mode=mode,
allowed=real_decision.allowed,
code=real_decision.code,
access_source="entitlement",
entitlement_status=entitlement_status,
entitlement_code=real_decision.code,
entitlement=real_decision.entitlement,
)
def _subscription_manage_url() -> str:
base_url = str(getattr(settings, "PUBLIC_BASE_URL", "") or "").rstrip("/")
return f"{base_url}/subscription" if base_url else "/subscription"
def _effective_plan(access):
entitlement = access.entitlement
if entitlement is not None:
plan_code = (
f"plan-{entitlement.source_plan_id}"
if entitlement.source_plan_id
else f"entitlement-{entitlement.id}"
)
return {
"code": plan_code,
"display_name": entitlement.plan_name,
# T-630 compatibility fields; new clients use the top-level dates.
"name": entitlement.plan_name,
"expires_at": entitlement.expires_at.isoformat(),
"grace_expires_at": entitlement.grace_expires_at.isoformat(),
}
if entitlement is not None
else None,
}, entitlement.expires_at, entitlement.grace_expires_at
if access.access_source in {"open_mode", "shadow_fallback"}:
expires_at = datetime(2099, 12, 31, 23, 59, 59, tzinfo=datetime_timezone.utc)
plan_code = "development-open" if access.access_source == "open_mode" else "shadow-fallback"
display_name = "开发测试长期会员" if access.access_source == "open_mode" else "订阅过渡会员"
return {
"code": plan_code,
"display_name": display_name,
"name": display_name,
"expires_at": expires_at.isoformat(),
"grace_expires_at": None,
}, expires_at, None
return None, None, None
def software_subscription_status(*, user, product_code: str, now=None, mode=None) -> dict:
access = evaluate_subscription_access(
user=user,
product_code=product_code,
now=now,
mode=mode,
)
plan, expires_at, grace_expires_at = _effective_plan(access)
username = user.get_username()
return {
"product_code": product_code,
"status": (
access.entitlement_status
if access.access_source == "entitlement"
else "active"
),
"allowed": access.allowed,
"code": access.code or None,
"account": {
"display_name": user.get_full_name() or username,
"username": username,
},
"plan": plan,
"expires_at": expires_at.isoformat() if expires_at else None,
"grace_expires_at": grace_expires_at.isoformat() if grace_expires_at else None,
"manage_url": _subscription_manage_url(),
"notice_id": (
f"{product_code}-subscription-{access.mode}-{access.entitlement_status}-v1"
),
"access_source": access.access_source,
"entitlement_status": access.entitlement_status,
}
apps/licensing/tests.py
+142 -9
View File
@@ -2,10 +2,11 @@ from datetime import timedelta
from decimal import Decimal
from concurrent.futures import ThreadPoolExecutor
from types import SimpleNamespace
from unittest.mock import patch
from django.contrib import admin
from django.db import close_old_connections
from django.test import TestCase, TransactionTestCase, override_settings
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
@@ -29,6 +30,7 @@ from apps.licensing.services import (
LicensingError,
SoftwareOrderAmountMismatchError,
SoftwareOrderTransactionMismatchError,
SubscriptionAuthorizationDecision,
apply_software_payment,
assign_license_seat,
confirm_migration_request,
@@ -36,16 +38,139 @@ from apps.licensing.services import (
create_migration_request,
create_software_order,
evaluate_device_authorization,
evaluate_subscription_access,
get_subscription_mode,
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 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(
@@ -550,22 +675,22 @@ class SoftwareEntitlementAdminTests(TestCase):
).exists()
)
def test_admin_hides_legacy_license_models_but_keeps_current_models_visible(self):
def test_admin_index_only_exposes_membership_models(self):
hidden_models = (
ClientDevice,
DeviceSession,
DeviceBindingAudit,
LicenseSeat,
LegacyMigrationGrant,
MigrationRequest,
DeviceCredential,
)
visible_models = (
ClientDevice,
DeviceSession,
DeviceBindingAudit,
SoftwarePlan,
SoftwareEntitlement,
SoftwareOrder,
LicenseEvent,
)
visible_models = (
SoftwarePlan,
SoftwareEntitlement,
)
request = SimpleNamespace(user=self.operator)
for model in hidden_models:
@@ -576,6 +701,14 @@ class SoftwareEntitlementAdminTests(TestCase):
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):