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

feat(subscription): add membership access preflight

Browse Source
This commit is contained in:
chengma
2026-07-21 17:38:05 +08:00
parent ca46c34dfe
commit 9ee48dc2aa
14 changed files with 788 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_appconfig.py
+23
View File
@@ -480,6 +480,29 @@ class AppConfigTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_subscription_notice_id_persists_without_credentials(self):
with self.make_temp_dir() as temp_dir:
config_path = os.path.join(temp_dir, "config.json")
loaded = appconfig.load_config(config_path)
self.assertEqual("", appconfig.subscription_notice_id(loaded))
saved = appconfig.save_subscription_notice_id(
"subscription-notice-20260721",
path=config_path,
)
self.assertEqual(
"subscription-notice-20260721",
appconfig.subscription_notice_id(saved),
)
reloaded = appconfig.load_config(config_path)
self.assertEqual(
"subscription-notice-20260721",
appconfig.subscription_notice_id(reloaded),
)
self.assert_removed(temp_dir)
def test_cmhub_base_url_normalizes_to_gateway_root(self):
cases = {
"https://cmhub.example.com/": "https://cmhub.example.com",
tests/test_gui.py
+57
View File
@@ -22,6 +22,7 @@ from app import (
image_studio,
product_status,
prompts,
subscription,
update_check,
update_installer,
)
@@ -11154,6 +11155,62 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_main_window_displays_active_subscription_in_header(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
self.addCleanup(window.close)
status = subscription.SubscriptionStatus(
subscription.STATUS_ACTIVE,
account_name="主账号",
plan_name="专业版",
expires_at="2026-08-20T23:59:59+08:00",
)
window._apply_subscription_status(status, show_prompt=False)
self.assertIn("主账号 · 专业版 · 有效至2026-08-20", window._subscription_label.text())
self.assertTrue(
all(window.tabs.isTabEnabled(index) for index in range(window.tabs.count()))
)
def test_main_window_restricts_tabs_when_subscription_is_invalid(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
self.addCleanup(window.close)
status = subscription.SubscriptionStatus(subscription.STATUS_EXPIRED)
window._apply_subscription_status(status, show_prompt=False)
for index, title in enumerate(TAB_TITLES):
self.assertEqual(title == "设置", window.tabs.isTabEnabled(index))
self.assertEqual(TAB_TITLES.index("设置"), window.tabs.currentIndex())
def test_generate_and_suite_preflight_can_stop_new_submissions(self):
with self.make_temp_dir() as temp_dir:
config = self.make_config(temp_dir)
actions = []
def deny(action):
actions.append(action)
return False
generate_tab = GenerateTab(
config=config,
subscription_preflight_callback=deny,
)
self.addCleanup(generate_tab.close)
generate_tab.start_generate()
suite_tab = ProductSuiteTab(
config=config,
db_path=config["db_path"],
subscription_preflight_callback=deny,
)
self.addCleanup(suite_tab.close)
self.assertFalse(suite_tab.start_generation(None))
self.assertEqual(["开始 AI 生成", "生成商品套图"], actions)
def test_accounts_tab_launch_failure_restores_controls(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
tests/test_subscription.py
+156
View File
@@ -0,0 +1,156 @@
import os
import unittest
from _helpers import TempDirMixin
from app import ai, appconfig, subscription
class SubscriptionTests(TempDirMixin, unittest.TestCase):
def _config(self, temp_dir):
cmhub_path = os.path.join(temp_dir, "cmhub.json")
appconfig.save_cmhub_config({"api_key": "test-key"}, path=cmhub_path)
return {
"cmhub_config_path": cmhub_path,
"ai": {
"use_system_proxy": False,
"cmhub": {
"base_url": "https://cm.example.com",
"connect_timeout": 66,
},
},
}
def test_missing_key_requires_configuration(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
appconfig.save_cmhub_config({"api_key": ""}, path=config["cmhub_config_path"])
result = subscription.check_status(config)
self.assertEqual(subscription.STATUS_NOT_CONFIGURED, result.state)
self.assertFalse(result.allows_product_workflows)
def test_active_response_returns_safe_display_fields(self):
with self.make_temp_dir() as temp_dir:
calls = []
def request_json(*args, **kwargs):
calls.append((args, kwargs))
return {
"product_code": "cmshopee",
"account": {"display_name": "主账号"},
"plan": {"code": "pro", "display_name": "专业版"},
"status": "active",
"expires_at": "2026-08-20T23:59:59+08:00",
"grace_expires_at": None,
"manage_url": "https://cm.example.com/user/subscriptions/cmshopee",
"notice_id": "notice-1",
}
result = subscription.check_status(
self._config(temp_dir),
request_json=request_json,
)
self.assertEqual(subscription.STATUS_ACTIVE, result.state)
self.assertTrue(result.allows_product_workflows)
self.assertEqual("主账号", result.account_name)
self.assertEqual("专业版", result.plan_name)
self.assertEqual("2026-08-20", subscription.format_expiry(result.expires_at))
self.assertEqual(
"https://cm.example.com/user/subscriptions/cmshopee",
result.manage_url,
)
self.assertEqual("GET", calls[0][0][0])
self.assertNotIn("test-key", repr(result))
def test_grace_and_expired_states_are_distinct(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
def grace_request(*args, **kwargs):
return {
"product_code": "cmshopee",
"account": {"display_name": "主账号"},
"plan": {"display_name": "专业版"},
"status": "grace",
"expires_at": "2026-08-20T23:59:59+08:00",
"grace_expires_at": "2026-08-23T23:59:59+08:00",
}
grace = subscription.check_status(config, request_json=grace_request)
self.assertEqual(subscription.STATUS_GRACE, grace.state)
self.assertTrue(grace.allows_product_workflows)
def expired_request(*args, **kwargs):
return {
"product_code": "cmshopee",
"account": {"display_name": "主账号"},
"plan": {"display_name": "专业版"},
"status": "expired",
"expires_at": "2026-08-20T23:59:59+08:00",
}
expired = subscription.check_status(config, request_json=expired_request)
self.assertEqual(subscription.STATUS_EXPIRED, expired.state)
self.assertFalse(expired.allows_product_workflows)
def test_legacy_404_keeps_existing_workflows_available(self):
with self.make_temp_dir() as temp_dir:
def request_json(*args, **kwargs):
raise ai.CMHubError("not_found", "接口不存在", status=404)
result = subscription.check_status(
self._config(temp_dir),
request_json=request_json,
)
self.assertEqual(subscription.STATUS_LEGACY, result.state)
self.assertTrue(result.allows_product_workflows)
self.assertFalse(result.interface_available)
def test_auth_error_and_network_failure_have_different_states(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
def invalid_key(*args, **kwargs):
raise ai.CMHubError("unauthorized", "不应展示", status=401)
result = subscription.check_status(config, request_json=invalid_key)
self.assertEqual(subscription.STATUS_KEY_INVALID, result.state)
self.assertNotIn("不应展示", result.user_message)
def network_failure(*args, **kwargs):
raise ai.CMHubError("network_error", "不应展示")
result = subscription.check_status(config, request_json=network_failure)
self.assertEqual(subscription.STATUS_UNAVAILABLE, result.state)
self.assertNotIn("不应展示", result.user_message)
def test_invalid_product_or_external_manage_url_is_not_trusted(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
def wrong_product(*args, **kwargs):
return {"product_code": "another-product", "status": "active"}
result = subscription.check_status(config, request_json=wrong_product)
self.assertEqual(subscription.STATUS_UNAVAILABLE, result.state)
def external_manage_url(*args, **kwargs):
return {
"product_code": "cmshopee",
"account": {"display_name": "主账号"},
"plan": {"display_name": "专业版"},
"status": "required",
"manage_url": "https://other.example.com/account",
}
result = subscription.check_status(config, request_json=external_manage_url)
self.assertEqual(subscription.STATUS_REQUIRED, result.state)
self.assertEqual("", result.manage_url)
if __name__ == "__main__":
unittest.main()