tessl/pypi-falcon

The ultra-reliable, fast ASGI+WSGI framework for building data plane APIs at scale.

—

Pending

Overview

Eval results

Files

Testing Framework

Name: tessl/pypi-falcon
Author: tessl

Comprehensive testing utilities for both WSGI and ASGI Falcon applications. Provides request simulation, response validation, WebSocket testing, and mock objects for building robust test suites.

Capabilities

WSGI Test Client

Full-featured test client for simulating HTTP requests against WSGI applications.

class TestClient:
    def __init__(self, app: object, headers: dict = None):
        """
        Create WSGI test client.
        
        Args:
            app: WSGI Falcon application instance
            headers: Default headers for all requests
        """
    
    def simulate_request(
        self,
        method: str,
        path: str,
        query_string: str = None,
        headers: dict = None,
        body: str = None,
        json: object = None,
        **kwargs
    ) -> Result:
        """
        Simulate generic HTTP request.
        
        Args:
            method: HTTP method (GET, POST, PUT, etc.)
            path: Request path (e.g., '/users/123')
            query_string: URL query parameters
            headers: Request headers
            body: Raw request body
            json: JSON request body (auto-serialized)
            **kwargs: Additional WSGI environment variables
            
        Returns:
            Result object with response data
        """
    
    def simulate_get(self, path: str, **kwargs) -> Result:
        """Simulate HTTP GET request"""
    
    def simulate_post(self, path: str, **kwargs) -> Result:
        """Simulate HTTP POST request"""
    
    def simulate_put(self, path: str, **kwargs) -> Result:
        """Simulate HTTP PUT request"""
    
    def simulate_patch(self, path: str, **kwargs) -> Result:
        """Simulate HTTP PATCH request"""
    
    def simulate_delete(self, path: str, **kwargs) -> Result:
        """Simulate HTTP DELETE request"""
    
    def simulate_head(self, path: str, **kwargs) -> Result:
        """Simulate HTTP HEAD request"""
    
    def simulate_options(self, path: str, **kwargs) -> Result:
        """Simulate HTTP OPTIONS request"""

Basic Testing Example

import falcon
from falcon.testing import TestClient

class UserResource:
    def on_get(self, req, resp, user_id=None):
        if user_id:
            resp.media = {'id': user_id, 'name': 'Test User'}
        else:
            resp.media = {'users': []}
    
    def on_post(self, req, resp):
        user_data = req.media
        resp.status = falcon.HTTP_201
        resp.media = {'created': user_data}

# Create app and test client
app = falcon.App()
app.add_route('/users', UserResource())
app.add_route('/users/{user_id}', UserResource())

client = TestClient(app)

# Test GET request
result = client.simulate_get('/users/123')
assert result.status_code == 200
assert result.json['name'] == 'Test User'

# Test POST request  
result = client.simulate_post('/users', json={'name': 'New User'})
assert result.status_code == 201
assert result.json['created']['name'] == 'New User'

ASGI Test Client

ASGI testing conductor for async applications and WebSocket testing.

class ASGIConductor:
    def __init__(self, app: object):
        """
        Create ASGI test conductor.
        
        Args:
            app: ASGI Falcon application instance
        """
    
    async def simulate_request(
        self,
        method: str,
        path: str,
        query_string: str = None,
        headers: dict = None,
        body: bytes = None,
        json: object = None,
        **kwargs
    ) -> Result:
        """
        Simulate async HTTP request.
        
        Args:
            method: HTTP method
            path: Request path
            query_string: URL query parameters
            headers: Request headers
            body: Raw request body bytes
            json: JSON request body (auto-serialized)
            **kwargs: Additional ASGI scope variables
            
        Returns:
            Result object with response data
        """
    
    async def simulate_get(self, path: str, **kwargs) -> Result:
        """Simulate async HTTP GET request"""
    
    async def simulate_post(self, path: str, **kwargs) -> Result:
        """Simulate async HTTP POST request"""
    
    async def simulate_put(self, path: str, **kwargs) -> Result:
        """Simulate async HTTP PUT request"""
    
    async def simulate_patch(self, path: str, **kwargs) -> Result:
        """Simulate async HTTP PATCH request"""
    
    async def simulate_delete(self, path: str, **kwargs) -> Result:
        """Simulate async HTTP DELETE request"""
    
    async def simulate_head(self, path: str, **kwargs) -> Result:
        """Simulate async HTTP HEAD request"""
    
    async def simulate_options(self, path: str, **kwargs) -> Result:
        """Simulate async HTTP OPTIONS request"""

WebSocket Testing

Utilities for testing WebSocket connections in ASGI applications.

class ASGIWebSocketSimulator:
    def __init__(self, app: object, path: str, headers: dict = None):
        """
        Create WebSocket test simulator.
        
        Args:
            app: ASGI Falcon application
            path: WebSocket path
            headers: Connection headers
        """
    
    async def __aenter__(self):
        """Async context manager entry"""
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit"""
    
    async def send_text(self, text: str):
        """
        Send text message to WebSocket.
        
        Args:
            text: Text message to send
        """
    
    async def send_data(self, data: bytes):
        """
        Send binary data to WebSocket.
        
        Args:
            data: Binary data to send
        """
    
    async def receive_text(self) -> str:
        """
        Receive text message from WebSocket.
        
        Returns:
            Text message received
        """
    
    async def receive_data(self) -> bytes:
        """
        Receive binary data from WebSocket.
        
        Returns:
            Binary data received
        """
    
    async def close(self, code: int = 1000):
        """
        Close WebSocket connection.
        
        Args:
            code: WebSocket close code
        """

ASGI and WebSocket Testing Example

import falcon.asgi
from falcon.testing import ASGIConductor, ASGIWebSocketSimulator

class AsyncUserResource:
    async def on_get(self, req, resp, user_id=None):
        # Simulate async database call
        user = await fetch_user(user_id)
        resp.media = user

class WebSocketEcho:
    async def on_websocket(self, req, ws):
        await ws.accept()
        while True:
            try:
                message = await ws.receive_text()
                await ws.send_text(f"Echo: {message}")
            except falcon.WebSocketDisconnected:
                break

# Create ASGI app
app = falcon.asgi.App()
app.add_route('/users/{user_id}', AsyncUserResource())
app.add_route('/echo', WebSocketEcho())

# Test HTTP endpoint
conductor = ASGIConductor(app)
result = await conductor.simulate_get('/users/123')
assert result.status_code == 200

# Test WebSocket
async with ASGIWebSocketSimulator(app, '/echo') as ws:
    await ws.send_text('Hello')
    response = await ws.receive_text()
    assert response == 'Echo: Hello'

Test Result Objects

Objects representing HTTP response data for assertions and validation.

class Result:
    def __init__(self, status: str, headers: list, body: bytes):
        """
        HTTP response result.
        
        Args:
            status: HTTP status line
            headers: Response headers list
            body: Response body bytes
        """
    
    # Properties
    status: str  # Full status line (e.g., '200 OK')
    status_code: int  # Status code only (e.g., 200)
    headers: dict  # Headers as case-insensitive dict
    text: str  # Response body as text
    content: bytes  # Raw response body
    json: object  # Parsed JSON response (if applicable)
    cookies: list  # Response cookies
    
    # Methods
    def __len__(self) -> int:
        """Get response body length"""
    
    def __iter__(self):
        """Iterate over response body bytes"""

class ResultBodyStream:
    def __init__(self, result: Result):
        """
        Streaming wrapper for large response bodies.
        
        Args:
            result: Result object to stream
        """
    
    def read(self, size: int = -1) -> bytes:
        """Read bytes from response body"""

class StreamedResult:
    def __init__(self, headers: dict, stream: object):
        """
        Result for streamed responses.
        
        Args:
            headers: Response headers
            stream: Response body stream
        """
    
    # Properties
    headers: dict
    stream: object

class Cookie:
    def __init__(self, name: str, value: str, **attributes):
        """
        Response cookie representation.
        
        Args:
            name: Cookie name
            value: Cookie value
            **attributes: Cookie attributes (path, domain, secure, etc.)
        """
    
    # Properties
    name: str
    value: str
    path: str
    domain: str
    secure: bool
    http_only: bool
    max_age: int
    expires: str

Test Helper Functions

Utility functions for creating test environments and mock objects.

# WSGI test helpers
def create_environ(
    method: str = 'GET',
    path: str = '/',
    query_string: str = '',
    headers: dict = None,
    body: str = '',
    **kwargs
) -> dict:
    """
    Create WSGI environment dictionary for testing.
    
    Args:
        method: HTTP method
        path: Request path
        query_string: Query parameters
        headers: Request headers
        body: Request body
        **kwargs: Additional environment variables
        
    Returns:
        WSGI environment dict
    """

def create_req(app: object, **kwargs) -> object:
    """
    Create Request object for testing.
    
    Args:
        app: Falcon application
        **kwargs: Environment parameters
        
    Returns:
        Request object
    """

# ASGI test helpers
def create_scope(
    type: str = 'http',
    method: str = 'GET',
    path: str = '/',
    query_string: str = '',
    headers: list = None,
    **kwargs
) -> dict:
    """
    Create ASGI scope dictionary for testing.
    
    Args:
        type: ASGI scope type ('http' or 'websocket')
        method: HTTP method
        path: Request path  
        query_string: Query parameters
        headers: Request headers as list of tuples
        **kwargs: Additional scope variables
        
    Returns:
        ASGI scope dict
    """

def create_asgi_req(app: object, **kwargs) -> object:
    """
    Create ASGI Request object for testing.
    
    Args:
        app: ASGI Falcon application
        **kwargs: Scope parameters
        
    Returns:
        ASGI Request object
    """

# Event simulation
class ASGIRequestEventEmitter:
    def __init__(self, body: bytes = b''):
        """
        ASGI request event emitter for testing.
        
        Args:
            body: Request body bytes
        """
    
    async def emit(self) -> dict:
        """Emit ASGI request events"""

class ASGIResponseEventCollector:
    def __init__(self):
        """ASGI response event collector for testing."""
    
    async def collect(self, message: dict):
        """Collect ASGI response events"""
    
    def get_response(self) -> tuple:
        """Get collected response data"""

# Utility functions
def get_unused_port(family: int = socket.AF_INET, type: int = socket.SOCK_STREAM) -> int:
    """
    Get unused network port for testing.
    
    Args:
        family: Socket family
        type: Socket type
        
    Returns:
        Unused port number
    """

def rand_string(length: int, alphabet: str = None) -> str:
    """
    Generate random string for testing.
    
    Args:
        length: String length
        alphabet: Character set to use
        
    Returns:
        Random string
    """

Test Resources and Mocks

Pre-built test resources and mock objects for common testing scenarios.

class SimpleTestResource:
    def __init__(self, status: str = '200 OK', body: str = 'Test'):
        """
        Simple test resource for basic testing.
        
        Args:
            status: HTTP status to return
            body: Response body
        """
    
    def on_get(self, req, resp):
        """Handle GET requests"""
    
    def on_post(self, req, resp):
        """Handle POST requests"""

class SimpleTestResourceAsync:
    def __init__(self, status: str = '200 OK', body: str = 'Test'):
        """
        Simple async test resource.
        
        Args:
            status: HTTP status to return
            body: Response body
        """
    
    async def on_get(self, req, resp):
        """Handle async GET requests"""

def capture_responder_args(*args, **kwargs) -> dict:
    """
    Capture responder method arguments for testing.
    
    Args:
        *args: Positional arguments
        **kwargs: Keyword arguments
        
    Returns:
        Captured arguments dict
    """

def set_resp_defaults(resp: object, base_resp: object):
    """
    Set default response values for testing.
    
    Args:
        resp: Response object to configure
        base_resp: Base response with defaults
    """

class StartResponseMock:
    def __init__(self):
        """Mock WSGI start_response callable for testing."""
    
    def __call__(self, status: str, headers: list, exc_info: tuple = None):
        """WSGI start_response interface"""
    
    # Properties
    status: str
    headers: list
    exc_info: tuple

Test Case Base Class

Base test case class with Falcon-specific helper methods.

class TestCase(unittest.TestCase):
    def setUp(self):
        """Test setup with Falcon app creation."""
        self.app = falcon.App()
        self.client = TestClient(self.app)
    
    def simulate_request(self, *args, **kwargs):
        """Simulate request using test client"""
        return self.client.simulate_request(*args, **kwargs)
    
    def simulate_get(self, *args, **kwargs):
        """Simulate GET request"""
        return self.client.simulate_get(*args, **kwargs)
    
    def simulate_post(self, *args, **kwargs):
        """Simulate POST request"""
        return self.client.simulate_post(*args, **kwargs)
    
    # Additional simulate_* methods for other HTTP verbs

Test Case Usage Example

import unittest
import falcon
from falcon.testing import TestCase

class UserResourceTest(TestCase):
    def setUp(self):
        super().setUp()
        self.app.add_route('/users/{user_id}', UserResource())
    
    def test_get_user(self):
        """Test getting user by ID"""
        result = self.simulate_get('/users/123')
        self.assertEqual(result.status_code, 200)
        self.assertEqual(result.json['id'], '123')
    
    def test_user_not_found(self):
        """Test user not found scenario"""
        result = self.simulate_get('/users/999')
        self.assertEqual(result.status_code, 404)
        
    def test_create_user(self):
        """Test user creation"""
        user_data = {'name': 'Test User', 'email': 'test@example.com'}
        result = self.simulate_post('/users', json=user_data)
        self.assertEqual(result.status_code, 201)
        self.assertEqual(result.json['name'], 'Test User')

Types

# Test clients
TestClient: type  # WSGI test client
ASGIConductor: type  # ASGI test conductor

# WebSocket testing
ASGIWebSocketSimulator: type

# Result objects
Result: type
ResultBodyStream: type  
StreamedResult: type
Cookie: type

# Test helpers
create_environ: callable
create_req: callable
create_scope: callable
create_asgi_req: callable

# Event simulation
ASGIRequestEventEmitter: type
ASGIResponseEventCollector: type

# Utilities
get_unused_port: callable
rand_string: callable

# Test resources
SimpleTestResource: type
SimpleTestResourceAsync: type
capture_responder_args: callable
set_resp_defaults: callable
StartResponseMock: type

# Test case base
TestCase: type

Install with Tessl CLI