1. 项目概述一次关于Agent测试工程的深度复盘最近在推进一个基于Agent架构的复杂系统项目我们内部称之为“Harness工程”。这个项目本质上是一套包裹在AI Agent核心推理逻辑之外的基础设施层你可以把它想象成给一个聪明但可能“手忙脚乱”的大脑Agent穿上了一套得体的宇航服。这套“宇航服”不负责代替大脑思考而是为它提供稳定、可靠、可观测的执行环境处理工具调用、状态管理、错误恢复、安全沙箱等一系列繁琐但至关重要的事务。项目临近一个关键里程碑我们需要对这套Harness基础设施进行一次全面的质量验证。测试团队领到的任务听起来很明确设计并执行45个测试用例。但“45个测试”这个数字背后远不是拍脑袋决定的。它涉及到如何系统性地覆盖一个多模块、异步、状态复杂的分布式系统的核心风险点。最终我们不仅完成了测试设计更通过这45个测试挖出了几个非常典型且具有启发性的缺陷。今天我就把这整个过程的设计思路、实操踩坑和发现的问题进行一次完整的复盘和分享。无论你是正在构建AI Agent应用还是从事传统的分布式系统测试相信其中的一些思路和教训都能带来参考。2. 测试策略与框架选型为什么是Pytest面对“Harness测试工程”这个命题首要任务是确立测试策略和选择趁手的工具。我们的系统主要由Python构建因此测试框架自然在pytest和unittest之间选择。最终我们全面采用了pytest原因有几个核心考量。2.1 摒弃unittest拥抱pytest的灵活与强大虽然Python标准库的unittest足够经典但pytest在编写测试的体验和功能扩展性上优势明显。对于Harness这种接口多、场景杂、需要大量参数化测试的项目pytest的几个特性成了决定性因素更简洁的语法不需要继承特定的类函数即测试用例用assert语句直接断言代码更清晰。强大的Fixture机制这是pytest的灵魂。Harness测试中我们需要频繁地初始化模拟的Agent核心、创建工具执行环境、建立临时的状态存储。这些都可以定义为pytest.fixture通过参数注入的方式优雅地在各个测试用例间共享和复用极大减少了重复的样板代码。出色的参数化测试支持pytest.mark.parametrize装饰器允许我们用一个测试函数覆盖多组输入数据和预期输出。这对于测试工具调用接口的不同参数组合、不同错误码返回等场景非常高效。丰富的插件生态我们后续集成了pytest-asyncio用于异步测试、pytest-cov生成代码覆盖率报告、pytest-html生成HTML测试报告这些都能无缝集成。2.2 测试金字塔与Harness的适配我们遵循测试金字塔模型但根据Harness的特点做了调整单元测试占比~50%针对Harness内部各个独立模块如工具路由、状态管理器、安全策略检查器。使用pytest配合unittest.mock对依赖进行隔离和模拟确保每个单元逻辑正确。集成测试占比~35%关注模块间的交互。例如测试“工具路由”模块是否能正确调用“安全沙箱”模块进行检查然后将任务派发给真实的“工具执行器”。这里会使用部分真实的组件但可能用模拟的Agent核心或简化版工具。端到端E2E测试占比~15%模拟完整的用户场景。启动一个完整的Harness服务向其发送一个模拟的Agent请求包含思考过程、工具调用列表验证Harness是否能协调所有组件最终返回正确的结果或错误处理信息。这部分测试成本高、速度慢但不可或缺。2.3 测试环境隔离虚拟环境与Docker为了保证测试的可重复性我们严格隔离测试环境。每个开发者和CI/CD流水线都使用venv或poetry创建独立的Python虚拟环境来安装依赖。对于需要特定外部服务如Redis用于状态缓存、某个特定版本的工具运行时的集成测试和E2E测试我们使用docker-compose在测试前拉起一套临时的服务集群测试结束后销毁。这避免了本地环境差异导致的“在我机器上好好的”问题。注意Fixture在这里再次发挥巨大作用。我们可以编写一个pytest.fixture(scope”session”)来启动整个Docker compose环境并在所有测试结束后自动清理。scope”session”确保整个测试会话只启动一次提高了测试效率。3. 45个测试用例的设计思路与拆解“45个测试”不是凭空而来的。我们采用系统化的方法从需求、架构和风险三个维度进行推导确保测试覆盖无遗漏。3.1 基于需求与功能点的正向推导首先我们从Harness工程的需求文档和API接口定义出发进行等价类划分和边界值分析。核心功能工具调用代理。我们设计了正常调用、异步工具调用、批量工具调用、调用超时、工具不存在等场景。状态管理状态设置、获取、更新、删除。设计了状态键冲突、状态过期、状态回滚等用例。安全沙箱权限检查文件读写、网络访问、资源限制CPU、内存、执行时间、恶意代码拦截。错误处理与恢复网络异常、工具执行失败、状态存储失败、部分失败后的重试与补偿机制。3.2 基于架构与数据流的逆向分析接着我们审视系统架构图和数据流寻找关键集成点和潜在薄弱环节。组件间通信消息队列如Redis Streams的消息丢失、重复消费、顺序保证测试。并发与竞态多个Agent请求同时修改同一状态同一个工具被并发调用时的锁机制。依赖服务故障模拟Redis宕机、数据库连接超时验证系统的降级和容错能力。3.3 基于风险与变更影响的分析最后我们进行风险评估对历史上出过问题的模块、近期代码改动大的模块、以及复杂度高的模块给予更多测试关注。历史问题回归将过去线上出现过的Bug转化为自动化测试用例防止回归。新功能与重构针对本次迭代新增的“工具调用链路追踪”功能设计了从发起到结束的全链路追踪验证测试。复杂度高的模块安全策略引擎规则复杂我们使用pytest的参数化对其规则组合进行了穷举测试。3.4 测试用例清单示例部分下面以表格形式展示部分测试用例的设计你可以看到其与上述思路的对应关系测试ID测试类别测试描述设计思路来源关键断言test_tool_invoke_happy_path单元测试使用合法参数调用一个模拟工具验证返回结果正确。需求-核心功能工具被调用一次返回结果与预期一致。test_tool_invoke_with_invalid_params单元测试传入不符合工具Schema的参数验证Harness能正确拒绝并返回验证错误。需求-错误处理收到ValidationError工具未被实际执行。test_concurrent_state_update集成测试模拟两个并发请求同时更新同一个状态键。架构-并发竞态最终状态值正确无数据损坏日志显示有锁竞争。test_sandbox_file_access_violation单元测试工具尝试读取其权限范围外的文件路径。需求-安全沙箱安全策略引擎拦截该操作返回SecurityViolation错误。test_harness_recovery_after_redis_downE2E测试在测试执行中重启Redis服务验证Harness能否检测到并尝试重连后续请求能恢复正常。架构-依赖故障服务日志记录连接中断和重连重连后新请求成功。test_tool_timeout_and_cancellation集成测试调用一个会长时间运行的模拟工具并设置短超时。需求-错误处理在超时时间到达后Harness发送了取消信号并返回TimeoutError。通过这三个维度的交叉分析我们最终梳理出了45个测试用例它们像一张网覆盖了Harness工程的主要功能面和风险点。4. 测试实现与核心环节详解有了设计接下来就是落地。我们以几个典型的测试为例看看如何用pytest实现并解释其中的关键点。4.1 使用Fixture构建测试脚手架这是提高测试代码复用性和可维护性的关键。我们在conftest.py文件中定义全局或模块级的Fixture。# conftest.py import pytest import asyncio from your_harness_sdk import HarnessClient, MockAgentCore from your_state_backend import RedisStateBackend import redis pytest.fixture def mock_agent_core(): 提供一个模拟的Agent核心可以预设其‘思考’结果。 core MockAgentCore() core.set_next_actions([{tool: calculator, args: {a: 5, b: 3}}]) return core pytest.fixture async def redis_client(): 创建并清理一个测试用的Redis客户端。 client redis.Redis(hosttest-redis, port6379, decode_responsesTrue) await client.ping() # 异步ping确保连接 yield client await client.flushdb() # 测试结束后清空数据库 await client.close() pytest.fixture async def harness_client(mock_agent_core, redis_client): 构建一个配置了模拟核心和真实Redis的Harness客户端这是很多测试的起点。 state_backend RedisStateBackend(redis_client) client HarnessClient(agent_coremock_agent_core, state_backendstate_backend) await client.initialize() yield client await client.shutdown()4.2 异步测试的处理Harness大量使用asyncio测试也必须适配。我们使用pytest-asyncio插件。import pytest pytest.mark.asyncio async def test_async_tool_invocation(harness_client): 测试异步工具的调用。 模拟一个需要网络请求的异步工具验证Harness能正确处理其生命周期。 # 1. 准备一个模拟的异步工具 async def mock_async_tool(delay: int): await asyncio.sleep(delay) # 模拟耗时操作 return {status: done, delay: delay} # 2. 将这个工具注册到Harness客户端测试环境允许动态注册 harness_client.register_tool(async_demo, mock_async_tool) # 3. 执行调用 task_id await harness_client.submit_task({tool_calls: [{name: async_demo, args: {delay: 0.1}}]}) # 4. 等待并获取结果 result await harness_client.get_task_result(task_id, timeout2.0) # 5. 断言 assert result[status] completed assert result[output][0][result][status] done assert result[output][0][result][delay] 0.1 # 同时可以断言工具确实被异步执行了没有阻塞主线程可以通过时间戳粗略判断4.3 模拟Mock与打桩Stub的精准应用单元测试的核心是隔离。我们使用unittest.mock来模拟外部依赖。from unittest.mock import AsyncMock, MagicMock, patch import pytest pytest.mark.asyncio async def test_state_rollback_on_tool_failure(): 测试当工具执行失败时Harness能否正确回滚在此次会话中设置的状态。 这是一个关键的原子性保证测试。 # 1. 模拟一个会在执行时失败的工具 failing_tool AsyncMock(side_effectRuntimeError(Tool crashed!)) # 2. 模拟一个状态后端我们将监视它的set_state和delete_state调用 mock_state_backend MagicMock() mock_state_backend.set_state AsyncMock() mock_state_backend.delete_state AsyncMock() # 3. 使用patch将Harness内部的状态后端替换为我们的mock with patch(your_harness_module.StateManager._backend, mock_state_backend): harness HarnessClient(state_backendmock_state_backend) harness.register_tool(bad_tool, failing_tool) # 4. 提交一个任务这个任务会先设置一个状态然后调用失败的工具 task { pre_actions: [{action: set_state, key: temp, value: data}], tool_calls: [{name: bad_tool}] } # 5. 执行并期待失败 with pytest.raises(RuntimeError, matchTool crashed!): await harness.execute_task(task) # 6. 关键断言验证状态被设置过因为pre_action但在失败后又被删除了回滚 mock_state_backend.set_state.assert_called_once_with(temp, data) mock_state_backend.delete_state.assert_called_once_with(temp) # 断言set_state和delete_state的调用顺序如果需要可以用 call_args_list 进一步检查。这个测试验证了Harness事务性的核心逻辑通过Mock我们无需关心真实的状态存储只关注行为是否符合预期。5. 测试执行与发现的典型Bug实录当我们运行这45个测试组成的套件时CI流水线变成了一个高效的“缺陷筛子”。以下是我们发现的几个最具代表性的Bug以及它们反映出的设计问题。5.1 Bug 1资源泄漏——未关闭的异步生成器测试用例test_streaming_tool_response(测试流式输出工具)。现象当连续运行多次该测试或与其他异步测试混合运行时偶尔会抛出asyncio任务警告或内存缓慢增长。排查过程使用pytest的--tbshort减少日志干扰聚焦错误信息。在测试中增加asyncio事件循环的调试信息发现存在“任务未被正确收集”的提示。审查Harness中处理流式工具的代码发现一个async generator在工具提前退出或异常时没有在finally块中执行aclose()。根本原因开发者在编写流式响应转发逻辑时只考虑了正常情况下的迭代没有在异常处理路径中确保异步生成器的显式关闭导致底层连接或缓冲区资源未被释放。修复在工具调用包装器中使用async with语句管理异步生成器或确保在所有退出路径上调用agen.aclose()。5.2 Bug 2竞态条件——状态缓存与数据库不一致测试用例test_concurrent_state_update(上文提到的并发状态更新测试)。现象测试并非每次失败但大约有10%的概率会失败。断言发现最终状态值与基于操作顺序的预期值不符。排查过程这是一个典型的“海森堡Bug”观察时行为会改变。我们首先在测试中增加了更详细的日志记录每个并发操作的精确时间戳和操作内容。分析日志发现两个并发的“读-改-写”操作出现了交错线程A读取状态值X线程B也读取XA计算后写入X1B计算后也写入X1而不是预期的X2。检查状态管理器的实现发现虽然对Redis的单个操作是原子的但“读取-计算-写入”这个业务逻辑组合并非原子操作。代码中使用了本地内存缓存为了性能来减少Redis读取但更新缓存和更新Redis的时机存在窗口。根本原因缺乏分布式锁或乐观锁机制来保护“读-改-写”事务。本地缓存的存在加剧了问题因为一个线程更新了Redis后另一个线程的本地缓存还是旧值。修复引入基于Redis的分布式锁redlock算法或使用SETNX来保护关键的状态更新区域。或者采用乐观锁在状态对象中增加版本号更新时检查版本号是否变化。5.3 Bug 3错误处理黑洞——异常被过度捕获且未日志测试用例test_harness_graceful_shutdown(测试Harness服务优雅关闭)。现象测试通过但在CI日志中发现了未被捕获的异常痕迹然而Harness的主流程并没有因此失败或打印错误。排查过程查看测试中模拟的“关闭信号”处理代码。发现Harness在关闭时会取消所有正在运行的后台asyncio.Task。进一步检查发现这些Task被取消时其内部可能正在进行的网络请求或资源清理操作会抛出CancelledError或其他异常。而Harness的通用任务包装器使用了一个过于宽泛的except Exception:仅仅记录了一条“任务结束”的DEBUG级别日志就把异常吞没了。根本原因错误处理策略不清晰。CancelledError是asyncio的正常控制流不应作为错误处理。而其他运行时异常被捕获后仅记录低级别日志导致运维人员难以从日志中发现问题。修复区分CancelledError通常直接放过或记录为INFO。对于其他异常至少记录为WARNING或ERROR级别并包含完整的异常堆栈。考虑是否将某些严重异常向上传播以便在服务层面触发告警。5.4 Bug 4配置敏感度——环境变量依赖导致测试不稳定测试用例多个依赖外部API URL或密钥的集成测试。现象在CI环境中测试时好时坏排查发现是因为CI机器上的某个环境变量未被设置或值不同。排查过程这其实是一个测试设计问题。我们过于依赖“完美”的预配置环境。根本原因测试代码硬编码了从环境变量读取配置但没有为测试环境提供默认值或回退机制。当CI流水线重构或新同事搭建环境时极易遗漏。修复在测试的conftest.py或setUp方法中使用monkeypatchpytest的一个内置Fixture来动态设置所需的环境变量。def test_with_env(monkeypatch): monkeypatch.setenv(API_KEY, test-key-not-a-real-one) monkeypatch.setenv(API_URL, http://localhost:9999/mock-api) # ... 执行测试这样保证了测试在一个确定性的配置下运行与外部环境解耦。6. 测试框架工程化与CI/CD集成单个测试能运行还不够我们需要一套可持续、可观测的测试体系。6.1 测试分类与标记我们使用pytest的标记mark功能对测试进行分类方便选择性运行。# 在测试文件中 import pytest pytest.mark.integration pytest.mark.slow def test_full_harness_workflow(): pass pytest.mark.unit pytest.mark.fast def test_single_component(): pass # 在命令行中运行 # 只运行单元测试: pytest -m unit # 运行除慢测试外的所有测试: pytest -m not slow # 运行集成测试且生成覆盖率报告: pytest -m integration --cov.6.2 测试报告与可视化我们配置pytest-html插件来生成美观的HTML报告并与Allure报告集成提供历史趋势和详细的分析。pytest --htmlreport.html --self-contained-html在CI流水线中我们将这个HTML报告作为制品保存每次构建结果一目了然。结合pytest-cov生成的覆盖率报告我们可以清晰地看到代码的哪一部分没有被测试到驱动我们补充用例。6.3 持续集成流水线我们在GitLab CI其他如Jenkins、GitHub Actions同理中定义了多阶段的测试流水线代码检查阶段运行black,isort,flake8进行代码格式化与静态检查。快速测试阶段运行所有标记为fast的测试主要是单元测试通常在几分钟内完成给予开发者快速反馈。完整测试阶段在合并请求Merge Request时触发运行全部45个测试包括需要Docker的集成测试和E2E测试。这个阶段会生成HTML报告和覆盖率报告。发布前测试阶段在打标签准备发布时触发可能还会增加一些性能测试或安全扫描。6.4 测试数据管理对于需要真实数据模式的测试如测试数据库迁移脚本我们使用pytest的Fixture来管理测试数据库的迁移和种子数据。通常采用以下模式每个测试用例运行在一个事务中测试结束后自动回滚保证数据库干净。或者为每个测试用例生成一个唯一的数据库Schema如test_uuid彻底隔离。7. 经验总结与避坑指南回顾这45个测试从设计到执行的全过程有几个深刻的体会和技巧值得分享7.1 测试设计要先于代码实现TDD的思维不一定非要严格遵循测试驱动开发TDD但在编写一个模块或功能前先思考“它应该怎么被测试”能极大改善接口设计。你会自然而然地考虑模块的职责是否单一、依赖是否明确、错误情况是否可观测。这次Harness中状态管理器的清晰接口就部分得益于我们早期对测试用例的构思。7.2 Mock要适度过度Mock会掩盖集成问题Mock是单元测试的利器但过度使用会导致测试与实现耦合过紧且无法发现组件间的真实交互问题。我们的原则是单元测试充分Mock集成测试部分Mock或使用真实组件E2E测试尽量真实。例如测试工具路由逻辑时可以Mock安全沙箱和工具执行器但测试“工具调用全链路”时就应该使用真实的安全沙箱也许是测试模式和模拟工具。7.3 异步测试的稳定性是关键难点异步代码的测试容易遇到“测试通过了但日志里有异常”或者偶发性的超时失败。除了之前提到的资源清理还要注意使用足够的超时asyncio.wait_for或pytest-asyncio的默认超时要设置合理避免因CI机器负载导致假失败。小心事件循环确保每个测试都运行在干净的事件循环上。pytest-asyncio默认处理得很好但如果混用其他异步库或自己管理循环要格外小心。验证异步行为不仅要验证结果有时还要验证顺序。例如测试流式输出时可以用asyncio.Queue来收集输出片段再断言其顺序和内容。7.4 测试本身也需要维护和重构测试代码也是代码会随着生产代码的演进而腐化。定期比如每个季度回顾测试套件删除过时的测试针对已移除功能的测试。合并重复的测试特别是那些通过不同参数测试同一逻辑的用例。重构冗长的测试将复杂的Setup提取成更小、更可复用的Fixture。检查测试速度优化慢测试看是否能通过更好的Mock或使用更轻量的集成方式来提速。7.5 让测试失败信息一目了然断言失败时错误信息应该直接告诉你哪里不对。善用pytest的内置断言重写或者使用像pytest-assume允许一个测试中多个断言都执行这样的插件。对于复杂对象的比较可以使用pytest的-vv详细模式或者自定义对象的__repr__方法让差异在失败信息中高亮显示。这次针对Harness工程的45个测试不仅是一个质量保障活动更是一次对系统设计的深度审查。每一个暴露出来的Bug都促使我们反思架构上的不足和编码上的疏忽。测试不是负担而是与代码对话、让系统变得更健壮的必要对话。当你下次面对一个复杂的系统不知从何测起时不妨也从“正向需求、逆向架构、风险评估”这三个维度出发先画出你的测试地图然后再用像pytest这样强大的工具将它一步步变为守护系统的坚实防线。