# # This file is part of gunicorn released under the MIT license. # See the NOTICE for more information. """Test WSGI application for HTTP/2 Docker integration tests.""" import json def app(environ, start_response): """Simple WSGI app for testing HTTP/2 functionality.""" path = environ.get('PATH_INFO', '/') method = environ.get('REQUEST_METHOD', 'GET') if path == '/': body = b'Hello HTTP/2!' status = '200 OK' content_type = 'text/plain' elif path == '/health': body = b'OK' status = '200 OK' content_type = 'text/plain' elif path == '/echo': # Echo back the request body content_length = int(environ.get('CONTENT_LENGTH', 0) or 0) body = environ['wsgi.input'].read(content_length) status = '200 OK' content_type = 'application/octet-stream' elif path == '/headers': # Return all HTTP headers as JSON headers = {} for key, value in environ.items(): if key.startswith('HTTP_'): headers[key] = value # Also include some important non-HTTP_ headers for key in ['CONTENT_TYPE', 'CONTENT_LENGTH', 'REQUEST_METHOD', 'PATH_INFO', 'QUERY_STRING', 'SERVER_PROTOCOL']: if key in environ: headers[key] = str(environ[key]) body = json.dumps(headers, indent=2).encode('utf-8') status = '200 OK' content_type = 'application/json' elif path == '/version': # Return HTTP version info server_protocol = environ.get('SERVER_PROTOCOL', 'HTTP/1.1') body = server_protocol.encode('utf-8') status = '200 OK' content_type = 'text/plain' elif path == '/large': # Return a large response (1MB) for testing streaming body = b'X' * (1024 * 1024) status = '200 OK' content_type = 'application/octet-stream' elif path == '/stream': # Return a streaming response def generate(): for i in range(10): yield f'chunk-{i}\n'.encode('utf-8') start_response('200 OK', [ ('Content-Type', 'text/plain'), ('Transfer-Encoding', 'chunked') ]) return generate() elif path == '/status': # Return a specific status code based on query string query = environ.get('QUERY_STRING', '') try: code = int(query.split('=')[1]) if '=' in query else 200 except (ValueError, IndexError): code = 200 status_messages = { 200: 'OK', 201: 'Created', 204: 'No Content', 400: 'Bad Request', 404: 'Not Found', 500: 'Internal Server Error', } status = f'{code} {status_messages.get(code, "Unknown")}' body = f'Status: {code}'.encode('utf-8') content_type = 'text/plain' elif path == '/delay': # Simulate a slow response import time query = environ.get('QUERY_STRING', '') try: delay = float(query.split('=')[1]) if '=' in query else 1.0 delay = min(delay, 5.0) # Cap at 5 seconds except (ValueError, IndexError): delay = 1.0 time.sleep(delay) body = f'Delayed {delay}s'.encode('utf-8') status = '200 OK' content_type = 'text/plain' elif path == '/method': # Return the request method body = method.encode('utf-8') status = '200 OK' content_type = 'text/plain' elif path == '/early-hints': # Test endpoint for 103 Early Hints # Send early hints if the callback is available if 'wsgi.early_hints' in environ: environ['wsgi.early_hints']([ ('Link', '; rel=preload; as=style'), ('Link', '; rel=preload; as=script'), ]) body = b'Early hints sent!' status = '200 OK' content_type = 'text/plain' elif path == '/early-hints-multiple': # Test endpoint for multiple 103 Early Hints responses if 'wsgi.early_hints' in environ: # First early hints environ['wsgi.early_hints']([ ('Link', '; rel=preload; as=style'), ]) # Second early hints environ['wsgi.early_hints']([ ('Link', '; rel=preload; as=script'), ]) body = b'Multiple early hints sent!' status = '200 OK' content_type = 'text/plain' else: body = b'Not Found' status = '404 Not Found' content_type = 'text/plain' response_headers = [ ('Content-Type', content_type), ('Content-Length', str(len(body))), ('X-Request-Path', path), ('X-Request-Method', method), ] start_response(status, response_headers) return [body] # For running directly with python if __name__ == '__main__': from wsgiref.simple_server import make_server server = make_server('localhost', 8000, app) print('Serving on http://localhost:8000') server.serve_forever()