nonsocket http server def initself serveraddress requesthandlerclass baseserver initself serveraddress requesthandlerclass self serverbind def serverbindself host port self serveraddress self servername host self serverport port self setupenviron class mockhandlerwsgirequesthandler utility to compare a generic iterator with an iterable this tests the iterator using iternext makeit must be a function returning a fresh iterator to be tested since this may test the iterator twice it makeit if not iterit is it raise assertionerror for item in match if not nextit item raise assertionerror try nextit except stopiteration pass else raise assertionerrortoo many items from next it class integrationteststestcase def checkhelloself out haslengthtrue pyver pythonimplementation sys version split0 self assertequalout http1 0 200 okrn server wsgiserver0 2 pyver rn contenttype textplainrn date mon 05 jun 2006 18 49 54 gmtrn haslength and contentlength 13rn or rn hello world encodeiso88591 def testplainhelloself out err runamock self checkhelloout def testenvironself request bget p61th querytest http1 0n bxtestheader python test n bxtestheader python test 2n bcontentlength 0nn out err runamockheaderapp request self assertequal out splitlines1 bpython test python test 2 querytest path def testrequestlengthself out err runamockdatabget bx 65537 b http1 0nn self assertequalout splitlines0 bhttp1 0 414 requesturi too long def testvalidatedhelloself out err runamockvalidatorhelloapp the middleware doesn t support len so contentlength isn t there self checkhelloout haslengthfalse def testsimplevalidationerrorself def badappenviron startresponse startresponse200 ok contenttype textplain return hello world out err runamockvalidatorbadapp self asserttrueout endswith ba server error occurred please contact the administrator self assertequal err splitlines2 assertionerror headers contenttype textplain must be of type list class tuple def teststatusvalidationerrorsself def createbadappstatus def badappenviron startresponse startresponsestatus contenttype textplain charsetutf8 return bhello world return badapp tests 200 assertionerror status must be at least 4 characters 20x ok assertionerror status message must begin w3digit code 200ok assertionerror status message must have a space after code for status excmessage in tests with self subteststatusstatus out err runamockcreatebadappstatus self asserttrueout endswith ba server error occurred please contact the administrator self assertequalerr splitlines2 excmessage def testwsgiinputself def badappe s ewsgi input read s200 ok contenttype textplain charsetutf8 return bdata out err runamockvalidatorbadapp self asserttrueout endswith ba server error occurred please contact the administrator self assertequal err splitlines2 assertionerror def testbytesvalidationself def appe s s200 ok contenttype textplain charsetutf8 date wed 24 dec 2008 13 29 32 gmt return bdata out err runamockvalidatorapp self asserttrueerr endswith get http1 0 200 4n ver sys version split0 encode ascii py pythonimplementation encode ascii pyver py b ver self assertequal bhttp1 0 200 okrn bserver wsgiserver0 2 pyver brn bcontenttype textplain charsetutf8rn bdate wed 24 dec 2008 13 29 32 gmtrn brn bdata out def testcp1252urlself def appe s s200 ok contenttype textplain date wed 24 dec 2008 13 29 32 gmt pep3333 says environ variables are decoded as latin1 encode as latin1 to get original bytes return epathinfo encodelatin1 out err runamock validatorapp databget x8080 http1 0 self assertequal bhttp1 0 200 ok mock any bcontenttype textplain bdate wed 24 dec 2008 13 29 32 gmt b bx80x80 out splitlines def testinterruptedwriteself basehandler write and flush have to write all data even if it takes multiple send calls test this by interrupting a send call with a unix signal pthreadkill support getattributesignal pthreadkill def appenviron startresponse startresponse200 ok return b 0 support sockmaxsize class wsgihandlernologrequesthandler wsgirequesthandler pass server makeserversockethelper host 0 app handlerclasswsgihandler self addcleanupserver serverclose interrupted threading event def signalhandlersignum frame interrupted set original signal signalsignal sigusr1 signalhandler self addcleanupsignal signal signal sigusr1 original received none mainthread threading getident def runclient http httpconnectionserver serveraddress http requestget with http getresponse as response response read100 the main thread should now be blocking in a send system call but in theory it could get interrupted by other signals and then retried so keep sending the signal in a loop in case an earlier signal happens to be delivered at an inconvenient moment while true pthreadkillmainthread signal sigusr1 if interrupted waittimeoutfloat1 break nonlocal received received lenresponse read http close background threading threadtargetrunclient background start server handlerequest background join self assertequalreceived support sockmaxsize 100 class utilityteststestcase def checkshiftself snin piin part snout piout env scriptname snin pathinfo piin util setuptestingdefaultsenv self assertequalutil shiftpathinfoenv part self assertequalenv pathinfo piout self assertequalenv scriptname snout return env def checkdefaultself key value altnone check defaulting when empty env util setuptestingdefaultsenv if isinstancevalue stringio self assertisinstanceenvkey stringio elif isinstancevalue bytesio self assertisinstanceenvkey bytesio else self assertequalenvkey value check existing value env key alt util setuptestingdefaultsenv self assertisenvkey alt def checkcrossdefaultself key value kw util setuptestingdefaultskw self assertequalkwkey value def checkappuriself uri kw util setuptestingdefaultskw self assertequalutil applicationurikw uri def checkrequriself uri query1 kw util setuptestingdefaultskw self assertequalutil requesturikw query uri def checkfwself text size match def makeittexttext sizesize return util filewrapperstringiotext size comparegenericitermakeit match it makeit self assertfalseit filelike closed for item in it pass self assertfalseit filelike closed it close self asserttrueit filelike closed def testsimpleshiftsself self checkshift self checkshift x x x self checkshift none self checkshift a xy x ax y self checkshift a x x ax def testnormalizedshiftsself self checkshift ab y a y self checkshift y y self checkshift ab y y aby self checkshift ab y y aby self checkshift ab y y aby self checkshift ab y y aby self checkshift ab y a y self checkshift ab ab self checkshift ab ab self checkshift ab x x abx self checkshift ab none ab def testdefaultsself for key value in servername 127 0 0 1 serverport 80 serverprotocol http1 0 httphost 127 0 0 1 requestmethod get scriptname pathinfo wsgi version 1 0 wsgi runonce 0 wsgi multithread 0 wsgi multiprocess 0 wsgi input bytesio wsgi errors stringio wsgi urlscheme http self checkdefaultkey value def testcrossdefaultsself self checkcrossdefault httphost foo bar servernamefoo bar self checkcrossdefault wsgi urlscheme https httpson self checkcrossdefault wsgi urlscheme https https1 self checkcrossdefault wsgi urlscheme https httpsyes self checkcrossdefault wsgi urlscheme http httpsfoo self checkcrossdefault serverport 80 httpsfoo self checkcrossdefault serverport 443 httpson def testguessschemeself self assertequalutil guessscheme http self assertequalutil guessscheme https foo http self assertequalutil guessscheme https on https self assertequalutil guessscheme https yes https self assertequalutil guessscheme https 1 https def testappurisself self checkappurihttp 127 0 0 1 self checkappurihttp 127 0 0 1spam scriptnamespam self checkappurihttp 127 0 0 1spe4m scriptnamespxe4m self checkappurihttp spam example com 2071 httphostspam example com 2071 serverport2071 self checkappurihttp spam example com servernamespam example com self checkappurihttp 127 0 0 1 httphost127 0 0 1 servernamespam example com self checkappurihttps 127 0 0 1 httpson self checkappurihttp 127 0 0 1 8000 serverport8000 httphostnone def testrequrisself self checkrequrihttp 127 0 0 1 self checkrequrihttp 127 0 0 1spam scriptnamespam self checkrequrihttp 127 0 0 1spe4m scriptnamespxe4m self checkrequrihttp 127 0 0 1spammityspam scriptnamespammity pathinfospam self checkrequrihttp 127 0 0 1spammityspe4m scriptnamespammity pathinfospxe4m self checkrequrihttp 127 0 0 1spammityspam ham scriptnamespammity pathinfospam ham self checkrequrihttp 127 0 0 1spammityspam cookie1234 5678 scriptnamespammity pathinfospam cookie1234 5678 self checkrequrihttp 127 0 0 1spammityspam sayni scriptnamespammity pathinfospam querystringsayni self checkrequrihttp 127 0 0 1spammityspam se4yni scriptnamespammity pathinfospam querystringse4yni self checkrequrihttp 127 0 0 1spammityspam 0 scriptnamespammity pathinfospam querystringsayni def testfilewrapperself self checkfwxyz50 120 xyz40 xyz10 def testhopbyhopself for hop in connection keepalive proxyauthenticate proxyization te trailers transferencoding upgrade split for alt in hop hop title hop upper hop lower self asserttrueutil ishopbyhopalt not comprehensive just a few random header names for hop in accept cachecontrol date pragma trailer via warning split for alt in hop hop title hop upper hop lower self assertfalseutil ishopbyhopalt class headerteststestcase def testmappinginterfaceself test x y self assertequallenheaders 0 self assertequallenheaders 0 self assertequallenheaderstest 1 self assertequalheaderstest keys x self assertequalheaderstest values y self assertequalheaderstest items test self assertisnotheaderstest items test must be copy h headers del h foo should not raise an error h foo bar for m in h contains h get h getall h getitem self asserttruem foo self asserttruem foo self asserttruem foo self assertfalsem bar self assertequalh foo bar h foo baz self assertequalh foo baz self assertequalh getall foo baz self assertequalh getfoo whee baz self assertequalh getzoo whee whee self assertequalh setdefaultfoo whee baz self assertequalh setdefaultzoo whee whee self assertequalhfoo baz self assertequalhzoo whee def testrequirelistself self assertraisestypeerror headers foo def testextrasself h headers self assertequalstrh rn h addheader foo bar bazspam self assertequalh foo bar bazspam self assertequalstrh foo bar bazspamrnrn h addheader foo bar cheesenone self assertequalh getall foo bar bazspam bar cheese self assertequalstrh foo bar bazspamrn foo bar cheesern rn class errorhandlerbasecgihandler basehandler records the os environment at import time but envvars might have been changed later by other tests which trips up handlertests testenviron simple handler subclass for testing basehandler werror passthru def handleerrorself raise for testing we want to see what s happening class handlerteststestcase testenviron can produce long error message maxdiff 80 50 def testenvironself osenviron very basic environment home myhome path mypath lang frfr utf8 set some wsgi variables scriptname testscriptname servername testservername with support swapattrtesthandler osenviron osenviron override x and home variables handler testhandlerxy homeoverridehome handler setupenviron check that wsgixxx attributes are copied to wsgi xxx variables of handler environ for attr in version multithread multiprocess runonce filewrapper self assertequalgetattrhandler wsgi attr handler environ wsgi attr test handler environ as a dict expected setuptestingdefaultsexpected handler inherits osenviron variables which are not overridden by simplehandler addcgivars simplehandler baseenv for key value in osenviron items if key not in expected expectedkey value expected update x doesn t exist in osenviron x y home is overridden by testhandler home overridehome overridden by setuptestingdefaults scriptname servername 127 0 0 1 set by basehandler setupenviron wsgi input handler getstdin wsgi errors handler getstderr wsgi version 1 0 wsgi runonce false wsgi urlscheme http wsgi multithread true wsgi multiprocess true wsgi filewrapper util filewrapper self assertdictequalhandler environ expected def testcgienvironself h basecgihandlernone none none h setupenviron for key in wsgi urlscheme wsgi input wsgi errors self assertinkey h environ def testschemeself htesthandlerhttpson h setupenviron self assertequalh environ wsgi urlscheme https htesthandler h setupenviron self assertequalh environ wsgi urlscheme http def testabstractmethodsself h basehandler for name in flush getstdin getstderr addcgivars self assertraisesnotimplementederror getattrh name self assertraisesnotimplementederror h write test def testcontentlengthself demo one reason iteration is better than write def trivialapp1e s s 200 ok return e wsgi urlscheme encode iso88591 def trivialapp2e s s 200 ok e wsgi urlscheme encode iso88591 return def trivialapp3e s s 200 ok return u0442u0435u0441u0442 encodeutf8 def trivialapp4e s simulate a response to a head request s 200 ok contentlength 12345 return h testhandler h runtrivialapp1 self assertequalh stdout getvalue status 200 okrn contentlength 4rn rn http encodeiso88591 h testhandler h runtrivialapp2 self assertequalh stdout getvalue status 200 okrn rn http encodeiso88591 h testhandler h runtrivialapp3 self assertequalh stdout getvalue b status 200 okrn b contentlength 8rn b rn b xd1x82xd0xb5xd1x81xd1x82 h testhandler h runtrivialapp4 self assertequalh stdout getvalue b status 200 okrn b contentlength 12345rn b rn def testbasicerroroutputself def nonerrorappe s s 200 ok return def errorappe s raise assertionerrorthis should be caught by handler h errorhandler h runnonerrorapp self assertequalh stdout getvalue status 200 okrn contentlength 0rn rn encodeiso88591 self assertequalh stderr getvalue h errorhandler h runerrorapp self assertequalh stdout getvalue status srn contenttype textplainrn contentlength drn rn h errorstatus lenh errorbody encode iso88591 h errorbody self assertinassertionerror h stderr getvalue def testerrorafteroutputself msg bsome output has been sent def errorappe s s200 ok msg raise assertionerrorthis should be caught by handler h errorhandler h runerrorapp self assertequalh stdout getvalue status 200 okrn rn encodeiso88591msg self assertinassertionerror h stderr getvalue def testheaderformatsself def nonerrorappe s s 200 ok return stdpat rhttps 200 okrn rdate w3 0123d w3 d4 dd dd dd gmtrn rs rcontentlength 0rn rrn shortpat status 200 okrn contentlength 0rn rn encodeiso88591 for ssw in foobar1 0 none sw ssw and server srn ssw or for version in 1 0 1 1 for proto in http0 9 http1 0 http1 1 h testhandlerserverprotocolproto h originserver false h httpversion version h serversoftware ssw h runnonerrorapp self assertequalshortpat h stdout getvalue h testhandlerserverprotocolproto h originserver true h httpversion version h serversoftware ssw h runnonerrorapp if protohttp0 9 self assertequalh stdout getvalue b else self asserttrue re matchstdpatversion sw encodeiso88591 h stdout getvalue stdpatversion sw encodeiso88591 h stdout getvalue def testbytesdataself def appe s s200 ok contenttype textplain charsetutf8 return bdata h testhandler h runapp self assertequalbstatus 200 okrn bcontenttype textplain charsetutf8rn bcontentlength 4rn brn bdata h stdout getvalue def testcloseonerrorself sideeffects closecalled false msg bsome output has been sent def errorappe s s200 ok msg class crashyiterableobject def iterself while true yield b blah raise assertionerrorthis should be caught by handler def closeself sideeffects closecalled true return crashyiterable h errorhandler h runerrorapp self assertequalsideeffects closecalled true def testpartialwriteself written bytearray class partialwriter def writeself b partial b 7 written extendpartial return lenpartial def flushself pass environ serverprotocol http1 0 h simplehandlerbytesio partialwriter sys stderr environ msg should not do partial writes with self assertwarnsregexdeprecationwarning msg h runhelloapp self assertequalbhttp1 0 200 okrn bcontenttype textplainrn bdate mon 05 jun 2006 18 49 54 gmtrn bcontentlength 13rn brn bhello world written def testclientconnectionterminationsself environ serverprotocol http1 0 for exception in connectionabortederror brokenpipeerror connectionreseterror with self subtestexceptionexception class abortingwriter def writeself b raise exception stderr stringio h simplehandlerbytesio abortingwriter stderr environ h runhelloapp self assertfalsestderr getvalue def testdontresetinternalstateonexceptionself class customexceptionvalueerror pass we are raising customexception here to trigger an exception during the execution of simplehandler finishresponse so we can easily test that the internal state of the handler is preserved in case of an exception class abortingwriter def writeself b raise customexception stderr stringio environ serverprotocol http1 0 h simplehandlerbytesio abortingwriter stderr environ h runhelloapp self assertincustomexception stderr getvalue test that the internal state of the handler is preserved self assertisnotnoneh result self assertisnotnoneh headers self assertisnotnoneh status self assertisnotnoneh environ if name main unittest main non socket http server non socket http handler utility to compare a generic iterator with an iterable this tests the iterator using iter next make_it must be a function returning a fresh iterator to be tested since this may test the iterator twice the middleware doesn t support len so content length isn t there pep3333 says environ variables are decoded as latin1 encode as latin1 to get original bytes basehandler _write and _flush have to write all data even if it takes multiple send calls test this by interrupting a send call with a unix signal the main thread should now be blocking in a send system call but in theory it could get interrupted by other signals and then retried so keep sending the signal in a loop in case an earlier signal happens to be delivered at an inconvenient moment check defaulting when empty check existing value not comprehensive just a few random header names must be copy should not raise an error simple handler subclass for testing basehandler basehandler records the os environment at import time but envvars might have been changed later by other tests which trips up handlertests testenviron simple handler subclass for testing basehandler w error passthru for testing we want to see what s happening testenviron can produce long error message very basic environment set some wsgi variables override x and home variables check that wsgi_xxx attributes are copied to wsgi xxx variables of handler environ test handler environ as a dict handler inherits os_environ variables which are not overridden by simplehandler add_cgi_vars simplehandler base_env x doesn t exist in os_environ home is overridden by testhandler overridden by setup_testing_defaults set by basehandler setup_environ demo one reason iteration is better than write simulate a response to a head request we are raising customexception here to trigger an exception during the execution of simplehandler finish_response so we can easily test that the internal state of the handler is preserved in case of an exception test that the internal state of the handler is preserved
from unittest import mock
from test import support
from test.support import socket_helper
from test.test_httpservers import NoLogRequestHandler
from unittest import TestCase
from wsgiref.util import setup_testing_defaults
from wsgiref.headers import Headers
from wsgiref.handlers import BaseHandler, BaseCGIHandler, SimpleHandler
from wsgiref import util
from wsgiref.validate import validator
from wsgiref.simple_server import WSGIServer, WSGIRequestHandler
from wsgiref.simple_server import make_server
from http.client import HTTPConnection
from io import StringIO, BytesIO, BufferedReader
from socketserver import BaseServer
from platform import python_implementation

import os
import re
import signal
import sys
import threading
import unittest


class MockServer(WSGIServer):
    

    def __init__(self, server_address, RequestHandlerClass):
        BaseServer.__init__(self, server_address, RequestHandlerClass)
        self.server_bind()

    def server_bind(self):
        host, port = self.server_address
        self.server_name = host
        self.server_port = port
        self.setup_environ()


class MockHandler(WSGIRequestHandler):
    
    def setup(self):
        self.connection = self.request
        self.rfile, self.wfile = self.connection

    def finish(self):
        pass


def hello_app(environ,start_response):
    start_response("200 OK", [
        ('Content-Type','text/plain'),
        ('Date','Mon, 05 Jun 2006 18:49:54 GMT')
    ])
    return [b"Hello, world!"]


def header_app(environ, start_response):
    start_response("200 OK", [
        ('Content-Type', 'text/plain'),
        ('Date', 'Mon, 05 Jun 2006 18:49:54 GMT')
    ])
    return [';'.join([
        environ['HTTP_X_TEST_HEADER'], environ['QUERY_STRING'],
        environ['PATH_INFO']
    ]).encode('iso-8859-1')]


def run_amock(app=hello_app, data=b"GET / HTTP/1.0\n\n"):
    server = make_server("", 80, app, MockServer, MockHandler)
    inp = BufferedReader(BytesIO(data))
    out = BytesIO()
    olderr = sys.stderr
    err = sys.stderr = StringIO()

    try:
        server.finish_request((inp, out), ("127.0.0.1",8888))
    finally:
        sys.stderr = olderr

    return out.getvalue(), err.getvalue()


def compare_generic_iter(make_it, match):
    

    it = make_it()
    if not iter(it) is it:
        raise AssertionError
    for item in match:
        if not next(it) == item:
            raise AssertionError
    try:
        next(it)
    except StopIteration:
        pass
    else:
        raise AssertionError("Too many items from .__next__()", it)


class IntegrationTests(TestCase):

    def check_hello(self, out, has_length=True):
        pyver = (python_implementation() + "/" +
                sys.version.split()[0])
        self.assertEqual(out,
            ("HTTP/1.0 200 OK\r\n"
            "Server: WSGIServer/0.2 " + pyver +"\r\n"
            "Content-Type: text/plain\r\n"
            "Date: Mon, 05 Jun 2006 18:49:54 GMT\r\n" +
            (has_length and  "Content-Length: 13\r\n" or "") +
            "\r\n"
            "Hello, world!").encode("iso-8859-1")
        )

    def test_plain_hello(self):
        out, err = run_amock()
        self.check_hello(out)

    def test_environ(self):
        request = (
            b"GET /p%61th/?query=test HTTP/1.0\n"
            b"X-Test-Header: Python test \n"
            b"X-Test-Header: Python test 2\n"
            b"Content-Length: 0\n\n"
        )
        out, err = run_amock(header_app, request)
        self.assertEqual(
            out.splitlines()[-1],
            b"Python test,Python test 2;query=test;/path/"
        )

    def test_request_length(self):
        out, err = run_amock(data=b"GET " + (b"x" * 65537) + b" HTTP/1.0\n\n")
        self.assertEqual(out.splitlines()[0],
                         b"HTTP/1.0 414 Request-URI Too Long")

    def test_validated_hello(self):
        out, err = run_amock(validator(hello_app))
        
        self.check_hello(out, has_length=False)

    def test_simple_validation_error(self):
        def bad_app(environ,start_response):
            start_response("200 OK", ('Content-Type','text/plain'))
            return ["Hello, world!"]
        out, err = run_amock(validator(bad_app))
        self.assertTrue(out.endswith(
            b"A server error occurred.  Please contact the administrator."
        ))
        self.assertEqual(
            err.splitlines()[-2],
            "AssertionError: Headers (('Content-Type', 'text/plain')) must"
            " be of type list: <class 'tuple'>"
        )

    def test_status_validation_errors(self):
        def create_bad_app(status):
            def bad_app(environ, start_response):
                start_response(status, [("Content-Type", "text/plain; charset=utf-8")])
                return [b"Hello, world!"]
            return bad_app

        tests = [
            ('200', 'AssertionError: Status must be at least 4 characters'),
            ('20X OK', 'AssertionError: Status message must begin w/3-digit code'),
            ('200OK', 'AssertionError: Status message must have a space after code'),
        ]

        for status, exc_message in tests:
            with self.subTest(status=status):
                out, err = run_amock(create_bad_app(status))
                self.assertTrue(out.endswith(
                    b"A server error occurred.  Please contact the administrator."
                ))
                self.assertEqual(err.splitlines()[-2], exc_message)

    def test_wsgi_input(self):
        def bad_app(e,s):
            e["wsgi.input"].read()
            s("200 OK", [("Content-Type", "text/plain; charset=utf-8")])
            return [b"data"]
        out, err = run_amock(validator(bad_app))
        self.assertTrue(out.endswith(
            b"A server error occurred.  Please contact the administrator."
        ))
        self.assertEqual(
            err.splitlines()[-2], "AssertionError"
        )

    def test_bytes_validation(self):
        def app(e, s):
            s("200 OK", [
                ("Content-Type", "text/plain; charset=utf-8"),
                ("Date", "Wed, 24 Dec 2008 13:29:32 GMT"),
                ])
            return [b"data"]
        out, err = run_amock(validator(app))
        self.assertTrue(err.endswith('"GET / HTTP/1.0" 200 4\n'))
        ver = sys.version.split()[0].encode('ascii')
        py  = python_implementation().encode('ascii')
        pyver = py + b"/" + ver
        self.assertEqual(
                b"HTTP/1.0 200 OK\r\n"
                b"Server: WSGIServer/0.2 "+ pyver + b"\r\n"
                b"Content-Type: text/plain; charset=utf-8\r\n"
                b"Date: Wed, 24 Dec 2008 13:29:32 GMT\r\n"
                b"\r\n"
                b"data",
                out)

    def test_cp1252_url(self):
        def app(e, s):
            s("200 OK", [
                ("Content-Type", "text/plain"),
                ("Date", "Wed, 24 Dec 2008 13:29:32 GMT"),
                ])
            
            
            return [e["PATH_INFO"].encode("latin1")]

        out, err = run_amock(
            validator(app), data=b"GET /\x80%80 HTTP/1.0")
        self.assertEqual(
            [
                b"HTTP/1.0 200 OK",
                mock.ANY,
                b"Content-Type: text/plain",
                b"Date: Wed, 24 Dec 2008 13:29:32 GMT",
                b"",
                b"/\x80\x80",
            ],
            out.splitlines())

    def test_interrupted_write(self):
        
        
        
        pthread_kill = support.get_attribute(signal, "pthread_kill")

        def app(environ, start_response):
            start_response("200 OK", [])
            return [b'\0' * support.SOCK_MAX_SIZE]

        class WsgiHandler(NoLogRequestHandler, WSGIRequestHandler):
            pass

        server = make_server(socket_helper.HOST, 0, app, handler_class=WsgiHandler)
        self.addCleanup(server.server_close)
        interrupted = threading.Event()

        def signal_handler(signum, frame):
            interrupted.set()

        original = signal.signal(signal.SIGUSR1, signal_handler)
        self.addCleanup(signal.signal, signal.SIGUSR1, original)
        received = None
        main_thread = threading.get_ident()

        def run_client():
            http = HTTPConnection(*server.server_address)
            http.request("GET", "/")
            with http.getresponse() as response:
                response.read(100)
                
                
                
                
                
                while True:
                    pthread_kill(main_thread, signal.SIGUSR1)
                    if interrupted.wait(timeout=float(1)):
                        break
                nonlocal received
                received = len(response.read())
            http.close()

        background = threading.Thread(target=run_client)
        background.start()
        server.handle_request()
        background.join()
        self.assertEqual(received, support.SOCK_MAX_SIZE - 100)


class UtilityTests(TestCase):

    def checkShift(self,sn_in,pi_in,part,sn_out,pi_out):
        env = {'SCRIPT_NAME':sn_in,'PATH_INFO':pi_in}
        util.setup_testing_defaults(env)
        self.assertEqual(util.shift_path_info(env),part)
        self.assertEqual(env['PATH_INFO'],pi_out)
        self.assertEqual(env['SCRIPT_NAME'],sn_out)
        return env

    def checkDefault(self, key, value, alt=None):
        
        env = {}
        util.setup_testing_defaults(env)
        if isinstance(value, StringIO):
            self.assertIsInstance(env[key], StringIO)
        elif isinstance(value,BytesIO):
            self.assertIsInstance(env[key],BytesIO)
        else:
            self.assertEqual(env[key], value)

        
        env = {key:alt}
        util.setup_testing_defaults(env)
        self.assertIs(env[key], alt)

    def checkCrossDefault(self,key,value,**kw):
        util.setup_testing_defaults(kw)
        self.assertEqual(kw[key],value)

    def checkAppURI(self,uri,**kw):
        util.setup_testing_defaults(kw)
        self.assertEqual(util.application_uri(kw),uri)

    def checkReqURI(self,uri,query=1,**kw):
        util.setup_testing_defaults(kw)
        self.assertEqual(util.request_uri(kw,query),uri)

    def checkFW(self,text,size,match):

        def make_it(text=text,size=size):
            return util.FileWrapper(StringIO(text),size)

        compare_generic_iter(make_it,match)

        it = make_it()
        self.assertFalse(it.filelike.closed)

        for item in it:
            pass

        self.assertFalse(it.filelike.closed)

        it.close()
        self.assertTrue(it.filelike.closed)

    def testSimpleShifts(self):
        self.checkShift('','/', '', '/', '')
        self.checkShift('','/x', 'x', '/x', '')
        self.checkShift('/','', None, '/', '')
        self.checkShift('/a','/x/y', 'x', '/a/x', '/y')
        self.checkShift('/a','/x/',  'x', '/a/x', '/')

    def testNormalizedShifts(self):
        self.checkShift('/a/b', '/../y', '..', '/a', '/y')
        self.checkShift('', '/../y', '..', '', '/y')
        self.checkShift('/a/b', '//y', 'y', '/a/b/y', '')
        self.checkShift('/a/b', '//y/', 'y', '/a/b/y', '/')
        self.checkShift('/a/b', '/./y', 'y', '/a/b/y', '')
        self.checkShift('/a/b', '/./y/', 'y', '/a/b/y', '/')
        self.checkShift('/a/b', '///./..//y/.//', '..', '/a', '/y/')
        self.checkShift('/a/b', '///', '', '/a/b/', '')
        self.checkShift('/a/b', '/.//', '', '/a/b/', '')
        self.checkShift('/a/b', '/x//', 'x', '/a/b/x', '/')
        self.checkShift('/a/b', '/.', None, '/a/b', '')

    def testDefaults(self):
        for key, value in [
            ('SERVER_NAME','127.0.0.1'),
            ('SERVER_PORT', '80'),
            ('SERVER_PROTOCOL','HTTP/1.0'),
            ('HTTP_HOST','127.0.0.1'),
            ('REQUEST_METHOD','GET'),
            ('SCRIPT_NAME',''),
            ('PATH_INFO','/'),
            ('wsgi.version', (1,0)),
            ('wsgi.run_once', 0),
            ('wsgi.multithread', 0),
            ('wsgi.multiprocess', 0),
            ('wsgi.input', BytesIO()),
            ('wsgi.errors', StringIO()),
            ('wsgi.url_scheme','http'),
        ]:
            self.checkDefault(key,value)

    def testCrossDefaults(self):
        self.checkCrossDefault('HTTP_HOST',"foo.bar",SERVER_NAME="foo.bar")
        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="on")
        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="1")
        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="yes")
        self.checkCrossDefault('wsgi.url_scheme',"http",HTTPS="foo")
        self.checkCrossDefault('SERVER_PORT',"80",HTTPS="foo")
        self.checkCrossDefault('SERVER_PORT',"443",HTTPS="on")

    def testGuessScheme(self):
        self.assertEqual(util.guess_scheme({}), "http")
        self.assertEqual(util.guess_scheme({'HTTPS':"foo"}), "http")
        self.assertEqual(util.guess_scheme({'HTTPS':"on"}), "https")
        self.assertEqual(util.guess_scheme({'HTTPS':"yes"}), "https")
        self.assertEqual(util.guess_scheme({'HTTPS':"1"}), "https")

    def testAppURIs(self):
        self.checkAppURI("http://127.0.0.1/")
        self.checkAppURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam")
        self.checkAppURI("http://127.0.0.1/sp%E4m", SCRIPT_NAME="/sp\xe4m")
        self.checkAppURI("http://spam.example.com:2071/",
            HTTP_HOST="spam.example.com:2071", SERVER_PORT="2071")
        self.checkAppURI("http://spam.example.com/",
            SERVER_NAME="spam.example.com")
        self.checkAppURI("http://127.0.0.1/",
            HTTP_HOST="127.0.0.1", SERVER_NAME="spam.example.com")
        self.checkAppURI("https://127.0.0.1/", HTTPS="on")
        self.checkAppURI("http://127.0.0.1:8000/", SERVER_PORT="8000",
            HTTP_HOST=None)

    def testReqURIs(self):
        self.checkReqURI("http://127.0.0.1/")
        self.checkReqURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam")
        self.checkReqURI("http://127.0.0.1/sp%E4m", SCRIPT_NAME="/sp\xe4m")
        self.checkReqURI("http://127.0.0.1/spammity/spam",
            SCRIPT_NAME="/spammity", PATH_INFO="/spam")
        self.checkReqURI("http://127.0.0.1/spammity/sp%E4m",
            SCRIPT_NAME="/spammity", PATH_INFO="/sp\xe4m")
        self.checkReqURI("http://127.0.0.1/spammity/spam;ham",
            SCRIPT_NAME="/spammity", PATH_INFO="/spam;ham")
        self.checkReqURI("http://127.0.0.1/spammity/spam;cookie=1234,5678",
            SCRIPT_NAME="/spammity", PATH_INFO="/spam;cookie=1234,5678")
        self.checkReqURI("http://127.0.0.1/spammity/spam?say=ni",
            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni")
        self.checkReqURI("http://127.0.0.1/spammity/spam?s%E4y=ni",
            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="s%E4y=ni")
        self.checkReqURI("http://127.0.0.1/spammity/spam", 0,
            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni")

    def testFileWrapper(self):
        self.checkFW("xyz"*50, 120, ["xyz"*40,"xyz"*10])

    def testHopByHop(self):
        for hop in (
            "Connection Keep-Alive Proxy-Authenticate Proxy-Authorization "
            "TE Trailers Transfer-Encoding Upgrade"
        ).split():
            for alt in hop, hop.title(), hop.upper(), hop.lower():
                self.assertTrue(util.is_hop_by_hop(alt))

        
        for hop in (
            "Accept Cache-Control Date Pragma Trailer Via Warning"
        ).split():
            for alt in hop, hop.title(), hop.upper(), hop.lower():
                self.assertFalse(util.is_hop_by_hop(alt))

class HeaderTests(TestCase):

    def testMappingInterface(self):
        test = [('x','y')]
        self.assertEqual(len(Headers()), 0)
        self.assertEqual(len(Headers([])),0)
        self.assertEqual(len(Headers(test[:])),1)
        self.assertEqual(Headers(test[:]).keys(), ['x'])
        self.assertEqual(Headers(test[:]).values(), ['y'])
        self.assertEqual(Headers(test[:]).items(), test)
        self.assertIsNot(Headers(test).items(), test)  

        h = Headers()
        del h['foo']   

        h['Foo'] = 'bar'
        for m in h.__contains__, h.get, h.get_all, h.__getitem__:
            self.assertTrue(m('foo'))
            self.assertTrue(m('Foo'))
            self.assertTrue(m('FOO'))
            self.assertFalse(m('bar'))

        self.assertEqual(h['foo'],'bar')
        h['foo'] = 'baz'
        self.assertEqual(h['FOO'],'baz')
        self.assertEqual(h.get_all('foo'),['baz'])

        self.assertEqual(h.get("foo","whee"), "baz")
        self.assertEqual(h.get("zoo","whee"), "whee")
        self.assertEqual(h.setdefault("foo","whee"), "baz")
        self.assertEqual(h.setdefault("zoo","whee"), "whee")
        self.assertEqual(h["foo"],"baz")
        self.assertEqual(h["zoo"],"whee")

    def testRequireList(self):
        self.assertRaises(TypeError, Headers, "foo")

    def testExtras(self):
        h = Headers()
        self.assertEqual(str(h),'\r\n')

        h.add_header('foo','bar',baz="spam")
        self.assertEqual(h['foo'], 'bar; baz="spam"')
        self.assertEqual(str(h),'foo: bar; baz="spam"\r\n\r\n')

        h.add_header('Foo','bar',cheese=None)
        self.assertEqual(h.get_all('foo'),
            ['bar; baz="spam"', 'bar; cheese'])

        self.assertEqual(str(h),
            'foo: bar; baz="spam"\r\n'
            'Foo: bar; cheese\r\n'
            '\r\n'
        )

class ErrorHandler(BaseCGIHandler):
    

    
    
    
    os_environ = dict(os.environ.items())

    def __init__(self,**kw):
        setup_testing_defaults(kw)
        BaseCGIHandler.__init__(
            self, BytesIO(), BytesIO(), StringIO(), kw,
            multithread=True, multiprocess=True
        )

class TestHandler(ErrorHandler):
    

    def handle_error(self):
        raise   


class HandlerTests(TestCase):
    
    maxDiff = 80 * 50

    def testEnviron(self):
        os_environ = {
            
            'HOME': '/my/home',
            'PATH': '/my/path',
            'LANG': 'fr_FR.UTF-8',

            
            'SCRIPT_NAME': 'test_script_name',
            'SERVER_NAME': 'test_server_name',
        }

        with support.swap_attr(TestHandler, 'os_environ', os_environ):
            
            handler = TestHandler(X="Y", HOME="/override/home")
            handler.setup_environ()

        
        
        for attr in ('version', 'multithread', 'multiprocess', 'run_once',
                     'file_wrapper'):
            self.assertEqual(getattr(handler, 'wsgi_' + attr),
                             handler.environ['wsgi.' + attr])

        
        expected = {}
        setup_testing_defaults(expected)
        
        
        for key, value in os_environ.items():
            if key not in expected:
                expected[key] = value
        expected.update({
            
            "X": "Y",
            
            'HOME': "/override/home",

            
            "SCRIPT_NAME": "",
            "SERVER_NAME": "127.0.0.1",

            
            'wsgi.input': handler.get_stdin(),
            'wsgi.errors': handler.get_stderr(),
            'wsgi.version': (1, 0),
            'wsgi.run_once': False,
            'wsgi.url_scheme': 'http',
            'wsgi.multithread': True,
            'wsgi.multiprocess': True,
            'wsgi.file_wrapper': util.FileWrapper,
        })
        self.assertDictEqual(handler.environ, expected)

    def testCGIEnviron(self):
        h = BaseCGIHandler(None,None,None,{})
        h.setup_environ()
        for key in 'wsgi.url_scheme', 'wsgi.input', 'wsgi.errors':
            self.assertIn(key, h.environ)

    def testScheme(self):
        h=TestHandler(HTTPS="on"); h.setup_environ()
        self.assertEqual(h.environ['wsgi.url_scheme'],'https')
        h=TestHandler(); h.setup_environ()
        self.assertEqual(h.environ['wsgi.url_scheme'],'http')

    def testAbstractMethods(self):
        h = BaseHandler()
        for name in [
            '_flush','get_stdin','get_stderr','add_cgi_vars'
        ]:
            self.assertRaises(NotImplementedError, getattr(h,name))
        self.assertRaises(NotImplementedError, h._write, "test")

    def testContentLength(self):
        

        def trivial_app1(e,s):
            s('200 OK',[])
            return [e['wsgi.url_scheme'].encode('iso-8859-1')]

        def trivial_app2(e,s):
            s('200 OK',[])(e['wsgi.url_scheme'].encode('iso-8859-1'))
            return []

        def trivial_app3(e,s):
            s('200 OK',[])
            return ['\u0442\u0435\u0441\u0442'.encode("utf-8")]

        def trivial_app4(e,s):
            
            s('200 OK',[('Content-Length', '12345')])
            return []

        h = TestHandler()
        h.run(trivial_app1)
        self.assertEqual(h.stdout.getvalue(),
            ("Status: 200 OK\r\n"
            "Content-Length: 4\r\n"
            "\r\n"
            "http").encode("iso-8859-1"))

        h = TestHandler()
        h.run(trivial_app2)
        self.assertEqual(h.stdout.getvalue(),
            ("Status: 200 OK\r\n"
            "\r\n"
            "http").encode("iso-8859-1"))

        h = TestHandler()
        h.run(trivial_app3)
        self.assertEqual(h.stdout.getvalue(),
            b'Status: 200 OK\r\n'
            b'Content-Length: 8\r\n'
            b'\r\n'
            b'\xd1\x82\xd0\xb5\xd1\x81\xd1\x82')

        h = TestHandler()
        h.run(trivial_app4)
        self.assertEqual(h.stdout.getvalue(),
            b'Status: 200 OK\r\n'
            b'Content-Length: 12345\r\n'
            b'\r\n')

    def testBasicErrorOutput(self):

        def non_error_app(e,s):
            s('200 OK',[])
            return []

        def error_app(e,s):
            raise AssertionError("This should be caught by handler")

        h = ErrorHandler()
        h.run(non_error_app)
        self.assertEqual(h.stdout.getvalue(),
            ("Status: 200 OK\r\n"
            "Content-Length: 0\r\n"
            "\r\n").encode("iso-8859-1"))
        self.assertEqual(h.stderr.getvalue(),"")

        h = ErrorHandler()
        h.run(error_app)
        self.assertEqual(h.stdout.getvalue(),
            ("Status: %s\r\n"
            "Content-Type: text/plain\r\n"
            "Content-Length: %d\r\n"
            "\r\n" % (h.error_status,len(h.error_body))).encode('iso-8859-1')
            + h.error_body)

        self.assertIn("AssertionError", h.stderr.getvalue())

    def testErrorAfterOutput(self):
        MSG = b"Some output has been sent"
        def error_app(e,s):
            s("200 OK",[])(MSG)
            raise AssertionError("This should be caught by handler")

        h = ErrorHandler()
        h.run(error_app)
        self.assertEqual(h.stdout.getvalue(),
            ("Status: 200 OK\r\n"
            "\r\n".encode("iso-8859-1")+MSG))
        self.assertIn("AssertionError", h.stderr.getvalue())

    def testHeaderFormats(self):

        def non_error_app(e,s):
            s('200 OK',[])
            return []

        stdpat = (
            r"HTTP/%s 200 OK\r\n"
            r"Date: \w{3}, [ 0123]\d \w{3} \d{4} \d\d:\d\d:\d\d GMT\r\n"
            r"%s" r"Content-Length: 0\r\n" r"\r\n"
        )
        shortpat = (
            "Status: 200 OK\r\n" "Content-Length: 0\r\n" "\r\n"
        ).encode("iso-8859-1")

        for ssw in "FooBar/1.0", None:
            sw = ssw and "Server: %s\r\n" % ssw or ""

            for version in "1.0", "1.1":
                for proto in "HTTP/0.9", "HTTP/1.0", "HTTP/1.1":

                    h = TestHandler(SERVER_PROTOCOL=proto)
                    h.origin_server = False
                    h.http_version = version
                    h.server_software = ssw
                    h.run(non_error_app)
                    self.assertEqual(shortpat,h.stdout.getvalue())

                    h = TestHandler(SERVER_PROTOCOL=proto)
                    h.origin_server = True
                    h.http_version = version
                    h.server_software = ssw
                    h.run(non_error_app)
                    if proto=="HTTP/0.9":
                        self.assertEqual(h.stdout.getvalue(),b"")
                    else:
                        self.assertTrue(
                            re.match((stdpat%(version,sw)).encode("iso-8859-1"),
                                h.stdout.getvalue()),
                            ((stdpat%(version,sw)).encode("iso-8859-1"),
                                h.stdout.getvalue())
                        )

    def testBytesData(self):
        def app(e, s):
            s("200 OK", [
                ("Content-Type", "text/plain; charset=utf-8"),
                ])
            return [b"data"]

        h = TestHandler()
        h.run(app)
        self.assertEqual(b"Status: 200 OK\r\n"
            b"Content-Type: text/plain; charset=utf-8\r\n"
            b"Content-Length: 4\r\n"
            b"\r\n"
            b"data",
            h.stdout.getvalue())

    def testCloseOnError(self):
        side_effects = {'close_called': False}
        MSG = b"Some output has been sent"
        def error_app(e,s):
            s("200 OK",[])(MSG)
            class CrashyIterable(object):
                def __iter__(self):
                    while True:
                        yield b'blah'
                        raise AssertionError("This should be caught by handler")
                def close(self):
                    side_effects['close_called'] = True
            return CrashyIterable()

        h = ErrorHandler()
        h.run(error_app)
        self.assertEqual(side_effects['close_called'], True)

    def testPartialWrite(self):
        written = bytearray()

        class PartialWriter:
            def write(self, b):
                partial = b[:7]
                written.extend(partial)
                return len(partial)

            def flush(self):
                pass

        environ = {"SERVER_PROTOCOL": "HTTP/1.0"}
        h = SimpleHandler(BytesIO(), PartialWriter(), sys.stderr, environ)
        msg = "should not do partial writes"
        with self.assertWarnsRegex(DeprecationWarning, msg):
            h.run(hello_app)
        self.assertEqual(b"HTTP/1.0 200 OK\r\n"
            b"Content-Type: text/plain\r\n"
            b"Date: Mon, 05 Jun 2006 18:49:54 GMT\r\n"
            b"Content-Length: 13\r\n"
            b"\r\n"
            b"Hello, world!",
            written)

    def testClientConnectionTerminations(self):
        environ = {"SERVER_PROTOCOL": "HTTP/1.0"}
        for exception in (
            ConnectionAbortedError,
            BrokenPipeError,
            ConnectionResetError,
        ):
            with self.subTest(exception=exception):
                class AbortingWriter:
                    def write(self, b):
                        raise exception

                stderr = StringIO()
                h = SimpleHandler(BytesIO(), AbortingWriter(), stderr, environ)
                h.run(hello_app)

                self.assertFalse(stderr.getvalue())

    def testDontResetInternalStateOnException(self):
        class CustomException(ValueError):
            pass

        
        
        
        
        class AbortingWriter:
            def write(self, b):
                raise CustomException

        stderr = StringIO()
        environ = {"SERVER_PROTOCOL": "HTTP/1.0"}
        h = SimpleHandler(BytesIO(), AbortingWriter(), stderr, environ)
        h.run(hello_app)

        self.assertIn("CustomException", stderr.getvalue())

        
        self.assertIsNotNone(h.result)
        self.assertIsNotNone(h.headers)
        self.assertIsNotNone(h.status)
        self.assertIsNotNone(h.environ)


if __name__ == "__main__":
    unittest.main()