cmshoppe/tests/test_subscription.py

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()