157 lines
6.3 KiB
Python
157 lines
6.3 KiB
Python
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()
|