在Python生态里做BDD测试behave是老牌选手了Gherkin语法写用例圈内基本都认。但用久了你会发现一个痛点步骤函数是同步执行的一旦被测系统里有外部API调用、数据库查询这类I/O操作整个测试就在那干等。最近我做了某跨平台订单系统的接口测试改造把核心步骤全部异步化一批原本跑一个多小时的用例压到十几分钟这篇文章就把完整的做法、背后的原理和踩过的坑一次性说清楚。适合手头测试用例多、执行慢或者正在调研behave异步方案的团队参考。1. 为什么BDD测试需要异步同步步骤的性能瓶颈1.1 测试耗时从哪来先看看同步测试慢在哪。假设一个下单场景的feature文件长这样Given 用户tom已登录 When 用户提交订单SO-2024001 Then 系统返回下单成功 And 数据库中存在订单SO-2024001每个步骤背后都是一个真实操作。登录要调认证接口提交订单要调订单服务校验结果要查一次数据库。单看一次请求正常也就50到300毫秒。但问题在于behave执行步骤是一个接一个串行的而且大多数项目里每个场景之间也串行。20个场景每个场景5个步骤每个步骤平均200毫秒算下来就是20秒纯I/O等待。这还只是单场景单请求的情况。我在某电商项目的实际压测里统计过一个完整的退货流程场景包含查订单、查库存、发起退款、确认退款结果、写审计日志五个步骤全是RPC调用累计耗时超过1.5秒。整个回归套件有140多个场景总共跑下来接近一个半小时。这个时间大家普遍觉得难以接受谁也不想每次提交代码都在CI里等这么久。1.2 behave原生步骤机制的局限behave本身对异步支持非常有限。步骤函数就是普通函数你直接写when(用户提交订单{order_id}) async def submit_order(context, order_id): await api.submit_order(order_id)运行时会直接报错或者产生RuntimeWarningcoroutine was never awaited。原因很简单behave在加载步骤定义后按名称直接调用函数拿到的是一个协程对象没人去await它自然也不会执行里面的逻辑。有人会想到在每个步骤里手动跑事件循环when(用户提交订单{order_id}) def submit_order(context, order_id): asyncio.run(api.submit_order(order_id))单看这个写法能跑但在一个场景里多个步骤都这么干每个asyncio.run都创建和销毁一个事件循环开销不小。更麻烦的是如果业务上需要在一个场景里保持同一个HTTP会话连接或者复用同一个数据库事务上下文这种写法根本做不到。异步的价值不只是单个请求不阻塞更重要的是在同一个事件循环里并发多个请求、复用连接池这两点在步骤级asyncio.run都拿不到。1.3 异步化的核心收益异步改造真正的收益是两条。一条是单步骤内的并发I/O比如一个断言要同时查订单状态、查库存流水、查支付回调三个查询本来串行要400毫秒用asyncio.gather并发一次跑完只要120毫秒。另一条是场景级别的快速执行步骤事件循环统一管理HTTP连接复用数据库连接池复用连接建立的开销被摊薄。我改造完成后做了个简单的基准对比同一套订单系统场景集同步版本平均每个场景耗时8.4秒异步版本平均2.7秒整体回归从85分钟压到28分钟。有些场景收益不明显但涉及多个外部服务调用的场景基本都有3倍以上提升。接下来进入正题讲具体怎么实现。2. 异步步骤的方案选型与前置准备2.1 架构选型绕开Django提供的快速部署路径先说清楚一个误区异步步骤不等于要引入pytest-asyncio或者把behave整个替换掉。behave的步骤机制虽然同步但我们可以通过一个薄封装层让步骤支持协程。核心思路是自定义一个async_step装饰器检测到被装饰的函数是协程函数时在调用阶段负责把协程调度到事件循环里执行并统一管理loop的生命周期。方案上有两条路。一条是步骤内部自己管理事件循环每个步骤独立asyncio.run好处是思路简单依赖少缺点是连接池和上下文无法跨步骤复用。另一条是在environment.py里创建全局事件循环通过context传给步骤步骤内部用asyncio.ensure_future或者loop.run_until_complete调度任务。我在实际项目中选了第二条虽然代码量多一点但收益更大。2.2 环境搭建与依赖安装基础依赖其实很少就behave本身加上aiohttp用于异步HTTP请求asyncpg用于异步PostgreSQL查询。如果被测系统是HTTP服务再装一个pytest做辅助断言也行但不是必须。pip install behave aiohttp asyncpg版本上我用的版本是behave 1.2.6aiohttp 3.9.xasyncpg 0.29.x。这几个库稳定性都还可以。如果你用得比较新注意behave从1.2.7之后好像就不怎么维护了但社区还在对小团队来说影响不大。目录结构按behave的标准来features/ ├── environment.py ├── order.feature └── steps/ ├── async_steps.py ├── order_steps.py └── db_steps.pyenvironment.py放钩子函数steps目录放步骤定义feature文件放Gherkin用例。异步相关的封装可以单独放一个模块也可以直接放environment.py里我习惯放一个独立的async_util.py方便多套件复用。2.3 全局事件循环的初始化环境钩子里初始化事件循环是关键。behave的钩子执行顺序是before_all最先执行此时context已经可用。我在before_all里创建事件循环并存入context确保整个测试进程只用一个loop。# features/environment.py import asyncio from async_util import AsyncStepRunner def before_all(context): loop asyncio.new_event_loop() asyncio.set_event_loop(loop) context.async_runner AsyncStepRunner(loop) def before_scenario(context, scenario): context.http_session None context.order_ids [] def after_scenario(context, scenario): if context.http_session is not None: asyncio.get_event_loop().run_until_complete( context.http_session.close() ) def after_all(context): context.async_runner.close() asyncio.get_event_loop().close()这里有个关键点asyncio.new_event_loop()和asyncio.set_event_loop()要成对出现。如果只set不new在有些环境下会拿到上一次测试残留的loop里面的任务状态、连接缓存都还在容易出奇怪问题。每次测试进程启动的时候新建loop跑完再关闭是一套干净的生命周期。3. 异步步骤的核心实现原理3.1 协程函数与普通函数的调度差异在写装饰器之前先理解一下协程的本质。一个async def定义的协程函数调用时并不执行函数体而是返回一个coroutine对象。只有把这个对象交给事件循环去await函数体才会真正运行。所以我们的装饰器要做的事是判断被装饰函数是什么类型。如果是普通函数直接原样调用如果是协程函数就用一个统一的执行器去调度它。这个调度器要能在步骤运行那个时刻把协程塞进当前事件循环并拿到最终返回值。实现上我去网上翻了各种写法比较干净的是用一个runner对象持有loop提供run和run_async两个接口。步骤装饰器内部根据函数类型决定调用哪个接口。3.2 异步步骤装饰器的两种实现模式先贴第一种也是团队里大多数人用的写法# async_util.py import asyncio import functools def async_step(step_func): if not asyncio.iscoroutinefunction(step_func): return step_func functools.wraps(step_func) def wrapper(context, *args, **kwargs): loop asyncio.get_event_loop() return loop.run_until_complete( step_func(context, *args, **kwargs) ) return wrapper这样步骤里就可以直接用async def写业务逻辑# steps/order_steps.py from async_util import async_step async_step when(用户提交订单{order_id}) async def submit_order(context, order_id): async with context.http_session.post( http://order-api.internal/submit, json{order_id: order_id} ) as resp: assert resp.status 200 body await resp.json() context.order_ids.append(body[order_id])注意装饰器的顺序async_step要放在最外层让它先于when执行这样behave拿到的才是包装后的同步函数。第二种模式是把事件循环的调度逻辑封装成一个基类适用场景是多个步骤需要共享同一个客户端连接池。做法是在before_all里创建aiohttp.ClientSession存入context步骤内部直接用这个session发请求。session本身是异步对象但它内部管理连接池跨多个步骤复用效率提升明显。# steps/order_steps.py async def _get_order(order_id, session): async with session.get( fhttp://order-api.internal/orders/{order_id} ) as resp: return await resp.json() async_step then(订单{order_id}状态为{status}) async def check_order_status(context, order_id, status): order await _get_order(order_id, context.http_session) assert order[status] status, f期望{status},实际{order[status]}第三步跨越多个步骤的共享场景也多多益善后面实操部分我会结合完整案例再说细节。3.3 超时控制与并发组合异步步骤最容易踩的坑是假设不会卡住。同步代码里一个requests.get如果不加timeout默认是等到底层socket超时可能扛住几分钟。异步代码更需要注意因为协程挂起时不占资源一旦某个远端服务一直不返回coroutine可以一直挂几分钟甚至更久整条测试会卡在loop.run_until_complete上。所以每个外部调用都必须加显式超时。aiohttp的ClientSession支持timeout参数from aiohttp import ClientSession, ClientTimeout timeout ClientTimeout(total5) context.http_session ClientSession(timeouttimeout)传输层超时只是一个底线业务层我还会用asyncio.wait_for再次兜底result await asyncio.wait_for( session.get(url), timeout3 )wait_for的好处是只作用于这一处调用不会影响同一个session里其他请求的超时配置。两层加在一起既防止单请求卡死又防止整个场景因为某个接口响应慢而拖垮。并发组合就要用到asyncio.gather。比如准备数据时需要同时创建三个不同的测试前置记录串行得300毫秒并发只要80毫秒results await asyncio.gather( create_order(SO-001), create_user(tom), create_payment(PAY-001), return_exceptionsFalse )gather默认只要其中一个抛异常整个gather立刻抛出但其他任务并不会被取消可能还在后台跑。如果希望一个失败不影响其他断言可以return_exceptionsTrue再逐个检查结果。这个参数在实际测试断言里很有用后面排查章节再展开。4. 完整实操订单系统接口测试异步化改造4.1 被测试系统与场景设计我改造的是某跨平台订单系统的回归套件。系统结构比较简单前端调订单API订单服务落库同时调支付服务和库存服务。测试的feature文件设计成一张场景表覆盖下单、支付回调、库存扣减、订单查询四个核心链路。feature文件里我用了Scenario Outline方便在一条用例里跑多组数据# features/order.feature Feature: 订单核心链路 Scenario Outline: 下单并支付 Given 用户user已登录 When 用户提交订单order_id And 用户完成支付 Then 系统返回支付成功 And 数据库订单状态为PAID Examples: | user | order_id | | tom | SO-2024001 | | jerry | SO-2024002 |每个步骤背后至少一次RPC调用这个场景在同步版本里平均跑1.2秒异步版本压到0.4秒。后面所有步骤定义都以这个feature为例展开。4.2 登录态与HTTP会话的建立登录是第一个步骤。这部分如果每次重登完整走一遍认证接口第二个场景又要重新走浪费明显。我做法是第一个场景登录后把token存到context里后续场景如果token还有效就复用。为了简单可靠实际上我用了更直接的方式在before_all阶段一次性获取token全部场景共享。# steps/order_steps.py from async_util import async_step async_step given(用户{user}已登录) async def user_logged_in(context, user): if context.token: return async with context.http_session.post( http://order-api.internal/auth, json{username: user, password: test-password} ) as resp: assert resp.status 200 body await resp.json() context.token body[token] context.http_session.headers.update( {Authorization: fBearer {context.token}} )这里有个容易忽略的细节如果多个场景都跑同一个Given步骤每个场景执行到这一步都会调用一次user_logged_in。而context在场景之间默认不共享token我是在environment.py的before_scenario里通过context.token getattr(context, token, None)保留上一场景的token。后来想更严谨一点直接把token的获取放到before_all阶段场景内步骤只负责设置Authorization头这样登录请求在整个测试套件只发生一次。这也是异步改造能提速的隐藏福利同步代码里每次登录是一次完整HTTP往返异步化之后连登录都只需要一次。4.3 步骤内部并发叫接口下单和支付两个步骤同步版本是串行等两个接口。改造后我在支付步骤里同时请求支付API和查询订单服务两个结果都拿到后一起断言。这里有个权衡支付步骤本身的业务含义就是调用支付接口额外并发查订单状态并不会影响业务正确性因为支付操作在单独的协程里执行。async_step when(用户提交订单{order_id}) async def submit_order(context, order_id): async def _do_submit(): async with context.http_session.post( http://order-api.internal/order, json{order_id: order_id} ) as resp: return resp.status, await resp.text() async def _do_sanity_check(): async with context.http_session.get( fhttp://order-api.internal/sandbox/order/{order_id} ) as resp: return resp.status status_code, text await asyncio.gather( _do_submit(), _do_sanity_check() ) assert status_code 201, f下单失败: {text}4.4 数据库校验步骤的异步查询数据库步骤是性能瓶颈最明显的地方。同步用psycopg2连PostgreSQL每次查询一个连接在CI环境里连接池经常不够用偶尔还报too many connections。换成asyncpg后连接池复用一个连接查询耗时从120毫秒降到30毫秒左右。# steps/db_steps.py import asyncpg from async_util import async_step async_step then(数据库订单状态为{status}) async def check_db_order_status(context, status): pool context.db_pool async with pool.acquire() as conn: row await conn.fetchrow( SELECT status FROM orders WHERE order_id $1, SO-2024001 ) assert row[status] status, f数据库状态为{row[status]}在environment.py的before_all里初始化连接池after_all里关闭async def _create_db_pool(): return await asyncpg.create_pool( hosttest-db.internal, databaseorder_test, usertester, passwordtest-password, min_size2, max_size10, ) def before_all(context): loop asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete( _setup_db_pool(context) )这里有个细节run_until_complete会在全部协程结束前阻塞事件循环如果pool的创建本身比较快用run_until_complete没问题。asyncpg的create_pool内部会建立多个连接如果网络环境慢这里可能会卡几分钟所以我额外加了asyncio.wait_for包一层超时设为15秒。4.5 完整步骤文件长这样把上面几块拼起来一个完整的steps/order_steps.py如下import asyncio from behave import given, when, then from async_util import async_step async_step given(用户{user}已登录) async def user_logged_in(context, user): if context.token: return async with context.http_session.post( http://order-api.internal/auth, json{username: user, password: test-password} ) as resp: assert resp.status 200 body await resp.json() context.token body[token] context.http_session.headers.update( {Authorization: fBearer {context.token}} ) async_step when(用户提交订单{order_id}) async def submit_order(context, order_id): async with context.http_session.post( http://order-api.internal/order, json{order_id: order_id} ) as resp: assert resp.status 201 body await resp.json() context.order_ids.append(body[order_id]) async_step when(用户完成支付) async def pay_order(context): if not context.order_ids: raise RuntimeError(没有可用订单步骤执行顺序有误) order_id context.order_ids[-1] async with context.http_session.post( http://order-api.internal/pay, json{order_id: order_id} ) as resp: assert resp.status 200 async_step then(数据库订单状态为{status}) async def check_db_status(context, status): order_id context.order_ids[-1] async with context.db_pool.acquire() as conn: row await conn.fetchrow( SELECT status FROM orders WHERE order_id$1, order_id ) assert row[status] status这个文件已经可以在behave里跑了。你可以把代码里订单接口域名、数据库连接串替换成自己mock的地址先跑通一遍再迁移到真实被测服务上。5. 常见问题与排查技巧实录5.1 事件循环冲突RuntimeError: Event loop is closed出现这个报错的频率很高常见场景上一个测试进程退出时没关闭loop下一个进程又尝试asyncio.get_event_loop()拿到一个半关闭状态的loop。排查方法是在environment.py里确认before_all创建新loop、after_all关闭不要依赖behave内部默认的loop。还有一个隐蔽场景某些版本的behave会调用pytest的fixture或者allure报告插件这些第三方库内部可能创建自己的loop发生嵌套。一旦发现报错和第三方库有关最省事的是把第三方库的调用封装成同步子进程或者升级到最新版。5.2 协程从未被await测试静默通过这是最坑的问题。步骤函数里写了async def但不小心漏了await函数体不会抛错behave只是拿到一个coroutine然后这个coroutine被丢弃步骤直接通过。等到后续步骤去查数据库时发现数据根本没写入才暴露问题。排查办法加一个告警过滤器把所有未等待的协程打印出来。import warnings def before_all(context): warnings.filterwarnings(always, categoryRuntimeWarning)跑测试时如果控制台出现RuntimeWarning: coroutine xxx was never awaited说明哪一步漏了await。通常漏在哪要看上下文你调用了异步函数但没写await或者调用了返回协程的库函数但没有await它。仔细检查每一处调用别偷懒。5.3 多个步骤之间状态隔离与传递behave默认的context是单个对象一个场景内的步骤共享context属性没问题但场景之间默认不隔离。我遇到过的问题场景A在context.order_ids里append了一个ID场景B执行时没有清空导致断言时拿到的是场景A的订单号整个测试结果错乱。解决方法是environment.py里before_scenario重置所有需要隔离的属性。def before_scenario(context, scenario): context.order_ids [] context.token getattr(context, token, None)token不清是因为跨场景复用order_ids必须清是因为每个场景独立。具体哪些清哪些不清需要结合业务判断没有通用答案。我的原则业务结果类数据一定要清认证凭据类复用。还有一种更清晰的方案是context本身支持嵌套就不展开了我项目里用属性命名区分也觉得够用。5.4 数据库连接池受限asyncpg默认连接池max_size如果配得太小比如2多个步骤并发查数据库时会排队。症状是某个数据库步骤莫名其妙的慢从30毫秒涨到300毫秒。排查方法是在db步骤里打印pool.stat()查看当前连接数如果一直跑满max_size说明并发度超了调大池子或者降低步骤并发。还有一种情况是连接泄漏没正常释放池子被占满。检查acquire()有没有用async with或者有没有在finally里release。5.5 gather里一个失败导致断言异常场景asyncio.gather并发三个请求其中一个接口500了其他两个正常返回。默认return_exceptionsFalse时gather会立刻抛异常但其他两个请求还在后台跑。如果这个步骤里还有断言异常会导致断言没执行行为上看起来像测试失败但失败原因不对。调试技巧是临时改用return_exceptionsTrue打印所有结果快速定位是哪个接口出错results await asyncio.gather( fetch_order(order_id), fetch_payment(order_id), return_exceptionsTrue ) print(results)定位后还是把return_exceptions改回False因为正常测试中一个失败就应该让整个场景失败吞异常反而掩盖问题。6. 性能调优与最佳实践6.1 并发度与限制异步化之后最大的风险是并发过高把被测系统打挂。测试环境通常资源有限如果10个场景同时向外发请求下游服务扛不住反而拖慢整体速度。我在订单系统测试里做了一组实验并发场景数从1调到20最明显的收益在5左右再往上收益很小而且偶发连接超时变多。控制并发度可以用asyncio.Semaphoresemaphore asyncio.Semaphore(5) async def limited_request(url): async with semaphore: async with context.http_session.get(url) as resp: return await resp.json()别忘了Semaphore本身要放进context跨场景复用。如果每个场景都重新创建信号量就失效了。6.2 资源生命周期管理HTTP session和数据库连接池是异步化收益的两个关键资源管理好它们性能才有保障。session创建时设置合理的超时和连接池大小不要用默认值。aiohttp的TCPConnector要限制总连接数默认是无限制的测试并发高的时候会创建大量socket超过系统文件描述符限制。from aiohttp import TCPConnector, ClientSession, ClientTimeout connector TCPConnector( limit50, ttl_dns_cache300 ) session ClientSession( connectorconnector, timeoutClientTimeout(total10) )after_all里要依次关闭session和连接池顺序反了容易报ConnectionResetError。先关session再关pool最后关loop。6.3 CI集成与报告输出CI里跑异步behave关键是要开junit报告方便和流水线集成。命令很简单behave -f junit -o reports/ features/如果你用gitlab或者jenkins把reports目录交给对应的报告插件解析就行。还有个小细节CI环境里并发度和本地不一样建议通过环境变量控制信号量大小。我在environment.py里是这么读的import os def before_all(context): concurrency int(os.getenv(BDD_CONCURRENCY, 5)) context.semaphore asyncio.Semaphore(concurrency)本地调大CI调小避免把测试环境搞崩。6.4 踩过坑后的经验总结最后分享几条压箱底的经验。第一异步步骤的装饰器一定要放在最外层。谁放在外面谁说了算一旦放错behave拿到的是协程而不是函数报错非常难排查。第二任何外部I/O都要有超时。没有超时的异步请求是个定时炸弹可能挂住整个测试进程CI卡死到超时才知道是哪一个步骤。宁可超时设短一点然后重试也不要设成无限等。第三异步化的改造不是一蹴而就的。比较稳妥的方式是先把高频、I/O密集的步骤改造成async_step其他步骤保持同步跑通后再逐步扩大范围。一次全改容易引入一堆并发问题调试成本反而高。第四我不建议为了异步而异步。如果被测系统纯内存计算步骤本身没有I/O硬套异步没收益还增加心智负担。异步化解决的是等I/O的问题不是快计算的问题方向上别搞反。我在实际项目中还养成了一个习惯每个feature文件跑完后把耗时最高的场景打印出来持续监控。异步步骤能提多少速不同系统差很多通过对比才有数。回到最初那个问题behave异步步骤到底值不值得做。我的答案很明确只要测试里有真实的外部服务交互值得做而且收益不是小数目。它不会让测试从逻辑上变聪明但能让同样多用例在更短时间里跑完对团队来说省下的是实实在在的等待时间。这套做法目前在我参与的订单、商品、库存三条业务线里都跑得不错后续如果能跟pytest的异步插件打通应该还能再进一步优化场景级并发那是另一个话题了。