tessl/pypi-tornado

Tornado is a Python web framework and asynchronous networking library designed for applications requiring long-lived connections to many users.

Overview

Eval results

Files

Testing Utilities

Name: tessl/pypi-tornado
Author: tessl

Comprehensive testing support for asynchronous code including test case classes, HTTP test servers, and async test decorators. Integrates with standard Python testing frameworks.

Capabilities

Async Test Cases

Base test case classes for testing asynchronous code with proper event loop management and timeout handling.

class AsyncTestCase(unittest.TestCase):
    """TestCase for async test methods."""
    
    def setUp(self):
        """Set up test case with fresh IOLoop."""
    
    def tearDown(self):
        """Clean up test case and close IOLoop."""
    
    def get_new_ioloop(self):
        """Create new IOLoop for testing."""
    
    def run_sync(self, func, timeout: float = None):
        """
        Run async function synchronously.
        
        Args:
            func: Async function to run
            timeout: Timeout in seconds
            
        Returns:
            Function result
        """
    
    def stop(self, _arg=None, **kwargs):
        """Stop the IOLoop."""
    
    def wait(self, condition=None, timeout: float = None):
        """
        Wait for condition or timeout.
        
        Args:
            condition: Condition function
            timeout: Timeout in seconds
        """

HTTP Test Cases

Test case classes that start HTTP servers for testing web applications and HTTP clients.

class AsyncHTTPTestCase(AsyncTestCase):
    """TestCase with HTTP server for testing web applications."""
    
    def setUp(self):
        """Set up test case with HTTP server."""
    
    def get_app(self):
        """
        Get application to test.
        
        Returns:
            tornado.web.Application instance
            
        Must be implemented by subclasses.
        """
    
    def get_server(self):
        """Get HTTP server instance."""
    
    def get_http_client(self):
        """Get HTTP client for making requests."""
    
    def get_http_port(self) -> int:
        """Get HTTP server port."""
    
    def get_protocol(self) -> str:
        """Get protocol (http or https)."""
    
    def get_url(self, path: str) -> str:
        """
        Get full URL for path.
        
        Args:
            path: URL path
            
        Returns:
            Full URL including protocol, host, and port
        """
    
    def fetch(self, path: str, **kwargs):
        """
        Fetch URL from test server.
        
        Args:
            path: URL path
            **kwargs: HTTPRequest arguments
            
        Returns:
            HTTPResponse
        """

class AsyncHTTPSTestCase(AsyncHTTPTestCase):
    """TestCase with HTTPS server."""
    
    def get_ssl_options(self):
        """
        Get SSL options for HTTPS server.
        
        Returns:
            SSL options dict
        """
    
    def get_protocol(self) -> str:
        """Get protocol (https)."""

Test Decorators

Decorators for async test methods and generator-based tests.

def gen_test(func=None, timeout: float = None):
    """
    Decorator for generator-based async tests.
    
    Args:
        func: Test function (when used without arguments)
        timeout: Test timeout in seconds
        
    Returns:
        Decorated test function
        
    Usage:
        @gen_test
        def test_async_operation(self):
            response = yield self.http_client.fetch(self.get_url('/'))
            self.assertEqual(response.code, 200)
    """

def timeout(timeout_seconds: float):
    """
    Decorator to set test timeout.
    
    Args:
        timeout_seconds: Timeout in seconds
        
    Returns:
        Decorator function
    """

Test Utilities

Utility functions for testing including port binding, log expectation, and async test configuration.

def bind_unused_port(reuse_port: bool = False) -> Tuple[socket.socket, int]:
    """
    Bind to unused port for testing.
    
    Args:
        reuse_port: Enable SO_REUSEPORT
        
    Returns:
        Tuple of (socket, port)
    """

def get_async_test_timeout() -> float:
    """
    Get default timeout for async tests.
    
    Returns:
        Timeout in seconds
    """

def setup_with_context_manager(test, context_manager):
    """
    Set up test with context manager.
    
    Args:
        test: Test case instance
        context_manager: Context manager to use
    """

def main(**kwargs):
    """
    Run tests with tornado-specific configuration.
    
    Args:
        **kwargs: Arguments passed to unittest.main()
    """

Log Testing

Context manager for testing expected log messages.

class ExpectLog(logging.Handler):
    """Context manager for expected log messages."""
    
    def __init__(self, logger, regex: str, required: bool = True, level: int = None):
        """
        Initialize log expectation.
        
        Args:
            logger: Logger to monitor
            regex: Regular expression for expected message
            required: Whether message is required
            level: Log level to expect
        """
    
    def __enter__(self):
        """Enter context manager."""
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit context manager and verify expectations."""

Usage Examples

Basic Async Test

import tornado.testing
import tornado.httpclient

class AsyncTestExample(tornado.testing.AsyncTestCase):
    @tornado.testing.gen_test
    def test_http_request(self):
        client = tornado.httpclient.AsyncHTTPClient()
        response = yield client.fetch("http://httpbin.org/get")
        self.assertEqual(response.code, 200)
        client.close()
    
    async def test_async_method(self):
        # Using async/await syntax
        client = tornado.httpclient.AsyncHTTPClient()
        try:
            response = await client.fetch("http://httpbin.org/get")
            self.assertEqual(response.code, 200)
        finally:
            client.close()

if __name__ == '__main__':
    tornado.testing.main()

HTTP Server Testing

import tornado.testing
import tornado.web

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write({"message": "Hello, World!"})
    
    def post(self):
        data = tornado.escape.json_decode(self.request.body)
        self.write({"echo": data})

class HTTPTestExample(tornado.testing.AsyncHTTPTestCase):
    def get_app(self):
        return tornado.web.Application([
            (r"/", MainHandler),
        ])
    
    def test_get_request(self):
        response = self.fetch('/')
        self.assertEqual(response.code, 200)
        data = tornado.escape.json_decode(response.body)
        self.assertEqual(data['message'], "Hello, World!")
    
    def test_post_request(self):
        body = tornado.escape.json_encode({"test": "data"})
        response = self.fetch('/', method='POST', body=body)
        self.assertEqual(response.code, 200)
        data = tornado.escape.json_decode(response.body)
        self.assertEqual(data['echo']['test'], "data")

if __name__ == '__main__':
    tornado.testing.main()

WebSocket Testing

import tornado.testing
import tornado.web
import tornado.websocket

class EchoWebSocket(tornado.websocket.WebSocketHandler):
    def on_message(self, message):
        self.write_message(f"Echo: {message}")

class WebSocketTestExample(tornado.testing.AsyncHTTPTestCase):
    def get_app(self):
        return tornado.web.Application([
            (r"/websocket", EchoWebSocket),
        ])
    
    @tornado.testing.gen_test
    def test_websocket(self):
        ws_url = f"ws://localhost:{self.get_http_port()}/websocket"
        ws = yield tornado.websocket.websocket_connect(ws_url)
        
        # Send message
        ws.write_message("Hello")
        
        # Read response
        msg = yield ws.read_message()
        self.assertEqual(msg, "Echo: Hello")
        
        ws.close()

if __name__ == '__main__':
    tornado.testing.main()

Testing with Expected Logs

import tornado.testing
import tornado.log
import logging

class LogTestExample(tornado.testing.AsyncTestCase):
    def test_expected_log(self):
        logger = logging.getLogger("test_logger")
        
        with tornado.testing.ExpectLog(logger, "Test message"):
            logger.info("Test message")
    
    def test_optional_log(self):
        logger = logging.getLogger("test_logger")
        
        # Log message is not required
        with tornado.testing.ExpectLog(logger, "Optional", required=False):
            pass  # No log message, but that's OK

if __name__ == '__main__':
    tornado.testing.main()

HTTPS Testing

import tornado.testing
import tornado.web
import ssl

class HTTPSTestExample(tornado.testing.AsyncHTTPSTestCase):
    def get_app(self):
        return tornado.web.Application([
            (r"/", MainHandler),
        ])
    
    def get_ssl_options(self):
        # For testing, use self-signed certificate
        return {
            "certfile": "test.crt",
            "keyfile": "test.key",
        }
    
    def test_https_request(self):
        # Disable SSL verification for self-signed cert
        response = self.fetch('/', validate_cert=False)
        self.assertEqual(response.code, 200)

if __name__ == '__main__':
    tornado.testing.main()

Generator-based Tests

import tornado.testing
import tornado.gen

class GeneratorTestExample(tornado.testing.AsyncTestCase):
    @tornado.testing.gen_test(timeout=10)
    def test_with_timeout(self):
        # Test with custom timeout
        yield tornado.gen.sleep(1)
        self.assertTrue(True)
    
    @tornado.testing.gen_test
    def test_multiple_operations(self):
        # Multiple async operations
        client = tornado.httpclient.AsyncHTTPClient()
        
        response1 = yield client.fetch("http://httpbin.org/get")
        response2 = yield client.fetch("http://httpbin.org/user-agent")
        
        self.assertEqual(response1.code, 200)
        self.assertEqual(response2.code, 200)
        
        client.close()

if __name__ == '__main__':
    tornado.testing.main()

Types

# Test function type
TestFunc = Callable[[], None]

# Async test function type
AsyncTestFunc = Callable[[], Awaitable[None]]

# Test timeout type
TestTimeout = float

# SSL options for testing
TestSSLOptions = Dict[str, Any]

# HTTP test response type
TestResponse = tornado.httpclient.HTTPResponse

Constants

# Default test timeout
_DEFAULT_ASYNC_TEST_TIMEOUT = 5.0

# Test certificate files
_TEST_CERT_FILE = "test.crt"
_TEST_KEY_FILE = "test.key"

Exceptions

class AsyncTestTimeoutError(Exception):
    """Exception for async test timeouts."""

class ExpectedLogNotFoundError(Exception):
    """Exception when expected log message is not found."""
    
    def __init__(self, logger_name: str, regex: str):
        """
        Initialize expected log error.
        
        Args:
            logger_name: Logger name
            regex: Expected message regex
        """

Install with Tessl CLI