Python Async Testing Skill

Python Async Testing

Problem Statement

Async testing requires specific patterns. pytest-asyncio has modes that affect behavior. Database tests need isolation. Mocking async functions differs from sync. Get these wrong and tests are flaky or don't catch bugs.

Pattern: pytest-asyncio Configuration

Problem: Pytest needs configuration for async tests.

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "strict"  # Requires explicit @pytest.mark.asyncio
# OR
asyncio_mode = "auto"    # All async tests run automatically

import pytest

# With asyncio_mode = "strict" (this codebase)
@pytest.mark.asyncio
async def test_something():
    result = await some_async_function()
    assert result == expected

# Without the marker = test won't run as async!

Pattern: Async Fixtures

Problem: Fixtures that provide async resources need specific handling.

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

# ✅ CORRECT: Async fixture for session
@pytest.fixture
async def session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session
        await session.rollback()  # Clean up after test

# ✅ CORRECT: Async fixture for test data
@pytest.fixture
async def test_user(session: AsyncSession) -> User:
    user = User(email="test@example.com", hashed_password="...")
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

# ✅ CORRECT: Using async fixtures
@pytest.mark.asyncio
async def test_get_user(session: AsyncSession, test_user: User):
    result = await session.execute(
        select(User).where(User.id == test_user.id)
    )
    user = result.scalar_one()
    assert user.email == "test@example.com"

Pattern: Database Test Isolation

Problem: Tests polluting each other's database state.

# ✅ CORRECT: Transaction rollback per test
@pytest.fixture
async def session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        # Start a transaction
        async with session.begin():
            yield session
            # Rollback happens automatically when we exit

# ✅ CORRECT: Nested transactions for complex tests
@pytest.fixture
async def session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        await session.begin()
        yield session
        await session.rollback()

# Alternative: Use separate test database
# conftest.py
@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def test_engine():
    # Use SQLite for tests, PostgreSQL for prod
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield engine
    await engine.dispose()

Pattern: Mocking Async Functions

Problem: Regular Mock doesn't work with async functions.

from unittest.mock import AsyncMock, patch

# ✅ CORRECT: AsyncMock for async functions
@pytest.mark.asyncio
async def test_with_mocked_service():
    mock_service = AsyncMock()
    mock_service.get_user.return_value = User(id=uuid4(), email="test@example.com")
    
    result = await mock_service.get_user(user_id)
    
    assert result.email == "test@example.com"
    mock_service.get_user.assert_called_once_with(user_id)

# ✅ CORRECT: Patching async functions
@pytest.mark.asyncio
@patch("app.services.user_service.send_email", new_callable=AsyncMock)
async def test_user_creation_sends_email(mock_send_email: AsyncMock, session: AsyncSession):
    mock_send_email.return_value = True
    
    user = await create_user(email="new@example.com", session=session)
    
    mock_send_email.assert_called_once_with(user.email, "Welcome!")

# ✅ CORRECT: AsyncMock with side_effect
mock_service = AsyncMock()
mock_service.get_user.side_effect = [
    User(id=uuid4(), email="first@example.com"),
    User(id=uuid4(), email="second@example.com"),
]

# First call returns first user, second call returns second user

Pattern: HTTP Client Testing

Problem: Testing FastAPI endpoints with async client.

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

@pytest.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client

@pytest.mark.asyncio
async def test_get_users(client: AsyncClient):
    response = await client.get("/api/users")
    
    assert response.status_code == 200
    data = response.json()
    assert isinstance(data, list)

@pytest.mark.asyncio
async def test_create_assessment(client: AsyncClient, auth_headers: dict):
    response = await client.post(
        "/api/assessments",
        json={"title": "Test Assessment", "skill_areas": ["fundamentals"]},
        headers=auth_headers,
    )
    
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Test Assessment"

# ✅ CORRECT: Auth fixture
@pytest.fixture
async def auth_headers(test_user: User) -> dict:
    token = create_access_token(user_id=test_user.id)
    return {"Authorization": f"Bearer {token}"}

Pattern: Testing Service Functions

@pytest.mark.asyncio
async def test_calculate_rating(session: AsyncSession, test_user: User):
    # Arrange: Create test data
    assessment = Assessment(user_id=test_user.id, title="Test")
    session.add(assessment)
    await session.commit()
    
    answers = [
        UserAnswer(user_id=test_user.id, question_id=q_id, value=4)
        for q_id in question_ids
    ]
    session.add_all(answers)
    await session.commit()
    
    # Act: Call the service
    result = await calculate_rating(assessment.id, session)
    
    # Assert: Check the result
    assert result.rating >= 1.0
    assert result.rating <= 5.5
    assert result.confidence > 0

@pytest.mark.asyncio
async def test_calculate_rating_no_answers(session: AsyncSession, test_user: User):
    assessment = Assessment(user_id=test_user.id, title="Empty")
    session.add(assessment)
    await session.commit()
    
    # Should raise or return specific result
    with pytest.raises(ValueError, match="No answers found"):
        await calculate_rating(assessment.id, session)

Pattern: Testing Multi-Step Flows

Same principle as frontend - test entire flows, not just units:

@pytest.mark.asyncio
async def test_complete_assessment_flow(session: AsyncSession, test_user: User):
    """Test full assessment flow: create -> answer -> submit -> results."""
    
    # Step 1: Create assessment
    assessment = await create_assessment(
        user_id=test_user.id,
        data=AssessmentCreate(title="Full Flow Test", skill_areas=["fundamentals"]),
        session=session,
    )
    assert assessment.id is not None
    
    # Step 2: Answer questions
    questions = await get_assessment_questions(assessment.id, session)
    for question in questions:
        await submit_answer(
            user_id=test_user.id,
            question_id=question.id,
            value=4,
            session=session,
        )
    
    # Step 3: Submit assessment
    result = await submit_assessment(assessment.id, session)
    assert result.status == "completed"
    
    # Step 4: Verify results
    rating = await get_assessment_rating(assessment.id, session)
    assert rating is not None
    assert rating.skill_area == "fundamentals"

Pattern: Fixture Scopes

Problem: Understanding when fixtures are recreated.

# function (default) - recreated for each test
@pytest.fixture
async def session():
    ...  # New session per test

# class - shared within test class
@pytest.fixture(scope="class")
async def shared_data():
    ...  # Created once per test class

# module - shared within test file
@pytest.fixture(scope="module")
async def module_setup():
    ...  # Created once per file

# session - shared across entire test run
@pytest.fixture(scope="session")
async def database():
    ...  # Created once, used by all tests

Best practices:

Database sessions: function scope (isolation)
Test data: function scope (clean state)
Database engine: session scope (expensive to create)

Pattern: Testing Error Cases

@pytest.mark.asyncio
async def test_get_nonexistent_user(session: AsyncSession):
    fake_id = uuid4()
    
    with pytest.raises(HTTPException) as exc_info:
        await get_user_or_404(fake_id, session)
    
    assert exc_info.value.status_code == 404
    assert str(fake_id) in exc_info.value.detail

@pytest.mark.asyncio
async def test_duplicate_email_rejected(session: AsyncSession, test_user: User):
    with pytest.raises(IntegrityError):
        duplicate = User(email=test_user.email, hashed_password="...")
        session.add(duplicate)
        await session.commit()

Pattern: Parameterized Tests

@pytest.mark.asyncio
@pytest.mark.parametrize("skill_area,expected_min,expected_max", [
    ("fundamentals", 1.0, 5.5),
    ("advanced", 1.0, 5.5),
    ("strategy", 1.0, 5.5),
])
async def test_rating_ranges(
    skill_area: str,
    expected_min: float,
    expected_max: float,
    session: AsyncSession,
):
    rating = await calculate_rating_for_area(skill_area, session)
    assert expected_min <= rating <= expected_max

@pytest.mark.asyncio
@pytest.mark.parametrize("invalid_input", [
    {"title": ""},  # Empty title
    {"title": "x" * 201},  # Too long
    {"skill_areas": []},  # Empty areas
])
async def test_assessment_validation(invalid_input: dict, client: AsyncClient):
    response = await client.post("/api/assessments", json=invalid_input)
    assert response.status_code == 422

Common Issues

| Issue | Likely Cause | Solution | |-------|--------------|----------| | "coroutine was never awaited" | Missing await in test | Add await | | Test not running async | Missing @pytest.mark.asyncio | Add marker or use asyncio_mode = "auto" | | Tests polluting each other | Missing rollback | Use transaction fixture with rollback | | "Event loop is closed" | Fixture scope mismatch | Check scope on async fixtures | | Mock not working | Using Mock instead of AsyncMock | Use AsyncMock for async |

Test Commands

# Run all tests
uv run pytest

# Verbose output
uv run pytest -v

# Specific file
uv run pytest tests/test_assessments.py

# Specific test
uv run pytest tests/test_assessments.py::test_create_assessment

# With coverage
uv run pytest --cov=app --cov-report=html

# Stop on first failure
uv run pytest -x

# Show print output
uv run pytest -s

Agent Skills: Python Async Testing

Install this agent skill to your local

Skill Files