tessl/pypi-treq

High-level Twisted HTTP Client API for asynchronous HTTP requests in Python

Overview

Eval results

Files

Testing Utilities

Name: tessl/pypi-treq
Author: tessl

In-memory testing framework for mocking HTTP requests without network access, enabling fast and reliable unit tests for HTTP-based applications. treq.testing provides a complete testing environment that simulates HTTP interactions.

Capabilities

StubTreq Class

Main testing client that provides all treq HTTP methods while routing requests to local Twisted resources instead of making network calls.

class StubTreq:
    def __init__(self, resource):
        """
        Create a testing HTTP client.
        
        Provides all the function calls exposed in treq.__all__ (get, post, put,
        delete, patch, head, request) plus content processing functions.
        
        Parameters:
        - resource: Resource - Twisted Resource object providing fake responses
        """
    
    # HTTP methods (same signatures as module-level functions)
    def get(self, url, **kwargs): ...
    def post(self, url, data=None, **kwargs): ...
    def put(self, url, data=None, **kwargs): ...
    def patch(self, url, data=None, **kwargs): ...
    def delete(self, url, **kwargs): ...
    def head(self, url, **kwargs): ...
    def request(self, method, url, **kwargs): ...
    
    # Content processing functions
    def collect(self, response, collector): ...
    def content(self, response): ...
    def text_content(self, response, encoding="ISO-8859-1"): ...
    def json_content(self, response, **kwargs): ...
    
    def flush(self):
        """
        Process all pending requests.
        
        StubTreq processes requests synchronously but may defer
        some operations. Call flush() to ensure all processing
        is complete.
        """

RequestTraversalAgent

Agent implementation that traverses Twisted resources to generate responses, used internally by StubTreq.

class RequestTraversalAgent:
    def __init__(self, resource):
        """
        Agent that traverses a resource tree to handle requests.
        
        Parameters:
        - resource: Resource - Root resource for request routing
        """
    
    def request(self, method, uri, headers=None, bodyProducer=None):
        """
        Process request through resource tree.
        
        Returns:
        Deferred that fires with IResponse
        """
    
    def flush(self):
        """Flush any pending operations."""

StringStubbingResource

Convenient resource for returning simple string responses with customizable status codes and headers.

class StringStubbingResource(Resource):
    def __init__(self, response_string):
        """
        Resource that returns a fixed string response.
        
        Parameters:
        - response_string: str or bytes - Response content to return
        """
    
    def render(self, request):
        """
        Handle request and return configured response.
        
        Returns:
        bytes - Response content
        """

Request Matching Utilities

Tools for validating that requests match expected patterns during testing.

class HasHeaders:
    def __init__(self, headers):
        """
        Matcher for validating request headers.
        
        Parameters:
        - headers: dict - Expected headers to match
        """

class RequestSequence:
    def __init__(self, requests, resource):
        """
        Resource that expects a specific sequence of requests.
        
        Parameters:
        - requests: list - Expected sequence of request patterns
        - resource: Resource - Resource to handle requests after validation
        """

Usage Examples

Basic Request Stubbing

from treq.testing import StubTreq, StringStubbingResource
from twisted.web.resource import Resource
from twisted.internet import defer
import json

# Create a simple resource
class APIResource(Resource):
    def render_GET(self, request):
        return json.dumps({'message': 'Hello, World!'}).encode('utf-8')
    
    def render_POST(self, request):
        # Echo back the posted data
        content = request.content.read()
        return json.dumps({'received': content.decode('utf-8')}).encode('utf-8')

@defer.inlineCallbacks
def test_basic_requests():
    # Create stub client
    resource = APIResource()
    client = StubTreq(resource)
    
    # Test GET request
    response = yield client.get('http://example.com/api')
    data = yield client.json_content(response)
    assert data['message'] == 'Hello, World!'
    
    # Test POST request
    response = yield client.post(
        'http://example.com/api',
        data='test data'
    )
    data = yield client.json_content(response)
    assert data['received'] == 'test data'
    
    # Ensure all processing is complete
    client.flush()

Testing with String Responses

from treq.testing import StubTreq, StringStubbingResource

@defer.inlineCallbacks
def test_string_responses():
    # Simple string resource
    resource = StringStubbingResource("Hello, World!")
    client = StubTreq(resource)
    
    response = yield client.get('http://example.com')
    text = yield client.text_content(response)
    assert text == "Hello, World!"
    
    # JSON string resource
    json_resource = StringStubbingResource('{"status": "ok"}')
    json_client = StubTreq(json_resource)
    
    response = yield json_client.get('http://example.com/status')
    data = yield json_client.json_content(response)
    assert data['status'] == 'ok'

Advanced Resource Testing

from twisted.web.resource import Resource
from twisted.web import server

class UserResource(Resource):
    def __init__(self):
        Resource.__init__(self)
        self.users = {}
        self.user_id_counter = 1
    
    def render_GET(self, request):
        # List all users
        return json.dumps(list(self.users.values())).encode('utf-8')
    
    def render_POST(self, request):
        # Create new user
        data = json.loads(request.content.read().decode('utf-8'))
        user_id = self.user_id_counter
        user = {'id': user_id, 'name': data['name'], 'email': data['email']}
        self.users[user_id] = user
        self.user_id_counter += 1
        
        request.setResponseCode(201)
        return json.dumps(user).encode('utf-8')

class UserDetailResource(Resource):
    def __init__(self, users):
        Resource.__init__(self)
        self.users = users
    
    def render_GET(self, request):
        user_id = int(request.path.decode('utf-8').split('/')[-1])
        if user_id in self.users:
            return json.dumps(self.users[user_id]).encode('utf-8')
        else:
            request.setResponseCode(404)
            return b'{"error": "User not found"}'

@defer.inlineCallbacks
def test_rest_api():
    # Set up resource tree
    root = Resource()
    user_resource = UserResource()
    root.putChild(b'users', user_resource)
    
    # Create client
    client = StubTreq(root)
    
    # Test creating user
    response = yield client.post(
        'http://example.com/users',
        json={'name': 'John Doe', 'email': 'john@example.com'}
    )
    assert response.code == 201
    user = yield client.json_content(response)
    assert user['name'] == 'John Doe'
    
    # Test listing users
    response = yield client.get('http://example.com/users')
    users = yield client.json_content(response)
    assert len(users) == 1
    assert users[0]['name'] == 'John Doe'

Error Response Testing

class ErrorResource(Resource):
    def render_GET(self, request):
        error_type = request.args.get(b'error', [b''])[0].decode('utf-8')
        
        if error_type == '404':
            request.setResponseCode(404)
            return b'Not Found'
        elif error_type == '500':
            request.setResponseCode(500)
            return b'Internal Server Error'
        elif error_type == 'timeout':
            # Simulate timeout by not responding
            return server.NOT_DONE_YET
        else:
            return b'OK'

@defer.inlineCallbacks
def test_error_handling():
    client = StubTreq(ErrorResource())
    
    # Test 404 error
    response = yield client.get('http://example.com?error=404')
    assert response.code == 404
    text = yield client.text_content(response)
    assert text == 'Not Found'
    
    # Test 500 error
    response = yield client.get('http://example.com?error=500')
    assert response.code == 500
    
    # Test successful response
    response = yield client.get('http://example.com')
    assert response.code == 200

Request Validation Testing

from treq.testing import HasHeaders

class ValidatingResource(Resource):
    def render_POST(self, request):
        # Validate headers
        if b'content-type' not in request.requestHeaders:
            request.setResponseCode(400)
            return b'Missing Content-Type'
        
        # Validate authentication
        auth_header = request.getHeader('authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            request.setResponseCode(401)
            return b'Unauthorized'
        
        return b'{"status": "success"}'

@defer.inlineCallbacks
def test_request_validation():
    client = StubTreq(ValidatingResource())
    
    # Test missing headers
    response = yield client.post('http://example.com/api', data='test')
    assert response.code == 400
    
    # Test with proper headers
    response = yield client.post(
        'http://example.com/api',
        json={'data': 'test'},
        headers={
            'Authorization': 'Bearer token123',
            'Content-Type': 'application/json'
        }
    )
    assert response.code == 200

Asynchronous Resource Testing

from twisted.internet import defer

class AsyncResource(Resource):
    @defer.inlineCallbacks
    def render_GET(self, request):
        # Simulate async operation
        yield defer.sleep(0.1)  # Simulated delay
        
        # Return response
        request.write(b'{"async": "response"}')
        request.finish()
        
        defer.returnValue(server.NOT_DONE_YET)

@defer.inlineCallbacks
def test_async_resource():
    client = StubTreq(AsyncResource())
    
    response = yield client.get('http://example.com/async')
    data = yield client.json_content(response)
    assert data['async'] == 'response'
    
    # Ensure async operations complete
    client.flush()

Integration Testing Pattern

class TestHTTPClient:
    def setUp(self):
        self.resource = self.create_test_resource()
        self.client = StubTreq(self.resource)
    
    def create_test_resource(self):
        root = Resource()
        api = Resource()
        root.putChild(b'api', api)
        
        # Add various endpoints
        api.putChild(b'users', UserResource())
        api.putChild(b'posts', PostResource())
        return root
    
    @defer.inlineCallbacks
    def test_user_workflow(self):
        # Create user
        user_response = yield self.client.post(
            'http://example.com/api/users',
            json={'name': 'Test User'}
        )
        user = yield self.client.json_content(user_response)
        
        # Create post for user
        post_response = yield self.client.post(
            'http://example.com/api/posts',
            json={'title': 'Test Post', 'user_id': user['id']}
        )
        post = yield self.client.json_content(post_response)
        
        # Verify relationships
        assert post['user_id'] == user['id']
        
        self.client.flush()

Types

Testing-related types:

# Resource types from Twisted
from twisted.web.resource import Resource

# Agent interface
from twisted.web.iweb import IAgent

# Request type
from twisted.web.server import Request

# Testing utilities
StubTreqType = StubTreq
RequestTraversalAgentType = RequestTraversalAgent

Testing Best Practices

Resource Design

Stateless resources: Design resources to be stateless when possible
Deterministic responses: Ensure consistent behavior across test runs
Error simulation: Include error conditions in test resources
Realistic delays: Use defer.sleep() for timing-sensitive tests

Test Organization

Setup/teardown: Use proper test setup and cleanup
Resource reuse: Share common resources across related tests
Isolation: Ensure tests don't interfere with each other
Assertions: Use clear, specific assertions for request/response validation

Performance Considerations

Memory usage: Clean up resources after tests
Test speed: StubTreq eliminates network latency for fast tests
Flush operations: Always call flush() to ensure completion
Resource complexity: Keep test resources simple and focused

Common Patterns

Mock external APIs: Replace external service calls with controlled responses
Error condition testing: Systematically test error scenarios
Authentication testing: Verify auth flows without real credentials
Integration testing: Test complete request/response workflows

Install with Tessl CLI