Python Pytest Framework: Advanced Testing Patterns for CI/CD Automation

Posted on August 25, 2024 • 6 minutes • 1121 words

Introduction

Writing tests is one thing. Writing maintainable, scalable tests that work seamlessly in CI/CD pipelines is another.

Pytest has become the gold standard for Python testing because of its flexibility, powerful fixtures, and plugin ecosystem. In this article, I’ll share advanced patterns I’ve used to build enterprise-grade test suites.

Why Pytest?

Pytest excels at:

✅ Simple syntax (plain assert statements)
✅ Powerful fixtures for setup/teardown
✅ Parametrization for testing multiple scenarios
✅ Rich plugin ecosystem
✅ Detailed failure reports
✅ Easy integration with CI/CD

Advanced Fixtures

1. Database Fixtures with Scope Control

# conftest.py
import pytest
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope="session")
def db_engine():
    """Create database engine (session scope = run once per test session)"""
    engine = create_engine("postgresql://user:password@localhost/testdb")
    
    # Create all tables
    Base.metadata.create_all(engine)
    
    yield engine
    
    # Cleanup
    Base.metadata.drop_all(engine)

@pytest.fixture(scope="function")
def db_session(db_engine):
    """Create database session (function scope = run for each test)"""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()
    
    yield session
    
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def sample_user(db_session):
    """Create a sample user for testing"""
    user = User(
        username="testuser",
        email="test@example.com",
        password_hash="hashed_password"
    )
    db_session.add(user)
    db_session.commit()
    
    yield user
    
    db_session.delete(user)
    db_session.commit()

2. Request Mocking Fixtures

# conftest.py
import pytest
from unittest.mock import Mock, patch
from requests_mock import Mocker

@pytest.fixture
def mock_http_requests():
    """Mock HTTP requests globally"""
    with Mocker() as m:
        m.get('https://api.example.com/users', json=[
            {'id': 1, 'name': 'John'},
            {'id': 2, 'name': 'Jane'}
        ])
        m.post('https://api.example.com/users', json={'id': 3, 'name': 'Bob'})
        yield m

@pytest.fixture
def aws_mock():
    """Mock AWS services"""
    with patch('boto3.client') as mock_client:
        mock_s3 = Mock()
        mock_client.return_value = mock_s3
        
        mock_s3.get_object.return_value = {
            'Body': Mock(read=Mock(return_value=b'file content'))
        }
        
        yield mock_s3

3. Environment Variable Fixtures

# conftest.py
import pytest
import os

@pytest.fixture
def env_vars(monkeypatch):
    """Set environment variables for testing"""
    env_config = {
        'DATABASE_URL': 'postgresql://localhost/testdb',
        'API_KEY': 'test_api_key_12345',
        'DEBUG': 'True',
        'LOG_LEVEL': 'DEBUG'
    }
    
    for key, value in env_config.items():
        monkeypatch.setenv(key, value)
    
    yield env_config

Parametrization: Testing Multiple Scenarios

1. Basic Parametrization

# tests/test_validators.py
import pytest
from validators import validate_email

@pytest.mark.parametrize("email,expected", [
    ("valid@example.com", True),
    ("invalid.email.com", False),
    ("user+tag@domain.co.uk", True),
    ("", False),
    ("@example.com", False),
])
def test_email_validation(email, expected):
    """Test email validation with multiple inputs"""
    assert validate_email(email) == expected

2. Complex Parametrization with Multiple Parameters

# tests/test_api_endpoints.py
import pytest

@pytest.mark.parametrize("method,endpoint,status_code,expected_keys", [
    ("GET", "/api/users", 200, ["id", "name", "email"]),
    ("GET", "/api/users/1", 200, ["id", "name", "email", "created_at"]),
    ("POST", "/api/users", 201, ["id"]),
    ("GET", "/api/invalid", 404, ["error"]),
    ("DELETE", "/api/users/999", 404, ["error"]),
])
def test_api_endpoints(client, method, endpoint, status_code, expected_keys):
    """Test multiple API endpoints with different scenarios"""
    if method == "GET":
        response = client.get(endpoint)
    elif method == "POST":
        response = client.post(endpoint, json={"name": "John"})
    elif method == "DELETE":
        response = client.delete(endpoint)
    
    assert response.status_code == status_code
    for key in expected_keys:
        assert key in response.json()

3. Indirect Parametrization (Using Fixtures)

# tests/test_database_operations.py
import pytest

@pytest.fixture
def user_data(request):
    """Parametrized fixture providing different user configurations"""
    users = {
        'admin': {'role': 'admin', 'permissions': ['read', 'write', 'delete']},
        'user': {'role': 'user', 'permissions': ['read']},
        'guest': {'role': 'guest', 'permissions': []}
    }
    return users[request.param]

@pytest.mark.parametrize("user_data", ['admin', 'user', 'guest'], indirect=True)
def test_user_permissions(user_data):
    """Test permissions for different user roles"""
    assert isinstance(user_data['permissions'], list)
    assert user_data['role'] in ['admin', 'user', 'guest']

Async Testing

Testing Async Functions

# tests/test_async_operations.py
import pytest
import asyncio

@pytest.fixture
def event_loop():
    """Create event loop for async tests"""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.mark.asyncio
async def test_async_api_call(mock_http_requests):
    """Test async HTTP request"""
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/users') as resp:
            data = await resp.json()
    
    assert len(data) == 2
    assert data[0]['name'] == 'John'

@pytest.mark.asyncio
async def test_concurrent_operations():
    """Test multiple concurrent operations"""
    async def fetch_user(user_id):
        await asyncio.sleep(0.1)
        return {'id': user_id, 'name': f'User {user_id}'}
    
    results = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3)
    )
    
    assert len(results) == 3
    assert results[0]['id'] == 1

Integration Testing

API Integration Tests

# tests/test_api_integration.py
import pytest
from fastapi.testclient import TestClient
from app import app

@pytest.fixture
def client():
    """FastAPI test client"""
    return TestClient(app)

class TestUserAPI:
    """User API integration tests"""
    
    def test_create_user_flow(self, client, db_session):
        """Test complete user creation workflow"""
        # 1. Create user
        response = client.post(
            "/api/users",
            json={"name": "John Doe", "email": "john@example.com"}
        )
        assert response.status_code == 201
        user_id = response.json()['id']
        
        # 2. Retrieve user
        response = client.get(f"/api/users/{user_id}")
        assert response.status_code == 200
        assert response.json()['name'] == "John Doe"
        
        # 3. Update user
        response = client.put(
            f"/api/users/{user_id}",
            json={"name": "John Updated"}
        )
        assert response.status_code == 200
        
        # 4. Delete user
        response = client.delete(f"/api/users/{user_id}")
        assert response.status_code == 204

Performance & Load Testing

Measuring Test Performance

# tests/test_performance.py
import pytest
import time

@pytest.fixture
def benchmark():
    """Simple benchmarking fixture"""
    def _benchmark(func, *args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start
        return result, elapsed
    return _benchmark

def test_algorithm_performance(benchmark):
    """Test algorithm performance"""
    def expensive_operation(n):
        return sum(i**2 for i in range(n))
    
    result, elapsed = benchmark(expensive_operation, 100000)
    
    assert result > 0
    assert elapsed < 1.0  # Should complete in under 1 second

Stress Testing

# tests/test_stress.py
import pytest
import concurrent.futures

def test_concurrent_requests(client):
    """Stress test with concurrent requests"""
    def make_request():
        return client.get("/api/health")
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
        futures = [executor.submit(make_request) for _ in range(1000)]
        results = [f.result() for f in concurrent.futures.as_completed(futures)]
    
    success_count = sum(1 for r in results if r.status_code == 200)
    assert success_count >= 990  # At least 99% success rate

CI/CD Integration

Pytest Configuration for CI/CD

# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = 
    -v
    --strict-markers
    --tb=short
    --cov=src
    --cov-report=xml
    --cov-report=html
    --junit-xml=test-results.xml
    -n auto
markers =
    slow: marks tests as slow
    integration: marks tests as integration tests
    unit: marks tests as unit tests

GitHub Actions Integration

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: [3.8, 3.9, '3.10', 3.11]
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ matrix.python-version }}
      
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements-dev.txt
      
      - name: Run tests
        run: pytest
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml
      
      - name: Publish test results
        if: always()
        uses: EnricoMi/publish-unit-test-result-action@v2
        with:
          files: test-results.xml

Best Practices

1. Use Clear Test Names

# ❌ Bad
def test_user():
    pass

# ✅ Good
def test_user_creation_with_valid_email_returns_user_id():
    pass

2. Follow AAA Pattern (Arrange, Act, Assert)

def test_user_login():
    # Arrange
    user = User(username="john", password="secret123")
    db_session.add(user)
    db_session.commit()
    
    # Act
    result = authenticate(username="john", password="secret123")
    
    # Assert
    assert result.username == "john"
    assert result.authenticated == True

3. Test Edge Cases

@pytest.mark.parametrize("input_value", [
    None,
    "",
    [],
    {},
    float('inf'),
    -999999,
    "a" * 10000
])
def test_function_with_edge_cases(input_value):
    """Test function with various edge cases"""
    result = my_function(input_value)
    assert result is not None

4. Use Markers for Test Organization

@pytest.mark.slow
def test_database_migration():
    pass

@pytest.mark.integration
def test_api_with_real_database():
    pass

# Run only fast tests:
# pytest -m "not slow"

# Run only integration tests:
# pytest -m integration

Conclusion

Pytest’s flexibility makes it perfect for enterprise testing. Combine fixtures, parametrization, and async support, and you have a testing framework that scales with your application.

Master these patterns, and you’ll write tests that catch bugs early and run efficiently in your CI/CD pipelines.

What’s your favorite Pytest feature? Share in the comments!