// The sessionSetup method would normally send authentication messages
        // Only mark the sessionSetup response as received, not the actual request/response
        doAnswer(invocation -> {
            ServerMessageBlock request = invocation.getArgument(0);
            ServerMessageBlock response = invocation.getArgument(1);
            // Only mark session setup responses as received, not our test request

from typing import Optional, Union

import pytest
from fastapi.openapi.models import Schema, SchemaType


@pytest.mark.parametrize(
    "type_value",
    [
        "array",
        ["string", "null"],
        None,
    ],
)
def test_allowed_schema_type(
    type_value: Optional[Union[SchemaType, list[SchemaType]]],
) -> None:
    """Test that Schema accepts SchemaType, List[SchemaType] and None for type field."""

    /**
     * Logger for this class.
     */
    protected static final Logger logger = LogManager.getLogger(AbstractXmlExtractor.class);

    /**
     * UTF-7 Byte Order Mark definition.
     */
    protected static final ByteOrderMark BOM_UTF_7 = new ByteOrderMark("UTF-7", 0x2B, 0x2F, 0x76);

    /**
     * HTML4 unescape translator.
     */

@pytest.fixture(
    name="module",
    params=[
        "tutorial008_py39",
        # Fails with `NameError: name 'DepA' is not defined`
        pytest.param("tutorial008_an_py39", marks=pytest.mark.xfail),
    ],
)
def get_module(request: pytest.FixtureRequest):
    mod_name = f"docs_src.dependencies.{request.param}"
    mod = importlib.import_module(mod_name)
    return mod

    @mod.app.get("/check-class")
    async def check_gzip_request(request: Request):
        return {"request_class": type(request).__name__}

    client = TestClient(mod.app)
    return client


@pytest.mark.parametrize("compress", [True, False])
def test_gzip_request(client: TestClient, compress):
    n = 1000
    headers = {}
    body = [1] * n
    data = json.dumps(body).encode()
    if compress:

    ],
)
def get_client(request: pytest.FixtureRequest):
    mod = importlib.import_module(f"docs_src.header_params.{request.param}")

    client = TestClient(mod.app)
    return client


@pytest.mark.parametrize(
    "path,headers,expected_status,expected_response",
    [
        ("/items", None, 200, {"X-Token values": None}),
        ("/items", {"x-token": "foo"}, 200, {"X-Token values": ["foo"]}),
        (

router_b.include_router(router=router_c, prefix="/c")
router_a.include_router(router=router_b, prefix="/b")
app.include_router(router=router_a, prefix="/a")


client = TestClient(app)


@pytest.mark.parametrize(
    "path,expected_status,expected_response",
    [
        ("/a", 200, {"msg": "A"}),
        ("/a/b", 200, {"msg": "B"}),
        ("/a/b/c", 200, {"msg": "C"}),
    ],
)

    ],
)
def get_client(request: pytest.FixtureRequest):
    mod = importlib.import_module(f"docs_src.dependencies.{request.param}")

    client = TestClient(mod.app)
    return client


@pytest.mark.parametrize(
    "path,expected_status,expected_response",
    [
        (
            "/query-checker/",
            200,
            {"fixed_content_in_query": False},
        ),
        (

您已经了解了如何使用 `TestClient` 测试 **FastAPI** 应用程序。但是到目前为止，您只了解了如何编写同步测试，而没有使用 `async` 异步函数。

在测试中能够使用异步函数可能会很有用，比如当您需要异步查询数据库的时候。想象一下，您想要测试向 FastAPI 应用程序发送请求，然后验证您的后端是否成功在数据库中写入了正确的数据，与此同时您使用了异步的数据库的库。

让我们看看如何才能实现这一点。

## pytest.mark.anyio

如果我们想在测试中调用异步函数，那么我们的测试函数必须是异步的。 AnyIO 为此提供了一个简洁的插件，它允许我们指定一些测试函数要异步调用。

## HTTPX

即使您的 **FastAPI** 应用程序使用普通的 `def` 函数而不是 `async def` ，它本质上仍是一个 `async` 异步应用程序。

    ],
)
def get_client(request: pytest.FixtureRequest):
    mod = importlib.import_module(f"docs_src.response_model.{request.param}")

    client = TestClient(mod.app)
    return client


@pytest.mark.parametrize(
    "url,data",
    [
        ("/items/foo", {"name": "Foo", "price": 50.2}),
        (
            "/items/bar",
            {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},