<s> import urlparse <EOL> import urllib2 <EOL> import BaseHTTPServer <EOL> import unittest <EOL> import hashlib <EOL> from test import test_support <EOL> if test_support . is_jython : <EOL> import socket <EOL> socket . _use_ipv4_addresses_only ( True ) <EOL> mimetools = test_support . import_module ( '<STR_LIT>' , deprecated = True ) <EOL> threading = test_support . import_module ( '<STR_LIT>' ) <EOL> class LoopbackHttpServer ( BaseHTTPServer . HTTPServer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , server_address , RequestHandlerClass ) : <EOL> BaseHTTPServer . HTTPServer . __init__ ( self , <EOL> server_address , <EOL> RequestHandlerClass , <EOL> True ) <EOL> host , port = self . socket . getsockname ( ) [ : <NUM_LIT:2> ] <EOL> self . server_name = socket . getfqdn ( host ) <EOL> self . server_port = port <EOL> self . socket . settimeout ( <NUM_LIT:1.0> ) <EOL> def get_request ( self ) : <EOL> """<STR_LIT>""" <EOL> request , client_address = self . socket . accept ( ) <EOL> request . settimeout ( <NUM_LIT> ) <EOL> return ( request , client_address ) <EOL> class LoopbackHttpServerThread ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , request_handler ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . _stop = False <EOL> self . ready = threading . Event ( ) <EOL> request_handler . protocol_version = "<STR_LIT>" <EOL> self . httpd = LoopbackHttpServer ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) , <EOL> request_handler ) <EOL> self . port = self . httpd . server_port <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _stop = True <EOL> self . join ( ) <EOL> def run ( self ) : <EOL> self . ready . set ( ) <EOL> while not self . _stop : <EOL> self . httpd . handle_request ( ) <EOL> class DigestAuthHandler : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _request_num = <NUM_LIT:0> <EOL> self . _nonces = [ ] <EOL> self . _users = { } <EOL> self . _realm_name = "<STR_LIT>" <EOL> self . _qop = "<STR_LIT>" <EOL> def set_qop ( self , qop ) : <EOL> self . _qop = qop <EOL> def set_users ( self , users ) : <EOL> assert isinstance ( users , dict ) <EOL> self . _users = users <EOL> def set_realm ( self , realm ) : <EOL> self . _realm_name = realm <EOL> def _generate_nonce ( self ) : <EOL> self . _request_num += <NUM_LIT:1> <EOL> nonce = hashlib . md5 ( str ( self . _request_num ) ) . hexdigest ( ) <EOL> self . _nonces . append ( nonce ) <EOL> return nonce <EOL> def _create_auth_dict ( self , auth_str ) : <EOL> first_space_index = auth_str . find ( "<STR_LIT:U+0020>" ) <EOL> auth_str = auth_str [ first_space_index + <NUM_LIT:1> : ] <EOL> parts = auth_str . split ( "<STR_LIT:U+002C>" ) <EOL> auth_dict = { } <EOL> for part in parts : <EOL> name , value = part . split ( "<STR_LIT:=>" ) <EOL> name = name . strip ( ) <EOL> if value [ <NUM_LIT:0> ] == '<STR_LIT:">' and value [ - <NUM_LIT:1> ] == '<STR_LIT:">' : <EOL> value = value [ <NUM_LIT:1> : - <NUM_LIT:1> ] <EOL> else : <EOL> value = value . strip ( ) <EOL> auth_dict [ name ] = value <EOL> return auth_dict <EOL> def _validate_auth ( self , auth_dict , password , method , uri ) : <EOL> final_dict = { } <EOL> final_dict . update ( auth_dict ) <EOL> final_dict [ "<STR_LIT:password>" ] = password <EOL> final_dict [ "<STR_LIT>" ] = method <EOL> final_dict [ "<STR_LIT>" ] = uri <EOL> HA1_str = "<STR_LIT>" % final_dict <EOL> HA1 = hashlib . md5 ( HA1_str ) . hexdigest ( ) <EOL> HA2_str = "<STR_LIT>" % final_dict <EOL> HA2 = hashlib . md5 ( HA2_str ) . hexdigest ( ) <EOL> final_dict [ "<STR_LIT>" ] = HA1 <EOL> final_dict [ "<STR_LIT>" ] = HA2 <EOL> response_str = "<STR_LIT>" "<STR_LIT>" % final_dict <EOL> response = hashlib . md5 ( response_str ) . hexdigest ( ) <EOL> return response == auth_dict [ "<STR_LIT>" ] <EOL> def _return_auth_challenge ( self , request_handler ) : <EOL> request_handler . send_response ( <NUM_LIT> , "<STR_LIT>" ) <EOL> request_handler . send_header ( "<STR_LIT:Content-Type>" , "<STR_LIT>" ) <EOL> request_handler . send_header ( <EOL> '<STR_LIT>' , '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( self . _realm_name , self . _qop , self . _generate_nonce ( ) ) ) <EOL> request_handler . end_headers ( ) <EOL> request_handler . wfile . write ( "<STR_LIT>" ) <EOL> return False <EOL> def handle_request ( self , request_handler ) : <EOL> """<STR_LIT>""" <EOL> if len ( self . _users ) == <NUM_LIT:0> : <EOL> return True <EOL> if '<STR_LIT>' not in request_handler . headers : <EOL> return self . _return_auth_challenge ( request_handler ) <EOL> else : <EOL> auth_dict = self . _create_auth_dict ( <EOL> request_handler . headers [ '<STR_LIT>' ] <EOL> ) <EOL> if auth_dict [ "<STR_LIT:username>" ] in self . _users : <EOL> password = self . _users [ auth_dict [ "<STR_LIT:username>" ] ] <EOL> else : <EOL> return self . _return_auth_challenge ( request_handler ) <EOL> if not auth_dict . get ( "<STR_LIT>" ) in self . _nonces : <EOL> return self . _return_auth_challenge ( request_handler ) <EOL> else : <EOL> self . _nonces . remove ( auth_dict [ "<STR_LIT>" ] ) <EOL> auth_validated = False <EOL> for path in [ request_handler . path , request_handler . short_path ] : <EOL> if self . _validate_auth ( auth_dict , <EOL> password , <EOL> request_handler . command , <EOL> path ) : <EOL> auth_validated = True <EOL> if not auth_validated : <EOL> return self . _return_auth_challenge ( request_handler ) <EOL> return True <EOL> class FakeProxyHandler ( BaseHTTPServer . BaseHTTPRequestHandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , digest_auth_handler , * args , ** kwargs ) : <EOL> self . digest_auth_handler = digest_auth_handler <EOL> BaseHTTPServer . BaseHTTPRequestHandler . __init__ ( self , * args , ** kwargs ) <EOL> def log_message ( self , format , * args ) : <EOL> pass <EOL> def do_GET ( self ) : <EOL> ( scm , netloc , path , params , query , fragment ) = urlparse . urlparse ( <EOL> self . path , '<STR_LIT:http>' ) <EOL> self . short_path = path <EOL> if self . digest_auth_handler . handle_request ( self ) : <EOL> self . send_response ( <NUM_LIT:200> , "<STR_LIT:OK>" ) <EOL> self . send_header ( "<STR_LIT:Content-Type>" , "<STR_LIT>" ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( "<STR_LIT>" % self . path ) <EOL> self . wfile . write ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class BaseTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . _threads = test_support . threading_setup ( ) <EOL> def tearDown ( self ) : <EOL> test_support . threading_cleanup ( * self . _threads ) <EOL> class ProxyAuthTests ( BaseTestCase ) : <EOL> URL = "<STR_LIT>" <EOL> USER = "<STR_LIT>" <EOL> PASSWD = "<STR_LIT>" <EOL> REALM = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ProxyAuthTests , self ) . setUp ( ) <EOL> self . digest_auth_handler = DigestAuthHandler ( ) <EOL> self . digest_auth_handler . set_users ( { self . USER : self . PASSWD } ) <EOL> self . digest_auth_handler . set_realm ( self . REALM ) <EOL> def create_fake_proxy_handler ( * args , ** kwargs ) : <EOL> return FakeProxyHandler ( self . digest_auth_handler , * args , ** kwargs ) <EOL> self . server = LoopbackHttpServerThread ( create_fake_proxy_handler ) <EOL> self . server . start ( ) <EOL> self . server . ready . wait ( ) <EOL> proxy_url = "<STR_LIT>" % self . server . port <EOL> handler = urllib2 . ProxyHandler ( { "<STR_LIT:http>" : proxy_url } ) <EOL> self . proxy_digest_handler = urllib2 . ProxyDigestAuthHandler ( ) <EOL> self . opener = urllib2 . build_opener ( handler , self . proxy_digest_handler ) <EOL> def tearDown ( self ) : <EOL> self . server . stop ( ) <EOL> super ( ProxyAuthTests , self ) . tearDown ( ) <EOL> def test_proxy_with_bad_password_raises_httperror ( self ) : <EOL> self . proxy_digest_handler . add_password ( self . REALM , self . URL , <EOL> self . USER , self . PASSWD + "<STR_LIT>" ) <EOL> self . digest_auth_handler . set_qop ( "<STR_LIT>" ) <EOL> self . assertRaises ( urllib2 . HTTPError , <EOL> self . opener . open , <EOL> self . URL ) <EOL> def test_proxy_with_no_password_raises_httperror ( self ) : <EOL> self . digest_auth_handler . set_qop ( "<STR_LIT>" ) <EOL> self . assertRaises ( urllib2 . HTTPError , <EOL> self . opener . open , <EOL> self . URL ) <EOL> def test_proxy_qop_auth_works ( self ) : <EOL> self . proxy_digest_handler . add_password ( self . REALM , self . URL , <EOL> self . USER , self . PASSWD ) <EOL> self . digest_auth_handler . set_qop ( "<STR_LIT>" ) <EOL> result = self . opener . open ( self . URL ) <EOL> while result . read ( ) : <EOL> pass <EOL> result . close ( ) <EOL> def test_proxy_qop_auth_int_works_or_throws_urlerror ( self ) : <EOL> self . proxy_digest_handler . add_password ( self . REALM , self . URL , <EOL> self . USER , self . PASSWD ) <EOL> self . digest_auth_handler . set_qop ( "<STR_LIT>" ) <EOL> try : <EOL> result = self . opener . open ( self . URL ) <EOL> except urllib2 . URLError : <EOL> result = None <EOL> if result : <EOL> while result . read ( ) : <EOL> pass <EOL> result . close ( ) <EOL> def GetRequestHandler ( responses ) : <EOL> class FakeHTTPRequestHandler ( BaseHTTPServer . BaseHTTPRequestHandler ) : <EOL> server_version = "<STR_LIT>" <EOL> requests = [ ] <EOL> headers_received = [ ] <EOL> port = <NUM_LIT> <EOL> def do_GET ( self ) : <EOL> body = self . send_head ( ) <EOL> if body : <EOL> self . wfile . write ( body ) <EOL> def do_POST ( self ) : <EOL> content_length = self . headers [ '<STR_LIT>' ] <EOL> post_data = self . rfile . read ( int ( content_length ) ) <EOL> self . do_GET ( ) <EOL> self . requests . append ( post_data ) <EOL> def send_head ( self ) : <EOL> FakeHTTPRequestHandler . headers_received = self . headers <EOL> self . requests . append ( self . path ) <EOL> response_code , headers , body = responses . pop ( <NUM_LIT:0> ) <EOL> self . send_response ( response_code ) <EOL> for ( header , value ) in headers : <EOL> self . send_header ( header , value % self . port ) <EOL> if body : <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . end_headers ( ) <EOL> return body <EOL> self . end_headers ( ) <EOL> def log_message ( self , * args ) : <EOL> pass <EOL> return FakeHTTPRequestHandler <EOL> class TestUrlopen ( BaseTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> proxy_handler = urllib2 . ProxyHandler ( { } ) <EOL> opener = urllib2 . build_opener ( proxy_handler ) <EOL> urllib2 . install_opener ( opener ) <EOL> super ( TestUrlopen , self ) . setUp ( ) <EOL> def start_server ( self , responses ) : <EOL> handler = GetRequestHandler ( responses ) <EOL> self . server = LoopbackHttpServerThread ( handler ) <EOL> self . server . start ( ) <EOL> self . server . ready . wait ( ) <EOL> port = self . server . port <EOL> handler . port = port <EOL> return handler <EOL> def test_redirection ( self ) : <EOL> expected_response = '<STR_LIT>' <EOL> responses = [ <EOL> ( <NUM_LIT> , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:200> , [ ] , expected_response ) <EOL> ] <EOL> handler = self . start_server ( responses ) <EOL> try : <EOL> f = urllib2 . urlopen ( '<STR_LIT>' % handler . port ) <EOL> data = f . read ( ) <EOL> f . close ( ) <EOL> self . assertEqual ( data , expected_response ) <EOL> self . assertEqual ( handler . requests , [ '<STR_LIT:/>' , '<STR_LIT>' ] ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_404 ( self ) : <EOL> expected_response = '<STR_LIT>' <EOL> handler = self . start_server ( [ ( <NUM_LIT> , [ ] , expected_response ) ] ) <EOL> try : <EOL> try : <EOL> urllib2 . urlopen ( '<STR_LIT>' % handler . port ) <EOL> except urllib2 . URLError , f : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> data = f . read ( ) <EOL> f . close ( ) <EOL> self . assertEqual ( data , expected_response ) <EOL> self . assertEqual ( handler . requests , [ '<STR_LIT>' ] ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_200 ( self ) : <EOL> expected_response = '<STR_LIT>' <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , expected_response ) ] ) <EOL> try : <EOL> f = urllib2 . urlopen ( '<STR_LIT>' % handler . port ) <EOL> data = f . read ( ) <EOL> f . close ( ) <EOL> self . assertEqual ( data , expected_response ) <EOL> self . assertEqual ( handler . requests , [ '<STR_LIT>' ] ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_200_with_parameters ( self ) : <EOL> expected_response = '<STR_LIT>' <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , expected_response ) ] ) <EOL> try : <EOL> f = urllib2 . urlopen ( '<STR_LIT>' % handler . port , '<STR_LIT>' ) <EOL> data = f . read ( ) <EOL> f . close ( ) <EOL> self . assertEqual ( data , expected_response ) <EOL> self . assertEqual ( handler . requests , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_sending_headers ( self ) : <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , "<STR_LIT>" ) ] ) <EOL> try : <EOL> req = urllib2 . Request ( "<STR_LIT>" % handler . port , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> urllib2 . urlopen ( req ) <EOL> self . assertEqual ( handler . headers_received [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_basic ( self ) : <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , "<STR_LIT>" ) ] ) <EOL> try : <EOL> open_url = urllib2 . urlopen ( "<STR_LIT>" % handler . port ) <EOL> for attr in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:info>" , "<STR_LIT>" ) : <EOL> self . assertTrue ( hasattr ( open_url , attr ) , "<STR_LIT>" <EOL> "<STR_LIT>" % attr ) <EOL> try : <EOL> self . assertTrue ( open_url . read ( ) , "<STR_LIT>" ) <EOL> finally : <EOL> open_url . close ( ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_info ( self ) : <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , "<STR_LIT>" ) ] ) <EOL> try : <EOL> open_url = urllib2 . urlopen ( "<STR_LIT>" % handler . port ) <EOL> info_obj = open_url . info ( ) <EOL> self . assertIsInstance ( info_obj , mimetools . Message , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( info_obj . getsubtype ( ) , "<STR_LIT>" ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_geturl ( self ) : <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , "<STR_LIT>" ) ] ) <EOL> try : <EOL> open_url = urllib2 . urlopen ( "<STR_LIT>" % handler . port ) <EOL> url = open_url . geturl ( ) <EOL> self . assertEqual ( url , "<STR_LIT>" % handler . port ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def test_bad_address ( self ) : <EOL> self . assertRaises ( IOError , <EOL> urllib2 . urlopen , "<STR_LIT>" ) <EOL> def test_iteration ( self ) : <EOL> expected_response = "<STR_LIT>" <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , expected_response ) ] ) <EOL> try : <EOL> data = urllib2 . urlopen ( "<STR_LIT>" % handler . port ) <EOL> for line in data : <EOL> self . assertEqual ( line , expected_response ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> def ztest_line_iteration ( self ) : <EOL> lines = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" * <NUM_LIT> + "<STR_LIT:\n>" ] <EOL> expected_response = "<STR_LIT>" . join ( lines ) <EOL> handler = self . start_server ( [ ( <NUM_LIT:200> , [ ] , expected_response ) ] ) <EOL> try : <EOL> data = urllib2 . urlopen ( "<STR_LIT>" % handler . port ) <EOL> for index , line in enumerate ( data ) : <EOL> self . assertEqual ( line , lines [ index ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( index , len ( lines [ index ] ) , len ( line ) ) ) <EOL> finally : <EOL> self . server . stop ( ) <EOL> self . assertEqual ( index + <NUM_LIT:1> , len ( lines ) ) <EOL> def test_main ( ) : <EOL> test_support . run_unittest ( ProxyAuthTests , TestUrlopen ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test_main ( ) </s>
