xiaoxia-saas/tests/unit/test_points_routes.py

"""积分/会员 API 路由对齐测试 — fix/1895-points-api-align

覆盖:
- P0-1: POST /points/recharge 返回 pay_params / points_amount / expire_at
- P0-2: POST /points/check 未知 scene_key 返回 400（非 500）
- P1-3: GET /points/rules 返回 description 字段
- P1-6: GET /subscription/plans 返回档位列表
- P1-7: multiplier 实际扣费一致（calculate_points_cost 统一应用）
"""

from __future__ import annotations

import os

from datetime import UTC, datetime, timedelta
from unittest.mock import MagicMock, patch

import pytest
from fastapi import HTTPException


def _make_cu(user_id="user-1", is_member=False, member_type=None):
    cu = MagicMock()
    cu.user.id = user_id
    cu.user.is_member = is_member
    cu.user.member_type = member_type
    cu.user.member_expires_at = None
    return cu


# ── P0-1: recharge response fields ────────────────────────────────────


class TestRechargeOrderResponse:
    def test_recharge_returns_pay_params_points_amount_expire_at(self):
        """recharge 响应必须包含 pay_params / points_amount / expire_at。"""
        from app.api.routes.points import create_recharge_order
        from app.schemas.points import PointsRechargeRequest

        svc = MagicMock()
        svc.create_order.return_value = {
            "id": "order-1",
            "order_type": "points",
            "product_code": "starter_pack",
            "amount_cents": 990,
            "status": "pending",
            "created_at": datetime.now(UTC).isoformat(),
        }
        db = MagicMock()
        cu = _make_cu()
        body = PointsRechargeRequest(package_id="starter_pack")

        before = datetime.now(UTC)
        with patch("app.api.routes.points._get_service", return_value=svc):
            resp = create_recharge_order(body=body, current_user=cu, db=db)
        after = datetime.now(UTC) + timedelta(hours=48)

        assert resp.points_amount == 100  # starter_pack 100 分
        assert isinstance(resp.pay_params, dict)
        assert resp.expire_at is not None
        expire_dt = datetime.fromisoformat(resp.expire_at)
        assert expire_dt >= before + timedelta(hours=47, minutes=55)
        assert expire_dt <= after

    def test_recharge_invalid_package_returns_400(self):
        from app.api.routes.points import create_recharge_order
        from app.schemas.points import PointsRechargeRequest

        svc = MagicMock()
        svc.create_order.side_effect = ValueError("invalid package")
        db = MagicMock()
        cu = _make_cu()
        body = PointsRechargeRequest(package_id="nonexistent")

        with pytest.raises(HTTPException) as exc, patch(
            "app.api.routes.points._get_service", return_value=svc
        ):
            create_recharge_order(body=body, current_user=cu, db=db)
        assert exc.value.status_code == 400


# ── P0-2: check unknown scene → 400 ───────────────────────────────────


class TestCheckPointsUnknownScene:
    def test_unknown_scene_returns_400_not_500(self):
        """未知 scene_key（如 ai_script）应返回 400 UNKNOWN_SCENE，而不是 500。"""
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest

        db = MagicMock()
        cu = _make_cu()
        body = PointsCheckRequest(scene_key="ai_script", quantity=1)

        with pytest.raises(HTTPException) as exc:
            check_points(body=body, current_user=cu, db=db)
        assert exc.value.status_code == 400
        detail = exc.value.detail
        assert detail["code"] == "UNKNOWN_SCENE"
        assert "ai_script" in detail["message"]
        assert "ai_voice" in detail["valid_scenes"]
        assert "ai_title" in detail["valid_scenes"]

    def test_known_scene_still_works(self):
        """合法 scene_key 正常返回，免费用户 ai_voice 1 分钟 = 2 积分。"""
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest

        svc = MagicMock()
        svc.check_daily_free_clip.return_value = False
        svc.get_or_create_account.return_value = {"balance": 50}
        db = MagicMock()
        cu = _make_cu()
        body = PointsCheckRequest(scene_key="ai_voice", quantity=1, duration_minutes=1)

        with patch("app.api.routes.points._get_service", return_value=svc):
            resp = check_points(body=body, current_user=cu, db=db)
        assert resp.required_points == 2  # ceil(1 * 1.15) = 2
        assert resp.current_balance == 50
        assert resp.allowed is True


# ── P1-3: rules include description ───────────────────────────────────


class TestPointsRulesDescription:
    def test_rules_have_description_field(self):
        from app.api.routes.points import get_rules

        resp = get_rules(_current_user=_make_cu())
        assert len(resp.rules) >= 9
        for rule in resp.rules:
            assert rule.description, f"{rule.scene_key} missing description"
            assert isinstance(rule.description, str)
            assert len(rule.description) > 0

    def test_free_user_multiplier_returned(self):
        from app.api.routes.points import get_rules

        resp = get_rules(_current_user=_make_cu())
        assert resp.free_user_multiplier == 1.15


# ── P1-6: GET /subscription/plans ─────────────────────────────────────


class TestSubscriptionPlans:
    @staticmethod
    def _import_plans_fn():
        """Import from the real file to avoid sys.modules shadowing by integration fixtures."""
        import importlib.util
        _route_path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            "..", "..", "apps", "api", "app", "api", "routes", "subscription.py",
        )
        _spec = importlib.util.spec_from_file_location(
            "_real_subscription_routes", os.path.abspath(_route_path)
        )
        _mod = importlib.util.module_from_spec(_spec)
        # inject settings before exec
        import os as _os
        _os.environ.setdefault("JWT_SECRET_KEY", "test-secret")
        _spec.loader.exec_module(_mod)
        return _mod.list_membership_plans

    def test_plans_endpoint_returns_three_tiers(self):
        import os  # noqa: F401  (used by _import_plans_fn)
        list_membership_plans = self._import_plans_fn()
        resp = list_membership_plans(current_user=_make_cu())
        plans = resp["plans"]
        plan_ids = {p["plan_id"] for p in plans}
        assert plan_ids == {"monthly", "quarterly", "yearly"}
        for p in plans:
            assert p["price_cents"] > 0
            assert p["duration_days"] in (30, 90, 365)
            assert 0 < p["points_discount"] <= 1.0
            assert "max_resolution" in p["features"]

    def test_longer_plans_cheaper_per_month(self):
        import os  # noqa: F401
        list_membership_plans = self._import_plans_fn()
        resp = list_membership_plans(current_user=_make_cu())
        plans = resp["plans"]
        monthly = next(p for p in plans if p["plan_id"] == "monthly")
        quarterly = next(p for p in plans if p["plan_id"] == "quarterly")
        yearly = next(p for p in plans if p["plan_id"] == "yearly")
        assert monthly["monthly_price_cents"] == 1990
        assert quarterly["monthly_price_cents"] < monthly["monthly_price_cents"]
        assert yearly["monthly_price_cents"] < quarterly["monthly_price_cents"]


# ── P1-7: multiplier consistency ──────────────────────────────────────


class TestMultiplierConsistency:
    def test_free_user_ai_title_costs_2(self):
        """ai_title base=1，免费用户 ceil(1*1.15)=2。"""
        from packages.domain.points_rules import calculate_points_cost

        assert calculate_points_cost("ai_title", is_member=False, quantity=1) == 2

    def test_check_matches_direct_calculation(self):
        """check 端点 required_points 与 calculate_points_cost 结果一致。"""
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest

        from packages.domain.points_rules import calculate_points_cost

        svc = MagicMock()
        svc.check_daily_free_clip.return_value = False
        svc.get_or_create_account.return_value = {"balance": 999}
        db = MagicMock()
        cu = _make_cu()

        for scene in ["ai_voice", "ai_title", "ai_cover", "ai_rewrite"]:
            body = PointsCheckRequest(scene_key=scene, quantity=1)
            with patch("app.api.routes.points._get_service", return_value=svc):
                resp = check_points(body=body, current_user=cu, db=db)
            expected = calculate_points_cost(scene, is_member=False, quantity=1)
            assert resp.required_points == expected, f"{scene}: got {resp.required_points}, expected {expected}"