Testing Guide

This guide covers testing practices and patterns for Request Forge.

Test Structure

Test Organization

tests/
├── __init__.py
├── conftest.py                  # Shared fixtures
├── test_client.py               # Client tests
├── test_config.py               # Configuration tests
├── test_models.py               # Model tests
├── test_retry.py                # Retry strategy tests
├── test_hooks.py                # Hook tests
├── test_token_manager.py        # Token manager tests
├── test_pipelines.py            # Pipeline tests
├── test_fetcher.py              # Fetcher tests
├── test_exceptions.py           # Exception tests
└── test_utils.py                # Utility tests

Test Markers

Tests are organized with pytest markers:

@pytest.mark.unit
def test_something():
    """Unit test."""
    pass

@pytest.mark.integration
def test_integration():
    """Integration test."""
    pass

@pytest.mark.slow
def test_slow_operation():
    """Slow test."""
    pass

Run tests by marker:

# Run only unit tests
pytest -m unit

# Run only integration tests
pytest -m integration

# Skip slow tests
pytest -m "not slow"

Common Fixtures

Basic Fixtures

# conftest.py
import pytest
from requestforge import HttpClientConfigBuilder, HttpClient

@pytest.fixture
def config():
    """Basic HTTP client configuration."""
    return (
        HttpClientConfigBuilder()
        .with_base_url('https://api.example.com')
        .with_timeout(30.0)
        .build()
    )

@pytest.fixture
def client(config):
    """HTTP client instance."""
    return HttpClient(config)

@pytest.fixture
def mock_token_data():
    """Mock token data."""
    from requestforge.config import TokenData
    from datetime import datetime, timedelta

    return TokenData(
        access_token='test-token-123',
        token_type='Bearer',
        expires_at=datetime.now() + timedelta(hours=1)
    )

responses Fixtures

import pytest
import responses

@pytest.fixture
def mock_api():
    """Activate responses mock."""
    with responses.RequestsMock() as rsps:
        yield rsps

@pytest.fixture
def mock_success_response(mock_api):
    """Mock successful API response."""
    mock_api.add(
        responses.GET,
        'https://api.example.com/users',
        json={'users': [{'id': 1, 'name': 'John'}]},
        status=200
    )
    return mock_api

Unit Testing

Testing Client

# tests/test_client.py
import pytest
import responses
from requestforge import HttpClient, HttpClientConfigBuilder

class TestHttpClient:
    """Test HTTP client functionality."""

    @responses.activate
    def test_get_request(self):
        # Mock response
        responses.add(
            responses.GET,
            'https://api.example.com/users',
            json={'users': [{'id': 1}]},
            status=200
        )

        # Create client
        config = HttpClientConfigBuilder().with_base_url('https://api.example.com').build()
        client = HttpClient(config)

        # Make request
        response = client.get('/users')

        # Assertions
        assert response.status_code == 200
        assert response.is_success
        assert len(response.json()['users']) == 1

    @responses.activate
    def test_post_request(self):
        responses.add(
            responses.POST,
            'https://api.example.com/users',
            json={'id': 1, 'name': 'John'},
            status=201
        )

        config = HttpClientConfigBuilder().with_base_url('https://api.example.com').build()
        client = HttpClient(config)

        response = client.post('/users', json_data={'name': 'John'})

        assert response.status_code == 201
        assert response.json()['name'] == 'John'

Testing Retry Logic

# tests/test_retry.py
import pytest
import responses
from requestforge import (
    HttpClient,
    HttpClientConfigBuilder,
    ExponentialBackoffRetryStrategy,
    MaxRetryException
)

class TestRetryStrategies:
    @responses.activate
    def test_retry_on_500(self):
        # First two calls fail, third succeeds
        responses.add(responses.GET, 'https://api.example.com/data', status=500)
        responses.add(responses.GET, 'https://api.example.com/data', status=500)
        responses.add(
            responses.GET,
            'https://api.example.com/data',
            json={'result': 'success'},
            status=200
        )

        strategy = ExponentialBackoffRetryStrategy(
            max_retries=3,
            base_delay=0.01  # Fast for testing
        )

        config = (
            HttpClientConfigBuilder()
            .with_base_url('https://api.example.com')
            .with_retry_strategy(strategy)
            .build()
        )

        client = HttpClient(config)
        response = client.get('/data')

        assert response.status_code == 200
        assert len(responses.calls) == 3

    @responses.activate
    def test_max_retries_exceeded(self):
        # All calls fail
        for _ in range(10):
            responses.add(responses.GET, 'https://api.example.com/data', status=500)

        strategy = ExponentialBackoffRetryStrategy(max_retries=3, base_delay=0.01)
        config = (
            HttpClientConfigBuilder()
            .with_base_url('https://api.example.com')
            .with_retry_strategy(strategy)
            .build()
        )

        client = HttpClient(config)

        with pytest.raises(MaxRetryException) as exc_info:
            client.get('/data')

        assert exc_info.value.attempts == 4  # 1 initial + 3 retries

Testing Hooks

# tests/test_hooks.py
import pytest
from unittest.mock import Mock
from requestforge import HttpClientConfigBuilder, HttpClient
from requestforge.interfaces import RequestHookInterface

class TestHooks:
    @responses.activate
    def test_request_hook_called(self):
        # Create mock hook
        mock_hook = Mock(spec=RequestHookInterface)
        mock_hook.before_request = Mock(side_effect=lambda req, ctx: req)

        # Configure client
        responses.add(responses.GET, 'https://api.example.com/test', status=200)

        config = (
            HttpClientConfigBuilder()
            .with_base_url('https://api.example.com')
            .with_request_hook(mock_hook)
            .build()
        )

        client = HttpClient(config)
        client.get('/test')

        # Verify hook was called
        assert mock_hook.before_request.called
        assert mock_hook.before_request.call_count == 1

Testing Token Manager

# tests/test_token_manager.py
import pytest
from unittest.mock import Mock
from requestforge import TokenManager
from requestforge.config import TokenData
from datetime import datetime, timedelta

class TestTokenManager:
    def test_token_caching(self):
        # Mock provider
        mock_provider = Mock()
        mock_provider.fetch_token.return_value = TokenData(
            access_token='token-123',
            token_type='Bearer',
            expires_at=datetime.now() + timedelta(hours=1)
        )
        mock_provider.service_name = 'test-service'

        # Create manager
        token_manager = TokenManager(mock_provider)

        # First call fetches token
        token1 = token_manager.get_token()
        assert token1.access_token == 'token-123'
        assert mock_provider.fetch_token.call_count == 1

        # Second call uses cache
        token2 = token_manager.get_token()
        assert token2.access_token == 'token-123'
        assert mock_provider.fetch_token.call_count == 1  # Still 1

    def test_force_refresh(self):
        mock_provider = Mock()
        mock_provider.fetch_token.return_value = TokenData(
            access_token='token-123',
            expires_at=datetime.now() + timedelta(hours=1)
        )
        mock_provider.service_name = 'test'

        token_manager = TokenManager(mock_provider)

        # Get token
        token_manager.get_token()

        # Force refresh
        new_token = token_manager.force_refresh()

        # Provider called twice
        assert mock_provider.fetch_token.call_count == 2

Integration Testing

Testing with Real Endpoints

import pytest
from requestforge import create_client

@pytest.mark.integration
@pytest.mark.skip(reason="Requires live API")
def test_real_api():
    """Test against real API (skipped by default)."""
    client = create_client('https://api.github.com')
    response = client.get('/users/octocat')

    assert response.is_success
    assert response.json()['login'] == 'octocat'

Testing with Test Server

import pytest
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
from requestforge import create_client

class SimpleHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-type', 'application/json')
        self.end_headers()
        self.wfile.write(b'{"status": "ok"}')

@pytest.fixture(scope='session')
def test_server():
    """Start test HTTP server."""
    server = HTTPServer(('localhost', 8888), SimpleHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.daemon = True
    thread.start()
    yield server
    server.shutdown()

@pytest.mark.integration
def test_with_test_server(test_server):
    client = create_client('http://localhost:8888')
    response = client.get('/')

    assert response.status_code == 200
    assert response.json()['status'] == 'ok'

Mocking Strategies

Using responses

import responses
from requestforge import create_client

@responses.activate
def test_api_with_responses():
    # Add mock responses
    responses.add(
        responses.GET,
        'https://api.example.com/users/1',
        json={'id': 1, 'name': 'John'},
        status=200
    )

    responses.add(
        responses.POST,
        'https://api.example.com/users',
        json={'id': 2, 'name': 'Jane'},
        status=201
    )

    # Make requests
    client = create_client('https://api.example.com')

    response1 = client.get('/users/1')
    assert response1.json()['name'] == 'John'

    response2 = client.post('/users', json_data={'name': 'Jane'})
    assert response2.status_code == 201

Using unittest.mock

from unittest.mock import Mock, patch
from requestforge import HttpClient, HttpResponse

def test_with_mock_patch():
    with patch('requestforge.HttpClient.get') as mock_get:
        # Configure mock
        mock_response = Mock(spec=HttpResponse)
        mock_response.status_code = 200
        mock_response.is_success = True
        mock_response.json.return_value = {'id': 1}
        mock_get.return_value = mock_response

        # Use client
        client = HttpClient(config)
        response = client.get('/users/1')

        # Assertions
        assert response.status_code == 200
        assert response.json()['id'] == 1

Testing Exceptions

Testing Error Handling

import pytest
import responses
from requestforge import (
    create_client,
    TimeoutException,
    NotFoundException,
    ServerErrorException
)

class TestExceptions:
    @responses.activate
    def test_timeout_exception(self):
        responses.add(
            responses.GET,
            'https://api.example.com/slow',
            body=TimeoutException()
        )

        client = create_client('https://api.example.com')

        with pytest.raises(TimeoutException):
            client.get('/slow', timeout=1.0)

    @responses.activate
    def test_not_found_exception(self):
        responses.add(
            responses.GET,
            'https://api.example.com/users/999',
            status=404
        )

        client = create_client('https://api.example.com')

        with pytest.raises(NotFoundException):
            client.get('/users/999')

    @responses.activate
    def test_server_error_exception(self):
        responses.add(
            responses.GET,
            'https://api.example.com/error',
            status=500
        )

        client = create_client('https://api.example.com')

        with pytest.raises(ServerErrorException) as exc_info:
            client.get('/error')

        assert exc_info.value.status_code == 500

Coverage Requirements

Minimum Coverage

• Overall coverage: 90%+

• New features: 100% coverage required

• Critical paths: 100% coverage required

Check Coverage

# Generate coverage report
pytest --cov=requestforge --cov-report=term-missing

# Generate HTML report
pytest --cov=requestforge --cov-report=html

# Fail if coverage below threshold
pytest --cov=requestforge --cov-fail-under=90

Coverage Configuration

# pyproject.toml
[tool.coverage.run]
source = ["src/requestforge"]
omit = [
    "*/tests/*",
    "*/__pycache__/*",
]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise AssertionError",
    "raise NotImplementedError",
    "if __name__ == .__main__.:",
    "if TYPE_CHECKING:",
    "@abstractmethod",
]

Performance Testing

Benchmark Tests

import pytest
import time
from requestforge import create_client

@pytest.mark.benchmark
def test_request_performance():
    """Benchmark request performance."""
    client = create_client('https://httpbin.org')

    start = time.time()
    for _ in range(100):
        client.get('/get')
    duration = time.time() - start

    # Should complete 100 requests in under 10 seconds
    assert duration < 10.0

@pytest.mark.benchmark
def test_concurrent_performance():
    """Benchmark concurrent requests."""
    from concurrent.futures import ThreadPoolExecutor

    client = create_client('https://httpbin.org')

    def make_request(i):
        return client.get('/get')

    start = time.time()
    with ThreadPoolExecutor(max_workers=10) as executor:
        list(executor.map(make_request, range(100)))
    duration = time.time() - start

    # Concurrent should be faster
    assert duration < 5.0

Best Practices

1. Write Tests First (TDD)

   # Write test first
   def test_new_feature():
       result = new_feature()
       assert result == expected
   
   # Then implement
   def new_feature():
       return expected
   

2. Use Descriptive Names

   # Good ✅
   def test_retry_strategy_respects_max_retries():
       ...
   
   # Avoid ❌
   def test_retry():
       ...
   

3. Test One Thing

   # Good ✅
   def test_get_request_returns_200():
       ...
   
   def test_get_request_parses_json():
       ...
   
   # Avoid ❌
   def test_get_request():
       # Tests multiple things
       ...
   

4. Use Fixtures

   # Good ✅
   @pytest.fixture
   def client():
       return create_client()
   
   def test_something(client):
       ...
   

5. Clean Up Resources

   # Good ✅
   @pytest.fixture
   def client():
       client = create_client()
       yield client
       client.close()
   

See Also

• Development Guide - Development guide

• Code Style Guide - Code style guide