"""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)