<s> """<STR_LIT>""" <EOL> import base64 <EOL> import os <EOL> import re <EOL> import shutil <EOL> import subprocess <EOL> import tempfile <EOL> import time <EOL> import unittest <EOL> import google <EOL> import mox <EOL> import portpicker <EOL> from google . appengine . api import appinfo <EOL> from google . appengine . tools . devappserver2 import http_proxy <EOL> from google . appengine . tools . devappserver2 import http_runtime <EOL> from google . appengine . tools . devappserver2 import instance <EOL> from google . appengine . tools . devappserver2 import login <EOL> from google . appengine . tools . devappserver2 import runtime_config_pb2 <EOL> from google . appengine . tools . devappserver2 import safe_subprocess <EOL> from google . appengine . tools . devappserver2 import wsgi_test_utils <EOL> class MockMessage ( object ) : <EOL> def __init__ ( self , headers ) : <EOL> self . headers = headers <EOL> def __iter__ ( self ) : <EOL> return iter ( set ( name for name , _ in self . headers ) ) <EOL> def getheaders ( self , name ) : <EOL> return [ value for header_name , value in self . headers if header_name == name ] <EOL> class FakeHttpResponse ( object ) : <EOL> def __init__ ( self , status , reason , headers , body ) : <EOL> self . body = body <EOL> self . has_read = False <EOL> self . partial_read_error = None <EOL> self . status = status <EOL> self . reason = reason <EOL> self . headers = headers <EOL> self . msg = MockMessage ( headers ) <EOL> def read ( self , amt = None ) : <EOL> if not self . has_read : <EOL> self . has_read = True <EOL> return self . body <EOL> elif self . partial_read_error : <EOL> raise self . partial_read_error <EOL> else : <EOL> return '<STR_LIT>' <EOL> def getheaders ( self ) : <EOL> return self . headers <EOL> class FakeTee ( object ) : <EOL> def __init__ ( self , buf ) : <EOL> self . buf = buf <EOL> def get_buf ( self ) : <EOL> return self . buf <EOL> def join ( self , unused_timeout ) : <EOL> pass <EOL> class ModuleConfigurationStub ( object ) : <EOL> def __init__ ( self , application_root = '<STR_LIT>' , error_handlers = None ) : <EOL> self . application_root = application_root <EOL> self . error_handlers = error_handlers <EOL> class HttpRuntimeProxyTest ( wsgi_test_utils . WSGITestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . tmpdir = tempfile . mkdtemp ( ) <EOL> module_configuration = ModuleConfigurationStub ( <EOL> application_root = self . tmpdir , <EOL> error_handlers = [ <EOL> appinfo . ErrorHandlers ( error_code = '<STR_LIT>' , file = '<STR_LIT>' ) , <EOL> appinfo . ErrorHandlers ( error_code = '<STR_LIT:default>' , file = '<STR_LIT>' ) , <EOL> ] ) <EOL> self . runtime_config = runtime_config_pb2 . Config ( ) <EOL> self . runtime_config . app_id = '<STR_LIT>' <EOL> self . runtime_config . version_id = '<STR_LIT:version>' <EOL> self . runtime_config . api_port = <NUM_LIT> <EOL> self . runtime_config . application_root = self . tmpdir <EOL> self . runtime_config . datacenter = '<STR_LIT>' <EOL> self . runtime_config . instance_id = '<STR_LIT>' <EOL> self . runtime_config . auth_domain = '<STR_LIT>' <EOL> self . runtime_config_getter = lambda : self . runtime_config <EOL> self . proxy = http_runtime . HttpRuntimeProxy ( <EOL> [ '<STR_LIT>' ] , self . runtime_config_getter , module_configuration , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . proxy . _port = <NUM_LIT> <EOL> self . process = self . mox . CreateMock ( subprocess . Popen ) <EOL> self . process . stdin = self . mox . CreateMockAnything ( ) <EOL> self . process . stdout = self . mox . CreateMockAnything ( ) <EOL> self . process . stderr = self . mox . CreateMockAnything ( ) <EOL> self . mox . StubOutWithMock ( safe_subprocess , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( login , '<STR_LIT>' ) <EOL> self . url_map = appinfo . URLMap ( url = r'<STR_LIT>' , <EOL> script = r'<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( http_proxy . HttpProxy , '<STR_LIT>' ) <EOL> http_proxy . HttpProxy . wait_for_connection ( ) <EOL> self . _saved_quit_with_sigterm = None <EOL> def tearDown ( self ) : <EOL> shutil . rmtree ( self . tmpdir ) <EOL> self . mox . UnsetStubs ( ) <EOL> if self . _saved_quit_with_sigterm is not None : <EOL> http_runtime . HttpRuntimeProxy . stop_runtimes_with_sigterm ( <EOL> self . _saved_quit_with_sigterm ) <EOL> def _test_start_and_quit ( self , quit_with_sigterm ) : <EOL> self . _saved_quit_with_sigterm = ( <EOL> http_runtime . HttpRuntimeProxy . stop_runtimes_with_sigterm ( <EOL> quit_with_sigterm ) ) <EOL> safe_subprocess . start_process ( <EOL> [ '<STR_LIT>' ] , <EOL> base64 . b64encode ( self . runtime_config . SerializeToString ( ) ) , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> cwd = self . tmpdir ) . AndReturn ( self . process ) <EOL> self . process . stdout . readline ( ) . AndReturn ( '<STR_LIT>' ) <EOL> self . proxy . _stderr_tee = FakeTee ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> self . assertEquals ( <NUM_LIT> , self . proxy . _proxy . _port ) <EOL> self . assertEquals ( '<STR_LIT:localhost>' , self . proxy . _proxy . _host ) <EOL> self . mox . VerifyAll ( ) <EOL> self . mox . ResetAll ( ) <EOL> if quit_with_sigterm : <EOL> self . process . terminate ( ) <EOL> else : <EOL> self . process . kill ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . quit ( ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_start_and_quit ( self ) : <EOL> self . _test_start_and_quit ( quit_with_sigterm = False ) <EOL> def test_start_and_quit_with_sigterm ( self ) : <EOL> self . _test_start_and_quit ( quit_with_sigterm = True ) <EOL> def test_start_with_host_port ( self ) : <EOL> safe_subprocess . start_process ( <EOL> [ '<STR_LIT>' ] , <EOL> base64 . b64encode ( self . runtime_config . SerializeToString ( ) ) , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> cwd = self . tmpdir ) . AndReturn ( self . process ) <EOL> self . process . stdout . readline ( ) . AndReturn ( '<STR_LIT>' ) <EOL> self . proxy . _stderr_tee = FakeTee ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> self . assertEquals ( <NUM_LIT> , self . proxy . _proxy . _port ) <EOL> self . assertEquals ( '<STR_LIT>' , self . proxy . _proxy . _host ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_start_bad_port ( self ) : <EOL> safe_subprocess . start_process ( <EOL> [ '<STR_LIT>' ] , <EOL> base64 . b64encode ( self . runtime_config . SerializeToString ( ) ) , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> cwd = self . tmpdir ) . AndReturn ( self . process ) <EOL> self . process . stdout . readline ( ) . AndReturn ( '<STR_LIT>' ) <EOL> header = "<STR_LIT>" <EOL> stderr0 = "<STR_LIT>" <EOL> stderr1 = "<STR_LIT>" <EOL> self . proxy . _stderr_tee = FakeTee ( stderr0 + stderr1 ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> expected_headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( len ( header ) + len ( stderr0 ) + len ( stderr1 ) ) , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , <EOL> header + stderr0 + stderr1 , <EOL> self . proxy . handle , { } , <EOL> url_map = self . url_map , <EOL> match = re . match ( self . url_map . url , '<STR_LIT>' ) , <EOL> request_id = '<STR_LIT>' , <EOL> request_type = instance . NORMAL_REQUEST ) <EOL> self . mox . VerifyAll ( ) <EOL> class HttpRuntimeProxyFileFlavorTest ( wsgi_test_utils . WSGITestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . tmpdir = tempfile . mkdtemp ( ) <EOL> module_configuration = ModuleConfigurationStub ( application_root = self . tmpdir ) <EOL> self . runtime_config = runtime_config_pb2 . Config ( ) <EOL> self . runtime_config . app_id = '<STR_LIT>' <EOL> self . runtime_config . version_id = '<STR_LIT:version>' <EOL> self . runtime_config . api_port = <NUM_LIT> <EOL> self . runtime_config . application_root = self . tmpdir <EOL> self . runtime_config . datacenter = '<STR_LIT>' <EOL> self . runtime_config . instance_id = '<STR_LIT>' <EOL> self . runtime_config . auth_domain = '<STR_LIT>' <EOL> self . runtime_config_getter = lambda : self . runtime_config <EOL> self . proxy = http_runtime . HttpRuntimeProxy ( <EOL> [ '<STR_LIT>' ] , self . runtime_config_getter , module_configuration , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> start_process_flavor = http_runtime . START_PROCESS_FILE ) <EOL> self . mox . StubOutWithMock ( self . proxy , '<STR_LIT>' ) <EOL> self . process = self . mox . CreateMock ( subprocess . Popen ) <EOL> self . process . stdin = self . mox . CreateMockAnything ( ) <EOL> self . process . stdout = self . mox . CreateMockAnything ( ) <EOL> self . process . stderr = self . mox . CreateMockAnything ( ) <EOL> self . process . child_out = self . mox . CreateMockAnything ( ) <EOL> self . mox . StubOutWithMock ( safe_subprocess , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( time , '<STR_LIT>' ) <EOL> self . url_map = appinfo . URLMap ( url = r'<STR_LIT>' , <EOL> script = r'<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( http_proxy . HttpProxy , '<STR_LIT>' ) <EOL> http_proxy . HttpProxy . wait_for_connection ( ) <EOL> def tearDown ( self ) : <EOL> shutil . rmtree ( self . tmpdir ) <EOL> self . mox . UnsetStubs ( ) <EOL> def test_basic ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . proxy . _process_lock : <EOL> safe_subprocess . start_process_file ( <EOL> args = [ '<STR_LIT>' ] , <EOL> input_string = self . runtime_config . SerializeToString ( ) , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> cwd = self . tmpdir , <EOL> stderr = subprocess . PIPE ) . AndReturn ( self . process ) <EOL> self . process . poll ( ) . AndReturn ( None ) <EOL> self . process . child_out . seek ( <NUM_LIT:0> ) . AndReturn ( None ) <EOL> self . process . child_out . read ( ) . AndReturn ( '<STR_LIT>' ) <EOL> self . process . child_out . close ( ) . AndReturn ( None ) <EOL> self . process . child_out . name = '<STR_LIT>' <EOL> os . remove ( '<STR_LIT>' ) . AndReturn ( None ) <EOL> self . proxy . _stderr_tee = FakeTee ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> self . assertEquals ( <NUM_LIT> , self . proxy . _proxy . _port ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_slow_shattered ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . proxy . _process_lock : <EOL> safe_subprocess . start_process_file ( <EOL> args = [ '<STR_LIT>' ] , <EOL> input_string = self . runtime_config . SerializeToString ( ) , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> cwd = self . tmpdir , <EOL> stderr = subprocess . PIPE ) . AndReturn ( self . process ) <EOL> for response , sleeptime in [ <EOL> ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , None ) ] : <EOL> self . process . poll ( ) . AndReturn ( None ) <EOL> self . process . child_out . seek ( <NUM_LIT:0> ) . AndReturn ( None ) <EOL> self . process . child_out . read ( ) . AndReturn ( response ) <EOL> if sleeptime is not None : <EOL> time . sleep ( sleeptime ) . AndReturn ( None ) <EOL> self . process . child_out . close ( ) . AndReturn ( None ) <EOL> self . process . child_out . name = '<STR_LIT>' <EOL> os . remove ( '<STR_LIT>' ) . AndReturn ( None ) <EOL> self . proxy . _stderr_tee = FakeTee ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> self . assertEquals ( <NUM_LIT> , self . proxy . _proxy . _port ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_runtime_instance_dies_immediately ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . proxy . _process_lock : <EOL> safe_subprocess . start_process_file ( <EOL> args = [ '<STR_LIT>' ] , <EOL> input_string = self . runtime_config . SerializeToString ( ) , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> cwd = self . tmpdir , <EOL> stderr = subprocess . PIPE ) . AndReturn ( self . process ) <EOL> self . process . poll ( ) . AndReturn ( <NUM_LIT:1> ) <EOL> self . process . child_out . close ( ) . AndReturn ( None ) <EOL> self . process . child_out . name = '<STR_LIT>' <EOL> os . remove ( '<STR_LIT>' ) . AndReturn ( None ) <EOL> header = "<STR_LIT>" <EOL> stderr0 = '<STR_LIT>' <EOL> self . proxy . _stderr_tee = FakeTee ( stderr0 ) <EOL> time . sleep ( <NUM_LIT> ) . AndReturn ( None ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> expected_headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( len ( header ) + len ( stderr0 ) ) , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , <EOL> header + stderr0 , <EOL> self . proxy . handle , { } , <EOL> url_map = self . url_map , <EOL> match = re . match ( self . url_map . url , '<STR_LIT>' ) , <EOL> request_id = '<STR_LIT>' , <EOL> request_type = instance . NORMAL_REQUEST ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_runtime_instance_invalid_response ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . proxy . _process_lock : <EOL> safe_subprocess . start_process_file ( <EOL> args = [ '<STR_LIT>' ] , <EOL> input_string = self . runtime_config . SerializeToString ( ) , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> cwd = self . tmpdir , <EOL> stderr = subprocess . PIPE ) . AndReturn ( self . process ) <EOL> for response , sleeptime in [ <EOL> ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT:1.0> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , None ) ] : <EOL> self . process . poll ( ) . AndReturn ( None ) <EOL> self . process . child_out . seek ( <NUM_LIT:0> ) . AndReturn ( None ) <EOL> self . process . child_out . read ( ) . AndReturn ( response ) <EOL> if sleeptime is not None : <EOL> time . sleep ( sleeptime ) . AndReturn ( None ) <EOL> self . process . child_out . close ( ) . AndReturn ( None ) <EOL> self . process . child_out . name = '<STR_LIT>' <EOL> os . remove ( '<STR_LIT>' ) . AndReturn ( None ) <EOL> header = "<STR_LIT>" <EOL> stderr0 = '<STR_LIT>' <EOL> self . proxy . _stderr_tee = FakeTee ( stderr0 ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> expected_headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( len ( header ) + len ( stderr0 ) ) , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , <EOL> header + stderr0 , <EOL> self . proxy . handle , { } , <EOL> url_map = self . url_map , <EOL> match = re . match ( self . url_map . url , '<STR_LIT>' ) , <EOL> request_id = '<STR_LIT>' , <EOL> request_type = instance . NORMAL_REQUEST ) <EOL> self . mox . VerifyAll ( ) <EOL> class HttpRuntimeProxyReverseFlavorTest ( wsgi_test_utils . WSGITestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . tmpdir = tempfile . mkdtemp ( ) <EOL> module_configuration = ModuleConfigurationStub ( application_root = self . tmpdir ) <EOL> self . runtime_config = runtime_config_pb2 . Config ( ) <EOL> self . runtime_config . app_id = '<STR_LIT>' <EOL> self . runtime_config . version_id = '<STR_LIT:version>' <EOL> self . runtime_config . api_port = <NUM_LIT> <EOL> self . runtime_config . application_root = self . tmpdir <EOL> self . runtime_config . datacenter = '<STR_LIT>' <EOL> self . runtime_config . instance_id = '<STR_LIT>' <EOL> self . runtime_config . auth_domain = '<STR_LIT>' <EOL> self . runtime_config_getter = lambda : self . runtime_config <EOL> self . proxy = http_runtime . HttpRuntimeProxy ( <EOL> [ '<STR_LIT>' ] , self . runtime_config_getter , module_configuration , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> start_process_flavor = http_runtime . START_PROCESS_REVERSE ) <EOL> self . mox . StubOutWithMock ( self . proxy , '<STR_LIT>' ) <EOL> self . process = self . mox . CreateMock ( subprocess . Popen ) <EOL> self . process . stdin = self . mox . CreateMockAnything ( ) <EOL> self . mox . StubOutWithMock ( safe_subprocess , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( time , '<STR_LIT>' ) <EOL> self . url_map = appinfo . URLMap ( url = r'<STR_LIT>' , <EOL> script = r'<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( http_proxy . HttpProxy , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( portpicker , '<STR_LIT>' ) <EOL> http_proxy . HttpProxy . wait_for_connection ( ) <EOL> def tearDown ( self ) : <EOL> shutil . rmtree ( self . tmpdir ) <EOL> self . mox . UnsetStubs ( ) <EOL> def test_basic ( self ) : <EOL> """<STR_LIT>""" <EOL> portpicker . PickUnusedPort ( ) . AndReturn ( <NUM_LIT> ) <EOL> with self . proxy . _process_lock : <EOL> safe_subprocess . start_process_file ( <EOL> args = [ '<STR_LIT>' ] , <EOL> input_string = self . runtime_config . SerializeToString ( ) , <EOL> env = { '<STR_LIT:foo>' : '<STR_LIT:bar>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> cwd = self . tmpdir , <EOL> stderr = subprocess . PIPE ) . AndReturn ( self . process ) <EOL> self . proxy . _stderr_tee = FakeTee ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> self . proxy . start ( ) <EOL> self . assertEquals ( <NUM_LIT> , self . proxy . _proxy . _port ) <EOL> self . mox . VerifyAll ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
