如何在Python中对FastAPI的纯ASGI中间件进行单元测试？

方式一：直接单元测试（模拟ASGI接口） #

这种方式无需依赖FastAPI应用，直接模拟ASGI规范中的scope、receive和send接口，精准验证中间件对请求体的修改逻辑。

测试代码示例 #

import json
import pytest
from starlette.types import Scope, Receive, Send
from your_module import MyMiddleware  # 替换为你的中间件所在模块

async def test_middleware_modifies_post_body():
    # 构造原始请求体
    original_body = json.dumps({"original_key": "original_value"}).encode("utf-8")
    received_messages = [
        {"type": "http.request", "body": original_body, "more_body": False}
    ]
    
    # 模拟receive函数，返回预设的请求消息
    async def mock_receive() -> dict:
        return received_messages.pop(0)
    
    # 模拟send函数，捕获下游应用收到的消息
    sent_messages = []
    async def mock_send(message: dict):
        sent_messages.append(message)
    
    # 构造ASGI scope对象
    scope: Scope = {
        "type": "http",
        "method": "POST",
        "path": "/test",
        "headers": [],
    }
    
    # 构造简单的下游ASGI应用，用于接收中间件处理后的请求
    async def mock_app(scope: Scope, receive: Receive, send: Send):
        message = await receive()
        body = json.loads(message["body"].decode("utf-8"))
        # 返回处理后的请求体作为响应
        await send({
            "type": "http.response.start",
            "status": 200,
            "headers": []
        })
        await send({
            "type": "http.response.body",
            "body": json.dumps(body).encode("utf-8"),
            "more_body": False
        })
    
    # 执行中间件逻辑
    middleware = MyMiddleware(mock_app)
    await middleware(scope, mock_receive, mock_send)
    
    # 验证结果：请求体已添加目标字段
    response_body = json.loads(sent_messages[1]["body"].decode("utf-8"))
    assert response_body["some_field"] == "foobar"
    assert response_body["original_key"] == "original_value"

async def test_middleware_ignores_non_post_requests():
    # 测试非POST请求，中间件不修改请求体
    scope: Scope = {
        "type": "http",
        "method": "GET",
        "path": "/test",
        "headers": [],
    }
    
    received_messages = [{"type": "http.request", "body": b""}]
    async def mock_receive():
        return received_messages.pop(0)
    
    sent_messages = []
    async def mock_send(message):
        sent_messages.append(message)
    
    async def mock_app(scope, receive, send):
        await send({
            "type": "http.response.start",
            "status": 200,
            "headers": []
        })
        await send({
            "type": "http.response.body",
            "body": b"",
            "more_body": False
        })
    
    middleware = MyMiddleware(mock_app)
    await middleware(scope, mock_receive, mock_send)
    
    # 验证请求体未被修改
    assert sent_messages[1]["body"] == b""

async def test_middleware_ignores_empty_body():
    # 测试空请求体场景，中间件不做修改
    scope: Scope = {
        "type": "http",
        "method": "POST",
        "path": "/test",
        "headers": [],
    }
    
    received_messages = [{"type": "http.request", "body": b""}]
    async def mock_receive():
        return received_messages.pop(0)
    
    sent_messages = []
    async def mock_send(message):
        sent_messages.append(message)
    
    async def mock_app(scope, receive, send):
        message = await receive()
        await send({
            "type": "http.response.start",
            "status": 200,
            "headers": []
        })
        await send({
            "type": "http.response.body",
            "body": message["body"],
            "more_body": False
        })
    
    middleware = MyMiddleware(mock_app)
    await middleware(scope, mock_receive, mock_send)
    
    assert sent_messages[1]["body"] == b""

方式二：使用FastAPI TestClient进行集成测试 #

这种方式模拟真实请求流程，通过创建带中间件的FastAPI应用，发送请求并验证返回结果，更贴近生产环境的使用场景。

测试代码示例 #

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient
from your_module import MyMiddleware

app = FastAPI()
app.add_middleware(MyMiddleware)

# 创建一个回显端点，直接返回请求体内容
@app.post("/echo")
async def echo(request: Request):
    body = await request.json()
    return body

client = TestClient(app)

def test_middleware_integration():
    response = client.post("/echo", json={"original_key": "original_value"})
    assert response.status_code == 200
    # 验证返回的请求体包含添加的字段
    assert response.json()["some_field"] == "foobar"
    assert response.json()["original_key"] == "original_value"

def test_integration_ignores_get_request():
    response = client.get("/echo")
    assert response.status_code == 405  # GET方法不符合端点定义，符合预期

def test_integration_empty_body():
    response = client.post("/echo", json={})
    assert response.status_code == 200
    assert response.json()["some_field"] == "foobar"

两种方式的对比 #

• 直接单元测试：聚焦中间件核心逻辑，不依赖FastAPI其他组件，测试速度快，适合单独验证中间件的功能正确性。
• 集成测试：覆盖完整请求链路，能验证中间件与FastAPI应用的配合是否正常，更贴近真实业务场景。

可根据测试需求选择合适的方式，通常建议两种都覆盖，确保中间件的稳定性。

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