# 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`

**实现**：
```python
@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`

**实现**：
```python
@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

```python
@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` 依赖

```python
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` 枚举值：
```python
provider=AIProvider.STABILITY  # ✅ 正确
provider='stability'           # ❌ 错误
```

### 问题 2：唯一约束冲突

**错误**：
```
UniqueViolationError: duplicate key value violates unique constraint "ai_models_model_name_key"
```

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

**解决**：
在 fixture 中添加存在性检查：
```python
# 先检查是否已存在
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 层的事务管理**（推荐）
```python
# 移除 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**
```python
# 使用 savepoint 代替 begin
async with self.db.begin_nested():  # 创建 savepoint
    consumption_log = await self.credit_service.consume_credits(...)
    job = await self.job_repository.create(...)
```

**方案 C：条件性事务**
```python
# 检查是否已在事务中
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 | 是否启用 |

**定价规则示例**：
```json
{
  "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 完整实现](./2026-01-29-ai-service-complete-implementation.md)
- [AI API 实现](./2026-01-29-ai-api-implementation.md)
- [AI API 测试套件](./2026-01-29-ai-api-test-suite.md)
- [测试执行结果](./2026-01-30-ai-api-test-execution-results.md)
- [测试修复策略](./2026-01-30-ai-api-test-fix-summary.md)