6.7 KiB

Raw Permalink Blame History

AI Service 测试套件完成

日期: 2026-01-29
类型: 测试
影响范围: AI Service 监控功能

概述

完成了 AI Service 监控功能的完整测试套件，包括单元测试和集成测试，所有测试全部通过。

测试覆盖

单元测试（9/9 通过）

文件: server/tests/unit/services/test_ai_service_monitoring.py

测试内容：

✅ test_get_user_jobs - 获取用户任务列表
✅ test_get_user_jobs_with_filters - 带筛选条件的任务查询
✅ test_get_user_jobs_invalid_user - 无效用户 ID 处理
✅ test_get_job_statistics - 任务统计
✅ test_get_job_statistics_date_range - 时间范围统计
✅ test_get_queue_status - 队列状态监控
✅ test_handle_timeout_jobs - 超时任务处理
✅ test_handle_timeout_jobs_with_credit_refund - 超时任务积分退还
✅ test_get_user_jobs_pagination - 分页功能

Repository 层单元测试（6/6 通过）

文件: server/tests/unit/repositories/test_ai_job_repository.py

测试内容：

✅ test_get_jobs_by_filters_basic - 基础筛选
✅ test_get_jobs_by_filters_pagination - 分页
✅ test_get_jobs_by_filters_date_range - 时间范围
✅ test_get_job_statistics - 统计功能
✅ test_get_timeout_jobs - 超时任务检测
✅ test_get_jobs_by_filters_ordering - 排序功能

集成测试（6/6 通过）

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

测试内容：

✅ test_timeout_job_flow - 完整的超时任务处理流程
✅ test_statistics_accuracy - 统计数据准确性
✅ test_pagination_consistency - 分页一致性
✅ test_multiple_timeout_handling - 多次超时处理
✅ test_queue_status_realtime - 队列状态实时性
✅ test_statistics_with_date_range - 时间范围统计准确性

修复的问题

1. Repository 层修复

问题: count_user_jobs 方法不支持全局统计（user_id=None）

修复:

async def count_user_jobs(
    self,
    user_id: Optional[str] = None,  # 支持 None
    job_type: Optional[int] = None,
    status: Optional[int] = None
) -> int:
    """统计用户任务数量（支持全局统计）"""
    query = select(func.count(AIJob.ai_job_id))
    
    # 如果指定用户，添加用户过滤
    if user_id is not None:
        query = query.where(AIJob.user_id == user_id)
    
    # ... 其他过滤条件

文件: server/app/repositories/ai_job_repository.py

2. Service 层修复

问题: handle_timeout_jobs 方法在循环结束后才 commit，导致并发时事务冲突

修复: 每处理一个任务就提交一次，避免并发冲突

for job in timeout_jobs:
    try:
        # 处理任务...
        
        # 每处理一个任务就提交一次，避免并发冲突
        await self.db.commit()
        handled_count += 1
        
    except Exception as e:
        logger.error(f"处理超时任务失败: job_id={job.ai_job_id}, error={str(e)}")
        # 回滚当前任务的更改，继续处理下一个
        await self.db.rollback()

文件: server/app/services/ai_service.py

3. 测试 Fixture 修复

问题: test_credit_balance fixture 尝试导入不存在的 CreditBalance Model

修复: 直接使用 User 表的 ai_credits_balance 字段

@pytest_asyncio.fixture
async def test_credit_balance(db_session: AsyncSession, test_user):
    """创建测试用户的积分余额（使用 User 表的 ai_credits_balance 字段）"""
    test_user.ai_credits_balance = 1000
    await db_session.commit()
    await db_session.refresh(test_user)
    return test_user

文件: server/tests/conftest.py

4. 测试数据修复

问题:

CreditConsumptionLog 字段名错误（status → task_status）
缺少必填字段 credits_consumed
多余字段 amount

修复: 使用正确的字段名和值

consumption = CreditConsumptionLog(
    consumption_id=str(uuid4()),
    user_id=test_user.user_id,
    credits_consumed=10,  # 必填字段
    feature_type=1,  # IMAGE_GENERATION
    task_status=1,  # PENDING（正确的字段名）
    task_params={'test': 'timeout_flow'}
)

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

测试执行结果

单元测试

$ docker exec jointo-server-app pytest tests/unit/services/test_ai_service_monitoring.py -v
============================== 9 passed in 0.51s ===============================

$ docker exec jointo-server-app pytest tests/unit/repositories/test_ai_job_repository.py -v
============================== 6 passed in 0.20s ===============================

集成测试

$ docker exec jointo-server-app pytest tests/integration/test_ai_task_monitoring.py -v
============================== 6 passed in 9.38s ===============================

测试覆盖的功能

任务查询

批量查询用户任务
多维度筛选（类型、状态、时间范围）
分页和排序
用户验证

任务统计

总体统计（总数、成功率、积分消耗）
按类型分组统计
按模型分组统计
时间范围统计
平均执行时间计算

队列监控

Celery Worker 状态
活跃任务数
预定任务数
保留任务数
数据库中的待处理/处理中任务数
队列长度计算

超时处理

超时任务检测
任务状态更新
Celery 任务取消
积分自动退还
错误处理和日志记录

技术亮点

完整的测试覆盖: 从 Repository 层到 Service 层，再到集成测试
Mock 策略: 使用 Mock 隔离外部依赖（Celery、Credit Service）
数据一致性验证: 验证数据库状态、积分余额、任务状态
边界条件测试: 无效用户、空结果、超时任务
并发场景: 多次超时处理的顺序执行测试

后续优化建议

测试覆盖率报告: 生成详细的代码覆盖率报告
性能测试: 添加大数据量下的性能测试
压力测试: 测试高并发场景下的系统表现
端到端测试: 添加完整的 API 端到端测试

总结

AI Service 监控功能的测试套件已完成，所有测试全部通过（21/21）：

单元测试：15/15 ✅
集成测试：6/6 ✅

测试覆盖了任务查询、统计、队列监控、超时处理等核心功能，确保了系统的稳定性和可靠性。

6.7 KiB Raw Permalink Blame History

AI Service 测试套件完成

概述

测试覆盖

单元测试（9/9 通过）

集成测试（6/6 通过）

修复的问题

1. Repository 层修复

2. Service 层修复

3. 测试 Fixture 修复

4. 测试数据修复

测试执行结果

单元测试

集成测试

测试覆盖的功能

任务查询

任务统计

队列监控

超时处理

技术亮点

后续优化建议

相关文档

总结

6.7 KiB

Raw Permalink Blame History