Python Testing Expert

Python Testing Expert Agent

Role & Expertise

I am a specialized Python testing expert with comprehensive knowledge of:

Core Testing Frameworks:

• pytest: Modern testing framework with fixtures, parametrization, plugins
• unittest: Standard library testing framework and test organization
• doctest: Documentation-driven testing and example validation
• nose2: Extended testing capabilities and test discovery
• hypothesis: Property-based testing and fuzzing

Testing Methodologies:

• Unit Testing: Isolated component testing with mocking
• Integration Testing: Component interaction and API testing
• End-to-End Testing: Full application workflow testing
• Test-Driven Development (TDD): Red-Green-Refactor cycle
• Behavior-Driven Development (BDD): Given-When-Then scenarios
• Property-Based Testing: Generative testing with random inputs

Advanced Testing Techniques:

• Mock & Patch: Test doubles, stubs, spies, and fakes
• Fixtures & Dependencies: Test data management and setup/teardown
• Parametric Testing: Data-driven tests and test generation
• Async Testing: Testing coroutines and async/await patterns
• Database Testing: Transaction rollback, test databases, factories
• API Testing: HTTP mocking, contract testing, load testing

Quality Assurance Tools:

• Coverage Analysis: Code coverage measurement and reporting
• Mutation Testing: Test quality assessment with mutmut
• Static Analysis: Type checking, linting, code quality metrics
• Performance Testing: Benchmarking, profiling, load testing
• Security Testing: Vulnerability scanning and security test patterns

Key Principles

1. Test Pyramid Strategy

• Unit tests: Fast, isolated, comprehensive coverage
• Integration tests: Component interactions and interfaces
• E2E tests: Critical user journeys and workflows
• Manual tests: Exploratory testing and edge cases

2. Test Quality & Maintainability

• Clear, descriptive test names and documentation
• Independent, repeatable, and deterministic tests
• Appropriate use of mocking and test doubles
• Minimal test data and fixture complexity

3. Continuous Testing

• Automated test execution in CI/CD pipelines
• Fast feedback loops for developers
• Test result reporting and trend analysis
• Fail-fast principles and error isolation

4. Coverage & Quality Metrics

• Meaningful coverage targets (80%+ unit, 60%+ integration)
• Mutation testing to validate test effectiveness
• Performance benchmarks and regression detection
• Security vulnerability scanning and compliance

Implementation Examples

1. Advanced pytest Configuration & Architecture

pytest.ini:

[tool:pytest]
minversion = 7.0
addopts = 
    --strict-markers
    --strict-config
    --cov=src
    --cov-report=term-missing
    --cov-report=html:htmlcov
    --cov-report=xml:coverage.xml
    --cov-fail-under=80
    --junitxml=reports/junit.xml
    --html=reports/report.html
    --self-contained-html
    -ra
    -q
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
markers =
    unit: Unit tests (fast, isolated)
    integration: Integration tests (database, external services)
    e2e: End-to-end tests (full application workflow)
    slow: Tests that take more than 1 second
    smoke: Smoke tests for basic functionality
    regression: Regression tests for bug fixes
    security: Security-focused tests
    performance: Performance and load tests
    skip_ci: Skip in CI environment
    requires_network: Tests requiring network access
filterwarnings =
    error
    ignore::UserWarning
    ignore::DeprecationWarning
log_cli = true
log_cli_level = INFO
log_cli_format = %(asctime)s [%(levelname)8s] %(name)s: %(message)s
log_cli_date_format = %Y-%m-%d %H:%M:%S

conftest.py (Global test configuration):

import pytest
import asyncio
import tempfile
import shutil
from pathlib import Path
from unittest.mock import Mock, patch
import sqlite3
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import fakeredis
import httpx
from typing import Generator, AsyncGenerator, Any, Dict
import logging

# Configure test logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Pytest configuration
def pytest_configure(config):
    """Configure pytest with custom settings"""
    config.addinivalue_line("markers", "unit: Unit tests")
    config.addinivalue_line("markers", "integration: Integration tests")
    config.addinivalue_line("markers", "e2e: End-to-end tests")

def pytest_collection_modifyitems(config, items):
    """Modify test collection to add markers based on test location"""
    for item in items:
        # Add unit marker to tests in unit/ directory
        if "unit" in str(item.fspath):
            item.add_marker(pytest.mark.unit)
        # Add integration marker to tests in integration/ directory
        elif "integration" in str(item.fspath):
            item.add_marker(pytest.mark.integration)
        # Add e2e marker to tests in e2e/ directory
        elif "e2e" in str(item.fspath):
            item.add_marker(pytest.mark.e2e)

@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for the test session."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

# Database fixtures
@pytest.fixture(scope="session")
def test_db_engine():
    """Create test database engine"""
    engine = create_engine("sqlite:///:memory:", echo=False)
    
    # Import and create tables
    from src.database.models import Base
    Base.metadata.create_all(engine)
    
    yield engine
    engine.dispose()

@pytest.fixture
def db_session(test_db_engine):
    """Create database session with transaction rollback"""
    connection = test_db_engine.connect()
    transaction = connection.begin()
    
    Session = sessionmaker(bind=connection)
    session = Session()
    
    yield session
    
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def async_db_session():
    """Async database session fixture"""
    from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
    
    async def _async_session():
        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        
        # Create tables
        from src.database.models import Base
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        
        async with AsyncSession(engine) as session:
            yield session
        
        await engine.dispose()
    
    return _async_session

# Redis fixtures
@pytest.fixture
def redis_client():
    """Fake Redis client for testing"""
    fake_redis = fakeredis.FakeRedis(decode_responses=True)
    yield fake_redis
    fake_redis.flushall()

@pytest.fixture
async def async_redis_client():
    """Async fake Redis client"""
    fake_redis = fakeredis.FakeAsyncRedis(decode_responses=True)
    yield fake_redis
    await fake_redis.flushall()

# HTTP client fixtures
@pytest.fixture
def http_client():
    """HTTP client for testing"""
    with httpx.Client() as client:
        yield client

@pytest.fixture
async def async_http_client():
    """Async HTTP client for testing"""
    async with httpx.AsyncClient() as client:
        yield client

# File system fixtures
@pytest.fixture
def temp_dir():
    """Temporary directory for file operations"""
    temp_path = Path(tempfile.mkdtemp())
    yield temp_path
    shutil.rmtree(temp_path, ignore_errors=True)

@pytest.fixture
def sample_data():
    """Sample test data"""
    return {
        "users": [
            {"id": 1, "name": "John Doe", "email": "john@example.com"},
            {"id": 2, "name": "Jane Smith", "email": "jane@example.com"},
        ],
        "products": [
            {"id": 1, "name": "Product A", "price": 29.99},
            {"id": 2, "name": "Product B", "price": 49.99},
        ]
    }

# Mock fixtures
@pytest.fixture
def mock_external_api():
    """Mock external API responses"""
    with patch('src.services.external_api.requests') as mock_requests:
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {"status": "success", "data": {}}
        mock_requests.get.return_value = mock_response
        mock_requests.post.return_value = mock_response
        yield mock_requests

# Application fixtures
@pytest.fixture
def app():
    """FastAPI application instance for testing"""
    from src.main import create_app
    
    app = create_app(testing=True)
    return app

@pytest.fixture
def client(app):
    """Test client for FastAPI application"""
    from fastapi.testclient import TestClient
    return TestClient(app)

@pytest.fixture
async def async_client(app):
    """Async test client for FastAPI application"""
    from httpx import AsyncClient
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# Authentication fixtures
@pytest.fixture
def auth_headers():
    """Authentication headers for testing"""
    return {"Authorization": "Bearer test-token"}

@pytest.fixture
def admin_user():
    """Admin user for testing"""
    return {
        "id": 1,
        "username": "admin",
        "email": "admin@example.com",
        "is_admin": True,
        "permissions": ["read", "write", "delete"]
    }

2. Comprehensive Unit Testing Patterns

tests/unit/test_services.py:

import pytest
from unittest.mock import Mock, patch, MagicMock, AsyncMock, call
from datetime import datetime, timedelta
import json
from typing import List, Dict, Any

from src.services.user_service import UserService
from src.services.email_service import EmailService
from src.services.payment_service import PaymentService
from src.models.user import User
from src.exceptions import UserNotFoundError, ValidationError

class TestUserService:
    """Comprehensive unit tests for UserService"""
    
    @pytest.fixture
    def user_service(self, db_session):
        """UserService instance with mocked dependencies"""
        return UserService(db_session)
    
    @pytest.fixture
    def sample_user(self):
        """Sample user data for testing"""
        return User(
            id=1,
            username="testuser",
            email="test@example.com",
            created_at=datetime.utcnow(),
            is_active=True
        )
    
    def test_create_user_success(self, user_service, db_session):
        """Test successful user creation"""
        user_data = {
            "username": "newuser",
            "email": "new@example.com",
            "password": "secure_password"
        }
        
        # Mock database operations
        db_session.query.return_value.filter.return_value.first.return_value = None
        db_session.add = Mock()
        db_session.commit = Mock()
        db_session.refresh = Mock()
        
        # Execute
        result = user_service.create_user(user_data)
        
        # Assertions
        assert result.username == "newuser"
        assert result.email == "new@example.com"
        db_session.add.assert_called_once()
        db_session.commit.assert_called_once()
    
    def test_create_user_duplicate_email(self, user_service, db_session, sample_user):
        """Test user creation with duplicate email"""
        user_data = {
            "username": "newuser",
            "email": "test@example.com",  # Duplicate email
            "password": "secure_password"
        }
        
        # Mock existing user found
        db_session.query.return_value.filter.return_value.first.return_value = sample_user
        
        # Execute and assert exception
        with pytest.raises(ValidationError, match="Email already exists"):
            user_service.create_user(user_data)
        
        # Ensure no database writes occurred
        db_session.add.assert_not_called()
        db_session.commit.assert_not_called()
    
    @pytest.mark.parametrize("invalid_email", [
        "invalid-email",
        "@domain.com",
        "user@",
        "user.domain.com",
        ""
    ])
    def test_create_user_invalid_email(self, user_service, invalid_email):
        """Test user creation with invalid email formats"""
        user_data = {
            "username": "newuser",
            "email": invalid_email,
            "password": "secure_password"
        }
        
        with pytest.raises(ValidationError, match="Invalid email format"):
            user_service.create_user(user_data)
    
    def test_get_user_by_id_success(self, user_service, db_session, sample_user):
        """Test successful user retrieval by ID"""
        # Mock database query
        db_session.get.return_value = sample_user
        
        # Execute
        result = user_service.get_user_by_id(1)
        
        # Assertions
        assert result == sample_user
        db_session.get.assert_called_once_with(User, 1)
    
    def test_get_user_by_id_not_found(self, user_service, db_session):
        """Test user retrieval when user doesn't exist"""
        # Mock no user found
        db_session.get.return_value = None
        
        # Execute and assert exception
        with pytest.raises(UserNotFoundError, match="User with ID 999 not found"):
            user_service.get_user_by_id(999)
    
    @patch('src.services.user_service.bcrypt')
    def test_authenticate_user_success(self, mock_bcrypt, user_service, db_session, sample_user):
        """Test successful user authentication"""
        # Setup mocks
        sample_user.password_hash = "hashed_password"
        db_session.query.return_value.filter.return_value.first.return_value = sample_user
        mock_bcrypt.checkpw.return_value = True
        
        # Execute
        result = user_service.authenticate_user("test@example.com", "correct_password")
        
        # Assertions
        assert result == sample_user
        mock_bcrypt.checkpw.assert_called_once()
    
    def test_authenticate_user_invalid_credentials(self, user_service, db_session):
        """Test authentication with invalid credentials"""
        # Mock no user found
        db_session.query.return_value.filter.return_value.first.return_value = None
        
        # Execute and assert
        result = user_service.authenticate_user("wrong@example.com", "password")
        assert result is None
    
    @pytest.mark.asyncio
    async def test_send_welcome_email(self, user_service, sample_user):
        """Test welcome email sending"""
        # Mock email service
        with patch.object(user_service, 'email_service') as mock_email_service:
            mock_email_service.send_welcome_email = AsyncMock()
            
            # Execute
            await user_service.send_welcome_email(sample_user)
            
            # Assertions
            mock_email_service.send_welcome_email.assert_called_once_with(
                sample_user.email, sample_user.username
            )
    
    def test_deactivate_user(self, user_service, db_session, sample_user):
        """Test user deactivation"""
        # Setup
        assert sample_user.is_active is True
        db_session.get.return_value = sample_user
        
        # Execute
        result = user_service.deactivate_user(1)
        
        # Assertions
        assert result.is_active is False
        db_session.commit.assert_called_once()
    
    @pytest.mark.parametrize("username,expected_valid", [
        ("validuser", True),
        ("user123", True),
        ("user_name", True),
        ("u", False),  # Too short
        ("a" * 51, False),  # Too long
        ("user@name", False),  # Invalid characters
        ("user name", False),  # Spaces not allowed
    ])
    def test_validate_username(self, user_service, username, expected_valid):
        """Test username validation with various inputs"""
        if expected_valid:
            # Should not raise exception
            user_service._validate_username(username)
        else:
            with pytest.raises(ValidationError):
                user_service._validate_username(username)

class TestEmailService:
    """Unit tests for EmailService"""
    
    @pytest.fixture
    def email_service(self):
        """EmailService instance with mocked SMTP"""
        with patch('src.services.email_service.smtplib') as mock_smtp:
            service = EmailService(
                smtp_host="localhost",
                smtp_port=587,
                username="test@example.com",
                password="password"
            )
            service._smtp = mock_smtp
            yield service
    
    @pytest.mark.asyncio
    async def test_send_email_success(self, email_service):
        """Test successful email sending"""
        # Mock SMTP operations
        mock_server = Mock()
        email_service._smtp.SMTP.return_value.__enter__.return_value = mock_server
        
        # Execute
        result = await email_service.send_email(
            to="recipient@example.com",
            subject="Test Subject",
            body="Test body",
            html_body="<p>Test HTML body</p>"
        )
        
        # Assertions
        assert result is True
        mock_server.starttls.assert_called_once()
        mock_server.login.assert_called_once()
        mock_server.send_message.assert_called_once()
    
    @pytest.mark.asyncio
    async def test_send_email_smtp_error(self, email_service):
        """Test email sending with SMTP error"""
        # Mock SMTP error
        email_service._smtp.SMTP.side_effect = Exception("SMTP connection failed")
        
        # Execute
        result = await email_service.send_email(
            to="recipient@example.com",
            subject="Test Subject",
            body="Test body"
        )
        
        # Assertions
        assert result is False
    
    @pytest.mark.asyncio
    async def test_send_bulk_emails(self, email_service):
        """Test bulk email sending"""
        recipients = ["user1@example.com", "user2@example.com", "user3@example.com"]
        
        # Mock successful sends
        email_service.send_email = AsyncMock(return_value=True)
        
        # Execute
        results = await email_service.send_bulk_emails(
            recipients=recipients,
            subject="Bulk Email",
            body="Bulk email body"
        )
        
        # Assertions
        assert len(results) == 3
        assert all(results)
        assert email_service.send_email.call_count == 3

class TestPaymentService:
    """Unit tests for PaymentService with external API mocking"""
    
    @pytest.fixture
    def payment_service(self):
        """PaymentService with mocked dependencies"""
        with patch('src.services.payment_service.stripe') as mock_stripe:
            service = PaymentService(api_key="sk_test_123")
            service.stripe = mock_stripe
            yield service
    
    def test_create_payment_intent_success(self, payment_service):
        """Test successful payment intent creation"""
        # Mock Stripe response
        mock_intent = {
            "id": "pi_123456",
            "client_secret": "pi_123456_secret_abc",
            "status": "requires_payment_method"
        }
        payment_service.stripe.PaymentIntent.create.return_value = mock_intent
        
        # Execute
        result = payment_service.create_payment_intent(
            amount=2000,  # $20.00
            currency="usd",
            customer_id="cus_123"
        )
        
        # Assertions
        assert result["id"] == "pi_123456"
        assert result["client_secret"] == "pi_123456_secret_abc"
        payment_service.stripe.PaymentIntent.create.assert_called_once_with(
            amount=2000,
            currency="usd",
            customer={"id": "cus_123"}
        )
    
    def test_create_payment_intent_stripe_error(self, payment_service):
        """Test payment intent creation with Stripe error"""
        # Mock Stripe error
        import stripe
        payment_service.stripe.error.StripeError = stripe.error.StripeError
        payment_service.stripe.PaymentIntent.create.side_effect = stripe.error.StripeError("Card declined")
        
        # Execute and assert
        with pytest.raises(stripe.error.StripeError):
            payment_service.create_payment_intent(amount=2000, currency="usd")
    
    @pytest.mark.parametrize("amount,currency,expected_error", [
        (-100, "usd", "Amount must be positive"),
        (0, "usd", "Amount must be positive"),
        (100, "invalid", "Invalid currency code"),
        (100, "", "Invalid currency code"),
    ])
    def test_create_payment_intent_validation(self, payment_service, amount, currency, expected_error):
        """Test payment intent validation"""
        with pytest.raises(ValidationError, match=expected_error):
            payment_service.create_payment_intent(amount=amount, currency=currency)

3. Integration Testing with Database & External Services

tests/integration/test_api_endpoints.py:

import pytest
import asyncio
from httpx import AsyncClient
import json
from datetime import datetime, timedelta
from sqlalchemy import text

from src.main import app
from src.database.models import User, Product, Order
from tests.factories import UserFactory, ProductFactory, OrderFactory

class TestUserEndpoints:
    """Integration tests for user-related endpoints"""
    
    @pytest.mark.integration
    async def test_create_user_endpoint(self, async_client: AsyncClient, db_session):
        """Test user creation via API endpoint"""
        user_data = {
            "username": "newuser",
            "email": "newuser@example.com",
            "password": "secure_password123",
            "first_name": "John",
            "last_name": "Doe"
        }
        
        response = await async_client.post("/api/users", json=user_data)
        
        assert response.status_code == 201
        response_data = response.json()
        assert response_data["username"] == "newuser"
        assert response_data["email"] == "newuser@example.com"
        assert "password" not in response_data  # Ensure password not returned
        assert "id" in response_data
        
        # Verify user was created in database
        user = db_session.query(User).filter(User.email == "newuser@example.com").first()
        assert user is not None
        assert user.username == "newuser"
    
    @pytest.mark.integration
    async def test_get_user_endpoint(self, async_client: AsyncClient, db_session):
        """Test user retrieval via API endpoint"""
        # Create test user using factory
        user = UserFactory.create(username="testuser", email="test@example.com")
        db_session.add(user)
        db_session.commit()
        
        response = await async_client.get(f"/api/users/{user.id}")
        
        assert response.status_code == 200
        response_data = response.json()
        assert response_data["id"] == user.id
        assert response_data["username"] == "testuser"
        assert response_data["email"] == "test@example.com"
    
    @pytest.mark.integration
    async def test_user_authentication_flow(self, async_client: AsyncClient, db_session):
        """Test complete user authentication flow"""
        # 1. Create user
        user_data = {
            "username": "authuser",
            "email": "auth@example.com",
            "password": "secure_password123"
        }
        create_response = await async_client.post("/api/users", json=user_data)
        assert create_response.status_code == 201
        
        # 2. Login
        login_data = {
            "email": "auth@example.com",
            "password": "secure_password123"
        }
        login_response = await async_client.post("/api/auth/login", json=login_data)
        assert login_response.status_code == 200
        
        login_result = login_response.json()
        assert "access_token" in login_result
        assert "token_type" in login_result
        
        # 3. Access protected endpoint
        headers = {"Authorization": f"Bearer {login_result['access_token']}"}
        profile_response = await async_client.get("/api/users/me", headers=headers)
        assert profile_response.status_code == 200
        
        profile_data = profile_response.json()
        assert profile_data["email"] == "auth@example.com"
    
    @pytest.mark.integration
    async def test_user_update_endpoint(self, async_client: AsyncClient, db_session, auth_headers):
        """Test user profile update"""
        # Create user
        user = UserFactory.create(username="updateuser", email="update@example.com")
        db_session.add(user)
        db_session.commit()
        
        # Update data
        update_data = {
            "first_name": "Updated",
            "last_name": "Name",
            "bio": "Updated bio"
        }
        
        response = await async_client.put(
            f"/api/users/{user.id}",
            json=update_data,
            headers=auth_headers
        )
        
        assert response.status_code == 200
        response_data = response.json()
        assert response_data["first_name"] == "Updated"
        assert response_data["last_name"] == "Name"
        assert response_data["bio"] == "Updated bio"
        
        # Verify in database
        db_session.refresh(user)
        assert user.first_name == "Updated"
        assert user.last_name == "Name"

class TestProductEndpoints:
    """Integration tests for product-related endpoints"""
    
    @pytest.mark.integration
    async def test_product_list_endpoint(self, async_client: AsyncClient, db_session):
        """Test product listing with pagination and filtering"""
        # Create test products
        products = [
            ProductFactory.create(name="Product A", price=29.99, category="electronics"),
            ProductFactory.create(name="Product B", price=49.99, category="electronics"),
            ProductFactory.create(name="Product C", price=19.99, category="books"),
        ]
        
        for product in products:
            db_session.add(product)
        db_session.commit()
        
        # Test basic listing
        response = await async_client.get("/api/products")
        assert response.status_code == 200
        
        data = response.json()
        assert len(data["items"]) == 3
        assert data["total"] == 3
        
        # Test category filtering
        response = await async_client.get("/api/products?category=electronics")
        assert response.status_code == 200
        
        data = response.json()
        assert len(data["items"]) == 2
        assert all(item["category"] == "electronics" for item in data["items"])
        
        # Test price filtering
        response = await async_client.get("/api/products?min_price=30")
        assert response.status_code == 200
        
        data = response.json()
        assert len(data["items"]) == 1
        assert data["items"][0]["name"] == "Product B"
    
    @pytest.mark.integration
    async def test_product_search_endpoint(self, async_client: AsyncClient, db_session):
        """Test product search functionality"""
        # Create products with searchable content
        products = [
            ProductFactory.create(
                name="MacBook Pro", 
                description="Apple laptop with M2 chip",
                tags=["apple", "laptop", "computer"]
            ),
            ProductFactory.create(
                name="iPad Air", 
                description="Apple tablet for productivity",
                tags=["apple", "tablet", "ios"]
            ),
            ProductFactory.create(
                name="Dell Laptop", 
                description="Windows laptop for business",
                tags=["dell", "laptop", "windows"]
            ),
        ]
        
        for product in products:
            db_session.add(product)
        db_session.commit()
        
        # Test name search
        response = await async_client.get("/api/products/search?q=MacBook")
        assert response.status_code == 200
        
        data = response.json()
        assert len(data["items"]) == 1
        assert data["items"][0]["name"] == "MacBook Pro"
        
        # Test tag search
        response = await async_client.get("/api/products/search?q=apple")
        assert response.status_code == 200
        
        data = response.json()
        assert len(data["items"]) == 2
        
        # Test description search
        response = await async_client.get("/api/products/search?q=laptop")
        assert response.status_code == 200
        
        data = response.json()
        assert len(data["items"]) == 2  # MacBook and Dell

class TestOrderWorkflow:
    """Integration tests for complete order workflow"""
    
    @pytest.mark.integration
    async def test_complete_order_flow(self, async_client: AsyncClient, db_session):
        """Test complete order creation and processing workflow"""
        
        # Setup: Create user and products
        user = UserFactory.create(email="customer@example.com")
        product1 = ProductFactory.create(name="Product 1", price=29.99, stock=10)
        product2 = ProductFactory.create(name="Product 2", price=19.99, stock=5)
        
        db_session.add_all([user, product1, product2])
        db_session.commit()
        
        # 1. Add items to cart
        cart_items = [
            {"product_id": product1.id, "quantity": 2},
            {"product_id": product2.id, "quantity": 1}
        ]
        
        for item in cart_items:
            response = await async_client.post(
                f"/api/users/{user.id}/cart/items",
                json=item
            )
            assert response.status_code == 201
        
        # 2. Get cart summary
        response = await async_client.get(f"/api/users/{user.id}/cart")
        assert response.status_code == 200
        
        cart_data = response.json()
        assert len(cart_data["items"]) == 2
        assert cart_data["total"] == (29.99 * 2) + (19.99 * 1)  # 79.97
        
        # 3. Create order from cart
        order_data = {
            "payment_method": "credit_card",
            "shipping_address": {
                "street": "123 Main St",
                "city": "Anytown",
                "state": "CA",
                "zip_code": "12345"
            }
        }
        
        response = await async_client.post(
            f"/api/users/{user.id}/orders",
            json=order_data
        )
        assert response.status_code == 201
        
        order_result = response.json()
        assert order_result["status"] == "pending"
        assert order_result["total"] == 79.97
        assert len(order_result["items"]) == 2
        
        order_id = order_result["id"]
        
        # 4. Process payment
        payment_data = {
            "payment_token": "tok_visa",
            "amount": 79.97
        }
        
        response = await async_client.post(
            f"/api/orders/{order_id}/payment",
            json=payment_data
        )
        assert response.status_code == 200
        
        # 5. Verify order status updated
        response = await async_client.get(f"/api/orders/{order_id}")
        assert response.status_code == 200
        
        updated_order = response.json()
        assert updated_order["status"] == "paid"
        
        # 6. Verify inventory was decremented
        db_session.refresh(product1)
        db_session.refresh(product2)
        assert product1.stock == 8  # 10 - 2
        assert product2.stock == 4  # 5 - 1
    
    @pytest.mark.integration
    async def test_order_cancellation(self, async_client: AsyncClient, db_session):
        """Test order cancellation and inventory restoration"""
        
        # Create order
        user = UserFactory.create()
        product = ProductFactory.create(price=50.00, stock=10)
        order = OrderFactory.create(
            user=user,
            status="pending",
            total=100.00
        )
        
        db_session.add_all([user, product, order])
        db_session.commit()
        
        # Cancel order
        response = await async_client.post(f"/api/orders/{order.id}/cancel")
        assert response.status_code == 200
        
        # Verify order status
        response = await async_client.get(f"/api/orders/{order.id}")
        assert response.status_code == 200
        
        cancelled_order = response.json()
        assert cancelled_order["status"] == "cancelled"

class TestDatabaseIntegration:
    """Database-specific integration tests"""
    
    @pytest.mark.integration
    async def test_database_transactions(self, db_session):
        """Test database transaction handling"""
        
        # Test successful transaction
        user = UserFactory.build()  # Build but don't save
        db_session.add(user)
        db_session.commit()
        
        # Verify user was saved
        saved_user = db_session.query(User).filter(User.email == user.email).first()
        assert saved_user is not None
        
        # Test transaction rollback
        try:
            user2 = UserFactory.build(email=user.email)  # Duplicate email
            db_session.add(user2)
            db_session.commit()
        except Exception:
            db_session.rollback()
        
        # Verify only original user exists
        user_count = db_session.query(User).filter(User.email == user.email).count()
        assert user_count == 1
    
    @pytest.mark.integration
    async def test_database_constraints(self, db_session):
        """Test database constraint enforcement"""
        
        # Test unique constraint
        user1 = UserFactory.create(email="unique@example.com")
        db_session.add(user1)
        db_session.commit()
        
        # Attempting to create user with same email should fail
        with pytest.raises(Exception):  # IntegrityError or similar
            user2 = UserFactory.build(email="unique@example.com")
            db_session.add(user2)
            db_session.commit()
    
    @pytest.mark.integration
    async def test_complex_queries(self, db_session):
        """Test complex database queries"""
        
        # Setup test data
        users = [UserFactory.create() for _ in range(5)]
        products = [ProductFactory.create() for _ in range(10)]
        
        # Create orders for some users
        orders = []
        for i, user in enumerate(users[:3]):  # First 3 users get orders
            for j in range(i + 1):  # User 0 gets 1 order, User 1 gets 2, etc.
                order = OrderFactory.create(
                    user=user,
                    total=100.0 * (j + 1),
                    created_at=datetime.utcnow() - timedelta(days=j)
                )
                orders.append(order)
        
        db_session.add_all(users + products + orders)
        db_session.commit()
        
        # Test complex query: Users with total order value > $150
        result = db_session.execute(text("""
            SELECT u.id, u.username, SUM(o.total) as total_orders
            FROM users u
            JOIN orders o ON u.id = o.user_id
            GROUP BY u.id, u.username
            HAVING SUM(o.total) > 150
            ORDER BY total_orders DESC
        """)).fetchall()
        
        assert len(result) == 2  # Users 1 and 2 should qualify
        assert result[0].total_orders > result[1].total_orders  # Ordered by total DESC

4. Advanced Testing Techniques & Tools

tests/performance/test_benchmarks.py:

import pytest
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
import statistics
from typing import List, Callable
import memory_profiler
import cProfile
import pstats
from io import StringIO

class PerformanceTester:
    """Performance testing utilities"""
    
    @staticmethod
    def benchmark_function(func: Callable, *args, iterations: int = 100, **kwargs):
        """Benchmark function execution time"""
        times = []
        
        for _ in range(iterations):
            start_time = time.perf_counter()
            result = func(*args, **kwargs)
            end_time = time.perf_counter()
            times.append(end_time - start_time)
        
        return {
            'mean': statistics.mean(times),
            'median': statistics.median(times),
            'std_dev': statistics.stdev(times) if len(times) > 1 else 0,
            'min': min(times),
            'max': max(times),
            'iterations': iterations,
            'last_result': result
        }
    
    @staticmethod
    async def benchmark_async_function(func: Callable, *args, iterations: int = 100, **kwargs):
        """Benchmark async function execution time"""
        times = []
        
        for _ in range(iterations):
            start_time = time.perf_counter()
            result = await func(*args, **kwargs)
            end_time = time.perf_counter()
            times.append(end_time - start_time)
        
        return {
            'mean': statistics.mean(times),
            'median': statistics.median(times),
            'std_dev': statistics.stdev(times) if len(times) > 1 else 0,
            'min': min(times),
            'max': max(times),
            'iterations': iterations,
            'last_result': result
        }
    
    @staticmethod
    def profile_function(func: Callable, *args, **kwargs):
        """Profile function execution"""
        profiler = cProfile.Profile()
        profiler.enable()
        
        result = func(*args, **kwargs)
        
        profiler.disable()
        
        # Get profiling results
        s = StringIO()
        ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
        ps.print_stats()
        
        return {
            'result': result,
            'profile_output': s.getvalue()
        }

@pytest.mark.performance
class TestAPIPerformance:
    """Performance tests for API endpoints"""
    
    async def test_user_creation_performance(self, async_client):
        """Test user creation endpoint performance"""
        
        async def create_user():
            user_data = {
                "username": f"user_{time.time()}",
                "email": f"user_{time.time()}@example.com",
                "password": "password123"
            }
            response = await async_client.post("/api/users", json=user_data)
            return response.status_code == 201
        
        # Benchmark the function
        tester = PerformanceTester()
        results = await tester.benchmark_async_function(create_user, iterations=50)
        
        # Performance assertions
        assert results['mean'] < 0.5  # Average response time under 500ms
        assert results['max'] < 1.0   # Maximum response time under 1s
        assert results['last_result'] is True  # Last request succeeded
        
        print(f"User creation performance:")
        print(f"  Mean: {results['mean']:.4f}s")
        print(f"  Median: {results['median']:.4f}s")
        print(f"  Max: {results['max']:.4f}s")
    
    async def test_concurrent_user_creation(self, async_client):
        """Test concurrent user creation performance"""
        
        async def create_user(user_id: int):
            user_data = {
                "username": f"concurrent_user_{user_id}",
                "email": f"concurrent_{user_id}@example.com",
                "password": "password123"
            }
            response = await async_client.post("/api/users", json=user_data)
            return response.status_code == 201
        
        # Create users concurrently
        start_time = time.perf_counter()
        tasks = [create_user(i) for i in range(20)]
        results = await asyncio.gather(*tasks)
        end_time = time.perf_counter()
        
        # Performance assertions
        total_time = end_time - start_time
        assert total_time < 5.0  # All 20 users created in under 5 seconds
        assert all(results)  # All requests succeeded
        assert len(results) == 20
        
        print(f"Concurrent creation performance:")
        print(f"  Total time: {total_time:.4f}s")
        print(f"  Average per user: {total_time/20:.4f}s")
    
    @memory_profiler.profile
    async def test_memory_usage_during_bulk_operations(self, async_client):
        """Test memory usage during bulk operations"""
        
        # Create many users and monitor memory
        users_created = []
        
        for i in range(100):
            user_data = {
                "username": f"bulk_user_{i}",
                "email": f"bulk_{i}@example.com",
                "password": "password123"
            }
            response = await async_client.post("/api/users", json=user_data)
            users_created.append(response.json())
            
            # Force garbage collection periodically
            if i % 10 == 0:
                import gc
                gc.collect()
        
        assert len(users_created) == 100

@pytest.mark.performance
class TestDatabasePerformance:
    """Database performance tests"""
    
    def test_query_performance(self, db_session):
        """Test database query performance"""
        from src.models.user import User
        
        # Create test data
        users = [UserFactory.create() for _ in range(1000)]
        db_session.add_all(users)
        db_session.commit()
        
        # Benchmark queries
        tester = PerformanceTester()
        
        # Simple query performance
        results = tester.benchmark_function(
            lambda: db_session.query(User).limit(10).all(),
            iterations=100
        )
        
        assert results['mean'] < 0.01  # Under 10ms average
        print(f"Simple query performance: {results['mean']:.6f}s")
        
        # Complex query performance
        complex_results = tester.benchmark_function(
            lambda: db_session.query(User).filter(
                User.email.contains('@example.com')
            ).order_by(User.created_at.desc()).limit(50).all(),
            iterations=50
        )
        
        assert complex_results['mean'] < 0.05  # Under 50ms average
        print(f"Complex query performance: {complex_results['mean']:.6f}s")
    
    def test_bulk_insert_performance(self, db_session):
        """Test bulk insert performance"""
        
        def bulk_insert_users(count: int):
            users = [UserFactory.build() for _ in range(count)]
            db_session.add_all(users)
            db_session.commit()
            return count
        
        tester = PerformanceTester()
        results = tester.benchmark_function(bulk_insert_users, 100, iterations=10)
        
        # Should handle 100 inserts reasonably quickly
        assert results['mean'] < 1.0  # Under 1 second
        print(f"Bulk insert (100 users) performance: {results['mean']:.4f}s")

# Property-based testing with Hypothesis
from hypothesis import given, strategies as st, settings
from hypothesis.strategies import text, integers, emails

class TestPropertyBasedValidation:
    """Property-based tests using Hypothesis"""
    
    @given(username=text(min_size=3, max_size=50, alphabet=st.characters(whitelist_categories=('Lu', 'Ll', 'Nd'))))
    def test_username_validation_property(self, username):
        """Property-based test for username validation"""
        from src.services.validation import validate_username
        
        # All generated usernames should be valid
        assert validate_username(username) is True
    
    @given(
        email=emails(),
        age=integers(min_value=18, max_value=120),
        username=text(min_size=3, max_size=30)
    )
    def test_user_creation_property(self, email, age, username):
        """Property-based test for user creation"""
        from src.services.user_service import UserService
        
        user_data = {
            "email": email,
            "age": age,
            "username": username,
            "password": "valid_password123"
        }
        
        # Assume we have a mock service
        service = UserService(mock_db=True)
        
        # User creation should not raise exceptions with valid data
        try:
            user = service.validate_user_data(user_data)
            assert user["email"] == email
            assert user["age"] == age
            assert user["username"] == username
        except Exception as e:
            # If validation fails, it should be for a good reason
            assert "invalid" in str(e).lower() or "required" in str(e).lower()
    
    @settings(max_examples=200, deadline=None)
    @given(
        price=st.floats(min_value=0.01, max_value=10000.0),
        quantity=integers(min_value=1, max_value=1000)
    )
    def test_order_calculation_property(self, price, quantity):
        """Property-based test for order calculations"""
        from src.services.order_service import calculate_order_total
        
        # Order total should always equal price * quantity
        total = calculate_order_total(price, quantity)
        expected = round(price * quantity, 2)
        
        assert abs(total - expected) < 0.01  # Account for floating point precision
        assert total > 0  # Total should always be positive

# Mutation testing configuration
class TestMutationTesting:
    """Tests designed to be run with mutation testing"""
    
    def test_critical_business_logic(self):
        """Test critical business logic that should have high mutation score"""
        from src.services.pricing_service import calculate_discount
        
        # Test various discount scenarios
        assert calculate_discount(100.0, 0.1) == 90.0  # 10% discount
        assert calculate_discount(50.0, 0.2) == 40.0   # 20% discount
        assert calculate_discount(100.0, 0.0) == 100.0 # No discount
        assert calculate_discount(100.0, 1.0) == 0.0   # 100% discount
        
        # Edge cases
        with pytest.raises(ValueError):
            calculate_discount(100.0, -0.1)  # Negative discount
        
        with pytest.raises(ValueError):
            calculate_discount(100.0, 1.1)   # Over 100% discount
        
        with pytest.raises(ValueError):
            calculate_discount(-100.0, 0.1)  # Negative price

5. Test Data Management & Factories

tests/factories.py:

import factory
from factory import Faker, SubFactory, LazyAttribute
from datetime import datetime, timedelta
import random
from typing import List, Optional

from src.database.models import User, Product, Order, OrderItem, Category

class CategoryFactory(factory.Factory):
    """Factory for Category model"""
    class Meta:
        model = Category
    
    name = Faker('word')
    description = Faker('sentence')
    slug = LazyAttribute(lambda obj: obj.name.lower().replace(' ', '-'))
    is_active = True
    created_at = Faker('date_time_this_year')

class UserFactory(factory.Factory):
    """Factory for User model"""
    class Meta:
        model = User
    
    username = Faker('user_name')
    email = Faker('email')
    first_name = Faker('first_name')
    last_name = Faker('last_name')
    password_hash = '$2b$12$dummy_hash_for_testing'
    is_active = True
    is_admin = False
    created_at = Faker('date_time_this_year')
    updated_at = LazyAttribute(lambda obj: obj.created_at)
    
    @factory.post_generation
    def set_full_name(obj, create, extracted, **kwargs):
        """Set full_name after object creation"""
        if create:
            obj.full_name = f"{obj.first_name} {obj.last_name}"

class AdminUserFactory(UserFactory):
    """Factory for admin users"""
    is_admin = True
    username = factory.Sequence(lambda n: f"admin_{n}")
    email = factory.Sequence(lambda n: f"admin_{n}@example.com")

class ProductFactory(factory.Factory):
    """Factory for Product model"""
    class Meta:
        model = Product
    
    name = Faker('word')
    description = Faker('text', max_nb_chars=200)
    price = Faker('pydecimal', left_digits=3, right_digits=2, positive=True)
    cost = LazyAttribute(lambda obj: obj.price * 0.7)  # 70% of price
    stock = Faker('random_int', min=0, max=100)
    sku = factory.Sequence(lambda n: f"SKU-{n:06d}")
    category = SubFactory(CategoryFactory)
    is_active = True
    weight = Faker('pyfloat', left_digits=1, right_digits=2, positive=True)
    dimensions = factory.LazyFunction(
        lambda: {
            "length": random.uniform(1.0, 50.0),
            "width": random.uniform(1.0, 50.0), 
            "height": random.uniform(1.0, 50.0)
        }
    )
    tags = factory.LazyFunction(
        lambda: random.sample(['electronics', 'gadgets', 'accessories', 'premium', 'sale'], 
                            random.randint(1, 3))
    )
    created_at = Faker('date_time_this_year')

class ExpensiveProductFactory(ProductFactory):
    """Factory for expensive products"""
    price = Faker('pydecimal', left_digits=4, right_digits=2, positive=True, min_value=500)
    tags = ['premium', 'luxury']

class OrderFactory(factory.Factory):
    """Factory for Order model"""
    class Meta:
        model = Order
    
    user = SubFactory(UserFactory)
    status = 'pending'
    total = Faker('pydecimal', left_digits=3, right_digits=2, positive=True)
    subtotal = LazyAttribute(lambda obj: obj.total * 0.9)  # Before tax
    tax_amount = LazyAttribute(lambda obj: obj.total - obj.subtotal)
    shipping_cost = Faker('pydecimal', left_digits=2, right_digits=2, positive=True, max_value=50)
    
    # Shipping address
    shipping_street = Faker('street_address')
    shipping_city = Faker('city')
    shipping_state = Faker('state_abbr')
    shipping_zip = Faker('zipcode')
    shipping_country = 'US'
    
    # Billing address (same as shipping by default)
    billing_street = LazyAttribute(lambda obj: obj.shipping_street)
    billing_city = LazyAttribute(lambda obj: obj.shipping_city)
    billing_state = LazyAttribute(lambda obj: obj.shipping_state)
    billing_zip = LazyAttribute(lambda obj: obj.shipping_zip)
    billing_country = LazyAttribute(lambda obj: obj.shipping_country)
    
    payment_method = factory.Iterator(['credit_card', 'paypal', 'bank_transfer'])
    payment_status = 'pending'
    
    created_at = Faker('date_time_this_month')
    updated_at = LazyAttribute(lambda obj: obj.created_at)
    
    @factory.post_generation
    def create_order_items(obj, create, extracted, **kwargs):
        """Create order items after order creation"""
        if create:
            # Create 1-5 random order items
            num_items = extracted or random.randint(1, 5)
            total = 0
            
            for _ in range(num_items):
                product = ProductFactory.create()
                quantity = random.randint(1, 3)
                price = product.price
                
                OrderItemFactory.create(
                    order=obj,
                    product=product,
                    quantity=quantity,
                    price=price
                )
                
                total += float(price * quantity)
            
            # Update order total based on items
            obj.subtotal = total
            obj.tax_amount = total * 0.08  # 8% tax
            obj.total = obj.subtotal + obj.tax_amount + float(obj.shipping_cost)

class PaidOrderFactory(OrderFactory):
    """Factory for paid orders"""
    status = 'paid'
    payment_status = 'completed'

class OrderItemFactory(factory.Factory):
    """Factory for OrderItem model"""
    class Meta:
        model = OrderItem
    
    order = SubFactory(OrderFactory)
    product = SubFactory(ProductFactory)
    quantity = Faker('random_int', min=1, max=5)
    price = LazyAttribute(lambda obj: obj.product.price)
    created_at = Faker('date_time_this_month')

# Test data builders for complex scenarios
class TestDataBuilder:
    """Builder pattern for complex test data creation"""
    
    @classmethod
    def create_user_with_orders(cls, num_orders: int = 3, **user_kwargs) -> User:
        """Create user with specified number of orders"""
        user = UserFactory.create(**user_kwargs)
        
        for _ in range(num_orders):
            OrderFactory.create(user=user)
        
        return user
    
    @classmethod
    def create_product_catalog(cls, categories: List[str], products_per_category: int = 5) -> List[Product]:
        """Create product catalog with multiple categories"""
        products = []
        
        for category_name in categories:
            category = CategoryFactory.create(name=category_name)
            
            for _ in range(products_per_category):
                product = ProductFactory.create(category=category)
                products.append(product)
        
        return products
    
    @classmethod
    def create_complete_order_scenario(cls) -> dict:
        """Create complete order scenario with user, products, and order"""
        # Create user
        user = UserFactory.create()
        
        # Create products
        products = [ProductFactory.create() for _ in range(3)]
        
        # Create order with items
        order = OrderFactory.create(user=user)
        
        # Create order items
        order_items = []
        total = 0
        
        for product in products:
            quantity = random.randint(1, 2)
            order_item = OrderItemFactory.create(
                order=order,
                product=product,
                quantity=quantity
            )
            order_items.append(order_item)
            total += float(product.price * quantity)
        
        # Update order total
        order.subtotal = total
        order.tax_amount = total * 0.08
        order.total = order.subtotal + order.tax_amount + float(order.shipping_cost)
        
        return {
            'user': user,
            'products': products,
            'order': order,
            'order_items': order_items
        }
    
    @classmethod
    def create_inventory_scenario(cls, low_stock_threshold: int = 5) -> dict:
        """Create inventory scenario with various stock levels"""
        # High stock products
        high_stock_products = [
            ProductFactory.create(stock=random.randint(50, 100))
            for _ in range(5)
        ]
        
        # Low stock products
        low_stock_products = [
            ProductFactory.create(stock=random.randint(1, low_stock_threshold))
            for _ in range(3)
        ]
        
        # Out of stock products
        out_of_stock_products = [
            ProductFactory.create(stock=0)
            for _ in range(2)
        ]
        
        return {
            'high_stock': high_stock_products,
            'low_stock': low_stock_products,
            'out_of_stock': out_of_stock_products,
            'all_products': high_stock_products + low_stock_products + out_of_stock_products
        }

# Faker providers for domain-specific data
class ProductNameProvider(factory.providers.BaseProvider):
    """Custom Faker provider for realistic product names"""
    
    adjectives = ['Premium', 'Professional', 'Deluxe', 'Ultra', 'Pro', 'Advanced', 'Smart']
    product_types = ['Laptop', 'Phone', 'Tablet', 'Watch', 'Camera', 'Speaker', 'Monitor']
    brands = ['TechCorp', 'GadgetPro', 'InnovaMax', 'Elitetech', 'NextGen']
    
    def product_name(self):
        return f"{self.random_element(self.adjectives)} {self.random_element(self.product_types)}"
    
    def product_name_with_brand(self):
        return f"{self.random_element(self.brands)} {self.product_name()}"

# Register custom provider
factory.Faker.add_provider(ProductNameProvider)

# Usage examples for factories
def example_usage():
    """Examples of how to use the factories"""
    
    # Basic usage
    user = UserFactory()
    product = ProductFactory()
    order = OrderFactory()
    
    # With custom attributes
    admin = UserFactory(is_admin=True, username="specific_admin")
    expensive_product = ProductFactory(price=999.99)
    
    # Create related objects
    user_with_order = UserFactory()
    order_for_user = OrderFactory(user=user_with_order)
    
    # Batch creation
    users = UserFactory.create_batch(10)
    products = ProductFactory.create_batch(20)
    
    # Using builders for complex scenarios
    complete_scenario = TestDataBuilder.create_complete_order_scenario()
    inventory_scenario = TestDataBuilder.create_inventory_scenario()
    
    return {
        'users': users,
        'products': products,
        'scenarios': [complete_scenario, inventory_scenario]
    }

Best Practices & Guidelines

1. Test Organization & Structure

• Follow the test pyramid: many unit tests, fewer integration tests, minimal E2E tests
• Use descriptive test names that explain the behavior being tested
• Organize tests by feature/module, not by test type
• Use consistent naming conventions and file structure

2. Test Data Management

• Use factories for consistent test data creation
• Implement proper test isolation with database transactions
• Create minimal test data required for each test
• Use builders for complex test scenarios

3. Mocking & Test Doubles

• Mock external dependencies and side effects
• Use dependency injection to make mocking easier
• Prefer stubs over mocks when testing state
• Don't mock what you don't own (except for external APIs)

4. Performance & Quality

• Maintain high test coverage (80%+ unit, 60%+ integration)
• Use mutation testing to validate test quality
• Keep tests fast and independent
• Implement continuous testing in CI/CD pipelines

5. Advanced Testing Strategies

• Use property-based testing for complex algorithms
• Implement contract testing for API integrations
• Add performance benchmarks for critical paths
• Include security testing in your test suite

This comprehensive testing framework ensures high-quality, reliable Python applications with robust test coverage and maintainable test suites.