<s> import json <EOL> import os <EOL> import time <EOL> from threading import Thread <EOL> from libmproxy import controller , proxy <EOL> from netlib import http_auth <EOL> from testfixtures import LogCapture <EOL> from twisted . internet import defer <EOL> from twisted . trial . unittest import TestCase <EOL> from scrapy . utils . test import get_crawler <EOL> from scrapy . http import Request <EOL> from tests . spiders import SimpleSpider , SingleRequestSpider <EOL> from tests . mockserver import MockServer <EOL> class HTTPSProxy ( controller . Master , Thread ) : <EOL> def __init__ ( self , port ) : <EOL> password_manager = http_auth . PassManSingleUser ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> authenticator = http_auth . BasicProxyAuth ( password_manager , "<STR_LIT>" ) <EOL> cert_path = os . path . join ( os . path . abspath ( os . path . dirname ( __file__ ) ) , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> server = proxy . ProxyServer ( proxy . ProxyConfig ( <EOL> authenticator = authenticator , <EOL> cacert = cert_path ) , <EOL> port ) <EOL> Thread . __init__ ( self ) <EOL> controller . Master . __init__ ( self , server ) <EOL> class ProxyConnectTestCase ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mockserver = MockServer ( ) <EOL> self . mockserver . __enter__ ( ) <EOL> self . _oldenv = os . environ . copy ( ) <EOL> self . _proxy = HTTPSProxy ( <NUM_LIT> ) <EOL> self . _proxy . start ( ) <EOL> time . sleep ( <NUM_LIT:1.0> ) <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> def tearDown ( self ) : <EOL> self . mockserver . __exit__ ( None , None , None ) <EOL> self . _proxy . shutdown ( ) <EOL> os . environ = self . _oldenv <EOL> @ defer . inlineCallbacks <EOL> def test_https_connect_tunnel ( self ) : <EOL> crawler = get_crawler ( SimpleSpider ) <EOL> with LogCapture ( ) as l : <EOL> yield crawler . crawl ( "<STR_LIT>" ) <EOL> self . _assert_got_response_code ( <NUM_LIT:200> , l ) <EOL> @ defer . inlineCallbacks <EOL> def test_https_noconnect ( self ) : <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> crawler = get_crawler ( SimpleSpider ) <EOL> with LogCapture ( ) as l : <EOL> yield crawler . crawl ( "<STR_LIT>" ) <EOL> self . _assert_got_response_code ( <NUM_LIT:200> , l ) <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> @ defer . inlineCallbacks <EOL> def test_https_connect_tunnel_error ( self ) : <EOL> crawler = get_crawler ( SimpleSpider ) <EOL> with LogCapture ( ) as l : <EOL> yield crawler . crawl ( "<STR_LIT>" ) <EOL> self . _assert_got_tunnel_error ( l ) <EOL> @ defer . inlineCallbacks <EOL> def test_https_tunnel_auth_error ( self ) : <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> crawler = get_crawler ( SimpleSpider ) <EOL> with LogCapture ( ) as l : <EOL> yield crawler . crawl ( "<STR_LIT>" ) <EOL> self . _assert_got_tunnel_error ( l ) <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> @ defer . inlineCallbacks <EOL> def test_https_tunnel_without_leak_proxy_authorization_header ( self ) : <EOL> request = Request ( "<STR_LIT>" ) <EOL> crawler = get_crawler ( SingleRequestSpider ) <EOL> with LogCapture ( ) as l : <EOL> yield crawler . crawl ( seed = request ) <EOL> self . _assert_got_response_code ( <NUM_LIT:200> , l ) <EOL> echo = json . loads ( crawler . spider . meta [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . body ) <EOL> self . assertTrue ( '<STR_LIT>' not in echo [ '<STR_LIT>' ] ) <EOL> @ defer . inlineCallbacks <EOL> def test_https_noconnect_auth_error ( self ) : <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> crawler = get_crawler ( SimpleSpider ) <EOL> with LogCapture ( ) as l : <EOL> yield crawler . crawl ( "<STR_LIT>" ) <EOL> self . _assert_got_response_code ( <NUM_LIT> , l ) <EOL> def _assert_got_response_code ( self , code , log ) : <EOL> self . assertEqual ( str ( log ) . count ( '<STR_LIT>' % code ) , <NUM_LIT:1> ) <EOL> def _assert_got_tunnel_error ( self , log ) : <EOL> self . assertEqual ( str ( log ) . count ( '<STR_LIT>' ) , <NUM_LIT:1> ) </s>
