selfsigned cert file for localhost selfsigned cert file for fakehostname loopback http server infrastructure http server w a few modifications that make it useful for loopback testing purposes set the timeout of our listening socket really low so that we can stop the server easily httpserver method overridden request clientaddress self socket accept it s a loopback connection so setting the timeout really low shouldn t affect anything but should make deadlocks less likely to occur request settimeout10 0 return request clientaddress class loopbackhttpserverthreadthreading thread stops the webserver if it s currently running self stopserver true self join self httpd serverclose def runself self ready set while not self stopserver self httpd handlerequest authentication infrastructure class digestauthhandler xxx not sure if we re supposed to add this next header or not requesthandler sendheader connection close performs digest authentication on the given http request handler returns true if authentication was successful false otherwise if no users have been set then digest auth is effectively disabled and this method will always return true msie uses shortpath in its validation but python s urllib request uses the full path so we re going to see if either of them works here handler for performing basic authentication server side values user testuser passwd testpass realm test userpasswd s s user passwd encodedauth base64 b64encodeuserpasswd encode ascii decode ascii def initself args kwargs http server basehttprequesthandler initself args kwargs def logmessageself format args suppress console log message pass def doheadself self sendresponse200 self sendheadercontenttype texthtml self endheaders def doauthheadself self sendresponse401 self sendheaderwwwauthenticate basic realms self realm self sendheadercontenttype texthtml self endheaders def dogetself if not self headers getization self doauthhead self wfile writebno auth header received elif self headers get ization basic self encodedauth self sendresponse200 self endheaders self wfile writebit works else request unized self doauthhead proxy test infrastructure class fakeproxyhandlerhttp server basehttprequesthandler def initself digestauthhandler args kwargs this has to be set before calling our parent s init which will try to call doget self digestauthhandler digestauthhandler http server basehttprequesthandler initself args kwargs def logmessageself format args uncomment the next line for debugging sys stderr writeformat args pass def dogetself scm netloc path params query fragment urllib parse urlparse self path http self shortpath path if self digestauthhandler handlerequestself self sendresponse200 ok self sendheadercontenttype texthtml self endheaders self wfile writebytesyou ve reached s br self path ascii self wfile writebour apologies but our server is down due to ba sudden zombie invasion test cases class basicauthtestsunittest testcase user testuser passwd testpass incorrectpasswd incorrect realm test def setupself superbasicauthtests self setup with basic authentication def httpserverwithbasicauthhandlerargs kwargs return basicauthhandlerargs kwargs self server loopbackhttpserverthreadhttpserverwithbasicauthhandler self addcleanupself stopserver self serverurl http 127 0 0 1 s self server port self server start self server ready wait def stopserverself self server stop self server none def teardownself superbasicauthtests self teardown def testbasicauthsuccessself ah urllib request httpbasicauthhandler ah addpasswordself realm self serverurl self user self passwd urllib request installopenerurllib request buildopenerah try self asserttrueurllib request urlopenself serverurl except urllib error httperror self failbasic auth failed for the url s self serverurl def testbasicauthhttperrorself ah urllib request httpbasicauthhandler ah addpasswordself realm self serverurl self user self incorrectpasswd urllib request installopenerurllib request buildopenerah self assertraisesurllib error httperror urllib request urlopen self serverurl hashlibhelper requireshashdigestmd5 openssltrue class proxyauthtestsunittest testcase url http localhost user tester passwd test123 realm testrealm def setupself superproxyauthtests self setup ignore proxy bypass settings in the environment def restoreenvironoldenviron os environ clear os environ updateoldenviron self addcleanuprestoreenviron os environ copy os environ noproxy os environ noproxy self digestauthhandler digestauthhandler self digestauthhandler setusersself user self passwd self digestauthhandler setrealmself realm with digest authentication def createfakeproxyhandlerargs kwargs return fakeproxyhandlerself digestauthhandler args kwargs self server loopbackhttpserverthreadcreatefakeproxyhandler self addcleanupself stopserver self server start self server ready wait proxyurl http 127 0 0 1 d self server port handler urllib request proxyhandlerhttp proxyurl self proxydigesthandler urllib request proxydigestauthhandler self opener urllib request buildopener handler self proxydigesthandler def stopserverself self server stop self server none def testproxywithbadpasswordraiseshttperrorself self proxydigesthandler addpasswordself realm self url self user self passwdbad self digestauthhandler setqopauth self assertraisesurllib error httperror self opener open self url def testproxywithnopasswordraiseshttperrorself self digestauthhandler setqopauth self assertraisesurllib error httperror self opener open self url def testproxyqopauthworksself self proxydigesthandler addpasswordself realm self url self user self passwd self digestauthhandler setqopauth with self opener openself url as result while result read pass def testproxyqopauthintworksorthrowsurlerrorself self proxydigesthandler addpasswordself realm self url self user self passwd self digestauthhandler setqopauthint try result self opener openself url except urllib error urlerror it s okay if we don t support authint but we certainly shouldn t receive any kind of exception here other than a urlerror pass else with result while result read pass def getrequesthandlerresponses class fakehttprequesthandlerhttp server basehttprequesthandler serverversion testhttp requests headersreceived port 80 def dogetself body self sendhead while body done self wfile writebody body bodydone def dopostself contentlength self headerscontentlength postdata self rfile readintcontentlength self doget self requests appendpostdata def sendheadself fakehttprequesthandler headersreceived self headers self requests appendself path responsecode headers body responses pop0 self sendresponseresponsecode for header value in headers self sendheaderheader value port self port if body self sendheadercontenttype textplain self endheaders return body self endheaders def logmessageself args pass return fakehttprequesthandler class testurlopenunittest testcase def setupself supertesturlopen self setup clear opener global variable self addcleanupurllib request urlcleanup ignore proxies for localhost tests def restoreenvironoldenviron os environ clear os environ updateoldenviron self addcleanuprestoreenviron os environ copy os environ noproxy os environ noproxy def urlopenself url datanone kwargs l f urllib request urlopenurl data kwargs try exercise various methods l extendf readlines200 l appendf readline l appendf read1024 l appendf read finally f close return b joinl def stopserverself self server stop self server none def startserverself responsesnone if responses is none responses 200 bwe don t care handler getrequesthandlerresponses self server loopbackhttpserverthreadhandler self addcleanupself stopserver self server start self server ready wait port self server port handler port port return handler def starthttpsserverself responsesnone kwargs if not hasattrurllib request httpshandler self skiptest ssl support required from test sslservers import makehttpsserver if responses is none responses 200 bwe care a bit handler getrequesthandlerresponses server makehttpsserverself handlerclasshandler kwargs handler port server port return handler def testredirectionself expectedresponse bwe got here responses 302 location http localhost portssomewhereelse 200 expectedresponse handler self startserverresponses data self urlopenhttp localhost s handler port self assertequaldata expectedresponse self assertequalhandler requests somewhereelse def testchunkedself expectedresponse bhello world chunkedstart b arn b hello worlrn b 1rn b drn b 0rn response 200 transferencoding chunked chunkedstart handler self startserverresponse data self urlopenhttp localhost s handler port self assertequaldata expectedresponse def test404self expectedresponse bbad bad bad handler self startserver404 expectedresponse try self urlopenhttp localhost sweeble handler port except urllib error urlerror as f data f read f close else self fail404 should raise urlerror self assertequaldata expectedresponse self assertequalhandler requests weeble def test200self expectedresponse bpycon 2008 handler self startserver200 expectedresponse data self urlopenhttp localhost sbizarre handler port self assertequaldata expectedresponse self assertequalhandler requests bizarre def test200withparametersself expectedresponse bpycon 2008 handler self startserver200 expectedresponse data self urlopenhttp localhost sbizarre handler port bgetwithfeeling self assertequaldata expectedresponse self assertequalhandler requests bizarre bgetwithfeeling def testhttpsself handler self starthttpsserver context ssl createdefaultcontextcafilecertlocalhost data self urlopenhttps localhost sbizarre handler port contextcontext self assertequaldata bwe care a bit def testhttpssniself if ssl is none self skiptestssl module required if not ssl hassni self skiptestsni support required in openssl sniname none def cbsnisslsock servername initialcontext nonlocal sniname sniname servername context ssl sslcontextssl protocoltlsserver context setservernamecallbackcbsni handler self starthttpsservercontextcontext certfilecertlocalhost context ssl createdefaultcontextcafilecertlocalhost self urlopenhttps localhost s handler port contextcontext self assertequalsniname localhost def testsendingheadersself handler self startserver req urllib request requesthttp localhost s handler port headersrange bytes2039 with urllib request urlopenreq pass self assertequalhandler headersreceivedrange bytes2039 def testsendingheaderscamelself handler self startserver req urllib request requesthttp localhost s handler port headersxsomeheader foobar with urllib request urlopenreq pass self assertinxsomeheader handler headersreceived keys self assertnotinxsomeheader handler headersreceived keys def testbasicself handler self startserver with urllib request urlopenhttp localhost s handler port as openurl for attr in read close info geturl self asserttruehasattropenurl attr object returned from urlopen lacks the s attribute attr self asserttrueopenurl read calling read failed def testinfoself handler self startserver openurl urllib request urlopen http localhost s handler port with openurl infoobj openurl info self assertisinstanceinfoobj email message message object returned by info is not an instance of email message message self assertequalinfoobj getcontentsubtype plain def testgeturlself make sure same url as opened is returned by geturl handler self startserver openurl urllib request urlopenhttp localhost s handler port with openurl url openurl geturl self assertequalurl http localhost s handler port def testiterationself expectedresponse bpycon 2008 handler self startserver200 expectedresponse data urllib request urlopenhttp localhost s handler port for line in data self assertequalline expectedresponse def testlineiterationself lines bwen bgotn bheren bverylong 8192 bn expectedresponse b joinlines handler self startserver200 expectedresponse data urllib request urlopenhttp localhost s handler port for index line in enumeratedata self assertequalline linesindex fetched line number s doesn t match expected n expected length was s got s index lenlinesindex lenline self assertequalindex 1 lenlines def testissue16464self see https bugs python orgissue16464 and https bugs python orgissue46648 handler self startserver 200 b any 200 b any opener urllib request buildopener request urllib request requesthttp localhost s handler port self assertequalnone request data opener openrequest 1 encodeusascii self assertequalb1 request data self assertequal1 request getheadercontentlength opener openrequest 1234567890 encodeusascii self assertequalb1234567890 request data self assertequal10 request getheadercontentlength def setupmodule threadinfo threadinghelper threadingsetup unittest addmodulecleanupthreadinghelper threadingcleanup threadinfo if name main unittest main self signed cert file for localhost self signed cert file for fakehostname loopback http server infrastructure http server w a few modifications that make it useful for loopback testing purposes set the timeout of our listening socket really low so that we can stop the server easily httpserver method overridden it s a loopback connection so setting the timeout really low shouldn t affect anything but should make deadlocks less likely to occur stoppable thread that runs a loopback http server stops the webserver if it s currently running authentication infrastructure handler for performing digest authentication xxx not sure if we re supposed to add this next header or not request_handler send_header connection close performs digest authentication on the given http request handler returns true if authentication was successful false otherwise if no users have been set then digest auth is effectively disabled and this method will always return true msie uses short_path in its validation but python s urllib request uses the full path so we re going to see if either of them works here handler for performing basic authentication server side values suppress console log message request unized proxy test infrastructure this is a fake proxy that makes it look like the entire internet has gone down due to a sudden zombie invasion it main utility is in providing us with authentication support for testing this has to be set before calling our parent s __init__ which will try to call do_get uncomment the next line for debugging sys stderr write format args test cases with basic authentication ignore proxy bypass settings in the environment with digest authentication it s okay if we don t support auth int but we certainly shouldn t receive any kind of exception here other than a urlerror tests urllib request urlopen using the network these tests are not exhaustive assuming that testing using files does a good job overall of some of the basic interface features there are no tests exercising the optional data and proxies arguments no tests for transparent redirection have been written clear _opener global variable ignore proxies for localhost tests exercise various methods make sure same url as opened is returned by geturl see https bugs python org issue16464 and https bugs python org issue46648
import base64
import os
import email
import urllib.parse
import urllib.request
import http.server
import threading
import unittest
import hashlib

from test import support
from test.support import hashlib_helper
from test.support import threading_helper

try:
    import ssl
except ImportError:
    ssl = None

support.requires_working_socket(module=True)

here = os.path.dirname(__file__)

CERT_localhost = os.path.join(here, 'certdata', 'keycert.pem')

CERT_fakehostname = os.path.join(here, 'certdata', 'keycert2.pem')




class LoopbackHttpServer(http.server.HTTPServer):
    

    def __init__(self, server_address, RequestHandlerClass):
        http.server.HTTPServer.__init__(self,
                                        server_address,
                                        RequestHandlerClass)

        
        
        self.socket.settimeout(0.1)

    def get_request(self):
        

        request, client_address = self.socket.accept()

        
        
        
        request.settimeout(10.0)

        return (request, client_address)

class LoopbackHttpServerThread(threading.Thread):
    

    def __init__(self, request_handler):
        threading.Thread.__init__(self)
        self._stop_server = False
        self.ready = threading.Event()
        request_handler.protocol_version = "HTTP/1.0"
        self.httpd = LoopbackHttpServer(("127.0.0.1", 0),
                                        request_handler)
        self.port = self.httpd.server_port

    def stop(self):
        

        self._stop_server = True

        self.join()
        self.httpd.server_close()

    def run(self):
        self.ready.set()
        while not self._stop_server:
            self.httpd.handle_request()



class DigestAuthHandler:
    

    def __init__(self):
        self._request_num = 0
        self._nonces = []
        self._users = {}
        self._realm_name = "Test Realm"
        self._qop = "auth"

    def set_qop(self, qop):
        self._qop = qop

    def set_users(self, users):
        assert isinstance(users, dict)
        self._users = users

    def set_realm(self, realm):
        self._realm_name = realm

    def _generate_nonce(self):
        self._request_num += 1
        nonce = hashlib.md5(str(self._request_num).encode("ascii")).hexdigest()
        self._nonces.append(nonce)
        return nonce

    def _create_auth_dict(self, auth_str):
        first_space_index = auth_str.find(" ")
        auth_str = auth_str[first_space_index+1:]

        parts = auth_str.split(",")

        auth_dict = {}
        for part in parts:
            name, value = part.split("=")
            name = name.strip()
            if value[0] == '"' and value[-1] == '"':
                value = value[1:-1]
            else:
                value = value.strip()
            auth_dict[name] = value
        return auth_dict

    def _validate_auth(self, auth_dict, password, method, uri):
        final_dict = {}
        final_dict.update(auth_dict)
        final_dict["password"] = password
        final_dict["method"] = method
        final_dict["uri"] = uri
        HA1_str = "%(username)s:%(realm)s:%(password)s" % final_dict
        HA1 = hashlib.md5(HA1_str.encode("ascii")).hexdigest()
        HA2_str = "%(method)s:%(uri)s" % final_dict
        HA2 = hashlib.md5(HA2_str.encode("ascii")).hexdigest()
        final_dict["HA1"] = HA1
        final_dict["HA2"] = HA2
        response_str = "%(HA1)s:%(nonce)s:%(nc)s:" \
                       "%(cnonce)s:%(qop)s:%(HA2)s" % final_dict
        response = hashlib.md5(response_str.encode("ascii")).hexdigest()

        return response == auth_dict["response"]

    def _return_auth_challenge(self, request_handler):
        request_handler.send_response(407, "Proxy Authentication Required")
        request_handler.send_header("Content-Type", "text/html")
        request_handler.send_header(
            'Proxy-Authenticate', 'Digest realm="%s", '
            'qop="%s",'
            'nonce="%s", ' % \
            (self._realm_name, self._qop, self._generate_nonce()))
        
        
        
        request_handler.end_headers()
        request_handler.wfile.write(b"Proxy Authentication Required.")
        return False

    def handle_request(self, request_handler):
        

        if len(self._users) == 0:
            return True

        if "Proxy-Authorization" not in request_handler.headers:
            return self._return_auth_challenge(request_handler)
        else:
            auth_dict = self._create_auth_dict(
                request_handler.headers["Proxy-Authorization"]
                )
            if auth_dict["username"] in self._users:
                password = self._users[ auth_dict["username"] ]
            else:
                return self._return_auth_challenge(request_handler)
            if not auth_dict.get("nonce") in self._nonces:
                return self._return_auth_challenge(request_handler)
            else:
                self._nonces.remove(auth_dict["nonce"])

            auth_validated = False

            
            
            

            for path in [request_handler.path, request_handler.short_path]:
                if self._validate_auth(auth_dict,
                                       password,
                                       request_handler.command,
                                       path):
                    auth_validated = True

            if not auth_validated:
                return self._return_auth_challenge(request_handler)
            return True


class BasicAuthHandler(http.server.BaseHTTPRequestHandler):
    
    
    USER = 'testUser'
    PASSWD = 'testPass'
    REALM = 'Test'
    USER_PASSWD = "%s:%s" % (USER, PASSWD)
    ENCODED_AUTH = base64.b64encode(USER_PASSWD.encode('ascii')).decode('ascii')

    def __init__(self, *args, **kwargs):
        http.server.BaseHTTPRequestHandler.__init__(self, *args, **kwargs)

    def log_message(self, format, *args):
        
        pass

    def do_HEAD(self):
        self.send_response(200)
        self.send_header("Content-type", "text/html")
        self.end_headers()

    def do_AUTHHEAD(self):
        self.send_response(401)
        self.send_header("WWW-Authenticate", "Basic realm=\"%s\"" % self.REALM)
        self.send_header("Content-type", "text/html")
        self.end_headers()

    def do_GET(self):
        if not self.headers.get("Authorization", ""):
            self.do_AUTHHEAD()
            self.wfile.write(b"No Auth header received")
        elif self.headers.get(
                "Authorization", "") == "Basic " + self.ENCODED_AUTH:
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"It works")
        else:
            
            self.do_AUTHHEAD()





class FakeProxyHandler(http.server.BaseHTTPRequestHandler):
    

    def __init__(self, digest_auth_handler, *args, **kwargs):
        
        
        self.digest_auth_handler = digest_auth_handler
        http.server.BaseHTTPRequestHandler.__init__(self, *args, **kwargs)

    def log_message(self, format, *args):
        
        
        pass

    def do_GET(self):
        (scm, netloc, path, params, query, fragment) = urllib.parse.urlparse(
            self.path, "http")
        self.short_path = path
        if self.digest_auth_handler.handle_request(self):
            self.send_response(200, "OK")
            self.send_header("Content-Type", "text/html")
            self.end_headers()
            self.wfile.write(bytes("You've reached %s!<BR>" % self.path,
                                   "ascii"))
            self.wfile.write(b"Our apologies, but our server is down due to "
                             b"a sudden zombie invasion.")



class BasicAuthTests(unittest.TestCase):
    USER = "testUser"
    PASSWD = "testPass"
    INCORRECT_PASSWD = "Incorrect"
    REALM = "Test"

    def setUp(self):
        super(BasicAuthTests, self).setUp()
        
        def http_server_with_basic_auth_handler(*args, **kwargs):
            return BasicAuthHandler(*args, **kwargs)
        self.server = LoopbackHttpServerThread(http_server_with_basic_auth_handler)
        self.addCleanup(self.stop_server)
        self.server_url = 'http://127.0.0.1:%s' % self.server.port
        self.server.start()
        self.server.ready.wait()

    def stop_server(self):
        self.server.stop()
        self.server = None

    def tearDown(self):
        super(BasicAuthTests, self).tearDown()

    def test_basic_auth_success(self):
        ah = urllib.request.HTTPBasicAuthHandler()
        ah.add_password(self.REALM, self.server_url, self.USER, self.PASSWD)
        urllib.request.install_opener(urllib.request.build_opener(ah))
        try:
            self.assertTrue(urllib.request.urlopen(self.server_url))
        except urllib.error.HTTPError:
            self.fail("Basic auth failed for the url: %s" % self.server_url)

    def test_basic_auth_httperror(self):
        ah = urllib.request.HTTPBasicAuthHandler()
        ah.add_password(self.REALM, self.server_url, self.USER, self.INCORRECT_PASSWD)
        urllib.request.install_opener(urllib.request.build_opener(ah))
        self.assertRaises(urllib.error.HTTPError, urllib.request.urlopen, self.server_url)


@hashlib_helper.requires_hashdigest("md5", openssl=True)
class ProxyAuthTests(unittest.TestCase):
    URL = "http://localhost"

    USER = "tester"
    PASSWD = "test123"
    REALM = "TestRealm"

    def setUp(self):
        super(ProxyAuthTests, self).setUp()
        
        def restore_environ(old_environ):
            os.environ.clear()
            os.environ.update(old_environ)
        self.addCleanup(restore_environ, os.environ.copy())
        os.environ['NO_PROXY'] = ''
        os.environ['no_proxy'] = ''

        self.digest_auth_handler = DigestAuthHandler()
        self.digest_auth_handler.set_users({self.USER: self.PASSWD})
        self.digest_auth_handler.set_realm(self.REALM)
        
        def create_fake_proxy_handler(*args, **kwargs):
            return FakeProxyHandler(self.digest_auth_handler, *args, **kwargs)

        self.server = LoopbackHttpServerThread(create_fake_proxy_handler)
        self.addCleanup(self.stop_server)
        self.server.start()
        self.server.ready.wait()
        proxy_url = "http://127.0.0.1:%d" % self.server.port
        handler = urllib.request.ProxyHandler({"http" : proxy_url})
        self.proxy_digest_handler = urllib.request.ProxyDigestAuthHandler()
        self.opener = urllib.request.build_opener(
            handler, self.proxy_digest_handler)

    def stop_server(self):
        self.server.stop()
        self.server = None

    def test_proxy_with_bad_password_raises_httperror(self):
        self.proxy_digest_handler.add_password(self.REALM, self.URL,
                                               self.USER, self.PASSWD+"bad")
        self.digest_auth_handler.set_qop("auth")
        self.assertRaises(urllib.error.HTTPError,
                          self.opener.open,
                          self.URL)

    def test_proxy_with_no_password_raises_httperror(self):
        self.digest_auth_handler.set_qop("auth")
        self.assertRaises(urllib.error.HTTPError,
                          self.opener.open,
                          self.URL)

    def test_proxy_qop_auth_works(self):
        self.proxy_digest_handler.add_password(self.REALM, self.URL,
                                               self.USER, self.PASSWD)
        self.digest_auth_handler.set_qop("auth")
        with self.opener.open(self.URL) as result:
            while result.read():
                pass

    def test_proxy_qop_auth_int_works_or_throws_urlerror(self):
        self.proxy_digest_handler.add_password(self.REALM, self.URL,
                                               self.USER, self.PASSWD)
        self.digest_auth_handler.set_qop("auth-int")
        try:
            result = self.opener.open(self.URL)
        except urllib.error.URLError:
            
            
            
            pass
        else:
            with result:
                while result.read():
                    pass


def GetRequestHandler(responses):

    class FakeHTTPRequestHandler(http.server.BaseHTTPRequestHandler):

        server_version = "TestHTTP/"
        requests = []
        headers_received = []
        port = 80

        def do_GET(self):
            body = self.send_head()
            while body:
                done = self.wfile.write(body)
                body = body[done:]

        def do_POST(self):
            content_length = self.headers["Content-Length"]
            post_data = self.rfile.read(int(content_length))
            self.do_GET()
            self.requests.append(post_data)

        def send_head(self):
            FakeHTTPRequestHandler.headers_received = self.headers
            self.requests.append(self.path)
            response_code, headers, body = responses.pop(0)

            self.send_response(response_code)

            for (header, value) in headers:
                self.send_header(header, value % {'port':self.port})
            if body:
                self.send_header("Content-type", "text/plain")
                self.end_headers()
                return body
            self.end_headers()

        def log_message(self, *args):
            pass


    return FakeHTTPRequestHandler


class TestUrlopen(unittest.TestCase):
    

    def setUp(self):
        super(TestUrlopen, self).setUp()

        
        self.addCleanup(urllib.request.urlcleanup)

        
        def restore_environ(old_environ):
            os.environ.clear()
            os.environ.update(old_environ)
        self.addCleanup(restore_environ, os.environ.copy())
        os.environ['NO_PROXY'] = '*'
        os.environ['no_proxy'] = '*'

    def urlopen(self, url, data=None, **kwargs):
        l = []
        f = urllib.request.urlopen(url, data, **kwargs)
        try:
            
            l.extend(f.readlines(200))
            l.append(f.readline())
            l.append(f.read(1024))
            l.append(f.read())
        finally:
            f.close()
        return b"".join(l)

    def stop_server(self):
        self.server.stop()
        self.server = None

    def start_server(self, responses=None):
        if responses is None:
            responses = [(200, [], b"we don't care")]
        handler = GetRequestHandler(responses)

        self.server = LoopbackHttpServerThread(handler)
        self.addCleanup(self.stop_server)
        self.server.start()
        self.server.ready.wait()
        port = self.server.port
        handler.port = port
        return handler

    def start_https_server(self, responses=None, **kwargs):
        if not hasattr(urllib.request, 'HTTPSHandler'):
            self.skipTest('ssl support required')
        from test.ssl_servers import make_https_server
        if responses is None:
            responses = [(200, [], b"we care a bit")]
        handler = GetRequestHandler(responses)
        server = make_https_server(self, handler_class=handler, **kwargs)
        handler.port = server.port
        return handler

    def test_redirection(self):
        expected_response = b"We got here..."
        responses = [
            (302, [("Location", "http://localhost:%(port)s/somewhere_else")],
             ""),
            (200, [], expected_response)
        ]

        handler = self.start_server(responses)
        data = self.urlopen("http://localhost:%s/" % handler.port)
        self.assertEqual(data, expected_response)
        self.assertEqual(handler.requests, ["/", "/somewhere_else"])

    def test_chunked(self):
        expected_response = b"hello world"
        chunked_start = (
                        b'a\r\n'
                        b'hello worl\r\n'
                        b'1\r\n'
                        b'd\r\n'
                        b'0\r\n'
                        )
        response = [(200, [("Transfer-Encoding", "chunked")], chunked_start)]
        handler = self.start_server(response)
        data = self.urlopen("http://localhost:%s/" % handler.port)
        self.assertEqual(data, expected_response)

    def test_404(self):
        expected_response = b"Bad bad bad..."
        handler = self.start_server([(404, [], expected_response)])

        try:
            self.urlopen("http://localhost:%s/weeble" % handler.port)
        except urllib.error.URLError as f:
            data = f.read()
            f.close()
        else:
            self.fail("404 should raise URLError")

        self.assertEqual(data, expected_response)
        self.assertEqual(handler.requests, ["/weeble"])

    def test_200(self):
        expected_response = b"pycon 2008..."
        handler = self.start_server([(200, [], expected_response)])
        data = self.urlopen("http://localhost:%s/bizarre" % handler.port)
        self.assertEqual(data, expected_response)
        self.assertEqual(handler.requests, ["/bizarre"])

    def test_200_with_parameters(self):
        expected_response = b"pycon 2008..."
        handler = self.start_server([(200, [], expected_response)])
        data = self.urlopen("http://localhost:%s/bizarre" % handler.port,
                             b"get=with_feeling")
        self.assertEqual(data, expected_response)
        self.assertEqual(handler.requests, ["/bizarre", b"get=with_feeling"])

    def test_https(self):
        handler = self.start_https_server()
        context = ssl.create_default_context(cafile=CERT_localhost)
        data = self.urlopen("https://localhost:%s/bizarre" % handler.port, context=context)
        self.assertEqual(data, b"we care a bit")

    def test_https_sni(self):
        if ssl is None:
            self.skipTest("ssl module required")
        if not ssl.HAS_SNI:
            self.skipTest("SNI support required in OpenSSL")
        sni_name = None
        def cb_sni(ssl_sock, server_name, initial_context):
            nonlocal sni_name
            sni_name = server_name
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.set_servername_callback(cb_sni)
        handler = self.start_https_server(context=context, certfile=CERT_localhost)
        context = ssl.create_default_context(cafile=CERT_localhost)
        self.urlopen("https://localhost:%s" % handler.port, context=context)
        self.assertEqual(sni_name, "localhost")

    def test_sending_headers(self):
        handler = self.start_server()
        req = urllib.request.Request("http://localhost:%s/" % handler.port,
                                     headers={"Range": "bytes=20-39"})
        with urllib.request.urlopen(req):
            pass
        self.assertEqual(handler.headers_received["Range"], "bytes=20-39")

    def test_sending_headers_camel(self):
        handler = self.start_server()
        req = urllib.request.Request("http://localhost:%s/" % handler.port,
                                     headers={"X-SoMe-hEader": "foobar"})
        with urllib.request.urlopen(req):
            pass
        self.assertIn("X-Some-Header", handler.headers_received.keys())
        self.assertNotIn("X-SoMe-hEader", handler.headers_received.keys())

    def test_basic(self):
        handler = self.start_server()
        with urllib.request.urlopen("http://localhost:%s" % handler.port) as open_url:
            for attr in ("read", "close", "info", "geturl"):
                self.assertTrue(hasattr(open_url, attr), "object returned from "
                             "urlopen lacks the %s attribute" % attr)
            self.assertTrue(open_url.read(), "calling 'read' failed")

    def test_info(self):
        handler = self.start_server()
        open_url = urllib.request.urlopen(
            "http://localhost:%s" % handler.port)
        with open_url:
            info_obj = open_url.info()
        self.assertIsInstance(info_obj, email.message.Message,
                              "object returned by 'info' is not an "
                              "instance of email.message.Message")
        self.assertEqual(info_obj.get_content_subtype(), "plain")

    def test_geturl(self):
        
        handler = self.start_server()
        open_url = urllib.request.urlopen("http://localhost:%s" % handler.port)
        with open_url:
            url = open_url.geturl()
        self.assertEqual(url, "http://localhost:%s" % handler.port)

    def test_iteration(self):
        expected_response = b"pycon 2008..."
        handler = self.start_server([(200, [], expected_response)])
        data = urllib.request.urlopen("http://localhost:%s" % handler.port)
        for line in data:
            self.assertEqual(line, expected_response)

    def test_line_iteration(self):
        lines = [b"We\n", b"got\n", b"here\n", b"verylong " * 8192 + b"\n"]
        expected_response = b"".join(lines)
        handler = self.start_server([(200, [], expected_response)])
        data = urllib.request.urlopen("http://localhost:%s" % handler.port)
        for index, line in enumerate(data):
            self.assertEqual(line, lines[index],
                             "Fetched line number %s doesn't match expected:\n"
                             "    Expected length was %s, got %s" %
                             (index, len(lines[index]), len(line)))
        self.assertEqual(index + 1, len(lines))

    def test_issue16464(self):
        
        
        handler = self.start_server([
            (200, [], b'any'),
            (200, [], b'any'),
        ])
        opener = urllib.request.build_opener()
        request = urllib.request.Request("http://localhost:%s" % handler.port)
        self.assertEqual(None, request.data)

        opener.open(request, "1".encode("us-ascii"))
        self.assertEqual(b"1", request.data)
        self.assertEqual("1", request.get_header("Content-length"))

        opener.open(request, "1234567890".encode("us-ascii"))
        self.assertEqual(b"1234567890", request.data)
        self.assertEqual("10", request.get_header("Content-length"))

def setUpModule():
    thread_info = threading_helper.threading_setup()
    unittest.addModuleCleanup(threading_helper.threading_cleanup, *thread_info)


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