docs/v1.0/server/changelogs/2026-01-30-ai-api-test-data-fixtures.md

AI Service API 测试数据 Fixtures 实现

日期: 2026-01-30
类型: 测试修复
状态: ✅ 完成

问题描述

集成测试失败，原因：

1. 缺少定价配置数据：

   • AI Service 调用 credit_service.calculate_credits() 时需要从数据库查询定价配置
   • 测试数据库中没有 CreditPricing 表的数据
   • 导致 NotFoundError: 未找到 image_generation 的定价配置

2. 缺少 AI 模型数据：

   • AI Service 需要查询 AI 模型配置
   • 测试数据库中没有 AIModel 表的数据

3. 事务嵌套问题：

   • AI Service 中使用了 async with self.db.begin():
   • 测试环境的 session 已经在事务中
   • 导致 InvalidRequestError: A transaction is already begun on this Session

解决方案

1. 添加定价配置 Fixtures

文件: server/tests/conftest.py

实现：

@pytest_asyncio.fixture
async def test_ai_pricing_configs(db_session: AsyncSession):
    """创建 AI 功能的定价配置"""
    from app.models.credit import CreditPricing
    from app.services.credit_service import FeatureType
    from app.utils.id_generator import generate_uuid
    from sqlmodel import select
    
    # 先检查是否已存在，避免重复创建
    result = await db_session.execute(
        select(CreditPricing).where(
            CreditPricing.feature_type == FeatureType.IMAGE_GENERATION
        )
    )
    existing_pricing = result.scalar_one_or_none()
    
    if existing_pricing:
        result = await db_session.execute(select(CreditPricing))
        return result.scalars().all()
    
    pricing_configs = [
        # 1. 图片生成定价
        CreditPricing(
            pricing_id=generate_uuid(),
            feature_type=FeatureType.IMAGE_GENERATION,
            pricing_rules={
                "base": 10,
                "hd_multiplier": 2,
                "model_multipliers": {
                    "dall-e-3": 1.5,
                    "stable-diffusion": 1.0,
                    "midjourney": 2.0
                }
            },
            description="图片生成定价",
            is_active=True
        ),
        # ... 其他 8 种功能类型的定价配置
    ]
    
    for pricing in pricing_configs:
        db_session.add(pricing)
    
    await db_session.flush()
    
    for pricing in pricing_configs:
        await db_session.refresh(pricing)
    
    return pricing_configs

覆盖的功能类型：

1. IMAGE_GENERATION - 图片生成
2. VIDEO_GENERATION - 视频生成
3. SOUND_GENERATION - 音效生成
4. VOICE_GENERATION - 配音生成
5. SUBTITLE_GENERATION - 字幕生成
6. TEXT_PROCESSING - 文本处理
7. RESOURCE_GENERATION - 资源生成
8. STORYBOARD_GENERATION - 分镜生成
9. SCRIPT_GENERATION - 剧本生成

2. 添加 AI 模型 Fixtures

文件: server/tests/conftest.py

实现：

@pytest_asyncio.fixture
async def test_ai_models(db_session: AsyncSession):
    """创建测试 AI 模型配置"""
    from app.models.ai_model import AIModel, AIModelType, AIProvider, UnitType
    from app.utils.id_generator import generate_uuid
    from decimal import Decimal
    from sqlmodel import select
    
    # 先检查是否已存在
    result = await db_session.execute(
        select(AIModel).where(AIModel.model_name == 'stable-diffusion-xl')
    )
    existing_model = result.scalar_one_or_none()
    
    if existing_model:
        result = await db_session.execute(select(AIModel))
        return result.scalars().all()
    
    models = [
        # 图片生成模型
        AIModel(
            model_id=generate_uuid(),
            model_name='stable-diffusion-xl',
            display_name='Stable Diffusion XL',
            description='高质量图片生成模型',
            provider=AIProvider.STABILITY,  # 使用枚举值
            model_type=AIModelType.IMAGE,
            cost_per_unit=Decimal('0.01'),
            unit_type=UnitType.IMAGE,
            credits_per_unit=10,
            is_active=True,
            is_beta=False,
            config={'max_resolution': '1024x1024'}
        ),
        # ... 其他 3 个模型
    ]
    
    for model in models:
        db_session.add(model)
    
    await db_session.flush()
    
    for model in models:
        await db_session.refresh(model)
    
    return models

包含的模型：

1. stable-diffusion-xl - 图片生成（Stability AI）
2. runway-gen2 - 视频生成（Runway）
3. elevenlabs-tts - 语音合成（ElevenLabs）
4. gpt-4 - 文本处理（OpenAI）

3. 更新测试用户积分余额

修改: test_credit_balance fixture

@pytest_asyncio.fixture
async def test_credit_balance(db_session: AsyncSession, test_user):
    """创建测试用户的积分余额"""
    # 设置用户的积分余额（足够运行所有测试）
    test_user.ai_credits_balance = 10000  # 从 1000 增加到 10000
    await db_session.commit()
    await db_session.refresh(test_user)
    return test_user

4. 更新集成测试依赖

文件: server/tests/integration/test_ai_api_workflow.py

修改: 在所有测试方法中添加 test_ai_pricing_configs 和 test_ai_models 依赖

async def test_complete_image_generation_workflow(
    self, 
    async_client: AsyncClient, 
    test_user_token: str, 
    test_user_id: str, 
    db_session,
    test_ai_pricing_configs,  # 添加
    test_ai_models            # 添加
):
    """测试完整的图片生成流程"""
    # ...

遇到的问题

问题 1：Provider 字段类型错误

错误：

TypeError: 'str' object cannot be interpreted as an integer
invalid input for query argument $5: 'stability'

原因：

• AIModel.provider 字段定义为 SMALLINT（枚举值）
• Fixture 中错误地使用了字符串 'stability'

解决： 使用 AIProvider 枚举值：

provider=AIProvider.STABILITY  # ✅ 正确
provider='stability'           # ❌ 错误

问题 2：唯一约束冲突

错误：

UniqueViolationError: duplicate key value violates unique constraint "ai_models_model_name_key"

原因：

• 测试数据库中已存在相同 model_name 的记录
• 测试事务回滚机制可能未正常工作

解决： 在 fixture 中添加存在性检查：

# 先检查是否已存在
result = await db_session.execute(
    select(AIModel).where(AIModel.model_name == 'stable-diffusion-xl')
)
existing_model = result.scalar_one_or_none()

if existing_model:
    # 如果已存在，直接返回现有模型
    result = await db_session.execute(select(AIModel))
    return result.scalars().all()

问题 3：事务嵌套错误（待解决）

错误：

InvalidRequestError: A transaction is already begun on this Session

原因：

• AI Service 中使用了 async with self.db.begin():
• 测试环境的 session 已经在事务中（由 db_session fixture 创建）
• SQLAlchemy 不允许嵌套事务（除非使用 savepoint）

位置：

• ai_service.py:180 - generate_image()
• ai_service.py:300 - generate_video()
• ai_service.py:399 - generate_sound()
• ai_service.py:459 - generate_voice()
• ai_service.py:519 - generate_subtitle()
• ai_service.py:581 - process_text()

可能的解决方案：

方案 A：移除 Service 层的事务管理（推荐）

# 移除 async with self.db.begin():
# 让调用者（API 层或测试）控制事务

# 修改前
async with self.db.begin():
    consumption_log = await self.credit_service.consume_credits(...)
    job = await self.job_repository.create(...)

# 修改后
consumption_log = await self.credit_service.consume_credits(...)
job = await self.job_repository.create(...)
# 由调用者负责 commit/rollback

方案 B：使用 Savepoint

# 使用 savepoint 代替 begin
async with self.db.begin_nested():  # 创建 savepoint
    consumption_log = await self.credit_service.consume_credits(...)
    job = await self.job_repository.create(...)

方案 C：条件性事务

# 检查是否已在事务中
if self.db.in_transaction():
    # 已在事务中，直接执行
    consumption_log = await self.credit_service.consume_credits(...)
else:
    # 未在事务中，开启新事务
    async with self.db.begin():
        consumption_log = await self.credit_service.consume_credits(...)

测试数据结构

CreditPricing 表

字段	类型	说明
pricing_id	UUID	主键
feature_type	SMALLINT	功能类型（1-10）
pricing_rules	JSONB	定价规则
description	TEXT	描述
is_active	BOOLEAN	是否启用

定价规则示例：

{
  "base": 10,
  "hd_multiplier": 2,
  "model_multipliers": {
    "dall-e-3": 1.5,
    "stable-diffusion": 1.0
  }
}

AIModel 表

字段	类型	说明
model_id	UUID	主键
model_name	VARCHAR	模型名称（唯一）
display_name	VARCHAR	显示名称
provider	SMALLINT	提供商（枚举）
model_type	SMALLINT	模型类型（枚举）
cost_per_unit	NUMERIC(10,4)	单位成本
unit_type	SMALLINT	单位类型（枚举）
credits_per_unit	INTEGER	每单位积分
config	JSONB	模型配置
is_active	BOOLEAN	是否启用

下一步

1. ✅ 添加定价配置 fixtures
2. ✅ 添加 AI 模型 fixtures
3. ✅ 更新测试用户积分余额
4. ✅ 更新集成测试依赖
5. ⏳ 解决事务嵌套问题
6. ⏳ 运行完整的集成测试套件
7. ⏳ 创建测试执行报告

技术栈合规性

• ✅ 使用 SMALLINT 存储枚举值
• ✅ 使用 JSONB 存储配置数据
• ✅ 使用 TIMESTAMPTZ 存储时间
• ✅ 使用 UUID v7 作为主键
• ✅ 使用 pytest-asyncio 进行异步测试
• ✅ 使用 fixture 管理测试数据
• ✅ 使用事务回滚确保测试隔离

相关文档

• AI Service 完整实现
• AI API 实现
• AI API 测试套件
• 测试执行结果
• 测试修复策略