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

feat(subscription): add expiry recovery flow

Browse Source
This commit is contained in:
chengma
2026-07-23 16:03:08 +08:00
parent b05c370250
commit c02866e610
7 changed files with 276 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_gui.py
+183 -6
View File
@@ -327,10 +327,23 @@ class GuiTests(TempDirMixin, unittest.TestCase):
def make_fake_message_box(self, selected_label):
boxes = []
class FakeButton:
def __init__(self):
self.enabled = True
self.tooltip = ""
def setEnabled(self, enabled):
self.enabled = bool(enabled)
def setToolTip(self, tooltip):
self.tooltip = str(tooltip)
class FakeMessageBox:
AcceptRole = object()
ActionRole = object()
DestructiveRole = object()
RejectRole = object()
Warning = object()
def __init__(self, parent=None):
self.parent = parent
@@ -338,8 +351,12 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.text = ""
self.buttons = {}
self.default_button = None
self.icon = None
boxes.append(self)
def setIcon(self, icon):
self.icon = icon
def setWindowTitle(self, title):
self.title = title
@@ -347,7 +364,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.text = text
def addButton(self, label, role):
button = object()
button = FakeButton()
self.buttons[label] = button
return button
@@ -373,10 +390,23 @@ class GuiTests(TempDirMixin, unittest.TestCase):
boxes = []
labels = list(selected_labels)
class FakeButton:
def __init__(self):
self.enabled = True
self.tooltip = ""
def setEnabled(self, enabled):
self.enabled = bool(enabled)
def setToolTip(self, tooltip):
self.tooltip = str(tooltip)
class FakeMessageBox:
AcceptRole = object()
ActionRole = object()
DestructiveRole = object()
RejectRole = object()
Warning = object()
def __init__(self, parent=None):
self.parent = parent
@@ -385,8 +415,12 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.buttons = {}
self.default_button = None
self.selected_label = labels.pop(0)
self.icon = None
boxes.append(self)
def setIcon(self, icon):
self.icon = icon
def setWindowTitle(self, title):
self.title = title
@@ -394,7 +428,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.text = text
def addButton(self, label, role):
button = object()
button = FakeButton()
self.buttons[label] = button
return button
@@ -11181,6 +11215,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
all(window.tabs.isTabEnabled(index) for index in range(window.tabs.count()))
)
@mock.patch.object(main_window, "SUBSCRIPTION_ENFORCEMENT_ENABLED", False)
def test_main_window_membership_title_uses_grace_and_resets_for_invalid_status(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
@@ -11212,22 +11247,163 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertIn("当前不影响使用", window.statusBar().currentMessage())
def test_main_window_restricts_tabs_when_subscription_is_invalid(self):
self.assertTrue(main_window.SUBSCRIPTION_ENFORCEMENT_ENABLED)
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)
with mock.patch.object(
main_window,
"SUBSCRIPTION_ENFORCEMENT_ENABLED",
True,
):
window,
"_show_expired_subscription_notice_once",
) as notice:
window._apply_subscription_status(status)
notice.assert_called_once_with(status)
for index, title in enumerate(TAB_TITLES):
self.assertEqual(title == "设置", window.tabs.isTabEnabled(index))
self.assertEqual(TAB_TITLES.index("设置"), window.tabs.currentIndex())
self.assertFalse(window.ensure_subscription_for_new_submit("开始 AI 生成"))
def test_expired_subscription_notice_opens_member_center_once_per_transition(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
self.addCleanup(window.close)
fake_box, boxes = self.make_sequence_message_box(
["前往会员中心", "前往会员中心"]
)
expired = subscription.SubscriptionStatus(
subscription.STATUS_EXPIRED,
manage_url="https://cm.example.com/user/subscriptions/cmshopee",
)
active = subscription.SubscriptionStatus(
subscription.STATUS_ACTIVE,
account_name="主账号",
plan_name="测试",
expires_at="2026-08-21T23:59:59+08:00",
)
with mock.patch.object(
main_window,
"QMessageBox",
fake_box,
), mock.patch.object(
main_window.QDesktopServices,
"openUrl",
side_effect=[True, False],
) as open_url:
window._apply_subscription_status(expired)
window._apply_subscription_status(expired)
window._apply_subscription_status(active)
window._apply_subscription_status(expired)
self.assertEqual(2, len(boxes))
self.assertTrue(all(box.title == "会员套餐已过期" for box in boxes))
self.assertTrue(all("业务功能已暂停" in box.text for box in boxes))
self.assertTrue(
all(
box.default_button is box.buttons["前往会员中心"]
for box in boxes
)
)
self.assertEqual(2, open_url.call_count)
self.assertIn(
"无法打开会员中心",
window.statusBar().currentMessage(),
)
self.assertTrue(
all(
call.args[0].toString()
== "https://cm.example.com/user/subscriptions/cmshopee"
for call in open_url.call_args_list
)
)
def test_expired_subscription_notice_disables_missing_url_and_can_exit(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
self.addCleanup(window.close)
fake_box, boxes = self.make_fake_message_box("退出程序")
expired = subscription.SubscriptionStatus(subscription.STATUS_EXPIRED)
with mock.patch.object(
main_window,
"QMessageBox",
fake_box,
), mock.patch.object(
main_window.QDesktopServices,
"openUrl",
) as open_url, mock.patch.object(window, "close") as close:
window._apply_subscription_status(expired)
self.assertEqual(1, len(boxes))
box = boxes[0]
self.assertFalse(box.buttons["前往会员中心"].enabled)
self.assertEqual(
"会员中心地址当前不可用",
box.buttons["前往会员中心"].tooltip,
)
self.assertIn("会员中心地址当前不可用", box.text)
self.assertIsNone(box.default_button)
open_url.assert_not_called()
close.assert_called_once_with()
def test_settings_subscription_recheck_uses_main_window_worker_state(self):
class _Signal:
def __init__(self):
self.callbacks = []
def connect(self, callback):
self.callbacks.append(callback)
class _Worker:
def __init__(self):
self.finished = _Signal()
self.cancelled = _Signal()
def cancel(self):
pass
class _Thread:
def __init__(self):
self.finished = _Signal()
self.started = False
def start(self):
self.started = True
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
self.addCleanup(window.close)
settings_tab = window._settings_tab()
worker = _Worker()
thread = _Thread()
with mock.patch(
"app.gui.main_window.SubscriptionCheckWorker",
return_value=worker,
), mock.patch(
"app.gui.main_window.run_worker",
return_value=thread,
):
settings_tab.subscription_check_button.click()
self.assertTrue(thread.started)
self.assertFalse(settings_tab.subscription_check_button.isEnabled())
self.assertEqual(
"正在检测会员状态...",
settings_tab.subscription_check_button.text(),
)
window._forget_subscription_thread(thread)
self.assertTrue(settings_tab.subscription_check_button.isEnabled())
self.assertEqual(
"重新检测会员状态",
settings_tab.subscription_check_button.text(),
)
@mock.patch.object(main_window, "SUBSCRIPTION_ENFORCEMENT_ENABLED", False)
def test_main_window_observation_mode_checks_without_restricting_workflows(self):
self.assertTrue(main_window.SUBSCRIPTION_CHECK_ENABLED)
self.assertFalse(main_window.SUBSCRIPTION_ENFORCEMENT_ENABLED)
@@ -11291,6 +11467,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
)
self.assertTrue(window.ensure_subscription_for_new_submit("开始 AI 生成"))
@mock.patch.object(main_window, "SUBSCRIPTION_ENFORCEMENT_ENABLED", False)
def test_main_window_observation_mode_suppresses_notice_and_rechecks_after_save(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
tests/test_subscription.py
+5
View File
@@ -90,11 +90,16 @@ class SubscriptionTests(TempDirMixin, unittest.TestCase):
"plan": {"display_name": "专业版"},
"status": "expired",
"expires_at": "2026-08-20T23:59:59+08:00",
"manage_url": "https://cm.example.com/user/subscriptions/cmshopee",
}
expired = subscription.check_status(config, request_json=expired_request)
self.assertEqual(subscription.STATUS_EXPIRED, expired.state)
self.assertFalse(expired.allows_product_workflows)
self.assertEqual(
"https://cm.example.com/user/subscriptions/cmshopee",
expired.manage_url,
)
def test_legacy_404_keeps_existing_workflows_available(self):
with self.make_temp_dir() as temp_dir: