<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import socket <EOL> import unittest <EOL> from SocketServer import TCPServer <EOL> import mock <EOL> from pants . java . nailgun_protocol import ChunkType , NailgunProtocol <EOL> from pants . pantsd . pailgun_server import PailgunHandler , PailgunServer <EOL> PATCH_OPTS = dict ( autospec = True , spec_set = True ) <EOL> class TestPailgunServer ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mock_handler_inst = mock . Mock ( ) <EOL> self . mock_runner_factory = mock . Mock ( side_effect = Exception ( '<STR_LIT>' ) ) <EOL> self . mock_handler_class = mock . Mock ( return_value = self . mock_handler_inst ) <EOL> with mock . patch . object ( PailgunServer , '<STR_LIT>' ) , mock . patch . object ( PailgunServer , '<STR_LIT>' ) : <EOL> self . server = PailgunServer ( <EOL> server_address = ( '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> runner_factory = self . mock_runner_factory , <EOL> handler_class = self . mock_handler_class <EOL> ) <EOL> @ mock . patch . object ( TCPServer , '<STR_LIT>' , ** PATCH_OPTS ) <EOL> def test_server_bind ( self , mock_tcpserver_bind ) : <EOL> mock_sock = mock . Mock ( ) <EOL> mock_sock . getsockname . return_value = ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . server . socket = mock_sock <EOL> self . server . server_bind ( ) <EOL> self . assertEquals ( self . server . server_port , <NUM_LIT> ) <EOL> self . assertIs ( mock_tcpserver_bind . called , True ) <EOL> @ mock . patch . object ( PailgunServer , '<STR_LIT>' , ** PATCH_OPTS ) <EOL> def test_process_request ( self , mock_close_request ) : <EOL> mock_request = mock . Mock ( ) <EOL> self . server . process_request ( mock_request , ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> self . assertIs ( self . mock_handler_inst . handle_request . called , True ) <EOL> mock_close_request . assert_called_once_with ( self . server , mock_request ) <EOL> @ mock . patch . object ( PailgunServer , '<STR_LIT>' , ** PATCH_OPTS ) <EOL> def test_process_request_error ( self , mock_shutdown_request ) : <EOL> mock_request = mock . Mock ( ) <EOL> self . mock_handler_inst . handle_request . side_effect = AttributeError ( '<STR_LIT>' ) <EOL> self . server . process_request ( mock_request , ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> self . assertIs ( self . mock_handler_inst . handle_request . called , True ) <EOL> self . assertIs ( self . mock_handler_inst . handle_error . called , True ) <EOL> mock_shutdown_request . assert_called_once_with ( self . server , mock_request ) <EOL> class TestPailgunHandler ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . client_sock , self . server_sock = socket . socketpair ( ) <EOL> self . mock_socket = mock . Mock ( ) <EOL> self . mock_server = mock . create_autospec ( PailgunServer , spec_set = True ) <EOL> self . handler = PailgunHandler ( <EOL> self . server_sock , <EOL> self . client_sock . getsockname ( ) [ : <NUM_LIT:2> ] , <EOL> self . mock_server <EOL> ) <EOL> def test_handle_error ( self ) : <EOL> self . handler . handle_error ( ) <EOL> last_chunk_type , last_payload = list ( NailgunProtocol . iter_chunks ( self . client_sock ) ) [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( last_chunk_type , ChunkType . EXIT ) <EOL> self . assertEquals ( bytes ( last_payload ) , '<STR_LIT:1>' ) <EOL> @ mock . patch . object ( PailgunHandler , '<STR_LIT>' , ** PATCH_OPTS ) <EOL> def test_handle_request ( self , mock_run_pants ) : <EOL> NailgunProtocol . send_request ( self . client_sock , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . handler . handle_request ( ) <EOL> self . assertIs ( mock_run_pants . called , True ) </s>
