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_allowed_when_disabled(self):
        """任意 scene_key（含未知/已下线）系统关闭时都返回 allowed=True, cost=0。"""
        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}
        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=PointsCheckRequest(scene_key="nope"), current_user=_make_cu(), db=MagicMock())
        assert resp.allowed is True
        assert resp.required_points == 0

    def test_check_enabled_calculates_cost(self):
        """开关开启时保持原有计费校验（voice_clone_synth 正常计费）。"""
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest

        svc = MagicMock()
        svc.get_or_create_account.return_value = {"balance": 100}
        body = PointsCheckRequest(scene_key="voice_clone_synth", quantity=1, duration_minutes=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

    def test_retired_scene_free_when_enabled(self):
        """开关开启时，已下线场景返回 cost=0，直接放行。"""
        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}
        with (
            patch("app.api.routes.points._credits_enabled", return_value=True),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            for scene in ["ai_voice", "ai_title", "ai_video", "ai_digital_human", "nope"]:
                body = PointsCheckRequest(scene_key=scene, quantity=1)
                resp = check_points(body=body, current_user=_make_cu(), db=MagicMock())
                assert resp.required_points == 0, f"{scene} should be free"
                assert resp.allowed is True


# ── /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_has_no_points_logic(self):
        """lipsync 路由：已移除手动扣点代码（不导入 PointsService/calculate_points_cost）。"""
        import inspect

        from app.api.routes import lipsync as lipsync_mod

        src = inspect.getsource(lipsync_mod)
        assert "PointsService" not in src
        assert "calculate_points_cost" not in src
        assert "_points_deducted" not in src

    def test_tts_route_has_no_points_logic(self):
        """tts 路由：已移除手动扣点代码（不导入 PointsService/calculate_points_cost）。"""
        import inspect

        from app.api.routes import tts as tts_mod

        src = inspect.getsource(tts_mod)
        assert "PointsService" not in src
        assert "calculate_points_cost" not in src
        assert "_points_deducted" not in src