端到端刷题工具的集成测试：LLM、编译器、评测器的协同验证

端到端刷题工具的集成测试LLM、编译器、评测器的协同验证一、单独测试都通过合在一起就崩AI 刷题工具的各个组件——LLM 代码生成、沙箱执行、评测器——在各自的单元测试中都表现良好。但如果把它们串成一个完整的链路用户输入题目 → LLM 生成代码 → 沙箱编译 → 评测器打分 → 返回结果你会发现各种协同问题LLM 生成的代码有引号格式问题沙箱不支持评测器的输出格式和前端期望不一致高并发下沙箱的资源争抢导致超时集成测试的目的是验证组件间的协作是否正常而不是验证单个组件的内部逻辑。二、集成测试的覆盖范围flowchart LR A[集成测试场景] -- B[正常流程] A -- C[异常降级] A -- D[性能压测] A -- E[多语言兼容] B -- B1[Python 题: 生成 → 编译 → 评测 → 返回] B -- B2[Java 题: 需要编译步骤] B -- B3[Go 题: go run 直接执行] C -- C1[LLM 超时: 返回友好错误] C -- C2[沙箱执行失败: 重试一次] C -- C3[评测器异常: 降级为仅判对错] D -- D1[10 并发: 各组件负载分布] D -- D2[100 并发: 沙箱队列积压] D -- D3[1000 并发: 超时与限流] E -- E1[跨语言题解同时评测] E -- E2[同一题的 Python/Java/C 输出一致性]三、实现import pytest import time import asyncio from unittest.mock import AsyncMock, patch from concurrent.futures import ThreadPoolExecutor class TestE2EPipeline: 端到端集成测试验证完整链路的协同行为 pytest.fixture def pipeline(self): 创建完整的刷题流水线 from pipeline import solve_pipeline # 假设的流水线入口 return solve_pipeline pytest.mark.asyncio async def test_normal_flow_python(self, pipeline): 正常流程Python 题解从生成到返回 request { problem: { title: 两数之和, description: 给定整数数组和目标值找出和为目标值的两个数, constraints: 2 ≤ n ≤ 10^4, }, language: python, } result await pipeline.process(request) # 验证返回结构完整 assert result[status] success assert code in result assert analysis in result assert eval_result in result # 验证评测结果正确 assert result[eval_result][passed] is True # 验证复杂度分析包含必要信息 assert O( in result[analysis][time] assert O( in result[analysis][space] pytest.mark.asyncio async def test_llm_timeout_degradation(self, pipeline): LLM 超时时的降级行为 request { problem: {title: Hard 题, description: ...}, language: python, } # 模拟 LLM 超时 with patch(llm_client.generate, side_effectTimeoutError): result await pipeline.process(request) # 验证返回降级错误而非崩溃 assert result[status] error assert timeout in result[message].lower() assert result[retryable] is True # 应该允许重试 pytest.mark.asyncio async def test_sandbox_execution_failure(self, pipeline): 沙箱执行失败时的重试行为 request { problem: {title: 简单题, description: ...}, language: python, } # 模拟第一次沙箱执行失败第二次成功 call_count [0] async def flaky_sandbox(*args, **kwargs): call_count[0] 1 if call_count[0] 1: raise RuntimeError(沙箱临时故障) return {passed: True, output: 42} with patch(sandbox.execute, side_effectflaky_sandbox): result await pipeline.process(request) # 验证重试后成功 assert result[status] success assert call_count[0] 2 # 确认重试了一次 pytest.mark.asyncio async def test_cross_language_consistency(self, pipeline): 跨语言一致性验证同一题在不同语言中评测结果一致 problem { title: 反转链表, description: 反转一个单链表, } results {} for lang in [python, java, cpp]: request {problem: problem, language: lang} results[lang] await pipeline.process(request) # 验证所有语言的评测都通过 for lang, result in results.items(): assert result[status] success, f{lang} 失败 assert result[eval_result][passed], f{lang} 评测未通过 pytest.mark.asyncio async def test_concurrent_requests(self, pipeline): 并发请求测试验证系统在高并发下的稳定性 request { problem: {title: 斐波那契数列, description: 计算第 n 项}, language: python, } # 模拟 20 个并发请求 tasks [pipeline.process(request) for _ in range(20)] results await asyncio.gather(*tasks, return_exceptionsTrue) # 验证所有请求都成功完成无异常 success_count sum(1 for r in results if not isinstance(r, Exception) and r[status] success) error_count sum(1 for r in results if isinstance(r, Exception)) assert error_count 0, f有 {error_count} 个请求抛出异常 assert success_count 18, f成功率 {success_count}/20 低于预期 class TestComponentIntegration: 组件间接口契约测试 def test_llm_output_format_contract(self): LLM 输出格式必须符合下游组件的解析要求 # LLM 应输出合法的 JSON包含 code 和 analysis 字段 llm_output { code: def solve():\n return 42, analysis: { time: O(n), space: O(1), } } # 验证沙箱能否解析 LLM 的输出 from sandbox import parse_code code parse_code(llm_output) assert code is not None assert def solve in code # 验证评测器能否解析复杂度分析 from evaluator import parse_analysis complexity parse_analysis(llm_output) assert complexity[time] O(n) assert complexity[space] O(1) def test_evaluator_output_format_contract(self): 评测器输出必须符合前端展示要求 eval_output { passed: True, test_cases: [ {input: 1,2,3, expected: 6, actual: 6, status: pass}, ], time_ms: 42, memory_mb: 15.3, } # 前端期望的字段 required_fields [passed, test_cases, time_ms, memory_mb] for field in required_fields: assert field in eval_output, f缺少必要字段: {field}四、集成测试的关键原则4.1 不要 Mock 一切集成测试的价值在于发现组件间协作问题。如果每个组件都被 Mock那实际上还是在做单元测试。至少保留 2-3 个真实组件之间的真实交互。4.2 关注数据流转集成测试的重心不是「这个组件做了什么」而是「数据在组件间传递时格式是否一致、边界值是否被正确处理」。4.3 异常路径覆盖正常路径的测试价值有限大家都测过了。更应该关注上游超时、下游返回格式异常、中间件不可用等异常场景下的降级行为。五、总结端到端的集成测试不是为了验证「每个组件对不对」那是单元测试的事而是验证「组合起来的系统在真实场景下是否稳定」。刷题工具链涉及 LLM、沙箱、评测器三个异构组件它们之间的接口契约、错误传播、并发协同都是集成测试需要关注的焦点。