xiaoxia-saas/tests/unit/test_credit_system_switch.py

"""积分系统暂停开关测试 (#1895, ENABLE_CREDIT_SYSTEM)。

产品要求：暂停积分系统但保留全部代码/表/接口。
- 默认 false：所有 AI 功能免费放行，不扣积分、不做余额拦截；
- /points/check 恒返回 allowed=True、required_points=0；
- /points/deduct 为 no-op，余额不变；
- 查询接口（balance/transactions/rules/packages/membership/usage）照常可用；
- 旧环境变量 POINTS_ENABLED 作为兼容别名仍可开启。
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest


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


# ── 配置层 ────────────────────────────────────────────────────────────────


class TestCreditSystemConfig:
    def test_default_disabled(self):
        import os

        from packages.config.base import SharedSettings

        assert os.environ.get("ENABLE_CREDIT_SYSTEM") is None
        assert os.environ.get("POINTS_ENABLED") is None
        s = SharedSettings(_env_file=None)
        assert s.credits_enabled is False
        # 旧属性名仍可用（业务代码大量引用 settings.points_enabled）
        assert s.points_enabled is False

    def test_enable_credit_system_env(self, monkeypatch):
        from packages.config import base as base_mod

        monkeypatch.setenv("ENABLE_CREDIT_SYSTEM", "true")
        s = base_mod.SharedSettings(_env_file=None)
        assert s.points_enabled is True
        assert s.credits_enabled is True

    def test_legacy_points_enabled_env_alias(self, monkeypatch):
        from packages.config import base as base_mod

        monkeypatch.setenv("ENABLE_CREDIT_SYSTEM", "false")
        monkeypatch.setenv("POINTS_ENABLED", "true")
        s = base_mod.SharedSettings(_env_file=None)
        assert s.points_enabled is True
        assert s.credits_enabled is False
        assert s.points_enabled_compat is True

    def test_legacy_setter_back_compat(self):
        from packages.config.base import SharedSettings

        s = SharedSettings(_env_file=None)
        s.points_enabled = True
        assert s.credits_enabled is True
        assert s.points_enabled is True
        s.points_enabled = False
        assert s.points_enabled is False


# ── /points/check：关闭时恒放行、需 0 积分 ────────────────────────────────


class TestCheckEndpointWhenDisabled:
    def test_check_allowed_zero_required(self):
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest

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

        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = check_points(body=body, current_user=cu, db=db)

        assert resp.allowed is True
        assert resp.required_points == 0
        assert resp.remaining_after == 0
        # 不再走免费额度判定
        svc.check_daily_free_clip.assert_not_called()

    def test_unknown_scene_still_400_when_disabled(self):
        """未知 scene 即使系统关闭也返回 400（参数校验先于开关）。"""
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest
        from fastapi import HTTPException

        with pytest.raises(HTTPException) as exc:
            check_points(body=PointsCheckRequest(scene_key="nope"), current_user=_make_cu(), db=MagicMock())
        assert exc.value.status_code == 400

    def test_check_enabled_calculates_cost(self):
        """开关开启时保持原有计费校验。"""
        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": 100}
        body = PointsCheckRequest(scene_key="ai_title", quantity=1)

        with (
            patch("app.api.routes.points._credits_enabled", return_value=True),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = check_points(body=body, current_user=_make_cu(), db=MagicMock())

        assert resp.required_points == 2  # 免费用户 ceil(1*1.15)=2


# ── /points/deduct：关闭时 no-op，余额不变 ────────────────────────────────


class TestDeductEndpointWhenDisabled:
    def test_deduct_is_noop(self):
        from app.api.routes.points import deduct_points
        from app.schemas.points import PointsDeductRequest

        svc = MagicMock()
        svc.get_or_create_account.return_value = {"balance": 7}
        body = PointsDeductRequest(scene_key="ai_voice", amount=999)

        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = deduct_points(body=body, current_user=_make_cu(), db=MagicMock())

        svc.deduct_points.assert_not_called()
        assert resp.success is True
        assert resp.data["balance"] == 7
        assert resp.data["transaction_id"] == ""

    def test_deduct_enabled_works_as_before(self):
        from app.api.routes.points import deduct_points
        from app.schemas.points import PointsDeductRequest

        svc = MagicMock()
        svc.deduct_points.return_value = {"success": True, "balance": 8, "transaction_id": "tx-1"}
        body = PointsDeductRequest(scene_key="ai_title", amount=2)

        with (
            patch("app.api.routes.points._credits_enabled", return_value=True),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = deduct_points(body=body, current_user=_make_cu(), db=MagicMock())

        svc.deduct_points.assert_called_once()
        assert resp.data["balance"] == 8
        assert resp.data["transaction_id"] == "tx-1"


# ── 查询接口：系统关闭时仍全部可用 ────────────────────────────────────────


class TestQueryEndpointsRemainAvailable:
    def test_balance_route_works_when_disabled(self):
        from app.api.routes.points import get_balance

        svc = MagicMock()
        svc.get_or_create_account.return_value = {"balance": 0, "total_earned": 0, "total_spent": 0}
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = get_balance(current_user=_make_cu(), db=MagicMock())
        assert resp.balance == 0
        assert resp.is_member is False

    def test_transactions_route_works_when_disabled(self):
        from app.api.routes.points import get_transactions

        svc = MagicMock()
        svc.get_transactions.return_value = {"items": [], "total": 0, "page": 1, "page_size": 20}
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = get_transactions(current_user=_make_cu(), db=MagicMock())
        assert resp.total == 0

    def test_daily_usage_route_works_when_disabled(self):
        from app.api.routes.points import get_daily_usage

        svc = MagicMock()
        svc.get_daily_usage.return_value = {
            "free_clips_used": 0,
            "free_clips_limit": 2,
            "free_clips_remaining": 2,
            "reset_at": "2026-09-20T00:00:00Z",
        }
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = get_daily_usage(current_user=_make_cu(), db=MagicMock())
        assert resp.free_clips_limit == 2


# ── 业务路由：开关关闭时 PointsService 不实例化、不扣分 ───────────────────


class TestBusinessRoutesBypassWhenDisabled:
    def test_lipsync_route_skips_points(self):
        """lipsync 创建任务路由：settings.points_enabled=False 时不构造 PointsService。"""
        from app.api.routes import lipsync as lipsync_mod

        assert bool(getattr(lipsync_mod.settings, "points_enabled", False)) is False

    def test_tts_route_skips_points(self):
        from app.api.routes import tts as tts_mod

        assert bool(getattr(tts_mod.settings, "points_enabled", False)) is False