Pytest

pytest - Advanced Python Unit Testing for Customer Support

Overview

pytest is the industry-standard testing framework for Python applications, offering powerful features that enable comprehensive, maintainable, and scalable test suites. This skill focuses specifically on customer support tech enablement, providing patterns and practices for testing backend support systems, ticketing platforms, knowledge bases, and customer data platforms.

Customer support systems require rigorous testing due to their mission-critical nature. Downtime or bugs directly impact customer satisfaction, agent productivity, and business operations. This skill provides comprehensive guidance on testing all aspects of support systems using pytest.

Why pytest for Customer Support Systems

Unique Requirements

Customer support applications have specific testing needs:

1. High Reliability: Support systems are mission-critical; failures directly affect customer experience 2. Complex Data Relationships: Tickets, customers, agents, comments, attachments, and knowledge articles 3. External Integrations: Email services (SendGrid, AWS SES), CRM systems (Salesforce, HubSpot), payment processors 4. Async Operations: Background jobs, email queues, notification services, webhook deliveries 5. Data Validation: Strict validation for customer PII, ticket priorities, SLA requirements, escalation rules 6. Multi-tenancy: Isolation between different customer organizations or workspaces 7. API-First Design: RESTful APIs requiring comprehensive endpoint coverage 8. Real-time Features: WebSocket connections, live chat, real-time ticket updates 9. Compliance: GDPR, CCPA, data retention policies, audit logging

pytest Advantages

pytest addresses these needs through:

• Powerful Fixture System: Manage complex database setups, API clients, and test data
• Parametrization: Test multiple scenarios efficiently (various ticket types, priority levels, user roles)
• Rich Plugin Ecosystem: pytest-asyncio for async testing, pytest-mock for mocking, pytest-cov for coverage
• Excellent Integration: Works seamlessly with FastAPI, SQLAlchemy, PostgreSQL, Pydantic
• Clear Output: Readable test results and comprehensive error reporting
• Scalability: Handles small test suites to thousands of tests with parallel execution
• Flexibility: Supports unit, integration, and end-to-end testing in one framework

Core Competencies

1. Fixtures and Dependency Injection

Fixtures are pytest's killer feature, providing reusable setup/teardown logic and dependency injection. For customer support systems, fixtures manage databases, API clients, test data, and external service mocks.

Basic Fixtures

import pytest
from app.models import Ticket, Customer, Agent

@pytest.fixture
def support_ticket():
    """Provide a basic support ticket dictionary."""
    return {
        "id": 1,
        "title": "Cannot access account",
        "description": "User unable to login after password reset",
        "status": "open",
        "priority": "high",
        "customer_email": "user@example.com",
        "category": "authentication"
    }

def test_ticket_structure(support_ticket):
    """Test ticket has required fields."""
    assert support_ticket["status"] == "open"
    assert support_ticket["priority"] in ["low", "medium", "high", "critical"]
    assert "@" in support_ticket["customer_email"]

Fixture Scopes

Control when fixtures are created and destroyed using scopes:

• function (default): Created once per test function, destroyed after test completes
• class: Created once per test class, shared across all methods
• module: Created once per module file, shared across all tests in file
• package: Created once per package, shared across all tests in package
• session: Created once per entire test session, shared across all tests

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from typing import Generator

@pytest.fixture(scope="session")
def database_engine():
    """Create database engine once per session."""
    engine = create_engine(
        "postgresql://support_user:password@localhost/support_test",
        echo=False,
        pool_pre_ping=True,
        pool_size=10,
        max_overflow=20
    )
    # Create all tables
    from app.database import Base
    Base.metadata.create_all(engine)

    yield engine

    # Cleanup: drop all tables and dispose engine
    Base.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture(scope="function")
def db_session(database_engine) -> Generator[Session, None, None]:
    """Create a new database session for each test with automatic rollback."""
    SessionLocal = sessionmaker(bind=database_engine)
    session = SessionLocal()

    try:
        yield session
    finally:
        session.rollback()  # Rollback any changes
        session.close()

Autouse Fixtures

Fixtures that run automatically without being explicitly requested:

@pytest.fixture(autouse=True)
def reset_caches():
    """Clear all caches before each test."""
    from app.cache import ticket_cache, customer_cache, agent_cache
    ticket_cache.clear()
    customer_cache.clear()
    agent_cache.clear()
    yield
    # Optional cleanup after test
    ticket_cache.clear()
    customer_cache.clear()
    agent_cache.clear()

@pytest.fixture(autouse=True, scope="session")
def configure_logging():
    """Configure logging for test session."""
    import logging
    logging.basicConfig(level=logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.ERROR)

Fixture Dependencies and Composition

Fixtures can depend on other fixtures, creating dependency chains:

@pytest.fixture
def database_url():
    """Provide database URL for testing."""
    return "postgresql://test:test@localhost:5432/support_test"

@pytest.fixture
def engine(database_url):
    """Create SQLAlchemy engine from database URL."""
    from sqlalchemy import create_engine
    return create_engine(database_url, echo=False)

@pytest.fixture
def session(engine):
    """Create database session from engine."""
    from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()

@pytest.fixture
def sample_customer(session):
    """Create a sample customer using database session."""
    customer = Customer(
        email="test@example.com",
        name="Test Customer",
        tier="premium",
        company="Acme Corp"
    )
    session.add(customer)
    session.commit()
    return customer

@pytest.fixture
def sample_ticket(session, sample_customer):
    """Create a sample ticket linked to sample customer."""
    ticket = Ticket(
        title="Test Ticket",
        description="Test description",
        priority="high",
        status="open",
        customer_id=sample_customer.id
    )
    session.add(ticket)
    session.commit()
    return ticket

Factory Fixtures

Create fixtures that return factory functions for flexible test data creation:

@pytest.fixture
def ticket_factory(db_session):
    """Factory for creating tickets with custom attributes."""
    created_tickets = []

    def _create_ticket(
        title="Default Ticket",
        description="Default description",
        priority="medium",
        status="open",
        **kwargs
    ):
        ticket = Ticket(
            title=title,
            description=description,
            priority=priority,
            status=status,
            **kwargs
        )
        db_session.add(ticket)
        db_session.commit()
        created_tickets.append(ticket)
        return ticket

    yield _create_ticket

    # Cleanup: delete all created tickets
    for ticket in created_tickets:
        db_session.delete(ticket)
    db_session.commit()

def test_multiple_tickets(ticket_factory):
    """Test using factory to create multiple tickets."""
    high_priority = ticket_factory(priority="high", title="Urgent Issue")
    low_priority = ticket_factory(priority="low", title="Minor Request")

    assert high_priority.priority == "high"
    assert low_priority.priority == "low"

2. Parametrization

Parametrization allows running the same test with different inputs, essential for comprehensive testing of customer support systems with various priority levels, statuses, user roles, and edge cases.

Basic Parametrization

@pytest.mark.parametrize("priority,expected_sla_hours", [
    ("critical", 1),
    ("high", 4),
    ("medium", 24),
    ("low", 72)
])
def test_sla_calculation(priority, expected_sla_hours):
    """Test SLA hours calculation for different priorities."""
    from app.services.sla import calculate_sla_hours
    sla = calculate_sla_hours(priority)
    assert sla == expected_sla_hours

@pytest.mark.parametrize("email,is_valid", [
    ("user@example.com", True),
    ("user.name+tag@example.co.uk", True),
    ("user@localhost", True),
    ("invalid.email", False),
    ("@example.com", False),
    ("user@", False),
    ("", False),
    ("user @example.com", False),
    ("user@example .com", False),
])
def test_email_validation(email, is_valid):
    """Test email validation with various valid and invalid inputs."""
    from app.validators import is_valid_email
    assert is_valid_email(email) == is_valid

Multiple Parameters

@pytest.mark.parametrize("status,priority,assigned,expected_urgent", [
    ("open", "critical", False, True),   # Unassigned critical = urgent
    ("open", "high", False, True),       # Unassigned high = urgent
    ("open", "medium", False, False),    # Unassigned medium = not urgent
    ("open", "low", False, False),       # Unassigned low = not urgent
    ("in_progress", "critical", True, False),  # Assigned critical = not urgent
    ("in_progress", "high", True, False),      # Assigned high = not urgent
    ("closed", "critical", True, False),       # Closed tickets never urgent
    ("resolved", "high", True, False),         # Resolved tickets never urgent
])
def test_urgent_ticket_identification(status, priority, assigned, expected_urgent):
    """Test identification of urgent tickets requiring immediate attention."""
    from app.models import Ticket
    ticket = Ticket(status=status, priority=priority)
    if assigned:
        ticket.assigned_agent_id = 123

    assert ticket.is_urgent() == expected_urgent

Parametrizing Fixtures

@pytest.fixture(params=["sqlite", "postgresql"])
def database_type(request):
    """Parametrized fixture for different database types."""
    return request.param

def test_database_connection(database_type):
    """Test runs twice: once with SQLite, once with PostgreSQL."""
    from app.database import get_connection
    connection = get_connection(database_type)
    assert connection.is_connected()
    connection.close()

@pytest.fixture(params=[
    {"role": "admin", "can_delete": True, "can_reassign": True},
    {"role": "agent", "can_delete": False, "can_reassign": True},
    {"role": "viewer", "can_delete": False, "can_reassign": False},
])
def user_permissions(request):
    """Parametrized fixture for different user roles and permissions."""
    return request.param

def test_permission_checks(user_permissions):
    """Test permission validation for different roles."""
    from app.auth import check_permission

    can_delete = check_permission(user_permissions["role"], "delete_ticket")
    can_reassign = check_permission(user_permissions["role"], "reassign_ticket")

    assert can_delete == user_permissions["can_delete"]
    assert can_reassign == user_permissions["can_reassign"]

Using pytest.param for Complex Cases

@pytest.mark.parametrize("email,expected_valid,reason", [
    pytest.param("user@example.com", True, "valid", id="valid-standard-email"),
    pytest.param("user+tag@example.com", True, "valid", id="valid-with-plus"),
    pytest.param("invalid", False, "missing-at", id="no-at-symbol"),
    pytest.param("@example.com", False, "no-user", id="missing-username"),
    pytest.param("user@", False, "no-domain", id="missing-domain"),
    pytest.param("user@.com", False, "invalid-domain", id="dot-at-domain-start"),
])
def test_email_validation_detailed(email, expected_valid, reason):
    """Test email validation with detailed reasoning."""
    from app.validators import validate_email_detailed
    result = validate_email_detailed(email)
    assert result.is_valid == expected_valid
    if not expected_valid:
        assert reason in result.error_code.lower()

Parametrization with IDs

@pytest.mark.parametrize("ticket_data,expected_status", [
    ({"priority": "critical", "auto_assign": True}, "assigned"),
    ({"priority": "high", "auto_assign": True}, "assigned"),
    ({"priority": "medium", "auto_assign": False}, "open"),
    ({"priority": "low", "auto_assign": False}, "open"),
], ids=["critical-auto", "high-auto", "medium-manual", "low-manual"])
def test_auto_assignment(ticket_data, expected_status):
    """Test automatic ticket assignment based on priority."""
    from app.services.assignment import process_new_ticket
    ticket = process_new_ticket(**ticket_data)
    assert ticket.status == expected_status

3. Mocking with pytest-mock

Mocking is essential for isolating units of code from external dependencies like email services, payment gateways, CRM systems, and third-party APIs. The pytest-mock plugin provides a clean interface to Python's unittest.mock.

Basic Mocking

def test_ticket_notification(mocker, db_session):
    """Test that creating a ticket sends email notification."""
    # Mock the email service
    mock_send = mocker.patch('app.services.email.EmailService.send_email')
    mock_send.return_value = {"status": "sent", "message_id": "msg_123"}

    # Create ticket
    from app.services.ticket import TicketService
    service = TicketService(db_session)
    ticket = service.create_ticket(
        title="Login Issue",
        description="Cannot access account",
        customer_email="customer@example.com",
        priority="high"
    )

    # Verify email was sent
    mock_send.assert_called_once()
    call_args = mock_send.call_args
    assert call_args.kwargs['to'] == "customer@example.com"
    assert "Login Issue" in call_args.kwargs['subject']
    assert ticket.id is not None

Mocking Return Values

def test_customer_tier_lookup(mocker, api_client):
    """Test customer tier lookup from external CRM."""
    # Mock external CRM API response
    mock_response = mocker.Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {
        "customer_id": "CRM_12345",
        "tier": "enterprise",
        "support_plan": "24/7 premium",
        "account_manager": "jane@company.com"
    }

    mocker.patch('requests.get', return_value=mock_response)

    # Test endpoint that uses CRM data
    response = api_client.get("/api/customers/customer@example.com/tier")

    assert response.status_code == 200
    data = response.json()
    assert data["tier"] == "enterprise"
    assert data["support_plan"] == "24/7 premium"

Mocking Side Effects

def test_retry_logic_on_api_failure(mocker):
    """Test retry mechanism when external API fails temporarily."""
    from app.services.external import call_with_retry

    mock_api = mocker.patch('app.services.external.call_external_api')
    # First two calls fail, third succeeds
    mock_api.side_effect = [
        ConnectionError("Connection timeout"),
        ConnectionError("Connection timeout"),
        {"success": True, "data": "response"}
    ]

    result = call_with_retry(max_attempts=3, backoff_seconds=0)

    assert result["success"] is True
    assert result["data"] == "response"
    assert mock_api.call_count == 3

def test_email_fallback_on_primary_failure(mocker):
    """Test email service fallback when primary provider fails."""
    from app.services.email import send_email_with_fallback

    mock_primary = mocker.patch('app.services.email.sendgrid_send')
    mock_fallback = mocker.patch('app.services.email.ses_send')

    # Primary fails, fallback succeeds
    mock_primary.side_effect = Exception("SendGrid API error")
    mock_fallback.return_value = {"status": "sent", "provider": "ses"}

    result = send_email_with_fallback(
        to="customer@example.com",
        subject="Ticket Update",
        body="Your ticket has been updated"
    )

    assert result["status"] == "sent"
    assert result["provider"] == "ses"
    mock_primary.assert_called_once()
    mock_fallback.assert_called_once()

Spying on Real Methods

def test_cache_utilization(mocker, db_session):
    """Spy on cache to verify it's being used correctly."""
    from app.cache import ticket_cache

    spy_get = mocker.spy(ticket_cache, 'get')
    spy_set = mocker.spy(ticket_cache, 'set')

    from app.services.ticket import get_ticket

    # First call: cache miss, should query database
    ticket1 = get_ticket(123)
    assert spy_get.call_count == 1
    assert spy_set.call_count == 1

    # Second call: cache hit, should not query database
    ticket2 = get_ticket(123)
    assert spy_get.call_count == 2
    assert spy_set.call_count == 1  # Not called again

    assert ticket1.id == ticket2.id

Mocking Database Operations

def test_ticket_creation_business_logic(mocker):
    """Test ticket creation logic without database."""
    mock_session = mocker.Mock()
    mock_session.add = mocker.Mock()
    mock_session.commit = mocker.Mock()
    mock_session.refresh = mocker.Mock(side_effect=lambda obj: setattr(obj, 'id', 123))

    from app.services.ticket import create_ticket_entity
    ticket = create_ticket_entity(
        mock_session,
        title="Test Ticket",
        priority="high"
    )

    assert ticket.id == 123
    assert ticket.priority == "high"
    mock_session.add.assert_called_once()
    mock_session.commit.assert_called_once()

Async Mocking

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_async_notification_service(mocker):
    """Test async notification service with proper async mocking."""
    mock_send = mocker.patch(
        'app.services.notification.send_push_notification',
        new_callable=AsyncMock
    )
    mock_send.return_value = {
        "delivered": True,
        "notification_id": "notif_789"
    }

    from app.services.notification import NotificationService
    service = NotificationService()
    result = await service.notify_agent(
        agent_id=456,
        message="New high-priority ticket assigned to you",
        priority="high"
    )

    assert result["delivered"] is True
    assert result["notification_id"] is not None
    mock_send.assert_awaited_once()

    # Verify call arguments
    call_kwargs = mock_send.call_args.kwargs
    assert call_kwargs["agent_id"] == 456
    assert "high-priority" in call_kwargs["message"]

4. Coverage Reporting

Code coverage measures what percentage of your code is executed during testing. For customer support systems, comprehensive coverage ensures reliability and helps identify untested edge cases.

Basic Coverage Setup

# Install pytest-cov
pip install pytest-cov

# Run tests with coverage
pytest --cov=app tests/

# Generate HTML report
pytest --cov=app --cov-report=html tests/

# Show lines not covered
pytest --cov=app --cov-report=term-missing tests/

# Fail if coverage below threshold
pytest --cov=app --cov-fail-under=80 tests/

Configuration in pytest.ini

[pytest]
addopts =
    --cov=app
    --cov-report=html
    --cov-report=term-missing
    --cov-report=xml
    --cov-fail-under=80
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
markers =
    unit: Unit tests
    integration: Integration tests
    slow: Slow running tests

Coverage Configuration (.coveragerc)

[run]
source = app
omit =
    */tests/*
    */migrations/*
    */__init__.py
    */config.py
    */venv/*
    */.venv/*

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    def __str__
    raise AssertionError
    raise NotImplementedError
    if __name__ == .__main__.:
    if TYPE_CHECKING:
    @abstractmethod
    @abc.abstractmethod
    except ImportError
precision = 2
show_missing = True

[html]
directory = htmlcov
title = Customer Support API Coverage Report

Branch Coverage

[run]
source = app
branch = True  # Enable branch coverage

[report]
show_missing = True
skip_covered = False

# Run with branch coverage
pytest --cov=app --cov-branch --cov-report=html

5. Async Testing

Modern customer support systems use asynchronous operations for better performance and scalability. Testing async code requires special handling to properly await coroutines.

Basic Async Tests

import pytest

@pytest.mark.asyncio
async def test_async_ticket_creation():
    """Test asynchronous ticket creation service."""
    from app.services.ticket import create_ticket_async

    ticket = await create_ticket_async(
        title="Async Test Ticket",
        description="Testing async operations",
        priority="high",
        customer_email="async@example.com"
    )

    assert ticket.id is not None
    assert ticket.status == "open"
    assert ticket.priority == "high"

Async Fixtures

import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

@pytest.fixture
async def async_db_session():
    """Provide async database session."""
    engine = create_async_engine(
        "postgresql+asyncpg://user:pass@localhost/support_test",
        echo=False,
        future=True
    )

    async_session_maker = sessionmaker(
        engine,
        class_=AsyncSession,
        expire_on_commit=False
    )

    async with async_session_maker() as session:
        yield session
        await session.rollback()

    await engine.dispose()

@pytest.mark.asyncio
async def test_async_database_operation(async_db_session):
    """Test async database operations."""
    from app.models import Ticket
    from sqlalchemy import select

    # Create ticket
    ticket = Ticket(title="Async DB Test", priority="medium")
    async_db_session.add(ticket)
    await async_db_session.commit()

    # Query ticket
    result = await async_db_session.execute(
        select(Ticket).where(Ticket.title == "Async DB Test")
    )
    found_ticket = result.scalar_one()

    assert found_ticket.id is not None
    assert found_ticket.priority == "medium"

Testing Async Background Tasks

@pytest.mark.asyncio
async def test_background_email_task(mocker):
    """Test async background task for sending emails."""
    from app.tasks.email import process_ticket_notifications

    mock_send = mocker.patch(
        'app.tasks.email.send_email_async',
        new_callable=AsyncMock
    )
    mock_send.return_value = {"status": "sent", "message_id": "msg_456"}

    await process_ticket_notifications(ticket_id=123)

    mock_send.assert_awaited_once()
    call_kwargs = mock_send.call_args.kwargs
    assert "ticket_id" in call_kwargs or call_kwargs.get("ticket_id") == 123

Testing Concurrent Operations

import asyncio

@pytest.mark.asyncio
async def test_concurrent_ticket_updates():
    """Test handling of concurrent ticket updates."""
    from app.services.ticket import update_ticket_async, add_comment_async

    ticket_id = 1

    async def update_status():
        return await update_ticket_async(ticket_id, status="in_progress")

    async def add_comment():
        return await add_comment_async(ticket_id, "Working on this issue")

    async def assign_agent():
        return await update_ticket_async(ticket_id, assigned_agent_id=42)

    # Execute concurrently
    results = await asyncio.gather(
        update_status(),
        add_comment(),
        assign_agent()
    )

    assert results[0]["status"] == "in_progress"
    assert results[1]["comment_count"] >= 1
    assert results[2]["assigned_agent_id"] == 42

Async Test Configuration

# pytest.ini or pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # Automatically detect and run async tests

# Or in pytest.ini
[pytest]
asyncio_mode = auto

6. FastAPI Testing

FastAPI is widely used for customer support APIs. Testing endpoints ensures proper request/response handling, validation, authentication, and business logic.

Basic FastAPI Testing

from fastapi.testclient import TestClient
from app.main import app

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

def test_create_ticket_endpoint(client):
    """Test ticket creation via POST endpoint."""
    response = client.post(
        "/api/v1/tickets",
        json={
            "title": "Cannot login to application",
            "description": "User receives 'invalid credentials' error",
            "priority": "high",
            "customer_email": "frustrated@example.com",
            "category": "authentication"
        }
    )

    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Cannot login to application"
    assert data["priority"] == "high"
    assert "id" in data
    assert "created_at" in data

def test_get_ticket_by_id(client, sample_tickets):
    """Test retrieving specific ticket by ID."""
    ticket_id = sample_tickets[0].id

    response = client.get(f"/api/v1/tickets/{ticket_id}")

    assert response.status_code == 200
    data = response.json()
    assert data["id"] == ticket_id
    assert data["title"] == sample_tickets[0].title

def test_update_ticket_status(client, sample_tickets):
    """Test updating ticket status via PATCH."""
    ticket_id = sample_tickets[0].id

    response = client.patch(
        f"/api/v1/tickets/{ticket_id}",
        json={"status": "in_progress"}
    )

    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "in_progress"
    assert data["id"] == ticket_id

Dependency Override

FastAPI's dependency injection can be overridden for testing:

from app.dependencies import get_db, get_current_user
from app.models import User

@pytest.fixture
def authenticated_client(client, mocker):
    """Provide authenticated test client."""
    mock_user = User(
        id=1,
        email="agent@support.com",
        role="agent",
        name="Test Agent"
    )

    async def override_get_current_user():
        return mock_user

    app.dependency_overrides[get_current_user] = override_get_current_user
    yield client
    app.dependency_overrides.clear()

def test_protected_endpoint(authenticated_client):
    """Test endpoint requiring authentication."""
    response = authenticated_client.get("/api/v1/admin/users")
    assert response.status_code == 200
    data = response.json()
    assert isinstance(data, list)

@pytest.fixture
def admin_client(client, mocker):
    """Provide client authenticated as admin."""
    mock_admin = User(
        id=1,
        email="admin@support.com",
        role="admin",
        permissions=["read", "write", "delete", "admin"]
    )

    async def override_get_current_user():
        return mock_admin

    app.dependency_overrides[get_current_user] = override_get_current_user
    yield client
    app.dependency_overrides.clear()

def test_admin_only_endpoint(admin_client):
    """Test endpoint requiring admin privileges."""
    response = admin_client.delete("/api/v1/tickets/123")
    assert response.status_code in [200, 204, 404]  # Success or not found

Testing with Database

@pytest.fixture
def client_with_db(db_session):
    """Provide FastAPI client with database session override."""
    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()

def test_ticket_lifecycle(client_with_db, db_session):
    """Test complete ticket lifecycle with database persistence."""
    # Create ticket
    create_response = client_with_db.post(
        "/api/v1/tickets",
        json={
            "title": "Lifecycle Test",
            "description": "Testing full CRUD cycle",
            "priority": "medium"
        }
    )
    assert create_response.status_code == 201
    ticket_id = create_response.json()["id"]

    # Read ticket
    get_response = client_with_db.get(f"/api/v1/tickets/{ticket_id}")
    assert get_response.status_code == 200
    assert get_response.json()["title"] == "Lifecycle Test"

    # Update ticket
    update_response = client_with_db.patch(
        f"/api/v1/tickets/{ticket_id}",
        json={"status": "resolved", "priority": "low"}
    )
    assert update_response.status_code == 200
    assert update_response.json()["status"] == "resolved"

    # Verify in database
    from app.models import Ticket
    ticket = db_session.query(Ticket).filter_by(id=ticket_id).first()
    assert ticket is not None
    assert ticket.status == "resolved"
    assert ticket.priority == "low"

    # Delete ticket
    delete_response = client_with_db.delete(f"/api/v1/tickets/{ticket_id}")
    assert delete_response.status_code in [200, 204]

    # Verify deletion
    get_deleted = client_with_db.get(f"/api/v1/tickets/{ticket_id}")
    assert get_deleted.status_code == 404

Testing Request Validation

def test_invalid_ticket_creation(client):
    """Test that invalid ticket data returns proper validation errors."""
    # Missing required field
    response = client.post(
        "/api/v1/tickets",
        json={"description": "Missing title"}
    )
    assert response.status_code == 422

    # Invalid priority
    response = client.post(
        "/api/v1/tickets",
        json={
            "title": "Test",
            "description": "Test",
            "priority": "invalid_priority"
        }
    )
    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any("priority" in str(e).lower() for e in errors)

    # Invalid email format
    response = client.post(
        "/api/v1/tickets",
        json={
            "title": "Test",
            "description": "Test",
            "customer_email": "not_an_email"
        }
    )
    assert response.status_code == 422

7. SQLAlchemy Testing

Testing database models, queries, relationships, and operations ensures data integrity in customer support systems.

Model Testing

from app.models import Ticket, Comment, User, Customer

def test_ticket_model_creation(db_session):
    """Test creating and persisting ticket model."""
    customer = Customer(
        email="customer@example.com",
        name="John Doe",
        tier="enterprise"
    )
    db_session.add(customer)
    db_session.flush()

    ticket = Ticket(
        title="Database Connection Error",
        description="Cannot connect to production database",
        priority="critical",
        status="open",
        customer_id=customer.id,
        category="technical"
    )
    db_session.add(ticket)
    db_session.commit()

    assert ticket.id is not None
    assert ticket.created_at is not None
    assert ticket.customer.email == "customer@example.com"

def test_ticket_model_defaults(db_session):
    """Test default values on ticket model."""
    ticket = Ticket(title="Test", description="Test")
    db_session.add(ticket)
    db_session.commit()

    assert ticket.status == "open"  # Default status
    assert ticket.priority == "medium"  # Default priority
    assert ticket.created_at is not None
    assert ticket.updated_at is not None

Relationship Testing

def test_ticket_comments_relationship(db_session):
    """Test one-to-many relationship between tickets and comments."""
    ticket = Ticket(
        title="Test Ticket",
        description="Testing relationships",
        priority="low"
    )
    db_session.add(ticket)
    db_session.flush()

    # Add multiple comments
    comment1 = Comment(
        ticket_id=ticket.id,
        content="First comment",
        author_email="agent1@support.com"
    )
    comment2 = Comment(
        ticket_id=ticket.id,
        content="Second comment",
        author_email="agent2@support.com"
    )
    db_session.add_all([comment1, comment2])
    db_session.commit()

    # Reload ticket and verify relationship
    db_session.expire(ticket)
    assert len(ticket.comments) == 2
    assert ticket.comments[0].content == "First comment"
    assert ticket.comments[1].content == "Second comment"

def test_customer_tickets_relationship(db_session):
    """Test one-to-many relationship between customers and tickets."""
    customer = Customer(
        email="loyal@customer.com",
        name="Loyal Customer"
    )
    db_session.add(customer)
    db_session.flush()

    # Create multiple tickets for same customer
    for i in range(3):
        ticket = Ticket(
            title=f"Issue #{i+1}",
            description=f"Description {i+1}",
            customer_id=customer.id
        )
        db_session.add(ticket)

    db_session.commit()

    # Verify customer has all tickets
    db_session.expire(customer)
    assert len(customer.tickets) == 3
    assert all(t.customer_id == customer.id for t in customer.tickets)

Query Testing

def test_filter_tickets_by_status(db_session):
    """Test filtering tickets by status."""
    # Create tickets with different statuses
    tickets = [
        Ticket(title=f"Ticket {i}", status="open" if i % 2 == 0 else "closed")
        for i in range(10)
    ]
    db_session.add_all(tickets)
    db_session.commit()

    # Query open tickets
    open_tickets = db_session.query(Ticket).filter_by(status="open").all()
    assert len(open_tickets) == 5
    assert all(t.status == "open" for t in open_tickets)

    # Query closed tickets
    closed_tickets = db_session.query(Ticket).filter_by(status="closed").all()
    assert len(closed_tickets) == 5
    assert all(t.status == "closed" for t in closed_tickets)

def test_complex_ticket_query(db_session):
    """Test complex multi-filter ticket query."""
    from sqlalchemy import and_, or_
    from datetime import datetime, timedelta

    # Create diverse set of tickets
    old_critical = Ticket(
        title="Old Critical",
        priority="critical",
        status="open",
        created_at=datetime.utcnow() - timedelta(days=7)
    )
    recent_high = Ticket(
        title="Recent High",
        priority="high",
        status="open",
        created_at=datetime.utcnow() - timedelta(hours=2)
    )
    old_low = Ticket(
        title="Old Low",
        priority="low",
        status="open",
        created_at=datetime.utcnow() - timedelta(days=30)
    )

    db_session.add_all([old_critical, recent_high, old_low])
    db_session.commit()

    # Query: open tickets that are either critical OR created in last 24 hours
    cutoff = datetime.utcnow() - timedelta(days=1)
    urgent_tickets = db_session.query(Ticket).filter(
        and_(
            Ticket.status == "open",
            or_(
                Ticket.priority == "critical",
                Ticket.created_at >= cutoff
            )
        )
    ).all()

    assert len(urgent_tickets) == 2
    ticket_titles = [t.title for t in urgent_tickets]
    assert "Old Critical" in ticket_titles
    assert "Recent High" in ticket_titles
    assert "Old Low" not in ticket_titles

Transaction Testing

def test_transaction_rollback_on_error(db_session):
    """Test that failed transactions are properly rolled back."""
    from sqlalchemy.exc import IntegrityError

    # Create initial ticket
    ticket1 = Ticket(title="First Ticket", description="Test")
    db_session.add(ticket1)
    db_session.commit()

    initial_count = db_session.query(Ticket).count()

    # Attempt to create invalid ticket (will fail)
    try:
        ticket2 = Ticket(title="Second Ticket", description="Test")
        db_session.add(ticket2)
        db_session.flush()

        # Simulate constraint violation
        ticket3 = Ticket(
            title="Third Ticket",
            external_id="DUPLICATE_ID"  # Assuming unique constraint
        )
        db_session.add(ticket3)

        ticket4 = Ticket(
            title="Fourth Ticket",
            external_id="DUPLICATE_ID"  # Will violate unique constraint
        )
        db_session.add(ticket4)
        db_session.commit()
    except IntegrityError:
        db_session.rollback()

    # Verify count hasn't changed
    final_count = db_session.query(Ticket).count()
    assert final_count == initial_count

8. Database Fixtures

Well-designed database fixtures are crucial for testing customer support systems with complex data relationships.

Comprehensive Database Fixture

import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from app.database import Base

@pytest.fixture(scope="session")
def db_engine():
    """Create test database engine for entire session."""
    engine = create_engine(
        "postgresql://test:test@localhost:5432/support_test",
        echo=False,
        pool_pre_ping=True
    )

    # Create all tables
    Base.metadata.create_all(engine)

    yield engine

    # Cleanup: drop all tables
    Base.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture
def db_session(db_engine):
    """Provide clean database session with automatic rollback."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()

    # Enable savepoints for nested transactions
    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(session, transaction):
        if transaction.nested and not transaction._parent.nested:
            session.expire_all()
            session.begin_nested()

    session.begin_nested()

    yield session

    # Cleanup
    session.close()
    transaction.rollback()
    connection.close()

Sample Data Fixtures

@pytest.fixture
def sample_customers(db_session):
    """Create sample customers with various tiers."""
    customers = [
        Customer(email="free@example.com", name="Free User", tier="free"),
        Customer(email="basic@example.com", name="Basic User", tier="basic"),
        Customer(email="premium@example.com", name="Premium User", tier="premium"),
        Customer(email="enterprise@example.com", name="Enterprise User", tier="enterprise"),
    ]
    db_session.add_all(customers)
    db_session.commit()
    return customers

@pytest.fixture
def sample_agents(db_session):
    """Create sample support agents."""
    agents = [
        User(email="agent1@support.com", name="Agent One", role="agent"),
        User(email="agent2@support.com", name="Agent Two", role="agent"),
        User(email="senior@support.com", name="Senior Agent", role="senior_agent"),
    ]
    db_session.add_all(agents)
    db_session.commit()
    return agents

@pytest.fixture
def sample_tickets(db_session, sample_customers):
    """Create diverse set of sample tickets."""
    tickets = []
    priorities = ["low", "medium", "high", "critical"]
    statuses = ["open", "in_progress", "resolved", "closed"]

    for i in range(12):
        ticket = Ticket(
            title=f"Sample Ticket #{i+1}",
            description=f"Description for ticket {i+1}",
            priority=priorities[i % len(priorities)],
            status=statuses[i % len(statuses)],
            customer_id=sample_customers[i % len(sample_customers)].id,
            category=["technical", "billing", "account"][i % 3]
        )
        tickets.append(ticket)

    db_session.add_all(tickets)
    db_session.commit()
    return tickets

Factory Pattern Fixtures

@pytest.fixture
def customer_factory(db_session):
    """Factory for creating customers with custom attributes."""
    created = []

    def _create_customer(
        email=None,
        name=None,
        tier="basic",
        **kwargs
    ):
        import uuid
        customer = Customer(
            email=email or f"customer{uuid.uuid4().hex[:8]}@example.com",
            name=name or f"Customer {uuid.uuid4().hex[:8]}",
            tier=tier,
            **kwargs
        )
        db_session.add(customer)
        db_session.commit()
        created.append(customer)
        return customer

    yield _create_customer

    # Cleanup
    for customer in created:
        db_session.delete(customer)
    db_session.commit()

@pytest.fixture
def ticket_factory(db_session, customer_factory):
    """Factory for creating tickets with automatic customer creation."""
    created = []

    def _create_ticket(
        title=None,
        customer=None,
        priority="medium",
        status="open",
        **kwargs
    ):
        import uuid

        if customer is None:
            customer = customer_factory()

        ticket = Ticket(
            title=title or f"Ticket {uuid.uuid4().hex[:8]}",
            description=kwargs.pop("description", "Test ticket description"),
            customer_id=customer.id,
            priority=priority,
            status=status,
            **kwargs
        )
        db_session.add(ticket)
        db_session.commit()
        created.append(ticket)
        return ticket

    yield _create_ticket

    # Cleanup
    for ticket in created:
        db_session.delete(ticket)
    db_session.commit()

# Usage in tests
def test_with_factories(customer_factory, ticket_factory):
    """Test using factory fixtures for flexible test data creation."""
    # Create premium customer
    premium_customer = customer_factory(tier="enterprise", name="Big Corp")

    # Create multiple tickets for same customer
    critical_ticket = ticket_factory(
        customer=premium_customer,
        priority="critical",
        title="Production outage"
    )
    normal_ticket = ticket_factory(
        customer=premium_customer,
        priority="low",
        title="Feature request"
    )

    assert critical_ticket.customer_id == premium_customer.id
    assert normal_ticket.customer_id == premium_customer.id
    assert premium_customer.tier == "enterprise"

9. Customer Support Testing Scenarios

Real-world testing scenarios specific to customer support systems.

Testing Ticket Assignment Logic

def test_round_robin_assignment(db_session, sample_agents):
    """Test round-robin ticket assignment to available agents."""
    from app.services.assignment import assign_ticket_round_robin

    tickets = []
    for i in range(9):
        ticket = Ticket(
            title=f"Assignment Test {i}",
            priority="medium",
            status="open"
        )
        db_session.add(ticket)
        db_session.flush()

        assign_ticket_round_robin(ticket, db_session)
        tickets.append(ticket)

    db_session.commit()

    # Verify even distribution
    for agent in sample_agents:
        assigned_count = len([t for t in tickets if t.assigned_agent_id == agent.id])
        assert assigned_count == 3  # 9 tickets / 3 agents = 3 each

def test_priority_based_assignment(db_session, sample_agents):
    """Test that critical tickets go to senior agents."""
    from app.services.assignment import assign_ticket_by_priority

    critical_ticket = Ticket(
        title="Critical Issue",
        priority="critical",
        status="open"
    )
    db_session.add(critical_ticket)
    db_session.flush()

    assign_ticket_by_priority(critical_ticket, db_session)
    db_session.commit()

    # Find senior agent
    senior_agent = next(a for a in sample_agents if a.role == "senior_agent")
    assert critical_ticket.assigned_agent_id == senior_agent.id

Testing SLA Compliance

from datetime import datetime, timedelta

def test_sla_breach_detection(db_session):
    """Test detection of SLA breaches for different priorities."""
    from app.services.sla import find_sla_breaches

    now = datetime.utcnow()

    # Critical ticket created 2 hours ago (SLA: 1 hour) - BREACHED
    breached_critical = Ticket(
        title="Critical - Breached",
        priority="critical",
        status="open",
        created_at=now - timedelta(hours=2)
    )

    # High priority created 5 hours ago (SLA: 4 hours) - BREACHED
    breached_high = Ticket(
        title="High - Breached",
        priority="high",
        status="open",
        created_at=now - timedelta(hours=5)
    )

    # Medium priority created 2 hours ago (SLA: 24 hours) - OK
    ok_medium = Ticket(
        title="Medium - OK",
        priority="medium",
        status="open",
        created_at=now - timedelta(hours=2)
    )

    # Critical ticket but already assigned (not breached)
    assigned_critical = Ticket(
        title="Critical - Assigned",
        priority="critical",
        status="in_progress",
        created_at=now - timedelta(hours=2),
        assigned_agent_id=1
    )

    db_session.add_all([breached_critical, breached_high, ok_medium, assigned_critical])
    db_session.commit()

    breaches = find_sla_breaches(db_session)
    breach_titles = [b.title for b in breaches]

    assert len(breaches) == 2
    assert "Critical - Breached" in breach_titles
    assert "High - Breached" in breach_titles
    assert "Medium - OK" not in breach_titles
    assert "Critical - Assigned" not in breach_titles

def test_sla_time_remaining_calculation(db_session):
    """Test calculation of remaining SLA time."""
    from app.services.sla import calculate_sla_remaining

    now = datetime.utcnow()

    # High priority ticket created 1 hour ago (SLA: 4 hours)
    ticket = Ticket(
        title="SLA Test",
        priority="high",
        status="open",
        created_at=now - timedelta(hours=1)
    )
    db_session.add(ticket)
    db_session.commit()

    remaining = calculate_sla_remaining(ticket)

    # Should have ~3 hours remaining (4 hour SLA - 1 hour elapsed)
    assert 2.9 <= remaining.total_seconds() / 3600 <= 3.1

Testing Escalation Rules

def test_automatic_escalation(db_session):
    """Test automatic ticket escalation for overdue tickets."""
    from app.services.escalation import escalate_overdue_tickets

    now = datetime.utcnow()

    # Medium priority ticket open for 3 days without assignment
    overdue_ticket = Ticket(
        title="Overdue Ticket",
        priority="medium",
        status="open",
        created_at=now - timedelta(days=3)
    )

    # Recent ticket (should not escalate)
    recent_ticket = Ticket(
        title="Recent Ticket",
        priority="medium",
        status="open",
        created_at=now - timedelta(hours=2)
    )

    db_session.add_all([overdue_ticket, recent_ticket])
    db_session.commit()

    # Run escalation logic
    escalated_count = escalate_overdue_tickets(db_session)

    db_session.refresh(overdue_ticket)
    db_session.refresh(recent_ticket)

    assert escalated_count == 1
    assert overdue_ticket.priority == "high"  # Escalated from medium
    assert overdue_ticket.escalated is True
    assert recent_ticket.priority == "medium"  # Unchanged
    assert recent_ticket.escalated is False

Testing Customer Notifications

def test_status_change_notification(mocker, db_session):
    """Test email notification sent when ticket status changes."""
    from app.services.ticket import update_ticket_status

    mock_send_email = mocker.patch('app.services.email.send_email')
    mock_send_email.return_value = {"status": "sent"}

    ticket = Ticket(
        title="Notification Test",
        customer_email="customer@example.com",
        status="open"
    )
    db_session.add(ticket)
    db_session.commit()

    # Update status
    update_ticket_status(ticket.id, "resolved", db_session)

    # Verify notification sent
    mock_send_email.assert_called_once()
    call_kwargs = mock_send_email.call_args.kwargs
    assert call_kwargs["to"] == "customer@example.com"
    assert "resolved" in call_kwargs["template_name"].lower()
    assert call_kwargs["template_data"]["ticket_id"] == ticket.id

def test_assignment_notification(mocker, db_session, sample_agents):
    """Test notification sent to agent when ticket assigned."""
    from app.services.ticket import assign_ticket

    mock_notify = mocker.patch('app.services.notification.notify_agent')

    ticket = Ticket(title="Assignment Notification Test", status="open")
    db_session.add(ticket)
    db_session.commit()

    agent = sample_agents[0]
    assign_ticket(ticket.id, agent.id, db_session)

    mock_notify.assert_called_once()
    call_args = mock_notify.call_args
    assert call_args[0][0] == agent.id  # First arg is agent_id
    assert "assigned" in call_args[0][1].lower()  # Second arg is message

10. Integration Patterns

Testing integrations with external services and systems.

Testing External API Integration

@pytest.mark.integration
def test_crm_sync_integration(mocker, db_session):
    """Test synchronization with external CRM system."""
    from app.services.crm import sync_customer_to_crm

    # Mock external CRM API
    mock_response = mocker.Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {
        "crm_customer_id": "CRM_98765",
        "sync_status": "success",
        "last_sync": "2025-01-15T10:30:00Z"
    }

    mocker.patch('requests.post', return_value=mock_response)

    customer = Customer(
        email="crm.test@example.com",
        name="CRM Test Customer",
        tier="enterprise"
    )
    db_session.add(customer)
    db_session.commit()

    result = sync_customer_to_crm(customer.id, db_session)

    assert result["sync_status"] == "success"
    assert result["crm_customer_id"] == "CRM_98765"

    # Verify customer record updated
    db_session.refresh(customer)
    assert customer.crm_id == "CRM_98765"

Testing Webhook Delivery

def test_webhook_delivery_on_ticket_created(mocker, db_session):
    """Test webhook delivery when new ticket is created."""
    from app.services.webhooks import deliver_webhook

    # Setup webhook subscription
    webhook = WebhookSubscription(
        url="https://customer-app.example.com/webhooks/support",
        events=["ticket.created", "ticket.updated"],
        active=True
    )
    db_session.add(webhook)
    db_session.commit()

    mock_post = mocker.patch('requests.post')
    mock_post.return_value.status_code = 200

    ticket = Ticket(
        title="Webhook Test Ticket",
        priority="high"
    )
    db_session.add(ticket)
    db_session.commit()

    # Trigger webhook delivery
    from app.services.webhooks import trigger_webhooks
    trigger_webhooks("ticket.created", ticket.to_dict(), db_session)

    # Verify webhook called
    assert mock_post.call_count == 1
    call_kwargs = mock_post.call_args.kwargs
    assert call_kwargs["url"] == "https://customer-app.example.com/webhooks/support"
    assert call_kwargs["json"]["event"] == "ticket.created"
    assert call_kwargs["json"]["data"]["id"] == ticket.id

Testing Background Job Processing

@pytest.mark.asyncio
async def test_background_export_job(db_session, sample_tickets):
    """Test asynchronous ticket export background job."""
    from app.tasks.export import export_tickets_job

    job = Job(
        type="export_tickets",
        status="pending",
        params={
            "format": "csv",
            "date_range": "last_7_days",
            "filters": {"priority": ["high", "critical"]}
        }
    )
    db_session.add(job)
    db_session.commit()

    # Process job
    result = await export_tickets_job(job.id, db_session)

    db_session.refresh(job)

    assert job.status == "completed"
    assert job.result is not None
    assert "file_url" in job.result
    assert "row_count" in job.result
    assert result["success"] is True

11. Best Practices

Test Organization

Structure tests to mirror application architecture:

tests/
├── unit/                   # Fast, isolated unit tests
│   ├── test_models.py
│   ├── test_validators.py
│   ├── test_utils.py
│   └── services/
│       ├── test_ticket_service.py
│       ├── test_email_service.py
│       └── test_sla_service.py
├── integration/           # Tests with database/external deps
│   ├── test_api_tickets.py
│   ├── test_api_customers.py
│   ├── test_database_ops.py
│   └── test_external_apis.py
├── e2e/                   # End-to-end workflow tests
│   ├── test_ticket_lifecycle.py
│   └── test_customer_journey.py
├── performance/           # Performance/load tests
│   └── test_api_performance.py
└── conftest.py           # Shared fixtures

Descriptive Test Names

# Good - clear, specific test names
def test_high_priority_ticket_sends_immediate_notification_to_assigned_agent():
    pass

def test_ticket_auto_closes_after_7_days_without_customer_response():
    pass

def test_sla_breach_alert_sent_to_team_lead_when_critical_ticket_unassigned_for_1_hour():
    pass

# Bad - vague test names
def test_ticket_notification():
    pass

def test_auto_close():
    pass

def test_sla():
    pass

AAA Pattern (Arrange-Act-Assert)

def test_ticket_assignment():
    # Arrange: Set up test data and conditions
    agent = create_agent(name="Test Agent", role="agent")
    ticket = create_ticket(title="Test", priority="high", status="open")

    # Act: Perform the action being tested
    result = assign_ticket(ticket.id, agent.id)

    # Assert: Verify expected outcomes
    assert result.assigned_agent_id == agent.id
    assert result.status == "assigned"
    assert result.assigned_at is not None

Test Isolation

# Each test should be independent and not rely on others

# Bad - tests depend on execution order
def test_create_user():
    create_user("test@example.com")

def test_get_user():
    user = get_user("test@example.com")  # Relies on test_create_user
    assert user is not None

# Good - each test is self-contained
@pytest.fixture
def test_user(db_session):
    user = User(email="test@example.com")
    db_session.add(user)
    db_session.commit()
    return user

def test_get_user_by_email(db_session, test_user):
    found_user = get_user(test_user.email, db_session)
    assert found_user.email == test_user.email

Testing Guidelines

1. Test one thing per test:

# Good
def test_ticket_creation_sets_default_status():
    ticket = create_ticket(title="Test")
    assert ticket.status == "open"

def test_ticket_creation_sets_created_timestamp():
    ticket = create_ticket(title="Test")
    assert ticket.created_at is not None

# Bad
def test_ticket_creation():
    ticket = create_ticket(title="Test")
    assert ticket.status == "open"
    assert ticket.created_at is not None
    assert ticket.priority == "medium"
    assert ticket.category is None

2. Use fixtures for setup:

# Good
@pytest.fixture
def authenticated_user(db_session):
    user = User(email="auth@example.com", role="agent")
    db_session.add(user)
    db_session.commit()
    return user

def test_authenticated_endpoint(authenticated_user, api_client):
    response = api_client.get("/api/tickets", user=authenticated_user)
    assert response.status_code == 200

# Less ideal - setup in test
def test_authenticated_endpoint(api_client):
    user = User(email="auth@example.com", role="agent")
    # ... more setup ...
    response = api_client.get("/api/tickets", user=user)
    assert response.status_code == 200

3. Mock external dependencies:

def test_email_notification_on_ticket_creation(mocker):
    """Always mock external services like email."""
    mock_send = mocker.patch('app.services.email.send_email')
    mock_send.return_value = {"status": "sent"}

    ticket = create_ticket(title="Test", customer_email="user@example.com")

    mock_send.assert_called_once()
    assert ticket.id is not None

12. Common Pitfalls and Solutions

Problem: Database State Leakage

Tests affecting each other due to shared database state.

Solution: Use transaction rollback in fixtures.

@pytest.fixture
def db_session(db_engine):
    """Isolate database state per test with rollback."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()

    yield session

    session.close()
    transaction.rollback()  # Rollback all changes
    connection.close()

Problem: Async Test Hanging

Forgetting to mark async tests or improper event loop handling.

Solution: Always use @pytest.mark.asyncio and await coroutines.

# Wrong - test will hang
async def test_async_operation():
    result = await async_function()
    assert result is not None

# Correct
@pytest.mark.asyncio
async def test_async_operation():
    result = await async_function()
    assert result is not None

Problem: Fixture Scope Issues

Session-scoped fixtures with mutable state causing test pollution.

Solution: Use appropriate scope and reset state.

# Problematic
@pytest.fixture(scope="session")
def user_cache():
    return {}  # Shared dict across all tests

# Better
@pytest.fixture(scope="function")
def user_cache():
    return {}  # New dict per test

# Or reset state
@pytest.fixture(scope="session")
def user_cache():
    cache = {}
    yield cache
    cache.clear()  # Clean up after each test

Problem: Over-Mocking

Mocking too much makes tests meaningless.

Solution: Only mock external boundaries.

# Over-mocked - testing nothing real
def test_ticket_creation(mocker):
    mocker.patch('app.services.create_ticket', return_value=Ticket(id=1))
    ticket = create_ticket(title="Test")
    assert ticket.id == 1  # Just testing the mock

# Better - test actual logic
def test_ticket_creation(db_session):
    ticket = create_ticket_service(db_session, title="Test", priority="high")
    assert ticket.id is not None
    assert ticket.status == "open"
    assert ticket.priority == "high"

Problem: Slow Test Suite

Tests taking too long to run.

Solution: Use appropriate markers and parallel execution.

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

# Run tests in parallel
pytest -n auto  # Requires pytest-xdist

# Run specific test file
pytest tests/unit/test_validators.py -v

13. PostgreSQL Integration

Testing with Real PostgreSQL

@pytest.fixture(scope="session")
def postgres_engine():
    """Create PostgreSQL engine with test database."""
    from sqlalchemy import create_engine

    # Connect to default database to create test database
    admin_url = "postgresql://postgres:password@localhost/postgres"
    admin_engine = create_engine(admin_url, isolation_level="AUTOCOMMIT")

    with admin_engine.connect() as conn:
        # Drop and recreate test database
        conn.execute("DROP DATABASE IF EXISTS support_test")
        conn.execute("CREATE DATABASE support_test")

    # Connect to test database
    test_url = "postgresql://postgres:password@localhost/support_test"
    test_engine = create_engine(test_url)

    from app.database import Base
    Base.metadata.create_all(test_engine)

    yield test_engine

    # Cleanup
    test_engine.dispose()
    with admin_engine.connect() as conn:
        conn.execute("DROP DATABASE support_test")
    admin_engine.dispose()

Testing PostgreSQL-Specific Features

def test_full_text_search(db_session):
    """Test PostgreSQL full-text search on tickets."""
    from sqlalchemy import func

    tickets = [
        Ticket(title="Cannot login", description="Password reset not working"),
        Ticket(title="Slow performance", description="Dashboard loading slowly"),
        Ticket(title="Email issues", description="Not receiving notifications"),
    ]
    db_session.add_all(tickets)
    db_session.commit()

    # Search using PostgreSQL full-text search
    search_term = "login password"
    results = db_session.query(Ticket).filter(
        func.to_tsvector('english', Ticket.title + ' ' + Ticket.description).match(search_term)
    ).all()

    assert len(results) >= 1
    assert any("login" in t.title.lower() for t in results)

def test_jsonb_field_operations(db_session):
    """Test JSONB field operations in PostgreSQL."""
    from sqlalchemy.dialects.postgresql import JSONB

    ticket = Ticket(
        title="JSONB Test",
        metadata_={"tags": ["urgent", "billing"], "source": "email"}
    )
    db_session.add(ticket)
    db_session.commit()

    # Query using JSONB operations
    from sqlalchemy import cast
    results = db_session.query(Ticket).filter(
        Ticket.metadata_['tags'].astext.contains('urgent')
    ).all()

    assert len(results) >= 1
    assert results[0].metadata_["tags"] == ["urgent", "billing"]

14. CI/CD Integration

GitHub Actions Configuration

name: Test Suite

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

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_password
          POSTGRES_DB: support_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install dependencies
        run: |
          pip install --upgrade pip
          pip install -r requirements.txt
          pip install pytest pytest-cov pytest-asyncio pytest-mock

      - name: Run migrations
        env:
          DATABASE_URL: postgresql://test_user:test_password@localhost:5432/support_test
        run: |
          alembic upgrade head

      - name: Run tests
        env:
          DATABASE_URL: postgresql://test_user:test_password@localhost:5432/support_test
        run: |
          pytest -v \
            --cov=app \
            --cov-report=xml \
            --cov-report=term-missing \
            --cov-fail-under=80

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

Troubleshooting

Common Issues

1. Import errors: Ensure PYTHONPATH includes project root or use pip install -e . 2. Fixture not found: Check conftest.py location and fixture name spelling 3. Database locked: Use PostgreSQL instead of SQLite for concurrent tests 4. Async tests hanging: Verify pytest-asyncio installed and asyncio_mode configured 5. Mocks not working: Patch where function is used, not where it's defined

Summary

pytest provides comprehensive testing capabilities for customer support systems:

• Fixtures for reusable setup and dependency injection
• Parametrization for testing multiple scenarios
• Mocking to isolate external dependencies
• Async support for modern async applications
• Coverage to measure test completeness
• FastAPI integration for API testing
• SQLAlchemy/PostgreSQL for database testing

This skill enables support teams to build reliable, well-tested backend systems that serve customers effectively.

pytest Examples for Customer Support Systems

This document provides 15+ comprehensive, production-ready examples for testing customer support applications using pytest, FastAPI, SQLAlchemy, and PostgreSQL.

Table of Contents

1. Testing FastAPI Support Ticket Endpoints 2. Database Fixtures with SQLAlchemy 3. Mocking External Email Service 4. Parametrized Data Validation Tests 5. Testing Async Operations 6. Coverage Reporting Setup 7. Testing Authentication 8. Testing Database Transactions 9. Testing Error Handling 10. Fixture Scopes Demonstration 11. Testing Background Tasks 12. Testing Webhooks 13. Conftest.py Organization 14. Testing with pytest-asyncio 15. CI Integration Example 16. Testing SLA Calculations 17. Testing Ticket Assignment Logic 18. Testing Customer Data Validation

---

1. Testing FastAPI Support Ticket Endpoints

Complete example of testing CRUD operations for support tickets.

# tests/integration/test_api_tickets.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.models import Ticket, Customer
from sqlalchemy.orm import Session

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

@pytest.fixture
def client_with_db(db_session):
    """Provide client with database dependency override."""
    from app.dependencies import get_db

    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()

def test_create_ticket_success(client_with_db):
    """Test successful ticket creation via POST."""
    response = client_with_db.post(
        "/api/v1/tickets",
        json={
            "title": "Cannot login to dashboard",
            "description": "User gets 'invalid credentials' error",
            "priority": "high",
            "customer_email": "user@example.com",
            "category": "authentication"
        }
    )

    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Cannot login to dashboard"
    assert data["priority"] == "high"
    assert data["status"] == "open"
    assert "id" in data
    assert "created_at" in data

def test_get_tickets_list(client_with_db, db_session):
    """Test retrieving paginated list of tickets."""
    # Create test tickets
    for i in range(5):
        ticket = Ticket(
            title=f"Test Ticket {i}",
            description=f"Description {i}",
            priority=["low", "medium", "high"][i % 3]
        )
        db_session.add(ticket)
    db_session.commit()

    response = client_with_db.get("/api/v1/tickets?page=1&limit=10")

    assert response.status_code == 200
    data = response.json()
    assert isinstance(data["items"], list)
    assert len(data["items"]) == 5
    assert "total" in data
    assert "page" in data

def test_get_ticket_by_id(client_with_db, db_session):
    """Test retrieving specific ticket by ID."""
    ticket = Ticket(
        title="Specific Ticket",
        description="Test description",
        priority="medium"
    )
    db_session.add(ticket)
    db_session.commit()

    response = client_with_db.get(f"/api/v1/tickets/{ticket.id}")

    assert response.status_code == 200
    data = response.json()
    assert data["id"] == ticket.id
    assert data["title"] == "Specific Ticket"

def test_update_ticket_status(client_with_db, db_session):
    """Test updating ticket status via PATCH."""
    ticket = Ticket(title="Update Test", priority="high", status="open")
    db_session.add(ticket)
    db_session.commit()

    response = client_with_db.patch(
        f"/api/v1/tickets/{ticket.id}",
        json={"status": "in_progress"}
    )

    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "in_progress"

    # Verify in database
    db_session.refresh(ticket)
    assert ticket.status == "in_progress"

def test_delete_ticket(client_with_db, db_session):
    """Test deleting a ticket via DELETE."""
    ticket = Ticket(title="Delete Test", priority="low")
    db_session.add(ticket)
    db_session.commit()
    ticket_id = ticket.id

    response = client_with_db.delete(f"/api/v1/tickets/{ticket_id}")

    assert response.status_code == 204

    # Verify deletion
    deleted_ticket = db_session.query(Ticket).filter_by(id=ticket_id).first()
    assert deleted_ticket is None

def test_get_ticket_not_found(client_with_db):
    """Test 404 response for non-existent ticket."""
    response = client_with_db.get("/api/v1/tickets/99999")
    assert response.status_code == 404
    assert "not found" in response.json()["detail"].lower()

def test_filter_tickets_by_priority(client_with_db, db_session):
    """Test filtering tickets by priority level."""
    for priority in ["low", "medium", "high", "critical"]:
        for i in range(2):
            ticket = Ticket(
                title=f"{priority} priority {i}",
                priority=priority
            )
            db_session.add(ticket)
    db_session.commit()

    response = client_with_db.get("/api/v1/tickets?priority=high")

    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 2
    assert all(t["priority"] == "high" for t in data["items"])

---

2. Database Fixtures with SQLAlchemy

Comprehensive database fixture examples with proper isolation and cleanup.

# tests/conftest.py
import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker, Session
from app.database import Base
from app.models import Ticket, Customer, Agent, Comment
from typing import Generator

@pytest.fixture(scope="session")
def db_engine():
    """Create database engine for test session."""
    engine = create_engine(
        "postgresql://test:test@localhost:5432/support_test",
        echo=False,
        pool_pre_ping=True,
        pool_size=10
    )

    # Create all tables
    Base.metadata.create_all(engine)

    yield engine

    # Cleanup
    Base.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture
def db_session(db_engine) -> Generator[Session, None, None]:
    """Provide database session with automatic rollback for test isolation."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()

    # Enable savepoints for nested transactions
    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(session, transaction):
        if transaction.nested and not transaction._parent.nested:
            session.expire_all()
            session.begin_nested()

    session.begin_nested()

    yield session

    # Cleanup
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def customer_factory(db_session):
    """Factory for creating test customers."""
    created = []

    def _create_customer(
        email=None,
        name=None,
        tier="basic",
        **kwargs
    ):
        import uuid
        customer = Customer(
            email=email or f"customer{uuid.uuid4().hex[:8]}@example.com",
            name=name or f"Customer {uuid.uuid4().hex[:8]}",
            tier=tier,
            **kwargs
        )
        db_session.add(customer)
        db_session.commit()
        created.append(customer)
        return customer

    yield _create_customer

@pytest.fixture
def ticket_factory(db_session):
    """Factory for creating test tickets."""
    created = []

    def _create_ticket(
        title=None,
        customer_id=None,
        priority="medium",
        status="open",
        **kwargs
    ):
        import uuid
        ticket = Ticket(
            title=title or f"Ticket {uuid.uuid4().hex[:8]}",
            description=kwargs.pop("description", "Test ticket description"),
            customer_id=customer_id,
            priority=priority,
            status=status,
            **kwargs
        )
        db_session.add(ticket)
        db_session.commit()
        created.append(ticket)
        return ticket

    yield _create_ticket

# Usage example
def test_customer_ticket_relationship(customer_factory, ticket_factory):
    """Test relationship between customers and their tickets."""
    customer = customer_factory(tier="premium", name="VIP Customer")

    # Create multiple tickets for customer
    ticket1 = ticket_factory(
        customer_id=customer.id,
        title="First Issue",
        priority="high"
    )
    ticket2 = ticket_factory(
        customer_id=customer.id,
        title="Second Issue",
        priority="medium"
    )

    assert len(customer.tickets) == 2
    assert customer.tickets[0].priority == "high"
    assert customer.tickets[1].priority == "medium"

---

3. Mocking External Email Service

Testing email notifications without actually sending emails.

# tests/unit/services/test_email_service.py
import pytest
from app.services.email import EmailService, send_ticket_notification
from app.models import Ticket

def test_send_ticket_created_notification(mocker):
    """Test email sent when ticket is created."""
    mock_send = mocker.patch('app.services.email.EmailService.send_email')
    mock_send.return_value = {
        "status": "sent",
        "message_id": "msg_abc123",
        "provider": "sendgrid"
    }

    ticket = Ticket(
        id=1,
        title="Login Problem",
        customer_email="customer@example.com",
        priority="high"
    )

    result = send_ticket_notification(ticket, "created")

    # Verify email service called
    mock_send.assert_called_once()

    # Verify call arguments
    call_kwargs = mock_send.call_args.kwargs
    assert call_kwargs["to"] == "customer@example.com"
    assert call_kwargs["template"] == "ticket_created"
    assert call_kwargs["context"]["ticket_id"] == 1
    assert call_kwargs["context"]["title"] == "Login Problem"

    # Verify return value
    assert result["status"] == "sent"
    assert result["message_id"] == "msg_abc123"

def test_send_email_with_retry(mocker):
    """Test email retry logic on failure."""
    mock_send = mocker.patch('app.services.email.EmailService.send_email')

    # First two attempts fail, third succeeds
    mock_send.side_effect = [
        Exception("Connection timeout"),
        Exception("Service unavailable"),
        {"status": "sent", "message_id": "msg_retry123"}
    ]

    email_service = EmailService(max_retries=3)
    result = email_service.send_with_retry(
        to="user@example.com",
        subject="Test",
        body="Test message"
    )

    assert mock_send.call_count == 3
    assert result["status"] == "sent"

def test_email_fallback_provider(mocker):
    """Test fallback to secondary email provider on primary failure."""
    mock_sendgrid = mocker.patch('app.services.email.sendgrid_send')
    mock_ses = mocker.patch('app.services.email.ses_send')

    # Primary provider fails
    mock_sendgrid.side_effect = Exception("SendGrid API error")

    # Fallback succeeds
    mock_ses.return_value = {
        "status": "sent",
        "message_id": "ses_123",
        "provider": "aws_ses"
    }

    email_service = EmailService(use_fallback=True)
    result = email_service.send_email(
        to="customer@example.com",
        subject="Ticket Update",
        body="Your ticket has been updated"
    )

    # Verify primary called first
    mock_sendgrid.assert_called_once()

    # Verify fallback called
    mock_ses.assert_called_once()

    # Verify result from fallback
    assert result["provider"] == "aws_ses"
    assert result["status"] == "sent"

def test_batch_email_sending(mocker, ticket_factory):
    """Test sending batch notifications to multiple customers."""
    mock_send = mocker.patch('app.services.email.EmailService.send_email')
    mock_send.return_value = {"status": "sent"}

    tickets = [
        ticket_factory(customer_email=f"customer{i}@example.com")
        for i in range(5)
    ]

    from app.services.email import send_batch_notifications
    results = send_batch_notifications(tickets, "status_update")

    assert mock_send.call_count == 5
    assert len(results) == 5
    assert all(r["status"] == "sent" for r in results)

---

4. Parametrized Data Validation Tests

Testing data validation with multiple input scenarios.

# tests/unit/test_validators.py
import pytest
from app.validators import (
    validate_email,
    validate_ticket_priority,
    validate_phone_number,
    validate_ticket_title
)
from pydantic import ValidationError

@pytest.mark.parametrize("email,expected_valid", [
    ("user@example.com", True),
    ("user.name@example.com", True),
    ("user+tag@example.co.uk", True),
    ("user@subdomain.example.com", True),
    ("invalid.email", False),
    ("@example.com", False),
    ("user@", False),
    ("user @example.com", False),
    ("user@example .com", False),
    ("", False),
    (None, False),
])
def test_email_validation(email, expected_valid):
    """Test email validation with various formats."""
    result = validate_email(email)
    assert result.is_valid == expected_valid

@pytest.mark.parametrize("priority,should_raise", [
    ("low", False),
    ("medium", False),
    ("high", False),
    ("critical", False),
    ("invalid", True),
    ("", True),
    (None, True),
    ("LOW", True),  # Case sensitive
    ("urgent", True),
])
def test_priority_validation(priority, should_raise):
    """Test ticket priority validation."""
    if should_raise:
        with pytest.raises(ValueError, match="Invalid priority"):
            validate_ticket_priority(priority)
    else:
        assert validate_ticket_priority(priority) == priority

@pytest.mark.parametrize("phone,country,expected_valid", [
    ("+1-555-123-4567", "US", True),
    ("555-123-4567", "US", True),
    ("+44 20 7123 4567", "UK", True),
    ("invalid", "US", False),
    ("123", "US", False),
    ("", "US", False),
])
def test_phone_number_validation(phone, country, expected_valid):
    """Test phone number validation for different countries."""
    result = validate_phone_number(phone, country_code=country)
    assert result.is_valid == expected_valid

@pytest.mark.parametrize("title,expected_error", [
    pytest.param("Valid ticket title", None, id="valid"),
    pytest.param("", "empty", id="empty"),
    pytest.param("a" * 300, "too_long", id="too-long"),
    pytest.param("ab", "too_short", id="too-short"),
    pytest.param("<script>alert('xss')</script>", "invalid_chars", id="xss-attempt"),
    pytest.param("Title with\nnewline", "invalid_chars", id="newline"),
])
def test_ticket_title_validation(title, expected_error):
    """Test ticket title validation with edge cases."""
    result = validate_ticket_title(title)

    if expected_error is None:
        assert result.is_valid is True
        assert result.sanitized_value == title
    else:
        assert result.is_valid is False
        assert expected_error in result.error_code

@pytest.mark.parametrize("data,expected_errors", [
    (
        {"title": "Valid", "priority": "high", "customer_email": "user@example.com"},
        []
    ),
    (
        {"priority": "high", "customer_email": "user@example.com"},
        ["title"]  # Missing title
    ),
    (
        {"title": "Valid", "priority": "invalid", "customer_email": "user@example.com"},
        ["priority"]  # Invalid priority
    ),
    (
        {"title": "Valid", "priority": "high", "customer_email": "invalid"},
        ["customer_email"]  # Invalid email
    ),
    (
        {"title": "", "priority": "invalid", "customer_email": "invalid"},
        ["title", "priority", "customer_email"]  # Multiple errors
    ),
])
def test_ticket_creation_validation(data, expected_errors):
    """Test complete ticket creation validation."""
    from app.schemas import TicketCreate

    if expected_errors:
        with pytest.raises(ValidationError) as exc_info:
            TicketCreate(**data)

        errors = exc_info.value.errors()
        error_fields = [e["loc"][0] for e in errors]

        for expected_field in expected_errors:
            assert expected_field in error_fields
    else:
        ticket = TicketCreate(**data)
        assert ticket.title == data["title"]
        assert ticket.priority == data["priority"]

---

5. Testing Async Operations

Testing asynchronous functions and background tasks.

# tests/unit/services/test_async_operations.py
import pytest
import asyncio
from app.services.async_ticket import (
    create_ticket_async,
    process_tickets_batch_async,
    send_notification_async
)

@pytest.mark.asyncio
async def test_async_ticket_creation():
    """Test asynchronous ticket creation."""
    ticket = await create_ticket_async(
        title="Async Ticket Test",
        description="Testing async operations",
        priority="high",
        customer_email="async@example.com"
    )

    assert ticket.id is not None
    assert ticket.title == "Async Ticket Test"
    assert ticket.status == "open"

@pytest.mark.asyncio
async def test_concurrent_ticket_creation():
    """Test creating multiple tickets concurrently."""
    tasks = [
        create_ticket_async(
            title=f"Concurrent Ticket {i}",
            priority=["low", "medium", "high"][i % 3]
        )
        for i in range(10)
    ]

    tickets = await asyncio.gather(*tasks)

    assert len(tickets) == 10
    assert all(t.id is not None for t in tickets)
    assert all(t.status == "open" for t in tickets)

@pytest.mark.asyncio
async def test_async_notification_with_mock(mocker):
    """Test async notification service with mocking."""
    from unittest.mock import AsyncMock

    mock_send = mocker.patch(
        'app.services.notification.send_push_async',
        new_callable=AsyncMock
    )
    mock_send.return_value = {"delivered": True, "notification_id": "notif_123"}

    result = await send_notification_async(
        user_id=456,
        message="New ticket assigned",
        priority="high"
    )

    assert result["delivered"] is True
    mock_send.assert_awaited_once()

    call_kwargs = mock_send.call_args.kwargs
    assert call_kwargs["user_id"] == 456
    assert "ticket assigned" in call_kwargs["message"].lower()

@pytest.mark.asyncio
async def test_batch_processing_async(ticket_factory):
    """Test async batch processing of tickets."""
    tickets = [ticket_factory(status="open") for _ in range(20)]
    ticket_ids = [t.id for t in tickets]

    results = await process_tickets_batch_async(
        ticket_ids=ticket_ids,
        action="assign_to_agent",
        agent_id=1
    )

    assert len(results) == 20
    assert all(r["status"] == "success" for r in results)

---

6. Coverage Reporting Setup

Complete coverage configuration and usage examples.

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

[report]
precision = 2
show_missing = True
skip_covered = False
exclude_lines =
    pragma: no cover
    def __repr__
    def __str__
    raise AssertionError
    raise NotImplementedError
    if __name__ == .__main__.:
    if TYPE_CHECKING:
    @abstractmethod

[html]
directory = htmlcov
title = Support System Coverage

[xml]
output = coverage.xml

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

# Fail if coverage below 80%
pytest --cov=app --cov-fail-under=80

# Generate XML for CI/CD
pytest --cov=app --cov-report=xml

# Coverage for specific module
pytest tests/unit/services/ --cov=app.services --cov-report=term

# pytest.ini
[pytest]
addopts =
    --cov=app
    --cov-report=html
    --cov-report=term-missing
    --cov-fail-under=75
    -v

---

7. Testing Authentication

Testing authentication and authorization flows.

# tests/integration/test_auth.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.auth import create_access_token, verify_token
from datetime import datetime, timedelta

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

def test_login_success(client, db_session):
    """Test successful login returns JWT token."""
    from app.models import User
    from app.auth import hash_password

    # Create test user
    user = User(
        email="agent@support.com",
        password_hash=hash_password("securepass123"),
        role="agent"
    )
    db_session.add(user)
    db_session.commit()

    response = client.post(
        "/api/v1/auth/login",
        json={"email": "agent@support.com", "password": "securepass123"}
    )

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

def test_login_invalid_credentials(client):
    """Test login fails with invalid credentials."""
    response = client.post(
        "/api/v1/auth/login",
        json={"email": "invalid@example.com", "password": "wrongpass"}
    )

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

def test_protected_endpoint_without_token(client):
    """Test protected endpoint requires authentication."""
    response = client.get("/api/v1/tickets/my-tickets")
    assert response.status_code == 401

def test_protected_endpoint_with_valid_token(client, db_session):
    """Test access with valid JWT token."""
    from app.models import User
    user = User(email="agent@support.com", role="agent")
    db_session.add(user)
    db_session.commit()

    token = create_access_token({"sub": user.email, "role": user.role})

    response = client.get(
        "/api/v1/tickets/my-tickets",
        headers={"Authorization": f"Bearer {token}"}
    )

    assert response.status_code == 200

def test_expired_token(client):
    """Test that expired tokens are rejected."""
    # Create expired token
    token = create_access_token(
        {"sub": "user@example.com"},
        expires_delta=timedelta(seconds=-1)  # Already expired
    )

    response = client.get(
        "/api/v1/tickets/my-tickets",
        headers={"Authorization": f"Bearer {token}"}
    )

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

def test_role_based_access_control(client, db_session):
    """Test RBAC - agents cannot access admin endpoints."""
    from app.models import User
    agent = User(email="agent@support.com", role="agent")
    db_session.add(agent)
    db_session.commit()

    token = create_access_token({"sub": agent.email, "role": "agent"})

    # Agent trying to access admin endpoint
    response = client.delete(
        "/api/v1/admin/users/123",
        headers={"Authorization": f"Bearer {token}"}
    )

    assert response.status_code == 403
    assert "permission" in response.json()["detail"].lower()

def test_admin_access_allowed(client, db_session):
    """Test admin can access admin endpoints."""
    from app.models import User
    admin = User(email="admin@support.com", role="admin")
    db_session.add(admin)
    db_session.commit()

    token = create_access_token({"sub": admin.email, "role": "admin"})

    response = client.get(
        "/api/v1/admin/users",
        headers={"Authorization": f"Bearer {token}"}
    )

    assert response.status_code == 200

---

8. Testing Database Transactions

Testing transaction handling and rollback scenarios.

# tests/integration/test_transactions.py
import pytest
from sqlalchemy.exc import IntegrityError
from app.models import Ticket, Customer

def test_transaction_commit(db_session):
    """Test successful transaction commit."""
    ticket = Ticket(title="Transaction Test", priority="medium")
    db_session.add(ticket)
    db_session.commit()

    assert ticket.id is not None

    # Verify persistence
    found = db_session.query(Ticket).filter_by(id=ticket.id).first()
    assert found is not None
    assert found.title == "Transaction Test"

def test_transaction_rollback_on_error(db_session):
    """Test transaction rollback on constraint violation."""
    # Create first customer
    customer1 = Customer(email="unique@example.com", name="First")
    db_session.add(customer1)
    db_session.commit()

    initial_count = db_session.query(Customer).count()

    # Try to create duplicate (should fail on unique email constraint)
    try:
        customer2 = Customer(email="unique@example.com", name="Duplicate")
        db_session.add(customer2)
        db_session.commit()
    except IntegrityError:
        db_session.rollback()

    final_count = db_session.query(Customer).count()
    assert final_count == initial_count  # No new record added

def test_nested_transaction_savepoint(db_session):
    """Test nested transactions with savepoints."""
    ticket1 = Ticket(title="First", priority="high")
    db_session.add(ticket1)
    db_session.flush()

    # Create savepoint
    savepoint = db_session.begin_nested()

    try:
        # This will fail
        ticket2 = Ticket(title="Second", priority="invalid_priority")
        db_session.add(ticket2)
        db_session.flush()
    except:
        savepoint.rollback()

    # First ticket should still exist
    db_session.commit()
    assert db_session.query(Ticket).count() == 1

def test_bulk_insert_transaction(db_session):
    """Test bulk insert in single transaction."""
    tickets = [
        Ticket(title=f"Bulk Ticket {i}", priority="medium")
        for i in range(100)
    ]

    db_session.bulk_save_objects(tickets)
    db_session.commit()

    count = db_session.query(Ticket).count()
    assert count == 100

def test_transaction_isolation(db_engine):
    """Test transaction isolation between sessions."""
    from sqlalchemy.orm import sessionmaker

    Session = sessionmaker(bind=db_engine)
    session1 = Session()
    session2 = Session()

    # Session 1 creates ticket but doesn't commit
    ticket = Ticket(title="Isolation Test", priority="low")
    session1.add(ticket)
    session1.flush()

    # Session 2 shouldn't see uncommitted ticket
    found = session2.query(Ticket).filter_by(title="Isolation Test").first()
    assert found is None

    # After commit, session 2 can see it
    session1.commit()
    found = session2.query(Ticket).filter_by(title="Isolation Test").first()
    assert found is not None

    session1.close()
    session2.close()

---

9. Testing Error Handling

Testing error responses and exception handling.

# tests/integration/test_error_handling.py
import pytest
from fastapi import status
from app.exceptions import TicketNotFoundError, InvalidPriorityError

def test_404_not_found(client):
    """Test 404 response for non-existent resource."""
    response = client.get("/api/v1/tickets/99999")

    assert response.status_code == status.HTTP_404_NOT_FOUND
    data = response.json()
    assert "detail" in data
    assert "not found" in data["detail"].lower()

def test_422_validation_error(client):
    """Test 422 response for invalid data."""
    response = client.post(
        "/api/v1/tickets",
        json={
            "description": "Missing required title field",
            "priority": "high"
        }
    )

    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
    data = response.json()
    assert "detail" in data

    errors = data["detail"]
    assert any("title" in str(error).lower() for error in errors)

def test_400_bad_request(client):
    """Test 400 response for malformed request."""
    response = client.post(
        "/api/v1/tickets",
        data="invalid json",
        headers={"Content-Type": "application/json"}
    )

    assert response.status_code == status.HTTP_400_BAD_REQUEST

def test_custom_exception_handling(mocker, client):
    """Test custom exception handling."""
    mocker.patch(
        'app.services.ticket.get_ticket',
        side_effect=TicketNotFoundError("Ticket 123 not found")
    )

    response = client.get("/api/v1/tickets/123")

    assert response.status_code == status.HTTP_404_NOT_FOUND
    assert "Ticket 123" in response.json()["detail"]

def test_logging_on_error(caplog, db_session):
    """Test that errors are properly logged."""
    import logging
    from app.services.ticket import delete_ticket

    with caplog.at_level(logging.ERROR):
        with pytest.raises(TicketNotFoundError):
            delete_ticket(99999, db_session)

    assert len(caplog.records) > 0
    assert "Ticket" in caplog.text
    assert "99999" in caplog.text

@pytest.mark.parametrize("invalid_input,expected_error", [
    ({"title": "", "priority": "high"}, "title"),
    ({"title": "Valid", "priority": "invalid"}, "priority"),
    ({"title": "Valid", "customer_email": "not-email"}, "email"),
])
def test_validation_errors(client, invalid_input, expected_error):
    """Test various validation errors."""
    response = client.post("/api/v1/tickets", json=invalid_input)

    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any(expected_error in str(e).lower() for e in errors)

---

10. Fixture Scopes Demonstration

Examples of different fixture scopes and their use cases.

# tests/conftest.py
import pytest

@pytest.fixture(scope="session")
def session_data():
    """Session scope: Created once for entire test session."""
    print("\n[SESSION] Creating session data")
    data = {"sessions": 0}
    yield data
    print("\n[SESSION] Cleaning up session data")

@pytest.fixture(scope="module")
def module_data():
    """Module scope: Created once per test module."""
    print("\n[MODULE] Creating module data")
    data = {"modules": 0}
    yield data
    print("\n[MODULE] Cleaning up module data")

@pytest.fixture(scope="class")
def class_data():
    """Class scope: Created once per test class."""
    print("\n[CLASS] Creating class data")
    data = {"classes": 0}
    yield data
    print("\n[CLASS] Cleaning up class data")

@pytest.fixture(scope="function")
def function_data():
    """Function scope: Created for each test function."""
    print("\n[FUNCTION] Creating function data")
    data = {"functions": 0}
    yield data
    print("\n[FUNCTION] Cleaning up function data")

# tests/test_scopes.py
def test_function_scope_1(function_data, session_data):
    """First test using function and session fixtures."""
    function_data["functions"] += 1
    session_data["sessions"] += 1
    assert function_data["functions"] == 1
    assert session_data["sessions"] == 1

def test_function_scope_2(function_data, session_data):
    """Second test - function fixture reset, session persists."""
    function_data["functions"] += 1
    session_data["sessions"] += 1
    assert function_data["functions"] == 1  # Reset for this test
    assert session_data["sessions"] == 2  # Persists across tests

class TestClassScope:
    """Tests demonstrating class scope."""

    def test_class_1(self, class_data):
        class_data["classes"] += 1
        assert class_data["classes"] == 1

    def test_class_2(self, class_data):
        class_data["classes"] += 1
        assert class_data["classes"] == 2  # Persists within class

# Practical example: Database connection pool
@pytest.fixture(scope="session")
def db_connection_pool():
    """Session-scoped database connection pool."""
    from sqlalchemy import pool
    connection_pool = pool.QueuePool(
        create_connection,
        max_overflow=10,
        pool_size=5
    )
    yield connection_pool
    connection_pool.dispose()

@pytest.fixture(scope="function")
def db_connection(db_connection_pool):
    """Function-scoped connection from pool."""
    conn = db_connection_pool.connect()
    yield conn
    conn.close()

---

11. Testing Background Tasks

Testing async background tasks and job processing.

# tests/unit/tasks/test_background_tasks.py
import pytest
from app.tasks.background import (
    send_bulk_emails_task,
    cleanup_old_tickets_task,
    export_tickets_task
)

@pytest.mark.asyncio
async def test_send_bulk_emails_task(mocker):
    """Test bulk email sending background task."""
    from unittest.mock import AsyncMock

    mock_send = mocker.patch(
        'app.services.email.send_email_async',
        new_callable=AsyncMock
    )
    mock_send.return_value = {"status": "sent"}

    emails = [
        {"to": f"customer{i}@example.com", "subject": "Update"}
        for i in range(5)
    ]

    result = await send_bulk_emails_task(emails)

    assert mock_send.call_count == 5
    assert result["total"] == 5
    assert result["sent"] == 5
    assert result["failed"] == 0

def test_cleanup_old_tickets_task(db_session, ticket_factory):
    """Test background task that cleans up old closed tickets."""
    from datetime import datetime, timedelta

    # Create old closed tickets
    old_date = datetime.utcnow() - timedelta(days=365)
    for i in range(3):
        ticket = ticket_factory(
            status="closed",
            created_at=old_date
        )

    # Create recent tickets
    for i in range(2):
        ticket = ticket_factory(status="closed")

    initial_count = db_session.query(Ticket).count()
    assert initial_count == 5

    # Run cleanup task
    deleted_count = cleanup_old_tickets_task(
        db_session,
        days_old=180,
        status="closed"
    )

    assert deleted_count == 3
    final_count = db_session.query(Ticket).count()
    assert final_count == 2

@pytest.mark.asyncio
async def test_export_tickets_task(db_session, ticket_factory, mocker):
    """Test ticket export background task."""
    # Create test tickets
    for i in range(10):
        ticket_factory(priority=["low", "high"][i % 2])

    mock_upload = mocker.patch('app.storage.upload_file')
    mock_upload.return_value = "https://storage.example.com/export_123.csv"

    result = await export_tickets_task(
        filters={"priority": "high"},
        format="csv"
    )

    assert result["status"] == "completed"
    assert result["row_count"] == 5
    assert "file_url" in result
    mock_upload.assert_called_once()

def test_task_failure_handling(mocker, db_session):
    """Test background task failure handling and retry."""
    from app.tasks.background import process_with_retry

    mock_process = mocker.patch('app.tasks.process_function')
    mock_process.side_effect = [
        Exception("Temporary error"),
        Exception("Temporary error"),
        {"status": "success"}
    ]

    result = process_with_retry(task_id=123, max_retries=3)

    assert result["status"] == "success"
    assert mock_process.call_count == 3

---

12. Testing Webhooks

Testing webhook delivery and handling.

# tests/integration/test_webhooks.py
import pytest
from app.models import WebhookSubscription, Ticket

def test_webhook_delivery_on_event(mocker, db_session):
    """Test webhook delivered when event occurs."""
    # Create webhook subscription
    webhook = WebhookSubscription(
        url="https://customer.example.com/webhooks",
        events=["ticket.created", "ticket.updated"],
        secret="webhook_secret_123",
        active=True
    )
    db_session.add(webhook)
    db_session.commit()

    mock_post = mocker.patch('requests.post')
    mock_post.return_value.status_code = 200

    # Trigger event
    from app.services.webhooks import trigger_webhook
    ticket = Ticket(id=1, title="Test", priority="high")

    trigger_webhook("ticket.created", ticket.to_dict())

    # Verify webhook called
    assert mock_post.call_count == 1
    call_kwargs = mock_post.call_args.kwargs

    assert call_kwargs["url"] == "https://customer.example.com/webhooks"
    assert "ticket.created" in call_kwargs["json"]["event"]
    assert call_kwargs["json"]["data"]["id"] == 1
    assert "X-Webhook-Signature" in call_kwargs["headers"]

def test_webhook_retry_on_failure(mocker, db_session):
    """Test webhook retry mechanism on delivery failure."""
    webhook = WebhookSubscription(
        url="https://customer.example.com/webhooks",
        events=["ticket.created"],
        active=True,
        max_retries=3
    )
    db_session.add(webhook)
    db_session.commit()

    mock_post = mocker.patch('requests.post')
    # First two attempts fail, third succeeds
    mock_post.side_effect = [
        mocker.Mock(status_code=500),
        mocker.Mock(status_code=503),
        mocker.Mock(status_code=200)
    ]

    from app.services.webhooks import deliver_webhook_with_retry
    result = deliver_webhook_with_retry(
        webhook_id=webhook.id,
        event="ticket.created",
        data={"id": 1}
    )

    assert mock_post.call_count == 3
    assert result["status"] == "delivered"

def test_webhook_signature_validation(mocker):
    """Test webhook signature for security."""
    import hmac
    import hashlib

    secret = "webhook_secret_123"
    payload = '{"event": "ticket.created", "data": {"id": 1}}'

    # Generate signature
    signature = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()

    from app.services.webhooks import validate_webhook_signature
    is_valid = validate_webhook_signature(payload, signature, secret)

    assert is_valid is True

    # Test invalid signature
    is_valid = validate_webhook_signature(payload, "invalid_sig", secret)
    assert is_valid is False

def test_webhook_filtering_by_event(mocker, db_session):
    """Test webhooks only triggered for subscribed events."""
    webhook = WebhookSubscription(
        url="https://customer.example.com/webhooks",
        events=["ticket.created"],  # Only subscribed to created
        active=True
    )
    db_session.add(webhook)
    db_session.commit()

    mock_post = mocker.patch('requests.post')

    from app.services.webhooks import trigger_webhooks

    # Should trigger
    trigger_webhooks("ticket.created", {"id": 1}, db_session)
    assert mock_post.call_count == 1

    # Should not trigger (not subscribed)
    trigger_webhooks("ticket.updated", {"id": 1}, db_session)
    assert mock_post.call_count == 1  # Still 1, not called again

---

13. Conftest Organization

Best practices for organizing conftest.py files.

# tests/conftest.py (root conftest)
"""
Root conftest.py - Shared fixtures for all tests.
Fixtures defined here are available to all test modules.
"""
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Database fixtures
@pytest.fixture(scope="session")
def db_engine():
    """Session-scoped database engine."""
    engine = create_engine("postgresql://test:test@localhost/test_db")
    yield engine
    engine.dispose()

@pytest.fixture
def db_session(db_engine):
    """Function-scoped database session with rollback."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()
    yield session
    session.close()
    transaction.rollback()
    connection.close()

# API client fixtures
@pytest.fixture
def client():
    """FastAPI test client."""
    from fastapi.testclient import TestClient
    from app.main import app
    return TestClient(app)

# tests/unit/conftest.py (unit test specific)
"""
Unit test conftest - Fixtures specific to unit tests.
"""
import pytest

@pytest.fixture
def mock_db_session(mocker):
    """Mock database session for unit tests."""
    session = mocker.Mock()
    session.add = mocker.Mock()
    session.commit = mocker.Mock()
    session.query = mocker.Mock()
    return session

# tests/integration/conftest.py (integration test specific)
"""
Integration test conftest - Fixtures for integration tests.
"""
import pytest

@pytest.fixture
def client_with_db(client, db_session):
    """Client with real database for integration tests."""
    from app.main import app
    from app.dependencies import get_db

    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    yield client
    app.dependency_overrides.clear()

@pytest.fixture
def sample_data(db_session):
    """Create sample data for integration tests."""
    from app.models import Ticket, Customer

    customer = Customer(email="test@example.com", name="Test")
    db_session.add(customer)
    db_session.flush()

    tickets = [
        Ticket(title=f"Ticket {i}", customer_id=customer.id)
        for i in range(5)
    ]
    db_session.add_all(tickets)
    db_session.commit()

    return {"customer": customer, "tickets": tickets}

# tests/e2e/conftest.py (e2e test specific)
"""
E2E test conftest - Fixtures for end-to-end tests.
"""
import pytest

@pytest.fixture(scope="module")
def e2e_client():
    """Module-scoped client for e2e tests."""
    from fastapi.testclient import TestClient
    from app.main import app
    return TestClient(app)

@pytest.fixture
def authenticated_user(e2e_client):
    """Create and authenticate user for e2e tests."""
    # Register user
    register_response = e2e_client.post(
        "/api/auth/register",
        json={"email": "e2e@test.com", "password": "testpass123"}
    )

    # Login
    login_response = e2e_client.post(
        "/api/auth/login",
        json={"email": "e2e@test.com", "password": "testpass123"}
    )

    token = login_response.json()["access_token"]
    return {"email": "e2e@test.com", "token": token}

---

14. Testing with pytest-asyncio

Advanced async testing patterns.

# tests/unit/test_async_advanced.py
import pytest
import asyncio
from app.services.async_service import AsyncTicketService

@pytest.fixture
async def async_service():
    """Async fixture for service instance."""
    service = AsyncTicketService()
    await service.initialize()
    yield service
    await service.cleanup()

@pytest.mark.asyncio
async def test_async_service_initialization(async_service):
    """Test async service initialization."""
    assert async_service.is_initialized is True

@pytest.mark.asyncio
async def test_concurrent_operations(async_service):
    """Test handling multiple concurrent async operations."""
    async def create_ticket(title):
        return await async_service.create(title=title, priority="medium")

    # Create 10 tickets concurrently
    tasks = [create_ticket(f"Ticket {i}") for i in range(10)]
    tickets = await asyncio.gather(*tasks)

    assert len(tickets) == 10
    assert all(t.id is not None for t in tickets)

@pytest.mark.asyncio
async def test_async_error_handling(async_service):
    """Test async error handling."""
    with pytest.raises(ValueError, match="Invalid priority"):
        await async_service.create(
            title="Test",
            priority="invalid_priority"
        )

@pytest.mark.asyncio
async def test_async_timeout():
    """Test async operation with timeout."""
    async def slow_operation():
        await asyncio.sleep(10)
        return "completed"

    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(), timeout=1.0)

@pytest.mark.asyncio
async def test_async_context_manager():
    """Test async context manager."""
    from app.services.async_context import AsyncDatabaseSession

    async with AsyncDatabaseSession() as session:
        ticket = await session.create_ticket(title="Test")
        assert ticket.id is not None

@pytest.mark.asyncio
@pytest.mark.parametrize("count", [1, 5, 10, 50])
async def test_async_batch_processing(count):
    """Test async batch processing with different sizes."""
    from app.services.batch import process_batch_async

    items = [{"id": i} for i in range(count)]
    results = await process_batch_async(items)

    assert len(results) == count
    assert all(r["status"] == "processed" for r in results)

---

15. CI Integration Example

Complete CI/CD pipeline configuration with pytest.

# .github/workflows/test.yml
name: Test Suite

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

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_password
          POSTGRES_DB: support_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

      redis:
        image: redis:7
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 6379:6379

    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install -r requirements-dev.txt

      - name: Lint with flake8
        run: |
          flake8 app tests --count --show-source --statistics

      - name: Type check with mypy
        run: |
          mypy app

      - name: Run database migrations
        env:
          DATABASE_URL: postgresql://test_user:test_password@localhost:5432/support_test
        run: |
          alembic upgrade head

      - name: Run tests with coverage
        env:
          DATABASE_URL: postgresql://test_user:test_password@localhost:5432/support_test
          REDIS_URL: redis://localhost:6379
        run: |
          pytest -v \
            --cov=app \
            --cov-report=xml \
            --cov-report=term-missing \
            --cov-fail-under=80 \
            -n auto

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

      - name: Archive test results
        if: always()
        uses: actions/upload-artifact@v3
        with:
          name: test-results
          path: |
            htmlcov/
            .coverage

      - name: Run security checks
        run: |
          pip install bandit safety
          bandit -r app/
          safety check

# Makefile for local testing
.PHONY: test test-unit test-integration test-e2e test-coverage

test:
	pytest -v

test-unit:
	pytest tests/unit -v

test-integration:
	pytest tests/integration -v -m integration

test-e2e:
	pytest tests/e2e -v -m e2e --slow

test-coverage:
	pytest --cov=app --cov-report=html --cov-report=term-missing

test-fast:
	pytest -m "not slow" -v

test-parallel:
	pytest -n auto -v

lint:
	flake8 app tests
	black app tests --check
	isort app tests --check

format:
	black app tests
	isort app tests

typecheck:
	mypy app

---

16. Testing SLA Calculations

Testing Service Level Agreement logic.

# tests/unit/services/test_sla.py
import pytest
from datetime import datetime, timedelta
from app.services.sla import (
    calculate_sla_hours,
    is_sla_breached,
    calculate_time_remaining,
    get_business_hours
)

@pytest.mark.parametrize("priority,expected_hours", [
    ("critical", 1),
    ("high", 4),
    ("medium", 24),
    ("low", 72),
])
def test_sla_hours_by_priority(priority, expected_hours):
    """Test SLA hours calculation for each priority level."""
    sla_hours = calculate_sla_hours(priority)
    assert sla_hours == expected_hours

def test_sla_breach_detection():
    """Test detection of SLA breaches."""
    created_at = datetime.utcnow() - timedelta(hours=2)
    ticket = Ticket(
        priority="critical",  # 1 hour SLA
        created_at=created_at,
        status="open"
    )

    is_breached = is_sla_breached(ticket)
    assert is_breached is True

def test_sla_not_breached():
    """Test ticket within SLA."""
    created_at = datetime.utcnow() - timedelta(minutes=30)
    ticket = Ticket(
        priority="critical",  # 1 hour SLA
        created_at=created_at,
        status="open"
    )

    is_breached = is_sla_breached(ticket)
    assert is_breached is False

def test_sla_time_remaining():
    """Test calculation of remaining SLA time."""
    created_at = datetime.utcnow() - timedelta(hours=1)
    ticket = Ticket(
        priority="high",  # 4 hour SLA
        created_at=created_at
    )

    remaining = calculate_time_remaining(ticket)
    assert 2.9 <= remaining.total_seconds() / 3600 <= 3.1  # ~3 hours

def test_business_hours_calculation():
    """Test business hours vs calendar hours."""
    # Monday 9 AM to Tuesday 9 AM
    start = datetime(2025, 1, 6, 9, 0)  # Monday
    end = datetime(2025, 1, 7, 9, 0)    # Tuesday

    business_hours = get_business_hours(start, end)
    # Should be 8 hours (Monday 9-5), not 24 hours
    assert business_hours == 8

---

17. Testing Ticket Assignment Logic

Testing automatic ticket assignment algorithms.

# tests/unit/services/test_assignment.py
import pytest
from app.services.assignment import (
    assign_round_robin,
    assign_by_workload,
    assign_by_expertise
)
from app.models import Agent, Ticket

@pytest.fixture
def agents(db_session):
    """Create test agents."""
    agents = [
        Agent(name="Agent 1", email="agent1@support.com", expertise=["billing"]),
        Agent(name="Agent 2", email="agent2@support.com", expertise=["technical"]),
        Agent(name="Agent 3", email="agent3@support.com", expertise=["billing", "technical"]),
    ]
    db_session.add_all(agents)
    db_session.commit()
    return agents

def test_round_robin_assignment(db_session, agents):
    """Test round-robin assignment distributes evenly."""
    tickets = []
    for i in range(9):
        ticket = Ticket(title=f"Ticket {i}", priority="medium")
        db_session.add(ticket)
        db_session.flush()

        assigned_agent = assign_round_robin(ticket, db_session)
        ticket.assigned_agent_id = assigned_agent.id
        tickets.append(ticket)

    db_session.commit()

    # Verify even distribution (3 tickets per agent)
    for agent in agents:
        assigned_count = len([t for t in tickets if t.assigned_agent_id == agent.id])
        assert assigned_count == 3

def test_workload_based_assignment(db_session, agents):
    """Test assignment based on current workload."""
    # Give agents different workloads
    for i in range(5):
        ticket = Ticket(title=f"Existing {i}", assigned_agent_id=agents[0].id)
        db_session.add(ticket)

    for i in range(2):
        ticket = Ticket(title=f"Existing {i}", assigned_agent_id=agents[1].id)
        db_session.add(ticket)

    db_session.commit()

    # New ticket should go to agent with least workload (agents[2] with 0)
    new_ticket = Ticket(title="New Ticket", priority="high")
    assigned_agent = assign_by_workload(new_ticket, db_session)

    assert assigned_agent.id == agents[2].id

def test_expertise_based_assignment(db_session, agents):
    """Test assignment based on agent expertise."""
    billing_ticket = Ticket(
        title="Billing Issue",
        category="billing",
        priority="high"
    )

    assigned_agent = assign_by_expertise(billing_ticket, db_session)

    # Should be assigned to agent with billing expertise
    assert "billing" in assigned_agent.expertise

---

18. Testing Customer Data Validation

Testing comprehensive customer data validation.

# tests/unit/test_customer_validation.py
import pytest
from app.validators import validate_customer_data
from app.schemas import CustomerCreate
from pydantic import ValidationError

def test_valid_customer_data():
    """Test validation of valid customer data."""
    customer = CustomerCreate(
        email="customer@example.com",
        name="John Doe",
        company="Acme Corp",
        phone="+1-555-123-4567",
        tier="premium"
    )

    assert customer.email == "customer@example.com"
    assert customer.tier == "premium"

@pytest.mark.parametrize("email", [
    "invalid",
    "@example.com",
    "user@",
    "",
])
def test_invalid_email(email):
    """Test customer creation fails with invalid email."""
    with pytest.raises(ValidationError) as exc_info:
        CustomerCreate(
            email=email,
            name="Test User"
        )

    errors = exc_info.value.errors()
    assert any("email" in str(e["loc"]) for e in errors)

def test_tier_validation():
    """Test customer tier validation."""
    with pytest.raises(ValidationError):
        CustomerCreate(
            email="test@example.com",
            name="Test",
            tier="invalid_tier"
        )

def test_phone_number_normalization():
    """Test phone number is normalized."""
    customer = CustomerCreate(
        email="test@example.com",
        name="Test",
        phone="555-123-4567"
    )

    # Should be normalized to E.164 format
    assert customer.phone.startswith("+")

def test_company_name_sanitization():
    """Test company name XSS protection."""
    customer = CustomerCreate(
        email="test@example.com",
        name="Test",
        company="<script>alert('xss')</script>"
    )

    # Should be sanitized
    assert "<script>" not in customer.company

---

Summary

These 18 comprehensive examples cover:

1. FastAPI endpoint testing 2. Database fixtures and factories 3. External service mocking 4. Parametrized validation tests 5. Async operation testing 6. Coverage configuration 7. Authentication flows 8. Transaction handling 9. Error handling 10. Fixture scopes 11. Background tasks 12. Webhook testing 13. Conftest organization 14. Advanced async patterns 15. CI/CD integration 16. SLA calculations 17. Assignment algorithms 18. Data validation

Each example is production-ready and demonstrates real-world testing scenarios for customer support systems using pytest, FastAPI, SQLAlchemy, and PostgreSQL.