Pytest Fixtures for API Testing — Building Reusable Test Components

# tests/conftest.py — shared fixtures for all API tests
import pytest
import httpx
import jwt
import datetime
from typing import Generator

BASE_URL = "https://api.durga-antivirus.com"

@pytest.fixture(scope="session")
def api_base_url() -> str:
    return BASE_URL

@pytest.fixture(scope="session")
def jwt_secret() -> str:
    return "test-secret-for-testing-only"

@pytest.fixture(scope="session")
def auth_token(jwt_secret: str) -> str:
    """Session-scoped auth token — generated once for all tests."""
    return jwt.encode(
        {
            "sub": "test-analyst",
            "roles": ["analyst"],
            "scope": "threat:read threat:write",
            "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=24)
        },
        jwt_secret,
        algorithm="HS256"
    )

@pytest.fixture
def auth_headers(auth_token: str) -> dict:
    """Function-scoped headers — fresh dict per test."""
    return {"Authorization": f"Bearer {auth_token}"}

@pytest.fixture
def http_client() -> Generator[httpx.Client, None, None]:
    with httpx.Client(base_url=BASE_URL, timeout=30.0) as client:
        yield client

# tests/factories.py
import pytest
import random

@pytest.fixture
def threat_factory():
    """Factory fixture — creates threat data dicts."""

    def _create_threat(**overrides):
        threat = {
            "name": f"Test Threat {random.randint(1000, 9999)}",
            "severity": random.choice(["low", "medium", "high", "critical"]),
            "source_ip": f"{random.randint(1, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 255)}",
            "description": "Created by test factory"
        }
        threat.update(overrides)
        return threat

    return _create_threat

@pytest.fixture
def multiple_threats(threat_factory):
    """Create multiple threat fixtures."""

    def _create_threats(count: int, **shared_overrides):
        return [
            threat_factory(**shared_overrides)
            for _ in range(count)
        ]

    return _create_threats


# Usage in tests
class TestThreatCreation:
    def test_create_valid_threat(self, http_client, auth_headers, threat_factory):
        threat_data = threat_factory(severity="high")
        response = http_client.post(
            "/api/v1/threats",
            json=threat_data,
            headers=auth_headers
        )
        assert response.status_code == 201

    def test_create_multiple_threats(self, http_client, auth_headers, multiple_threats):
        threats = multiple_threats(5, severity="critical")
        for threat in threats:
            response = http_client.post(
                "/api/v1/threats",
                json=threat,
                headers=auth_headers
            )
            assert response.status_code == 201

# tests/db_fixtures.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:15") as postgres:
        yield postgres

@pytest.fixture
def db_engine(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)

@pytest.fixture
def db_session(db_engine):
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()

    yield session

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

@pytest.fixture
def threat_in_db(db_session, threat_factory):
    """Create a threat directly in the database."""
    from api.models import Threat
    data = threat_factory()
    threat = Threat(**data)
    db_session.add(threat)
    db_session.commit()
    db_session.refresh(threat)
    return threat