Ticket #32472: test_server.py

import random
import socket
import string
from wsgiref import simple_server
from wsgiref.util import setup_testing_defaults


class ServerHandler(simple_server.ServerHandler):
    http_version = '1.1'


class WSGIRequestHandler(simple_server.WSGIRequestHandler):
    protocol_version = 'HTTP/1.1'

    def handle(self):
        """Copy of WSGIRequestHandler.handle() but with different ServerHandler"""
        self.raw_requestline = self.rfile.readline(65537)
        if len(self.raw_requestline) > 65536:
            self.requestline = ''
            self.request_version = ''
            self.command = ''
            self.send_error(414)
            return

        if not self.parse_request():  # An error code has been sent, just exit
            return

        handler = ServerHandler(
            self.rfile, self.wfile, self.get_stderr(), self.get_environ()
        )
        handler.request_handler = self      # backpointer for logging & connection closing
        handler.run(self.server.get_app())


def large_response(environ, start_response):
    setup_testing_defaults(environ)

    status = '200 OK'
    headers = [('Content-type', 'text/plain; charset=utf-8')]

    start_response(status, headers)

    response_body = ''.join(random.choices(string.ascii_uppercase + string.digits, k=1024*1024))

    return [response_body.encode()]


with simple_server.make_server('0.0.0.0', 8001, large_response, handler_class=WSGIRequestHandler) as httpd:
    print("Serving on port 8001...")
    httpd.serve_forever()