Async Testing Interview Questions & Answers

TestClient is synchronous — it works for most FastAPI tests. But some scenarios require true async tests:

• Testing async dependencies directly (e.g., get_async_db).
• Using async database fixtures (creating tables, seeding data).
• Testing WebSocket connections with AsyncClient.
• Testing code that uses asyncio.gather or asyncio.create_task.

Plain pytest doesn't run async def test functions. You need a plugin:

pip install anyio[trio] pytest-anyio   # recommended with FastAPI/Starlette
# OR
pip install pytest-asyncio

Rule of thumb: use TestClient (sync) for HTTP endpoint tests — it's simpler; use async tests only when you genuinely need to await inside the test body.

Mark async tests with @pytest.mark.anyio or configure anyio as the default async mode:

# conftest.py — set anyio as default for the whole session
import pytest

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

# test_async.py
import pytest

@pytest.mark.anyio
async def test_async_endpoint():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get("/items")
        assert response.status_code == 200

Rule of thumb: set anyio_backend = "asyncio" globally in conftest.py to avoid repeating the marker on every async test.

Use ASGITransport to route requests directly to the ASGI app without a network:

import pytest
import httpx
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.mark.anyio
async def test_get_item():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver",
    ) as client:
        response = await client.get("/items/1")
        assert response.status_code == 200
        assert response.json()["id"] == 1

ASGITransport calls the app's ASGI interface directly — no TCP socket needed.

Rule of thumb: use AsyncClient for async tests; use TestClient for sync tests — AsyncClient requires async with and an event loop.

Set headers= in the AsyncClient constructor or use an httpx.Auth class:

@pytest.mark.anyio
async def test_authenticated():
    token = create_test_token(user_id=1)
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver",
        headers={"Authorization": f"Bearer {token}"},
    ) as client:
        resp = await client.get("/me")
        assert resp.status_code == 200
        resp2 = await client.get("/orders")
        assert resp2.status_code == 200

Rule of thumb: set auth headers in the constructor for tests that make multiple authenticated requests; pass headers= per-request when mixing auth states.

# conftest.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.db.base import Base
from app.main import app
from app.db.session import get_async_db

TEST_DB = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture(scope="session")
async def db_engine():
    engine = create_async_engine(TEST_DB)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest.fixture
async def db_session(db_engine):
    factory = async_sessionmaker(db_engine, expire_on_commit=False)
    async with factory() as session:
        yield session
        await session.rollback()   # undo test writes

@pytest.fixture
async def async_client(db_session):
    async def override_db():
        yield db_session
    app.dependency_overrides[get_async_db] = override_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

Rule of thumb: roll back after each test (await session.rollback()) rather than truncating tables — rollback is faster and avoids FK constraint issues.

Use TestClient with the websocket_connect() context manager:

from fastapi.testclient import TestClient

def test_websocket_echo():
    with TestClient(app) as client:
        with client.websocket_connect("/ws") as ws:
            ws.send_text("Hello")
            data = ws.receive_text()
            assert data == "Echo: Hello"

For async WebSocket tests with httpx:

@pytest.mark.anyio
async def test_ws_async():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        async with client.websocket_connect("/ws") as ws:
            await ws.send_text("ping")
            msg = await ws.receive_text()
            assert msg == "pong"

Rule of thumb: use TestClient.websocket_connect for simple sync WS tests; use async AsyncClient when the WS handler is complex and needs concurrent I/O.

Each pytest scope (function, module, session) can have its own event loop. By default, each async test gets a fresh event loop (scope="function").

This matters because:

• scope="session" async fixtures (like a DB engine) must share an event loop with the tests that use them.
• Mixing scopes can cause "Event loop is closed" or "attached to different loop" errors.

With pytest-anyio, configure the scope:

# conftest.py
@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

This uses a session-scoped event loop — all async fixtures and tests share one loop.

Rule of thumb: set scope="session" for anyio_backend when you have session-scoped async fixtures (DB engines, HTTP clients) — otherwise they can't be awaited from function-scoped tests.

Yes — FastAPI is built on Starlette which uses anyio, supporting both asyncio and Trio. Change the backend in conftest:

@pytest.fixture(scope="session")
def anyio_backend():
    return "trio"   # or "asyncio"

And mark tests:

@pytest.mark.anyio
async def test_endpoint():
    async with AsyncClient(...) as client:
        ...

Running on Trio can catch asyncio-specific bugs (e.g., code that relies on asyncio internals rather than anyio primitives).

Rule of thumb: run your test suite on both asyncio and Trio once in CI to catch portability issues; ship on asyncio for production (most libraries are asyncio-native).

Use unittest.mock.AsyncMock or override the dependency:

from unittest.mock import AsyncMock, patch

# Option A — patch the async function directly
@pytest.mark.anyio
async def test_with_mock():
    with patch("app.services.email.send_email", new_callable=AsyncMock) as mock_email:
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
            resp = await c.post("/signup", json={"email": "a@b.com"})
        assert resp.status_code == 201
        mock_email.assert_called_once_with("a@b.com", "Welcome!")

# Option B — dependency override
async def fake_send_email(*args, **kwargs):
    pass   # no-op

app.dependency_overrides[send_email_dep] = lambda: fake_send_email

Rule of thumb: use dependency_overrides for FastAPI-injected async deps; use AsyncMock + patch for module-level async functions not in the DI graph.

Verify it was scheduled — check the response immediately, background not run yet:

def test_signup_queues_email(client):
    resp = client.post("/signup", json={"email": "a@b.com"})
    assert resp.status_code == 201
    # email runs after response — we can't assert it ran synchronously

Verify it ran — TestClient runs background tasks before closing:

emails_sent = []

def fake_send_email(to: str, msg: str):
    emails_sent.append(to)

app.dependency_overrides[get_email_sender] = lambda: fake_send_email

def test_email_sent(client):
    with TestClient(app) as c:    # lifespan + BG tasks run
        c.post("/signup", json={"email": "a@b.com"})
    assert "a@b.com" in emails_sent

Rule of thumb: TestClient (sync) runs background tasks before __exit__ — use it to verify task execution; for async tests you may need to await explicitly.

Use syrupy (snapshot testing library) or assert field-by-field:

# Field-by-field (explicit)
def test_item_response(client):
    resp = client.get("/items/1")
    data = resp.json()
    assert data.keys() == {"id", "name", "price", "created_at"}
    assert isinstance(data["id"], int)
    assert isinstance(data["price"], float)

# Snapshot (syrupy)
def test_item_snapshot(client, snapshot):
    resp = client.get("/items/1")
    assert resp.json() == snapshot   # first run saves; subsequent runs compare

Rule of thumb: explicit field assertions are more maintainable than full JSON snapshots for frequently changing response shapes — snapshots are great for stable, complex nested responses.