如何在pytest测试环境中接收第三方服务触发的Webhook？

核心思路 #

不做模拟，直接在测试环境启动一个临时HTTP服务接收第三方发来的Webhook请求，等请求到达后再执行后续断言，实现完整的流程闭环。

具体实现步骤 #

1. 编写临时Webhook接收器的pytest夹具 #

用aiohttp搭建轻量临时服务，作为Webhook的接收端点，同时保存接收到的请求数据供测试验证：

import asyncio
import pytest
from aiohttp import web

@pytest.fixture
async def webhook_receiver():
    # 存储接收到的Webhook请求
    received_requests = []

    async def handle_webhook(request):
        # 读取请求体与头信息
        body = await request.json()
        received_requests.append({"body": body, "headers": dict(request.headers)})
        return web.Response(status=200)

    # 创建aiohttp应用并添加路由
    app = web.Application()
    app.add_routes([web.post("/webhook", handle_webhook)])

    # 启动临时服务，监听随机端口
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "localhost", 0)  # 0表示自动分配可用端口
    await site.start()

    # 生成临时服务的Webhook地址
    base_url = f"http://localhost:{site._server.sockets[0].getsockname()[1]}"
    webhook_url = f"{base_url}/webhook"

    yield webhook_url, received_requests

    # 测试结束后清理服务
    await runner.cleanup()

2. 在测试函数中使用夹具完成闭环测试 #

拿到临时Webhook的URL后，配置给第三方服务（比如通过环境变量、配置项），发送请求后等待Webhook被接收，最后验证数据与业务结果：

import asyncio
from your_app import app, current_user
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_user(client: AsyncClient, session, webhook_receiver):
    webhook_url, received_requests = webhook_receiver

    # 1. 配置第三方服务的Webhook地址为临时服务URL
    # 根据实际对接方式调整，比如设置环境变量或修改配置
    import os
    os.environ["THIRD_PARTY_WEBHOOK_URL"] = webhook_url

    # 2. 设置当前用户依赖
    app.dependency_overrides[current_user] = lambda: user1

    # 3. 发送请求触发第三方服务流程
    await client.post('/user/')

    # 4. 等待Webhook请求被接收（设置超时避免无限等待）
    try:
        await asyncio.wait_for(
            asyncio.get_event_loop().create_task(
                lambda: asyncio.sleep(0.1) while not received_requests
            ),
            timeout=10  # 超时时间根据第三方响应速度调整
        )
    except asyncio.TimeoutError:
        pytest.fail("未在规定时间内接收到Webhook请求")

    # 5. 验证Webhook内容是否符合预期
    received = received_requests[0]
    assert received["body"]["user_id"] == user1.id
    assert received["headers"].get("X-Signature") == "预期的签名串"  # 有签名验证时添加

    # 6. 验证Webhook处理后的业务结果，比如数据库状态
    db_user = await session.get(User, user1.id)
    assert db_user.status == "synced"

注意事项 #

• 超时时间需根据第三方服务的实际响应速度调整，避免误判
• 如果第三方服务需要Webhook有效性验证（如签名校验），临时接收器要对应实现逻辑
• 若第三方为公网服务，需用内网穿透工具（如ngrok）让公网能访问到本地临时服务

内容的提问来源于stack exchange，提问作者tobias