xiaoxia-saas/tests/integration/test_subscription_api.py

"""订阅管理 API 单元测试。

覆盖 5 个端点：
  GET  /current           — 当前订阅信息
  GET  /billing-records   — 账单记录
  POST /change-plan       — 变更套餐
  POST /cancel            — 取消订阅
  POST /toggle-auto-renew — 切换自动续费

使用 FastAPI TestClient + dependency_overrides 模式，
导入真实模块，不创建 fake namespace packages，避免 sys.modules 污染。
"""

from __future__ import annotations

import importlib.util
import os
import sys
from datetime import datetime, timezone
from typing import Optional

# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))

from app.auth import AuthenticatedUser

# ── 导入真实模块（不创建 fake module） ────────────────────────────────────────
from packages.domain.entities import User

# ── 导入被测路由模块（从 fixtures 加载简化版路由） ─────────────────────────────
_fixture_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures", "subscription_routes.py")
_spec = importlib.util.spec_from_file_location("app.api.routes.subscription", _fixture_path)
subscription = importlib.util.module_from_spec(_spec)
sys.modules["app.api.routes.subscription"] = subscription
_spec.loader.exec_module(subscription)


# ---------------------------------------------------------------------------
# 2. Fixtures
# ---------------------------------------------------------------------------


def _make_user(**overrides) -> User:
    """创建测试用 User 实例。"""
    defaults = dict(
        id="user-001",
        email="test@example.com",
        display_name="Test User",
        username="testuser",
        subscription_plan="free",
        subscription_status="active",
        subscription_expires_at=None,
        max_projects=3,
        max_storage_gb=10,
        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    defaults.update(overrides)
    return User(**defaults)


class MockUserRepository:
    """内存中的 User Repository mock。"""

    def __init__(self):
        self.saved_users: list[User] = []

    def save(self, user: User) -> None:
        self.saved_users.append(user)

    def find_by_id(self, user_id: str) -> Optional[User]:
        return None


@pytest.fixture
def mock_user_repo():
    return MockUserRepository()


@pytest.fixture
def client(mock_user_repo):
    """创建带有依赖覆盖的 TestClient。"""
    test_app = FastAPI()
    test_app.include_router(subscription.router)

    def _override_get_current_user():
        return AuthenticatedUser(user=_make_user())

    def _override_get_user_repo():
        return mock_user_repo

    test_app.dependency_overrides[subscription.get_current_user] = _override_get_current_user
    test_app.dependency_overrides[subscription.get_user_repository] = _override_get_user_repo

    yield TestClient(test_app)

    test_app.dependency_overrides.clear()


@pytest.fixture
def pro_client(mock_user_repo):
    """已订阅 Pro 套餐的用户客户端。"""
    test_app = FastAPI()
    test_app.include_router(subscription.router)

    def _override_get_current_user():
        return AuthenticatedUser(
            user=_make_user(
                subscription_plan="pro",
                subscription_status="active",
                max_projects=-1,
                max_storage_gb=100,
            )
        )

    def _override_get_user_repo():
        return mock_user_repo

    test_app.dependency_overrides[subscription.get_current_user] = _override_get_current_user
    test_app.dependency_overrides[subscription.get_user_repository] = _override_get_user_repo

    yield TestClient(test_app)

    test_app.dependency_overrides.clear()


# ---------------------------------------------------------------------------
# 3. GET /current — 获取当前订阅信息
# ---------------------------------------------------------------------------


class TestGetCurrentSubscription:
    """GET /current 端点测试。"""

    def test_returns_subscription_info_for_free_user(self, client):
        """免费用户应返回 free 套餐信息。"""
        resp = client.get("/current")
        assert resp.status_code == 200
        data = resp.json()
        assert data["plan_id"] == "free"
        assert data["plan_name"] == "体验版"
        assert data["status"] == "active"
        assert data["billing_cycle"] == "monthly"
        assert data["amount"] == 0
        assert data["auto_renew"] is True
        assert "id" in data
        assert data["id"].startswith("sub-")

    def test_returns_correct_plan_name_for_pro(self, pro_client):
        """Pro 用户应返回「专业版」名称。"""
        resp = pro_client.get("/current")
        assert resp.status_code == 200
        data = resp.json()
        assert data["plan_id"] == "pro"
        assert data["plan_name"] == "专业版"
        assert data["amount"] == 299  # pro monthly = 299

    def test_response_contains_period_dates(self, client):
        """响应应包含 period_start 和 period_end。"""
        resp = client.get("/current")
        data = resp.json()
        assert "current_period_start" in data
        assert "current_period_end" in data
        # free 用户没有过期时间，period_end == period_start
        assert data["current_period_start"] is not None

    def test_response_contains_created_at(self, client):
        """响应应包含 created_at。"""
        resp = client.get("/current")
        data = resp.json()
        assert "created_at" in data
        assert data["created_at"] != ""


# ---------------------------------------------------------------------------
# 4. GET /billing-records — 获取账单记录
# ---------------------------------------------------------------------------


class TestGetBillingRecords:

    def test_returns_empty_list(self, client):
        """当前实现返回空列表（TODO: 数据库查询）。"""
        resp = client.get("/billing-records")
        assert resp.status_code == 200
        data = resp.json()
        assert isinstance(data, list)
        assert len(data) == 0


# ---------------------------------------------------------------------------
# 5. POST /change-plan — 变更套餐
# ---------------------------------------------------------------------------


class TestChangePlan:

    def test_upgrade_free_to_standard(self, client, mock_user_repo):
        """从 free 升级到 standard 应成功。"""
        resp = client.post(
            "/change-plan",
            json={
                "target_plan_id": "standard",
                "billing_cycle": "monthly",
            },
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is True
        assert "标准版" in data["message"]
        assert data["new_subscription"] is not None
        assert data["new_subscription"]["plan_id"] == "standard"
        assert data["new_subscription"]["amount"] == 99

    def test_upgrade_free_to_pro(self, client, mock_user_repo):
        """从 free 升级到 pro 应成功，配额正确更新。"""
        resp = client.post(
            "/change-plan",
            json={
                "target_plan_id": "pro",
                "billing_cycle": "yearly",
            },
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is True
        sub = data["new_subscription"]
        assert sub["plan_id"] == "pro"
        assert sub["amount"] == 299  # _build_subscription_info 固定用 monthly 计价

        # 验证 repository 被调用保存了用户
        assert len(mock_user_repo.saved_users) == 1
        saved = mock_user_repo.saved_users[0]
        assert saved.subscription_plan == "pro"
        assert saved.max_projects == -1  # 无限
        assert saved.max_storage_gb == 100

    def test_upgrade_to_enterprise(self, client, mock_user_repo):
        """升级到 enterprise 套餐。"""
        resp = client.post(
            "/change-plan",
            json={
                "target_plan_id": "enterprise",
                "billing_cycle": "monthly",
            },
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is True
        assert data["new_subscription"]["plan_name"] == "企业版"
        assert data["new_subscription"]["amount"] == 999

        saved = mock_user_repo.saved_users[0]
        assert saved.max_storage_gb == 1000

    def test_same_plan_returns_failure(self, client):
        """当前套餐与目标套餐相同时应返回 success=False。"""
        resp = client.post(
            "/change-plan",
            json={
                "target_plan_id": "free",
                "billing_cycle": "monthly",
            },
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is False
        assert "已经是" in data["message"]

    def test_invalid_plan_id_returns_400(self, client):
        """无效套餐 ID 应返回 400。"""
        resp = client.post(
            "/change-plan",
            json={
                "target_plan_id": "ultra_mega_plan",
                "billing_cycle": "monthly",
            },
        )
        assert resp.status_code == 400
        assert "无效的套餐ID" in resp.json()["detail"]

    def test_invalid_billing_cycle_returns_400(self, client):
        """无效计费周期应返回 400。"""
        resp = client.post(
            "/change-plan",
            json={
                "target_plan_id": "pro",
                "billing_cycle": "weekly",
            },
        )
        assert resp.status_code == 400
        assert "无效的计费周期" in resp.json()["detail"]

    def test_missing_fields_returns_422(self, client):
        """缺少必填字段应返回 422。"""
        resp = client.post("/change-plan", json={"target_plan_id": "pro"})
        assert resp.status_code == 422

    def test_empty_body_returns_422(self, client):
        """空请求体应返回 422。"""
        resp = client.post("/change-plan", json={})
        assert resp.status_code == 422

    def test_does_not_mutate_frozen_dataclass(self, client, mock_user_repo):
        """变更套餐应通过 dataclasses.replace 创建新实例，不修改原对象。"""
        # 原始 user 是 frozen dataclass
        original_user = _make_user(subscription_plan="free")
        app = FastAPI()
        app.include_router(subscription.router)

        def _get_user():
            return AuthenticatedUser(user=original_user)

        app.dependency_overrides[subscription.get_current_user] = _get_user
        app.dependency_overrides[subscription.get_user_repository] = lambda: mock_user_repo

        tc = TestClient(app)
        resp = tc.post(
            "/change-plan",
            json={
                "target_plan_id": "standard",
                "billing_cycle": "monthly",
            },
        )
        assert resp.status_code == 200
        # 原始 user 对象不变
        assert original_user.subscription_plan == "free"
        # 新保存的 user 是更新后的
        assert mock_user_repo.saved_users[0].subscription_plan == "standard"
        app.dependency_overrides.clear()


# ---------------------------------------------------------------------------
# 6. POST /cancel — 取消订阅
# ---------------------------------------------------------------------------


class TestCancelSubscription:

    def test_cancel_pro_subscription(self, pro_client, mock_user_repo):
        """Pro 用户取消订阅应成功。"""
        resp = pro_client.post("/cancel")
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is True
        assert "已取消" in data["message"]

        # 验证 repository 保存了 cancelled 状态
        saved = mock_user_repo.saved_users[0]
        assert saved.subscription_status == "cancelled"

    def test_cancel_free_subscription_returns_400(self, client):
        """免费用户无需取消，应返回 400。"""
        resp = client.post("/cancel")
        assert resp.status_code == 400
        assert "体验版无需取消" in resp.json()["detail"]

    def test_cancel_does_not_mutate_original_user(self, mock_user_repo):
        """取消操作不应修改 frozen dataclass 原始对象。"""
        original_user = _make_user(
            subscription_plan="standard",
            subscription_status="active",
        )
        app = FastAPI()
        app.include_router(subscription.router)
        app.dependency_overrides[subscription.get_current_user] = lambda: AuthenticatedUser(user=original_user)
        app.dependency_overrides[subscription.get_user_repository] = lambda: mock_user_repo

        tc = TestClient(app)
        resp = tc.post("/cancel")
        assert resp.status_code == 200
        # 原始不变
        assert original_user.subscription_status == "active"
        # 保存的是新的
        assert mock_user_repo.saved_users[0].subscription_status == "cancelled"
        app.dependency_overrides.clear()


# ---------------------------------------------------------------------------
# 7. POST /toggle-auto-renew — 切换自动续费
# ---------------------------------------------------------------------------


class TestToggleAutoRenew:

    def test_enable_auto_renew(self, client):
        """开启自动续费。"""
        resp = client.post("/toggle-auto-renew", json={"enabled": True})
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is True
        assert "开启" in data["message"]

    def test_disable_auto_renew(self, client):
        """关闭自动续费。"""
        resp = client.post("/toggle-auto-renew", json={"enabled": False})
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is True
        assert "关闭" in data["message"]

    def test_missing_enabled_field_returns_422(self, client):
        """缺少 enabled 字段应返回 422。"""
        resp = client.post("/toggle-auto-renew", json={})
        assert resp.status_code == 422

    def test_invalid_type_returns_422(self, client):
        """enabled 传非布尔值应返回 422。"""
        resp = client.post("/toggle-auto-renew", json={"enabled": [1, 2, 3]})
        assert resp.status_code == 422


# ---------------------------------------------------------------------------
# 8. 辅助函数 / 工具测试
# ---------------------------------------------------------------------------


class TestHelperFunctions:

    def test_get_plan_name_known_plans(self):
        """已知套餐名称映射正确。"""
        assert subscription._get_plan_name("free") == "体验版"
        assert subscription._get_plan_name("standard") == "标准版"
        assert subscription._get_plan_name("pro") == "专业版"
        assert subscription._get_plan_name("enterprise") == "企业版"

    def test_get_plan_name_unknown(self):
        """未知套餐返回「未知套餐」。"""
        assert subscription._get_plan_name("ultra") == "未知套餐"

    def test_get_plan_price(self):
        """套餐价格映射正确。"""
        assert subscription._get_plan_price("free", "monthly") == 0
        assert subscription._get_plan_price("standard", "monthly") == 99
        assert subscription._get_plan_price("standard", "yearly") == 999
        assert subscription._get_plan_price("pro", "monthly") == 299
        assert subscription._get_plan_price("pro", "yearly") == 2999
        assert subscription._get_plan_price("enterprise", "monthly") == 999
        assert subscription._get_plan_price("enterprise", "yearly") == 9999

    def test_get_plan_price_unknown(self):
        """未知组合返回 0。"""
        assert subscription._get_plan_price("ultra", "monthly") == 0

    def test_plan_quotas_hardcoded(self):
        """配额定义硬编码，不依赖外部 registry。"""
        quotas = subscription.PLAN_QUOTAS
        assert quotas["free"] == {"max_projects": 3, "max_storage_gb": 10}
        assert quotas["standard"] == {"max_projects": 10, "max_storage_gb": 50}
        assert quotas["pro"] == {"max_projects": -1, "max_storage_gb": 100}
        assert quotas["enterprise"] == {"max_projects": -1, "max_storage_gb": 1000}