xiaoxia-saas/tests/unit/test_points_service.py

"""PointsService 单元测试 (#1895) — 使用 SQLite 内存数据库"""

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from unittest.mock import patch

import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session, sessionmaker

from packages.domain.points_service import PointsService


@pytest.fixture()
def db_session():
    """创建 SQLite 内存数据库 session，包含所有积分相关表。"""
    from packages.adapters.sqlalchemy_impl.models import Base

    engine = create_engine("sqlite://", echo=False)

    # SQLite 不支持 WITH FOR UPDATE，mock 掉
    @event.listens_for(engine, "connect")
    def _disable_for_update(dbapi_conn, connection_record):
        pass

    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine)
    session = SessionLocal()

    yield session

    session.close()


@pytest.fixture()
def service():
    return PointsService()


@pytest.fixture()
def user_id():
    return uuid.uuid4().hex


class TestGetOrCreateAccount:
    def test_creates_new_account(self, service, db_session, user_id):
        data = service.get_or_create_account(user_id, db_session)
        assert data["user_id"] == user_id
        assert data["balance"] == 0
        assert data["total_earned"] == 0
        assert data["total_spent"] == 0

    def test_returns_existing_account(self, service, db_session, user_id):
        service.get_or_create_account(user_id, db_session)
        data = service.get_or_create_account(user_id, db_session)
        assert data["user_id"] == user_id
        assert data["balance"] == 0


class TestCheckBalance:
    def test_sufficient_when_zero(self, service, db_session, user_id):
        result = service.check_balance(user_id, 0, db_session)
        assert result["sufficient"] is True

    def test_insufficient_when_new_account(self, service, db_session, user_id):
        result = service.check_balance(user_id, 10, db_session)
        assert result["sufficient"] is False
        assert result["remaining_after"] == -10


class TestDeductPoints:
    def test_deduct_fails_insufficient_balance(self, service, db_session, user_id):
        result = service.deduct_points(user_id, 100, "ai_voice", db_session)
        assert result["success"] is False
        assert result["transaction_id"] is None

    def test_deduct_after_recharge(self, service, db_session, user_id):
        # 先充值
        service.add_points(user_id, 50, "recharge", db_session)
        # 再扣减
        result = service.deduct_points(user_id, 20, "ai_voice", db_session)
        assert result["success"] is True
        assert result["balance"] == 30

    def test_deduct_creates_transaction(self, service, db_session, user_id):
        service.add_points(user_id, 100, "recharge", db_session)
        result = service.deduct_points(user_id, 30, "ai_voice", db_session)
        assert result["success"] is True

        txns = service.get_transactions(user_id, db_session)
        assert txns["total"] == 2  # 1 add + 1 deduct
        deduct_txn = [t for t in txns["items"] if t["type"] == "deduct"][0]
        assert deduct_txn["amount"] == 30
        assert deduct_txn["balance_after"] == 70


class TestAddPoints:
    def test_add_new_account(self, service, db_session, user_id):
        result = service.add_points(user_id, 100, "recharge:starter_pack", db_session)
        assert result["success"] is True
        assert result["balance"] == 100

    def test_add_accumulates(self, service, db_session, user_id):
        service.add_points(user_id, 50, "recharge", db_session)
        result = service.add_points(user_id, 30, "bonus", db_session)
        assert result["balance"] == 80


class TestRefundPoints:
    def test_refund_adds_back(self, service, db_session, user_id):
        service.add_points(user_id, 100, "recharge", db_session)
        service.deduct_points(user_id, 20, "ai_voice", db_session)
        result = service.refund_points(user_id, 20, "ai_voice", db_session)
        assert result["success"] is True
        assert result["balance"] == 100

    def test_refund_creates_refund_transaction(self, service, db_session, user_id):
        service.add_points(user_id, 100, "recharge", db_session)
        service.refund_points(user_id, 10, "ai_rewrite", db_session)

        txns = service.get_transactions(user_id, db_session)
        refund_txns = [t for t in txns["items"] if t["type"] == "add" and "refund" in t["source"]]
        assert len(refund_txns) == 1
        assert "refund:" in refund_txns[0]["source"]


class TestGetTransactions:
    def test_empty_for_new_user(self, service, db_session, user_id):
        result = service.get_transactions(user_id, db_session)
        assert result["total"] == 0
        assert result["items"] == []

    def test_pagination(self, service, db_session, user_id):
        for i in range(5):
            service.add_points(user_id, 10, f"batch_{i}", db_session)

        result = service.get_transactions(user_id, db_session, page=1, page_size=3)
        assert result["total"] == 5
        assert len(result["items"]) == 3

        result2 = service.get_transactions(user_id, db_session, page=2, page_size=3)
        assert len(result2["items"]) == 2


class TestGetDailyUsage:
    def test_zero_usage(self, service, db_session, user_id):
        with patch("packages.domain.points_service._get_redis_client", return_value=None):
            result = service.get_daily_usage(user_id, db_session)
        assert result["free_clips_used"] == 0
        assert result["free_clips_limit"] == 2
        assert result["free_clips_remaining"] == 2
        assert "reset_at" in result

    def test_after_recording(self, service, db_session, user_id):
        with patch("packages.domain.points_service._get_redis_client", return_value=None):
            service.record_daily_free_clip(user_id, db_session)
            result = service.get_daily_usage(user_id, db_session)
        assert result["free_clips_used"] == 1
        assert result["free_clips_remaining"] == 1


class TestCreateOrder:
    def test_points_order(self, service, db_session, user_id):
        result = service.create_order(user_id, "points", "starter_pack", db_session)
        assert result["order_type"] == "points"
        assert result["product_code"] == "starter_pack"
        assert result["amount_cents"] == 990
        assert result["status"] == "pending"

    def test_membership_order(self, service, db_session, user_id):
        result = service.create_order(user_id, "membership", "monthly", db_session)
        assert result["order_type"] == "membership"
        assert result["amount_cents"] == 1990

    def test_unknown_package_raises(self, service, db_session, user_id):
        with pytest.raises(ValueError, match="Unknown points package"):
            service.create_order(user_id, "points", "nonexistent", db_session)

    def test_unknown_membership_raises(self, service, db_session, user_id):
        with pytest.raises(ValueError, match="Unknown membership type"):
            service.create_order(user_id, "membership", "lifetime", db_session)

    def test_unknown_order_type_raises(self, service, db_session, user_id):
        with pytest.raises(ValueError, match="Unknown order type"):
            service.create_order(user_id, "insurance", "basic", db_session)