<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import errno <EOL> import inspect <EOL> import io <EOL> import os <EOL> import socket <EOL> import time <EOL> import unittest <EOL> import mock <EOL> from pants . java . nailgun_io import NailgunStreamReader , NailgunStreamWriter <EOL> from pants . java . nailgun_protocol import ChunkType , NailgunProtocol <EOL> PATCH_OPTS = dict ( autospec = True , spec_set = True ) <EOL> class FakeFile ( object ) : <EOL> def __init__ ( self ) : <EOL> self . content = b'<STR_LIT>' <EOL> def write ( self , val ) : <EOL> self . content += val <EOL> def fileno ( self ) : <EOL> return - <NUM_LIT:1> <EOL> def flush ( self ) : <EOL> return <EOL> class TestNailgunStreamReader ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . in_fd = FakeFile ( ) <EOL> self . mock_socket = mock . Mock ( ) <EOL> self . reader = NailgunStreamReader ( in_fd = self . in_fd , sock = self . mock_socket ) <EOL> def test_stop ( self ) : <EOL> self . assertFalse ( self . reader . is_stopped ) <EOL> self . reader . stop ( ) <EOL> self . assertTrue ( self . reader . is_stopped ) <EOL> self . reader . run ( ) <EOL> def test_startable ( self ) : <EOL> self . assertTrue ( inspect . ismethod ( self . reader . start ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_stop_on_error ( self , mock_select ) : <EOL> mock_select . return_value = ( [ ] , [ ] , [ self . in_fd ] ) <EOL> self . reader . run ( ) <EOL> self . assertTrue ( self . reader . is_stopped ) <EOL> self . assertEquals ( mock_select . call_count , <NUM_LIT:1> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( NailgunProtocol , '<STR_LIT>' ) <EOL> def test_run_read_write ( self , mock_writer , mock_select , mock_read ) : <EOL> mock_select . side_effect = [ <EOL> ( [ self . in_fd ] , [ ] , [ ] ) , <EOL> ( [ self . in_fd ] , [ ] , [ ] ) <EOL> ] <EOL> mock_read . side_effect = [ <EOL> b'<STR_LIT:A>' * <NUM_LIT> , <EOL> b'<STR_LIT>' <EOL> ] <EOL> with self . reader . running ( ) : <EOL> while not self . reader . is_stopped : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertTrue ( self . reader . is_stopped ) <EOL> mock_read . assert_called_with ( - <NUM_LIT:1> , io . DEFAULT_BUFFER_SIZE ) <EOL> self . assertEquals ( mock_read . call_count , <NUM_LIT:2> ) <EOL> self . mock_socket . shutdown . assert_called_once_with ( socket . SHUT_WR ) <EOL> mock_writer . assert_has_calls ( [ <EOL> mock . call ( mock . ANY , ChunkType . STDIN , b'<STR_LIT:A>' * <NUM_LIT> ) , <EOL> mock . call ( mock . ANY , ChunkType . STDIN_EOF ) <EOL> ] ) <EOL> class TestNailgunStreamWriter ( unittest . TestCase ) : <EOL> TEST_VALUE = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> self . chunk_type = ChunkType . STDERR <EOL> self . mock_socket = mock . Mock ( ) <EOL> self . writer = NailgunStreamWriter ( self . mock_socket , self . chunk_type ) <EOL> @ mock . patch . object ( NailgunProtocol , '<STR_LIT>' ) <EOL> def test_write ( self , mock_writer ) : <EOL> self . writer . write ( self . TEST_VALUE ) <EOL> mock_writer . assert_called_once_with ( self . mock_socket , self . chunk_type , self . TEST_VALUE ) <EOL> @ mock . patch . object ( NailgunProtocol , '<STR_LIT>' ) <EOL> def test_write_broken_pipe_unmasked ( self , mock_writer ) : <EOL> mock_writer . side_effect = IOError ( errno . EPIPE , os . strerror ( errno . EPIPE ) ) <EOL> with self . assertRaises ( IOError ) : <EOL> self . writer . write ( self . TEST_VALUE ) <EOL> @ mock . patch . object ( NailgunProtocol , '<STR_LIT>' ) <EOL> def test_write_broken_pipe_masked ( self , mock_writer ) : <EOL> self . writer = NailgunStreamWriter ( self . mock_socket , self . chunk_type , mask_broken_pipe = True ) <EOL> mock_writer . side_effect = IOError ( errno . EPIPE , os . strerror ( errno . EPIPE ) ) <EOL> self . writer . write ( self . TEST_VALUE ) <EOL> def test_isatty ( self ) : <EOL> self . assertTrue ( self . writer . isatty ( ) ) <EOL> def test_not_isatty ( self ) : <EOL> self . writer = NailgunStreamWriter ( self . mock_socket , self . chunk_type , isatty = False ) <EOL> self . assertFalse ( self . writer . isatty ( ) ) <EOL> def test_misc ( self ) : <EOL> self . writer . flush ( ) </s>
