Async API Tests Need Bounded Concurrency, Not Just gather()

Use concurrency only when the scenario allows it

Independent reads across many resources can overlap safely. A sequence that creates an order, pays it, and then reads its status cannot be parallelized without changing the workflow. Before introducing tasks, draw the dependency edges. Only operations with no required ordering should run concurrently.

Concurrency tests and performance tests are also different. A functional scenario with ten concurrent requests may reveal duplicate creation or missing idempotency protection. It does not establish throughput capacity or latency percentiles. Load testing requires controlled traffic generation, measurement, environment ownership, and statistical analysis. Keep claims aligned with what the test actually observes.

Reuse one scoped AsyncClient

HTTPX recommends avoiding repeated creation of clients inside a hot loop because a client owns connection pooling. Reusing one AsyncClient within the test or fixture allows keep-alive connections and centralizes base URL, authentication, timeouts, and limits. Close it through async with or fixture teardown so sockets return cleanly even after an assertion fails.

Do not share a client beyond the lifetime supported by its event loop and test runner. A session-scoped async object can conflict with function-scoped event loops depending on pytest configuration. Scope should follow both mutability and runtime ownership, not merely the desire to save setup time.

Bound work at the task and pool layers

A semaphore limits how many coroutine bodies enter the protected section. HTTPX Limits control maximum open connections and keep-alive connections. They solve related but different problems. A semaphore can protect the target from an enormous task burst; pool limits constrain connection resources. If thousands of tasks are created and all wait on a pool of ten, memory and scheduling overhead still exist even though only ten connections are active.

Choose numbers from the test environment and scenario, not from a generic recommendation. A shared QA service may require very conservative concurrency. A local stub can tolerate more. The goal of a functional suite is repeatability and a meaningful race window, not saturation.

import asyncio
import httpx

async def fetch_orders(order_ids):
    gate = asyncio.Semaphore(5)
    limits = httpx.Limits(max_connections=10, max_keepalive_connections=5)
    timeout = httpx.Timeout(8.0, connect=2.0, pool=1.0)

    async with httpx.AsyncClient(
        base_url="https://api.example.test",
        limits=limits,
        timeout=timeout,
    ) as client:
        async def fetch(order_id):
            async with gate:
                response = await client.get(f"/orders/{order_id}")
                response.raise_for_status()
                return response.json()

        return await asyncio.gather(*(fetch(order_id) for order_id in order_ids))

Make timeout categories and failures visible

A pool timeout means a task could not acquire a connection in time; a connect timeout means establishing a connection took too long; a read timeout concerns receiving response data. Those signals lead to different investigations. Catching every HTTPX exception and returning None destroys that information and can turn an infrastructure failure into a confusing assertion about missing fields.

asyncio.gather returns results in input order, even though requests may finish in another order. By default, the first propagated exception ends the await from the caller's perspective, while other awaitables have their own lifecycle semantics. Decide whether one failure should fail the whole scenario immediately or whether all outcomes must be collected for diagnostics. If collecting exceptions, inspect and fail on them explicitly; never let an exception object masquerade as a successful result. Structured concurrency through TaskGroup can be a clearer choice when supported by the project's Python version and desired cancellation policy.

Assert invariants designed for concurrency

Concurrent completion order is usually nondeterministic. Assertions should compare results by stable identifiers or sets unless order is part of the API contract. For idempotent creation, assert the documented invariant: perhaps one resource exists and repeated requests return the same identifier. Do not assume every response must have the same status unless the contract says so; one request might create while others observe an existing result.

Test-owned data matters even more under concurrency. Generate a unique idempotency key or namespace per test, and do not reuse it across workers. Cleanup should wait until all tasks have completed or been cancelled and should target only those resources. When a failure is timing-sensitive, attach request IDs and sanitized timing metadata so the server logs can be correlated without exposing credentials or personal data.

Keep a synchronous baseline

A small synchronous case is often the best diagnostic baseline. It proves the endpoint and assertion independently of scheduling. Then a focused concurrent case can protect the race-sensitive invariant. Converting every API test to async adds complexity without benefit when each test sends only one request.

Measure suite wall time before and after changes, but interpret it cautiously. Faster local execution may come from connection reuse rather than concurrency, while a remote environment may throttle requests and become slower. The strongest design is explicit: synchronous for simple behavior, asynchronous for genuinely independent waits or concurrency contracts, and dedicated load tooling for capacity questions.