0
# Testing Utilities
1

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

4
## Capabilities
5

6
### Async Test Cases
7

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

10
```python { .api }
11
class AsyncTestCase(unittest.TestCase):
12
    """TestCase for async test methods."""
13
    
14
    def setUp(self):
15
        """Set up test case with fresh IOLoop."""
16
    
17
    def tearDown(self):
18
        """Clean up test case and close IOLoop."""
19
    
20
    def get_new_ioloop(self):
21
        """Create new IOLoop for testing."""
22
    
23
    def run_sync(self, func, timeout: float = None):
24
        """
25
        Run async function synchronously.
26
        
27
        Args:
28
            func: Async function to run
29
            timeout: Timeout in seconds
30
            
31
        Returns:
32
            Function result
33
        """
34
    
35
    def stop(self, _arg=None, **kwargs):
36
        """Stop the IOLoop."""
37
    
38
    def wait(self, condition=None, timeout: float = None):
39
        """
40
        Wait for condition or timeout.
41
        
42
        Args:
43
            condition: Condition function
44
            timeout: Timeout in seconds
45
        """
46
```
47

48
### HTTP Test Cases
49

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

52
```python { .api }
53
class AsyncHTTPTestCase(AsyncTestCase):
54
    """TestCase with HTTP server for testing web applications."""
55
    
56
    def setUp(self):
57
        """Set up test case with HTTP server."""
58
    
59
    def get_app(self):
60
        """
61
        Get application to test.
62
        
63
        Returns:
64
            tornado.web.Application instance
65
            
66
        Must be implemented by subclasses.
67
        """
68
    
69
    def get_server(self):
70
        """Get HTTP server instance."""
71
    
72
    def get_http_client(self):
73
        """Get HTTP client for making requests."""
74
    
75
    def get_http_port(self) -> int:
76
        """Get HTTP server port."""
77
    
78
    def get_protocol(self) -> str:
79
        """Get protocol (http or https)."""
80
    
81
    def get_url(self, path: str) -> str:
82
        """
83
        Get full URL for path.
84
        
85
        Args:
86
            path: URL path
87
            
88
        Returns:
89
            Full URL including protocol, host, and port
90
        """
91
    
92
    def fetch(self, path: str, **kwargs):
93
        """
94
        Fetch URL from test server.
95
        
96
        Args:
97
            path: URL path
98
            **kwargs: HTTPRequest arguments
99
            
100
        Returns:
101
            HTTPResponse
102
        """
103

104
class AsyncHTTPSTestCase(AsyncHTTPTestCase):
105
    """TestCase with HTTPS server."""
106
    
107
    def get_ssl_options(self):
108
        """
109
        Get SSL options for HTTPS server.
110
        
111
        Returns:
112
            SSL options dict
113
        """
114
    
115
    def get_protocol(self) -> str:
116
        """Get protocol (https)."""
117
```
118

119
### Test Decorators
120

121
Decorators for async test methods and generator-based tests.
122

123
```python { .api }
124
def gen_test(func=None, timeout: float = None):
125
    """
126
    Decorator for generator-based async tests.
127
    
128
    Args:
129
        func: Test function (when used without arguments)
130
        timeout: Test timeout in seconds
131
        
132
    Returns:
133
        Decorated test function
134
        
135
    Usage:
136
        @gen_test
137
        def test_async_operation(self):
138
            response = yield self.http_client.fetch(self.get_url('/'))
139
            self.assertEqual(response.code, 200)
140
    """
141

142
def timeout(timeout_seconds: float):
143
    """
144
    Decorator to set test timeout.
145
    
146
    Args:
147
        timeout_seconds: Timeout in seconds
148
        
149
    Returns:
150
        Decorator function
151
    """
152
```
153

154
### Test Utilities
155

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

158
```python { .api }
159
def bind_unused_port(reuse_port: bool = False) -> Tuple[socket.socket, int]:
160
    """
161
    Bind to unused port for testing.
162
    
163
    Args:
164
        reuse_port: Enable SO_REUSEPORT
165
        
166
    Returns:
167
        Tuple of (socket, port)
168
    """
169

170
def get_async_test_timeout() -> float:
171
    """
172
    Get default timeout for async tests.
173
    
174
    Returns:
175
        Timeout in seconds
176
    """
177

178
def setup_with_context_manager(test, context_manager):
179
    """
180
    Set up test with context manager.
181
    
182
    Args:
183
        test: Test case instance
184
        context_manager: Context manager to use
185
    """
186

187
def main(**kwargs):
188
    """
189
    Run tests with tornado-specific configuration.
190
    
191
    Args:
192
        **kwargs: Arguments passed to unittest.main()
193
    """
194
```
195

196
### Log Testing
197

198
Context manager for testing expected log messages.
199

200
```python { .api }
201
class ExpectLog(logging.Handler):
202
    """Context manager for expected log messages."""
203
    
204
    def __init__(self, logger, regex: str, required: bool = True, level: int = None):
205
        """
206
        Initialize log expectation.
207
        
208
        Args:
209
            logger: Logger to monitor
210
            regex: Regular expression for expected message
211
            required: Whether message is required
212
            level: Log level to expect
213
        """
214
    
215
    def __enter__(self):
216
        """Enter context manager."""
217
        return self
218
    
219
    def __exit__(self, exc_type, exc_val, exc_tb):
220
        """Exit context manager and verify expectations."""
221
```
222

223
## Usage Examples
224

225
### Basic Async Test
226

227
```python
228
import tornado.testing
229
import tornado.httpclient
230

231
class AsyncTestExample(tornado.testing.AsyncTestCase):
232
    @tornado.testing.gen_test
233
    def test_http_request(self):
234
        client = tornado.httpclient.AsyncHTTPClient()
235
        response = yield client.fetch("http://httpbin.org/get")
236
        self.assertEqual(response.code, 200)
237
        client.close()
238
    
239
    async def test_async_method(self):
240
        # Using async/await syntax
241
        client = tornado.httpclient.AsyncHTTPClient()
242
        try:
243
            response = await client.fetch("http://httpbin.org/get")
244
            self.assertEqual(response.code, 200)
245
        finally:
246
            client.close()
247

248
if __name__ == '__main__':
249
    tornado.testing.main()
250
```
251

252
### HTTP Server Testing
253

254
```python
255
import tornado.testing
256
import tornado.web
257

258
class MainHandler(tornado.web.RequestHandler):
259
    def get(self):
260
        self.write({"message": "Hello, World!"})
261
    
262
    def post(self):
263
        data = tornado.escape.json_decode(self.request.body)
264
        self.write({"echo": data})
265

266
class HTTPTestExample(tornado.testing.AsyncHTTPTestCase):
267
    def get_app(self):
268
        return tornado.web.Application([
269
            (r"/", MainHandler),
270
        ])
271
    
272
    def test_get_request(self):
273
        response = self.fetch('/')
274
        self.assertEqual(response.code, 200)
275
        data = tornado.escape.json_decode(response.body)
276
        self.assertEqual(data['message'], "Hello, World!")
277
    
278
    def test_post_request(self):
279
        body = tornado.escape.json_encode({"test": "data"})
280
        response = self.fetch('/', method='POST', body=body)
281
        self.assertEqual(response.code, 200)
282
        data = tornado.escape.json_decode(response.body)
283
        self.assertEqual(data['echo']['test'], "data")
284

285
if __name__ == '__main__':
286
    tornado.testing.main()
287
```
288

289
### WebSocket Testing
290

291
```python
292
import tornado.testing
293
import tornado.web
294
import tornado.websocket
295

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

300
class WebSocketTestExample(tornado.testing.AsyncHTTPTestCase):
301
    def get_app(self):
302
        return tornado.web.Application([
303
            (r"/websocket", EchoWebSocket),
304
        ])
305
    
306
    @tornado.testing.gen_test
307
    def test_websocket(self):
308
        ws_url = f"ws://localhost:{self.get_http_port()}/websocket"
309
        ws = yield tornado.websocket.websocket_connect(ws_url)
310
        
311
        # Send message
312
        ws.write_message("Hello")
313
        
314
        # Read response
315
        msg = yield ws.read_message()
316
        self.assertEqual(msg, "Echo: Hello")
317
        
318
        ws.close()
319

320
if __name__ == '__main__':
321
    tornado.testing.main()
322
```
323

324
### Testing with Expected Logs
325

326
```python
327
import tornado.testing
328
import tornado.log
329
import logging
330

331
class LogTestExample(tornado.testing.AsyncTestCase):
332
    def test_expected_log(self):
333
        logger = logging.getLogger("test_logger")
334
        
335
        with tornado.testing.ExpectLog(logger, "Test message"):
336
            logger.info("Test message")
337
    
338
    def test_optional_log(self):
339
        logger = logging.getLogger("test_logger")
340
        
341
        # Log message is not required
342
        with tornado.testing.ExpectLog(logger, "Optional", required=False):
343
            pass  # No log message, but that's OK
344

345
if __name__ == '__main__':
346
    tornado.testing.main()
347
```
348

349
### HTTPS Testing
350

351
```python
352
import tornado.testing
353
import tornado.web
354
import ssl
355

356
class HTTPSTestExample(tornado.testing.AsyncHTTPSTestCase):
357
    def get_app(self):
358
        return tornado.web.Application([
359
            (r"/", MainHandler),
360
        ])
361
    
362
    def get_ssl_options(self):
363
        # For testing, use self-signed certificate
364
        return {
365
            "certfile": "test.crt",
366
            "keyfile": "test.key",
367
        }
368
    
369
    def test_https_request(self):
370
        # Disable SSL verification for self-signed cert
371
        response = self.fetch('/', validate_cert=False)
372
        self.assertEqual(response.code, 200)
373

374
if __name__ == '__main__':
375
    tornado.testing.main()
376
```
377

378
### Generator-based Tests
379

380
```python
381
import tornado.testing
382
import tornado.gen
383

384
class GeneratorTestExample(tornado.testing.AsyncTestCase):
385
    @tornado.testing.gen_test(timeout=10)
386
    def test_with_timeout(self):
387
        # Test with custom timeout
388
        yield tornado.gen.sleep(1)
389
        self.assertTrue(True)
390
    
391
    @tornado.testing.gen_test
392
    def test_multiple_operations(self):
393
        # Multiple async operations
394
        client = tornado.httpclient.AsyncHTTPClient()
395
        
396
        response1 = yield client.fetch("http://httpbin.org/get")
397
        response2 = yield client.fetch("http://httpbin.org/user-agent")
398
        
399
        self.assertEqual(response1.code, 200)
400
        self.assertEqual(response2.code, 200)
401
        
402
        client.close()
403

404
if __name__ == '__main__':
405
    tornado.testing.main()
406
```
407

408
## Types
409

410
```python { .api }
411
# Test function type
412
TestFunc = Callable[[], None]
413

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

417
# Test timeout type
418
TestTimeout = float
419

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

423
# HTTP test response type
424
TestResponse = tornado.httpclient.HTTPResponse
425
```
426

427
## Constants
428

429
```python { .api }
430
# Default test timeout
431
_DEFAULT_ASYNC_TEST_TIMEOUT = 5.0
432

433
# Test certificate files
434
_TEST_CERT_FILE = "test.crt"
435
_TEST_KEY_FILE = "test.key"
436
```
437

438
## Exceptions
439

440
```python { .api }
441
class AsyncTestTimeoutError(Exception):
442
    """Exception for async test timeouts."""
443

444
class ExpectedLogNotFoundError(Exception):
445
    """Exception when expected log message is not found."""
446
    
447
    def __init__(self, logger_name: str, regex: str):
448
        """
449
        Initialize expected log error.
450
        
451
        Args:
452
            logger_name: Logger name
453
            regex: Expected message regex
454
        """
455
```