Testing Guide

This comprehensive guide covers Zenith’s built-in testing framework designed for modern Python API development.

Overview

Zenith provides a robust testing framework with integrated tools:

Built-in Testing Features:

TestClient: Full async HTTP client with automatic app lifecycle
MockAuth: Skip authentication for testing protected endpoints
TestService: Test business logic without HTTP overhead
Auto-Rollback: Database changes automatically rolled back between tests
Async Support: Full async/await throughout the testing framework
Request Mocking: Mock external APIs and services easily
Performance Testing: Built-in benchmarking and timing utilities
Error Simulation: Test error conditions and edge cases

Integrated Components:

Built-in framework with zero configuration required
Comprehensive testing utilities in a single package

Testing Setup

Zenith Testing Installation

# Zenith testing utilities are built-in
# Optional: Add coverage and data generation tools
uv add --dev pytest-cov                # Coverage reporting (optional)
uv add --dev faker                     # If you need custom fake data

Available Tools:

TestClient for HTTP testing
MockAuth for authentication testing
TestService for business logic testing
Database transaction rollback
Async support throughout
Performance benchmarking
Fixture factories

Configuration

Create pytest.ini:

[tool:pytest]
# Test discovery
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# Async configuration
asyncio_mode = auto

# Coverage settings
addopts =
    --strict-markers
    --verbose
    --cov=app
    --cov-branch
    --cov-report=term-missing:skip-covered
    --cov-report=html
    --cov-report=xml
    --cov-fail-under=80

# Custom markers
markers =
    slow: marks tests as slow (deselect with '-m "not slow"')
    integration: marks tests as integration tests
    unit: marks tests as unit tests
    smoke: marks tests for smoke testing
    performance: marks tests for performance testing

# Environment variables
env =
    TESTING=true
    DATABASE_URL=sqlite+aiosqlite:///:memory:
    REDIS_URL=redis://localhost:6379/15
    SECRET_KEY=test-secret-key

# Warnings
filterwarnings =
    error
    ignore::UserWarning
    ignore::DeprecationWarning

Create pyproject.toml configuration:

[tool.coverage.run]
source = ["app"]
omit = [
    "*/tests/*",
    "*/migrations/*",
    "*/__init__.py",
    "*/config.py"
]

[tool.coverage.report]
precision = 2
show_missing = true
skip_covered = false

[tool.coverage.html]
directory = "htmlcov"

[tool.pytest.ini_options]
minversion = "7.0"
testpaths = ["tests"]

Test Structure

Project Organization

tests/
├── conftest.py           # Shared fixtures
├── factories.py          # Test data factories
├── unit/                 # Unit tests
│   ├── test_models.py
│   ├── test_services.py
│   └── test_utils.py
├── integration/          # Integration tests
│   ├── test_api.py
│   ├── test_auth.py
│   └── test_database.py
├── performance/          # Performance tests
│   └── test_benchmarks.py
└── e2e/                  # End-to-end tests
    └── test_workflows.py

Base Test Fixtures

Create tests/conftest.py:

"""
Shared test fixtures and configuration.

Fixtures defined here are available to all tests.
"""

import asyncio
from typing import AsyncGenerator, Generator
import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, AsyncEngine, create_async_engine
from sqlmodel import SQLModel

from app.main import app
from app.database import get_session
from app.config import settings

# Override settings for testing
settings.TESTING = True
settings.DATABASE_URL = "sqlite+aiosqlite:///:memory:"

# ============= Event Loop Configuration =============

@pytest.fixture(scope="session")
def event_loop() -> Generator:
    """Create an event loop for the test session."""
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    loop.close()

# ============= Database Fixtures =============

@pytest_asyncio.fixture
async def engine() -> AsyncEngine:
    """Create a test database engine."""
    engine = create_async_engine(
        settings.DATABASE_URL,
        echo=False,  # Set to True for SQL debugging
        future=True,
        connect_args={"check_same_thread": False}  # SQLite specific
    )

    # Create all tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield engine

    # Clean up
    await engine.dispose()

@pytest_asyncio.fixture
async def session(engine: AsyncEngine) -> AsyncGenerator[AsyncSession, None]:
    """
    Create a database session for tests.

    Each test gets its own session with automatic rollback.
    """
    async with AsyncSession(engine, expire_on_commit=False) as session:
        async with session.begin():
            yield session
            # Rollback any changes
            await session.rollback()

# ============= Client Fixtures =============

@pytest_asyncio.fixture
async def client(session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    """
    Create a test client with database override.

    This client sends requests to the app with a test database.
    """
    # Override the get_session dependency
    async def override_get_session():
        yield session

    app.dependency_overrides[get_session] = override_get_session

    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

    # Clean up
    app.dependency_overrides.clear()

@pytest_asyncio.fixture
async def authenticated_client(
    client: AsyncClient,
    test_user
) -> AsyncClient:
    """Create a client with authentication headers."""
    from app.auth import create_access_token

    token = create_access_token(
        data={"sub": test_user.email, "user_id": test_user.id}
    )
    client.headers["Authorization"] = f"Bearer {token}"
    return client

# ============= User Fixtures =============

@pytest_asyncio.fixture
async def test_user(session: AsyncSession):
    """Create a test user."""
    from app.models import User
    from app.auth import hash_password

    user = User(
        email="test@example.com",
        username="testuser",
        password_hash=hash_password("testpass123"),
        is_active=True,
        is_verified=True
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

@pytest_asyncio.fixture
async def admin_user(session: AsyncSession):
    """Create an admin user."""
    from app.models import User, UserRole
    from app.auth import hash_password

    user = User(
        email="admin@example.com",
        username="admin",
        password_hash=hash_password("adminpass123"),
        role=UserRole.ADMIN,
        is_active=True,
        is_verified=True
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

# ============= Utility Fixtures =============

@pytest.fixture
def anyio_backend():
    """Configure anyio for async tests."""
    return "asyncio"

@pytest.fixture(autouse=True)
async def reset_database(session: AsyncSession):
    """Reset database state before each test."""
    # This runs before each test automatically
    yield
    # Cleanup runs after test
    await session.rollback()

Unit Testing

Testing Models

import pytest
from datetime import datetime
from app.models import User, Post, ValidationError

class TestUserModel:
    """Test User model functionality."""

    def test_user_creation(self):
        """Test creating a user instance."""
        user = User(
            email="alice@example.com",
            username="alice",
            password_hash="hashed"
        )

        assert user.email == "alice@example.com"
        assert user.username == "alice"
        assert user.is_active is True  # Default value

    def test_user_validation(self):
        """Test model validation."""
        # Invalid email should raise error
        with pytest.raises(ValidationError):
            User(email="invalid", username="test", password_hash="hash")

        # Username too short
        with pytest.raises(ValidationError):
            User(email="test@example.com", username="ab", password_hash="hash")

    async def test_user_save(self, session):
        """Test saving user to database."""
        user = User(
            email="save@example.com",
            username="savetest",
            password_hash="hash"
        )

        session.add(user)
        await session.commit()

        assert user.id is not None
        assert user.created_at is not None

    async def test_user_relationships(self, session):
        """Test user relationships."""
        user = User(
            email="author@example.com",
            username="author",
            password_hash="hash"
        )
        post = Post(title="Test Post", content="Content", author=user)

        session.add(user)
        session.add(post)
        await session.commit()

        await session.refresh(user)
        assert len(user.posts) == 1
        assert user.posts[0].title == "Test Post"

    def test_computed_properties(self):
        """Test model computed properties."""
        user = User(
            email="test@example.com",
            username="testuser",
            first_name="John",
            last_name="Doe",
            password_hash="hash"
        )

        assert user.full_name == "John Doe"
        assert user.display_name == "John Doe"

        # Without full name
        user2 = User(
            email="test2@example.com",
            username="testuser2",
            password_hash="hash"
        )
        assert user2.display_name == "testuser2"

Testing Services

import pytest
from unittest.mock import Mock, AsyncMock, patch
from app.services.users import UserService
from app.models import UserCreate
from app.exceptions import ConflictError, NotFoundError

class TestUserService:
    """Test UserService business logic."""

    @pytest.fixture
    def service(self, session):
        """Create service instance."""
        return UserService(session)

    @pytest.fixture
    def mock_session(self):
        """Create mock session."""
        session = AsyncMock()
        session.exec = AsyncMock()
        session.add = Mock()
        session.commit = AsyncMock()
        session.refresh = AsyncMock()
        return session

    async def test_create_user_success(self, service, session):
        """Test successful user creation."""
        user_data = UserCreate(
            email="new@example.com",
            username="newuser",
            password="SecurePass123"
        )

        user = await service.create_user(user_data)

        assert user.id is not None
        assert user.email == "new@example.com"
        assert user.password_hash != "SecurePass123"  # Should be hashed

    async def test_create_user_duplicate_email(self, service, session, test_user):
        """Test duplicate email rejection."""
        user_data = UserCreate(
            email=test_user.email,  # Duplicate
            username="another",
            password="Password123"
        )

        with pytest.raises(ConflictError) as exc_info:
            await service.create_user(user_data)

        assert "already registered" in str(exc_info.value)

    async def test_get_user_not_found(self, service):
        """Test getting non-existent user."""
        with pytest.raises(NotFoundError):
            await service.get_user(99999)

    @patch('app.services.users.send_email')
    async def test_send_welcome_email(self, mock_send_email, service, session):
        """Test welcome email is sent."""
        mock_send_email.return_value = None

        user_data = UserCreate(
            email="welcome@example.com",
            username="welcome",
            password="Password123"
        )

        user = await service.create_user(user_data)

        # Verify email was called
        mock_send_email.assert_called_once()
        call_args = mock_send_email.call_args[0]
        assert call_args[0] == "welcome@example.com"

    async def test_list_users_pagination(self, service, session):
        """Test user listing with pagination."""
        # Create test users
        for i in range(15):
            user = User(
                email=f"user{i}@example.com",
                username=f"user{i}",
                password_hash="hash"
            )
            session.add(user)
        await session.commit()

        # Test pagination
        page1, total = await service.list_users(skip=0, limit=10)
        assert len(page1) == 10
        assert total == 15

        page2, total = await service.list_users(skip=10, limit=10)
        assert len(page2) == 5

    @pytest.mark.parametrize("password,expected_error", [
        ("short", "at least 8 characters"),
        ("alllowercase", "uppercase letter"),
        ("ALLUPPERCASE", "lowercase letter"),
        ("NoNumbers", "contain a number"),
        ("password", "too common"),
    ])
    def test_password_validation(self, service, password, expected_error):
        """Test password strength validation."""
        with pytest.raises(ValidationError) as exc_info:
            service._validate_password_strength(password)

        assert expected_error in str(exc_info.value)

Testing Utilities

import pytest
from app.utils import (
    generate_slug,
    paginate,
    calculate_hash,
    parse_datetime,
    sanitize_html
)

class TestUtilityFunctions:
    """Test utility functions."""

    @pytest.mark.parametrize("input_text,expected", [
        ("Hello World", "hello-world"),
        ("Python 3.12 Guide", "python-312-guide"),
        ("  Spaces  ", "spaces"),
        ("Special!@#Characters", "specialcharacters"),
        ("", ""),
    ])
    def test_generate_slug(self, input_text, expected):
        """Test slug generation."""
        assert generate_slug(input_text) == expected

    def test_paginate(self):
        """Test pagination helper."""
        items = list(range(100))

        page1 = paginate(items, page=1, per_page=10)
        assert len(page1.items) == 10
        assert page1.total == 100
        assert page1.pages == 10
        assert page1.has_next is True
        assert page1.has_prev is False

        page5 = paginate(items, page=5, per_page=10)
        assert page5.items[0] == 40
        assert page5.has_prev is True

    def test_calculate_hash(self):
        """Test hash calculation."""
        hash1 = calculate_hash("test string")
        hash2 = calculate_hash("test string")
        hash3 = calculate_hash("different")

        assert hash1 == hash2  # Same input, same hash
        assert hash1 != hash3  # Different input, different hash
        assert len(hash1) == 64  # SHA256 length

    @pytest.mark.parametrize("date_string,expected_year", [
        ("2024-01-01", 2024),
        ("2024-01-01T12:00:00", 2024),
        ("2024-01-01T12:00:00Z", 2024),
        ("2024-01-01T12:00:00+00:00", 2024),
    ])
    def test_parse_datetime(self, date_string, expected_year):
        """Test datetime parsing."""
        result = parse_datetime(date_string)
        assert result.year == expected_year

    def test_sanitize_html(self):
        """Test HTML sanitization."""
        dangerous = '<script>alert("XSS")</script><p>Safe content</p>'
        safe = sanitize_html(dangerous)

        assert "<script>" not in safe
        assert "<p>Safe content</p>" in safe

Integration Testing

Testing API Endpoints

import pytest
from httpx import AsyncClient

class TestUserAPI:
    """Test user API endpoints."""

    async def test_create_user_endpoint(self, client: AsyncClient):
        """Test POST /users endpoint."""
        response = await client.post(
            "/users",
            json={
                "email": "api@example.com",
                "username": "apiuser",
                "password": "SecurePass123"
            }
        )

        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "api@example.com"
        assert "password" not in data

    async def test_get_user_endpoint(self, client: AsyncClient, test_user):
        """Test GET /users/{id} endpoint."""
        response = await client.get(f"/users/{test_user.id}")

        assert response.status_code == 200
        data = response.json()
        assert data["id"] == test_user.id
        assert data["email"] == test_user.email

    async def test_list_users_pagination(self, client: AsyncClient, session):
        """Test GET /users with pagination."""
        # Create test data
        for i in range(25):
            user = User(
                email=f"user{i}@example.com",
                username=f"user{i}",
                password_hash="hash"
            )
            session.add(user)
        await session.commit()

        # Test pagination headers
        response = await client.get("/users?page=1&per_page=10")
        assert response.status_code == 200
        assert response.headers["X-Total-Count"] == "25"
        assert response.headers["X-Page"] == "1"
        assert response.headers["X-Per-Page"] == "10"

        data = response.json()
        assert len(data["items"]) == 10

    async def test_unauthorized_access(self, client: AsyncClient):
        """Test unauthorized access to protected endpoint."""
        response = await client.get("/admin/users")
        assert response.status_code == 401
        assert "unauthorized" in response.json()["detail"].lower()

    async def test_rate_limiting(self, client: AsyncClient):
        """Test rate limiting."""
        # Make many requests quickly
        responses = []
        for _ in range(100):
            response = await client.get("/public/data")
            responses.append(response.status_code)

        # Should have some rate limited responses
        assert 429 in responses  # Too Many Requests

    @pytest.mark.parametrize("method", ["GET", "POST", "PUT", "DELETE"])
    async def test_cors_headers(self, client: AsyncClient, method):
        """Test CORS headers are present."""
        response = await client.request(
            method,
            "/api/test",
            headers={"Origin": "http://example.com"}
        )

        assert "Access-Control-Allow-Origin" in response.headers
        assert "Access-Control-Allow-Methods" in response.headers

Testing Authentication

import pytest
from datetime import datetime, timedelta
from jose import jwt

class TestAuthentication:
    """Test authentication flows."""

    async def test_login_success(self, client, test_user):
        """Test successful login."""
        response = await client.post(
            "/auth/login",
            json={
                "email": test_user.email,
                "password": "testpass123"
            }
        )

        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert "refresh_token" in data
        assert data["token_type"] == "bearer"

    async def test_login_wrong_password(self, client, test_user):
        """Test login with wrong password."""
        response = await client.post(
            "/auth/login",
            json={
                "email": test_user.email,
                "password": "wrongpassword"
            }
        )

        assert response.status_code == 401
        assert "invalid" in response.json()["detail"].lower()

    async def test_token_refresh(self, client, test_user):
        """Test refresh token flow."""
        # Login to get tokens
        login_response = await client.post(
            "/auth/login",
            json={
                "email": test_user.email,
                "password": "testpass123"
            }
        )
        refresh_token = login_response.json()["refresh_token"]

        # Refresh access token
        response = await client.post(
            "/auth/refresh",
            json={"refresh_token": refresh_token}
        )

        assert response.status_code == 200
        assert "access_token" in response.json()

    async def test_logout(self, authenticated_client):
        """Test logout."""
        response = await authenticated_client.post("/auth/logout")
        assert response.status_code == 200

        # Token should no longer work
        response = await authenticated_client.get("/users/me")
        assert response.status_code == 401

    async def test_protected_endpoint(self, client, authenticated_client):
        """Test accessing protected endpoints."""
        # Without auth
        response = await client.get("/users/me")
        assert response.status_code == 401

        # With auth
        response = await authenticated_client.get("/users/me")
        assert response.status_code == 200

Testing Database Operations

import pytest
from sqlalchemy import select
from app.models import User, Post, Tag

class TestDatabaseOperations:
    """Test database operations and transactions."""

    async def test_transaction_rollback(self, session):
        """Test transaction rollback on error."""
        user = User(
            email="transaction@example.com",
            username="transaction",
            password_hash="hash"
        )

        session.add(user)

        # Simulate error before commit
        with pytest.raises(Exception):
            await session.commit()
            raise Exception("Simulated error")

        # Rollback should have occurred
        await session.rollback()

        # User should not exist
        stmt = select(User).where(User.email == "transaction@example.com")
        result = await session.exec(stmt)
        assert result.first() is None

    async def test_cascade_delete(self, session):
        """Test cascade delete relationships."""
        user = User(
            email="cascade@example.com",
            username="cascade",
            password_hash="hash"
        )
        post = Post(title="Test", content="Content", author=user)

        session.add(user)
        session.add(post)
        await session.commit()

        # Delete user should cascade to posts
        await session.delete(user)
        await session.commit()

        # Post should be deleted too
        stmt = select(Post).where(Post.title == "Test")
        result = await session.exec(stmt)
        assert result.first() is None

    async def test_many_to_many(self, session):
        """Test many-to-many relationships."""
        post = Post(title="Tagged Post", content="Content")
        tag1 = Tag(name="python")
        tag2 = Tag(name="async")

        post.tags = [tag1, tag2]

        session.add(post)
        await session.commit()

        # Query through relationship
        await session.refresh(post)
        assert len(post.tags) == 2
        assert "python" in [tag.name for tag in post.tags]

    async def test_unique_constraint(self, session):
        """Test unique constraint violation."""
        user1 = User(
            email="unique@example.com",
            username="unique1",
            password_hash="hash"
        )
        user2 = User(
            email="unique@example.com",  # Duplicate
            username="unique2",
            password_hash="hash"
        )

        session.add(user1)
        await session.commit()

        session.add(user2)
        with pytest.raises(IntegrityError):
            await session.commit()

Test Factories

Creating Test Data

import factory
from factory import Faker, SubFactory, LazyAttribute
from datetime import datetime, timedelta
import random

class BaseFactory(factory.Factory):
    """Base factory with common configuration."""

    class Meta:
        abstract = True

    @classmethod
    def _create(cls, model_class, *args, **kwargs):
        """Create instance without saving to database."""
        return model_class(*args, **kwargs)

class UserFactory(BaseFactory):
    """Factory for creating test users."""

    class Meta:
        model = User

    email = Faker("email")
    username = Faker("user_name")
    first_name = Faker("first_name")
    last_name = Faker("last_name")
    password_hash = LazyAttribute(lambda obj: hash_password("testpass123"))
    is_active = True
    is_verified = True
    created_at = Faker("date_time_this_year")

    @factory.post_generation
    def posts(self, create, extracted, **kwargs):
        """Add posts to user if specified."""
        if not create:
            return

        if extracted:
            for post in extracted:
                self.posts.append(post)

    @classmethod
    def create_batch_with_posts(cls, size, posts_per_user=3):
        """Create users with posts."""
        users = []
        for _ in range(size):
            user = cls()
            posts = PostFactory.create_batch(posts_per_user, author=user)
            user.posts = posts
            users.append(user)
        return users

class PostFactory(BaseFactory):
    """Factory for creating test posts."""

    class Meta:
        model = Post

    title = Faker("sentence", nb_words=6)
    content = Faker("text", max_nb_chars=500)
    slug = LazyAttribute(lambda obj: generate_slug(obj.title))
    published = Faker("boolean", chance_of_getting_true=75)
    author = SubFactory(UserFactory)
    created_at = Faker("date_time_this_month")
    view_count = Faker("random_int", min=0, max=10000)

    @classmethod
    def create_published(cls, **kwargs):
        """Create a published post."""
        return cls(published=True, **kwargs)

    @classmethod
    def create_draft(cls, **kwargs):
        """Create a draft post."""
        return cls(published=False, **kwargs)

class CommentFactory(BaseFactory):
    """Factory for creating test comments."""

    class Meta:
        model = Comment

    content = Faker("paragraph", nb_sentences=3)
    post = SubFactory(PostFactory)
    author = SubFactory(UserFactory)
    created_at = Faker("date_time_this_week")
    is_approved = True

    @classmethod
    def create_thread(cls, post, depth=3, width=2):
        """Create a comment thread."""
        comments = []
        for i in range(width):
            parent = cls(post=post)
            comments.append(parent)

            if depth > 1:
                children = cls.create_thread(post, depth-1, width)
                for child in children:
                    child.parent = parent
                comments.extend(children)

        return comments

# Usage examples
def test_factories():
    """Example usage of factories."""
    # Create single user
    user = UserFactory()

    # Create user with specific values
    admin = UserFactory(
        email="admin@example.com",
        role=UserRole.ADMIN
    )

    # Create multiple users
    users = UserFactory.create_batch(10)

    # Create users with posts
    authors = UserFactory.create_batch_with_posts(5, posts_per_user=3)

    # Create published posts
    published_posts = PostFactory.create_batch(10, published=True)

    # Create comment thread
    post = PostFactory()
    comments = CommentFactory.create_thread(post, depth=3, width=2)

Mocking and Stubbing

Mocking External Services

import pytest
from unittest.mock import Mock, AsyncMock, patch
import httpx

class TestExternalServices:
    """Test external service interactions."""

    @patch('httpx.AsyncClient.get')
    async def test_api_call_success(self, mock_get):
        """Test successful API call."""
        # Mock response
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {"status": "success"}
        mock_get.return_value = mock_response

        # Call function that uses external API
        from app.services.external import fetch_data
        result = await fetch_data("https://api.example.com/data")

        assert result["status"] == "success"
        mock_get.assert_called_once_with("https://api.example.com/data")

    @patch('app.services.email.send_email')
    async def test_email_sending(self, mock_send_email):
        """Test email sending without actually sending."""
        mock_send_email.return_value = AsyncMock()

        from app.services.notifications import notify_user
        await notify_user(user_id=1, message="Test notification")

        mock_send_email.assert_called_once()
        call_args = mock_send_email.call_args
        assert "Test notification" in str(call_args)

    @patch('app.services.storage.S3Client')
    async def test_file_upload(self, mock_s3):
        """Test S3 file upload."""
        mock_client = AsyncMock()
        mock_client.upload_file.return_value = "https://s3.example.com/file.pdf"
        mock_s3.return_value = mock_client

        from app.services.files import upload_document
        url = await upload_document("test.pdf", b"file content")

        assert url == "https://s3.example.com/file.pdf"
        mock_client.upload_file.assert_called_once()

    @pytest.fixture
    def mock_redis(self, monkeypatch):
        """Mock Redis client."""
        mock_redis_client = AsyncMock()
        mock_redis_client.get = AsyncMock(return_value=None)
        mock_redis_client.set = AsyncMock(return_value=True)
        mock_redis_client.delete = AsyncMock(return_value=1)

        monkeypatch.setattr(
            "app.cache.redis_client",
            mock_redis_client
        )
        return mock_redis_client

    async def test_caching(self, mock_redis):
        """Test caching with mocked Redis."""
        from app.cache import get_cached, set_cached

        # Cache miss
        result = await get_cached("test_key")
        assert result is None
        mock_redis.get.assert_called_with("test_key")

        # Cache set
        await set_cached("test_key", "test_value", ttl=300)
        mock_redis.set.assert_called_with(
            "test_key",
            "test_value",
            ex=300
        )

Performance Testing

Benchmarking

import pytest
import time
import asyncio
from memory_profiler import profile

class TestPerformance:
    """Performance and benchmark tests."""

    @pytest.mark.benchmark(group="database")
    async def test_database_query_performance(self, benchmark, session):
        """Benchmark database queries."""
        # Setup test data
        users = UserFactory.create_batch(1000)
        for user in users:
            session.add(user)
        await session.commit()

        # Benchmark query
        async def query_users():
            stmt = select(User).where(User.is_active == True)
            result = await session.exec(stmt)
            return result.all()

        result = benchmark(query_users)
        assert len(result) > 0

    @pytest.mark.slow
    async def test_api_throughput(self, client):
        """Test API throughput under load."""
        start_time = time.time()
        tasks = []

        async def make_request():
            response = await client.get("/health")
            return response.status_code

        # Make 100 concurrent requests
        for _ in range(100):
            tasks.append(make_request())

        results = await asyncio.gather(*tasks)
        elapsed = time.time() - start_time

        # All should succeed
        assert all(status == 200 for status in results)

        # Should complete in reasonable time
        assert elapsed < 5  # 100 requests in under 5 seconds
        requests_per_second = 100 / elapsed
        print(f"Throughput: {requests_per_second:.2f} req/s")

    @pytest.mark.parametrize("size", [10, 100, 1000, 10000])
    async def test_scaling(self, size):
        """Test performance with different data sizes."""
        data = list(range(size))

        start = time.perf_counter()
        result = process_data(data)  # Your function
        elapsed = time.perf_counter() - start

        # Performance should scale linearly or better
        expected_max = size * 0.001  # 1ms per item max
        assert elapsed < expected_max

    @profile
    def test_memory_usage(self):
        """Test memory usage."""
        # Create large dataset
        users = UserFactory.create_batch(10000)

        # Process data
        results = []
        for user in users:
            results.append(process_user(user))

        # Memory should be reasonable
        import psutil
        process = psutil.Process()
        memory_mb = process.memory_info().rss / 1024 / 1024
        assert memory_mb < 500  # Less than 500MB

Test Coverage

Coverage Configuration

[run]
branch = True
source = app
omit =
    */tests/*
    */migrations/*
    */__init__.py
    */config.py
    */main.py

[report]
precision = 2
skip_empty = True
show_missing = True

exclude_lines =
    pragma: no cover
    def __repr__
    raise AssertionError
    raise NotImplementedError
    if __name__ == .__main__.:
    if TYPE_CHECKING:
    @abstractmethod
    @abc.abstractmethod

[html]
directory = htmlcov
title = Zenith Test Coverage

[xml]
output = coverage.xml

Running Coverage

# Run tests with coverage
pytest --cov=app --cov-report=html --cov-report=term

# Generate coverage badge
coverage-badge -o coverage.svg -f

# Check coverage thresholds
pytest --cov=app --cov-fail-under=80

# Coverage for specific modules
pytest --cov=app.services tests/unit/test_services.py

CI/CD Integration

GitHub Actions

name: Tests

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    strategy:
      matrix:
        python-version: ["3.11", "3.12"]
        database: [sqlite, postgres]

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: postgres
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
    - uses: actions/checkout@v3

    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}

    - name: Cache dependencies
      uses: actions/cache@v3
      with:
        path: ~/.cache/uv
        key: ${{ runner.os }}-uv-${{ hashFiles('**/pyproject.toml') }}

    - name: Install dependencies
      run: |
        pip install uv
        uv pip install -r requirements-dev.txt

    - name: Run tests
      env:
        DATABASE_URL: ${{ matrix.database == 'postgres' && 'postgresql://postgres:postgres@localhost/test' || 'sqlite:///:memory:' }}
      run: |
        pytest --cov=app --cov-report=xml

    - name: Upload coverage
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml
        fail_ci_if_error: true

Testing Best Practices

1. Test Organization

# Group related tests
class TestUserAuthentication:
    """All authentication-related tests."""

    async def test_login(self): ...
    async def test_logout(self): ...
    async def test_refresh(self): ...

# Use descriptive names
async def test_user_can_update_own_profile_but_not_others(): ...

# Not: async def test_update(): ...

2. Test Independence

# BAD: Tests depend on order
async def test_create_user(self):
    self.user_id = create_user()

async def test_delete_user(self):
    delete_user(self.user_id)  # Depends on previous test

# GOOD: Independent tests
async def test_create_user(self):
    user_id = create_user()
    assert user_id is not None

async def test_delete_user(self):
    user_id = create_user()  # Create own test data
    result = delete_user(user_id)
    assert result is True

3. Fixtures vs Factories

# Use fixtures for common setup
@pytest.fixture
async def authenticated_user(session):
    """User that's logged in for all tests."""
    return await create_and_login_user()

# Use factories for varied test data
def test_pagination():
    users = UserFactory.create_batch(50)  # Different each time

4. Assertion Messages

# Provide context in assertions
assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"

# Use pytest.approx for floats
assert result == pytest.approx(0.1, rel=1e-3)

Common Issues and Solutions

Issue: Async test not running

Solution: Use pytest-asyncio and mark tests with @pytest.mark.asyncio or configure asyncio_mode = auto

Issue: Database state leaking between tests

Solution: Use transactions with rollback, or create new database for each test

Issue: Slow test suite

Solution: Use in-memory database, parallelize with pytest-xdist, mark slow tests

Issue: Flaky tests

Solution: Remove time dependencies, use fixed random seeds, mock external services

Next Steps

Explore Performance Guide for optimization
Learn about Security Guide for secure testing
Check Deployment Guide for production testing

Security Guide

Learn about security testing and best practices Security Guide →

Additional Resources

Need help? Check our FAQ or ask in GitHub Discussions.