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", "real_entitlement_allowed": True, "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)) self.assertTrue(result.credentials_accepted) def test_in_memory_key_override_is_used_without_persisting_it(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"]) calls = [] def request_json(*args, **kwargs): calls.append((args, kwargs)) return { "product_code": "cmshopee", "account": {"display_name": "新账号"}, "plan": {"display_name": "测试套餐"}, "status": "active", "real_entitlement_allowed": True, "expires_at": "2026-08-20T23:59:59+08:00", } result = subscription.check_status( config, base_url="https://activation.example.com/api/v1", api_key="temporary-test-key", request_json=request_json, ) self.assertEqual(subscription.STATUS_ACTIVE, result.state) self.assertEqual("https://activation.example.com", calls[0][0][1]) self.assertEqual("temporary-test-key", calls[0][0][3]) self.assertEqual("", appconfig.get_cmhub_api_key(config["cmhub_config_path"])) 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", "real_entitlement_allowed": True, "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", "real_entitlement_allowed": False, "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_does_not_claim_real_entitlement(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.assertFalse(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) self.assertFalse(result.credentials_accepted) 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) self.assertFalse(result.credentials_accepted) 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", "real_entitlement_allowed": False, "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) def test_real_entitlement_field_is_required_and_consistent(self): with self.make_temp_dir() as temp_dir: config = self._config(temp_dir) common = { "product_code": "cmshopee", "account": {"display_name": "主账号"}, "plan": {"display_name": "专业版"}, "status": "active", "expires_at": "2026-08-20T23:59:59+08:00", } missing = subscription.check_status( config, request_json=lambda *args, **kwargs: dict(common), ) inconsistent = subscription.check_status( config, request_json=lambda *args, **kwargs: { **common, "real_entitlement_allowed": False, }, ) self.assertEqual(subscription.STATUS_UNAVAILABLE, missing.state) self.assertEqual(subscription.STATUS_UNAVAILABLE, inconsistent.state) self.assertFalse(missing.allows_product_workflows) if __name__ == "__main__": unittest.main()