<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import copy <EOL> import shutil <EOL> import time <EOL> import json <EOL> import random <EOL> import logging <EOL> import tempfile <EOL> import platform <EOL> import contextlib <EOL> import string <EOL> from pprint import pformat <EOL> from subprocess import Popen , PIPE <EOL> from awscli . compat import StringIO <EOL> try : <EOL> import mock <EOL> except ImportError as e : <EOL> mock = None <EOL> from awscli . compat import six <EOL> from botocore . hooks import HierarchicalEmitter <EOL> from botocore . session import Session <EOL> from botocore . exceptions import ClientError <EOL> import botocore . loaders <EOL> from botocore . vendored import requests <EOL> import awscli . clidriver <EOL> from awscli . plugin import load_plugins <EOL> from awscli . clidriver import CLIDriver <EOL> from awscli import EnvironmentVariables <EOL> if sys . version_info [ : <NUM_LIT:2> ] == ( <NUM_LIT:2> , <NUM_LIT:6> ) : <EOL> import unittest2 as unittest <EOL> else : <EOL> import unittest <EOL> if six . PY2 : <EOL> unittest . TestCase . assertCountEqual = unittest . TestCase . assertItemsEqual <EOL> _LOADER = botocore . loaders . Loader ( ) <EOL> INTEG_LOG = logging . getLogger ( '<STR_LIT>' ) <EOL> AWS_CMD = None <EOL> def skip_if_windows ( reason ) : <EOL> """<STR_LIT>""" <EOL> def decorator ( func ) : <EOL> return unittest . skipIf ( <EOL> platform . system ( ) not in [ '<STR_LIT>' , '<STR_LIT>' ] , reason ) ( func ) <EOL> return decorator <EOL> def create_clidriver ( ) : <EOL> driver = awscli . clidriver . create_clidriver ( ) <EOL> session = driver . session <EOL> data_path = session . get_config_variable ( '<STR_LIT>' ) . split ( os . pathsep ) <EOL> if not data_path : <EOL> data_path = [ ] <EOL> _LOADER . search_paths . extend ( data_path ) <EOL> session . register_component ( '<STR_LIT>' , _LOADER ) <EOL> return driver <EOL> def get_aws_cmd ( ) : <EOL> global AWS_CMD <EOL> import awscli <EOL> if AWS_CMD is None : <EOL> repo_root = os . path . dirname ( os . path . abspath ( awscli . __file__ ) ) <EOL> aws_cmd = os . path . join ( repo_root , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if not os . path . isfile ( aws_cmd ) : <EOL> aws_cmd = _search_path_for_cmd ( '<STR_LIT>' ) <EOL> if aws_cmd is None : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> AWS_CMD = aws_cmd <EOL> return AWS_CMD <EOL> def _search_path_for_cmd ( cmd_name ) : <EOL> for path in os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) . split ( os . pathsep ) : <EOL> full_cmd_path = os . path . join ( path , cmd_name ) <EOL> if os . path . isfile ( full_cmd_path ) : <EOL> return full_cmd_path <EOL> return None <EOL> def set_aws_cmd ( aws_cmd ) : <EOL> global AWS_CMD <EOL> AWS_CMD = aws_cmd <EOL> @ contextlib . contextmanager <EOL> def temporary_file ( mode ) : <EOL> """<STR_LIT>""" <EOL> temporary_directory = tempfile . mkdtemp ( ) <EOL> basename = '<STR_LIT>' % ( int ( time . time ( ) ) , random . randint ( <NUM_LIT:1> , <NUM_LIT:1000> ) ) <EOL> full_filename = os . path . join ( temporary_directory , basename ) <EOL> open ( full_filename , '<STR_LIT:w>' ) . close ( ) <EOL> try : <EOL> with open ( full_filename , mode ) as f : <EOL> yield f <EOL> finally : <EOL> shutil . rmtree ( temporary_directory ) <EOL> def create_bucket ( session , name = None , region = None ) : <EOL> """<STR_LIT>""" <EOL> if not region : <EOL> region = '<STR_LIT>' <EOL> client = session . create_client ( '<STR_LIT>' , region_name = region ) <EOL> if name : <EOL> bucket_name = name <EOL> else : <EOL> rand1 = '<STR_LIT>' . join ( random . sample ( string . ascii_lowercase + string . digits , <EOL> <NUM_LIT:10> ) ) <EOL> bucket_name = '<STR_LIT>' + str ( rand1 ) <EOL> params = { '<STR_LIT>' : bucket_name } <EOL> if region != '<STR_LIT>' : <EOL> params [ '<STR_LIT>' ] = { '<STR_LIT>' : region } <EOL> try : <EOL> client . meta . events . unregister ( <EOL> '<STR_LIT>' , unique_id = '<STR_LIT>' ) <EOL> client . create_bucket ( ** params ) <EOL> except ClientError as e : <EOL> if e . response [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> pass <EOL> else : <EOL> raise <EOL> return bucket_name <EOL> class BaseCLIDriverTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . environ = { <EOL> '<STR_LIT>' : os . environ [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . environ_patch = mock . patch ( '<STR_LIT>' , self . environ ) <EOL> self . environ_patch . start ( ) <EOL> self . driver = create_clidriver ( ) <EOL> self . session = self . driver . session <EOL> def tearDown ( self ) : <EOL> self . environ_patch . stop ( ) <EOL> class BaseAWSHelpOutputTest ( BaseCLIDriverTest ) : <EOL> def setUp ( self ) : <EOL> super ( BaseAWSHelpOutputTest , self ) . setUp ( ) <EOL> self . renderer_patch = mock . patch ( '<STR_LIT>' ) <EOL> self . renderer_mock = self . renderer_patch . start ( ) <EOL> self . renderer = CapturedRenderer ( ) <EOL> self . renderer_mock . return_value = self . renderer <EOL> def tearDown ( self ) : <EOL> super ( BaseAWSHelpOutputTest , self ) . tearDown ( ) <EOL> self . renderer_patch . stop ( ) <EOL> def assert_contains ( self , contains ) : <EOL> if contains not in self . renderer . rendered_contents : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> contains , self . renderer . rendered_contents ) ) <EOL> def assert_contains_with_count ( self , contains , count ) : <EOL> r_count = self . renderer . rendered_contents . count ( contains ) <EOL> if r_count != count : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> contains , count , self . renderer . rendered_contents , r_count ) ) <EOL> def assert_not_contains ( self , contents ) : <EOL> if contents in self . renderer . rendered_contents : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> contents , self . renderer . rendered_contents ) ) <EOL> def assert_text_order ( self , * args , ** kwargs ) : <EOL> starting_from = kwargs . pop ( '<STR_LIT>' ) <EOL> args = list ( args ) <EOL> contents = self . renderer . rendered_contents <EOL> self . assertIn ( starting_from , contents ) <EOL> start_index = contents . find ( starting_from ) <EOL> arg_indices = [ contents . find ( arg , start_index ) for arg in args ] <EOL> previous = arg_indices [ <NUM_LIT:0> ] <EOL> for i , index in enumerate ( arg_indices [ <NUM_LIT:1> : ] , <NUM_LIT:1> ) : <EOL> if index == - <NUM_LIT:1> : <EOL> self . fail ( '<STR_LIT>' <EOL> % ( args [ index ] , contents ) ) <EOL> if index < previous : <EOL> self . fail ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( args [ i ] , args [ i - <NUM_LIT:1> ] , contents ) ) <EOL> previous = index <EOL> class CapturedRenderer ( object ) : <EOL> def __init__ ( self ) : <EOL> self . rendered_contents = '<STR_LIT>' <EOL> def render ( self , contents ) : <EOL> self . rendered_contents = contents . decode ( '<STR_LIT:utf-8>' ) <EOL> class CapturedOutput ( object ) : <EOL> def __init__ ( self , stdout , stderr ) : <EOL> self . stdout = stdout <EOL> self . stderr = stderr <EOL> @ contextlib . contextmanager <EOL> def capture_output ( ) : <EOL> stderr = six . StringIO ( ) <EOL> stdout = six . StringIO ( ) <EOL> with mock . patch ( '<STR_LIT>' , stderr ) : <EOL> with mock . patch ( '<STR_LIT>' , stdout ) : <EOL> yield CapturedOutput ( stdout , stderr ) <EOL> class BaseAWSCommandParamsTest ( unittest . TestCase ) : <EOL> maxDiff = None <EOL> def setUp ( self ) : <EOL> self . last_params = { } <EOL> self . environ = { <EOL> '<STR_LIT>' : os . environ [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . environ_patch = mock . patch ( '<STR_LIT>' , self . environ ) <EOL> self . environ_patch . start ( ) <EOL> self . http_response = requests . models . Response ( ) <EOL> self . http_response . status_code = <NUM_LIT:200> <EOL> self . parsed_response = { } <EOL> self . make_request_patch = mock . patch ( '<STR_LIT>' ) <EOL> self . make_request_is_patched = False <EOL> self . operations_called = [ ] <EOL> self . parsed_responses = None <EOL> self . driver = create_clidriver ( ) <EOL> def tearDown ( self ) : <EOL> self . environ_patch . stop ( ) <EOL> if self . make_request_is_patched : <EOL> self . make_request_patch . stop ( ) <EOL> self . make_request_is_patched = False <EOL> def before_call ( self , params , ** kwargs ) : <EOL> self . _store_params ( params ) <EOL> def _store_params ( self , params ) : <EOL> self . last_request_dict = params <EOL> self . last_params = params [ '<STR_LIT:body>' ] <EOL> def patch_make_request ( self ) : <EOL> if self . make_request_is_patched : <EOL> self . make_request_patch . stop ( ) <EOL> self . make_request_is_patched = False <EOL> make_request_patch = self . make_request_patch . start ( ) <EOL> if self . parsed_responses is not None : <EOL> make_request_patch . side_effect = lambda * args , ** kwargs : ( self . http_response , self . parsed_responses . pop ( <NUM_LIT:0> ) ) <EOL> else : <EOL> make_request_patch . return_value = ( self . http_response , self . parsed_response ) <EOL> self . make_request_is_patched = True <EOL> def assert_params_for_cmd ( self , cmd , params = None , expected_rc = <NUM_LIT:0> , <EOL> stderr_contains = None , ignore_params = None ) : <EOL> stdout , stderr , rc = self . run_cmd ( cmd , expected_rc ) <EOL> if stderr_contains is not None : <EOL> self . assertIn ( stderr_contains , stderr ) <EOL> if params is not None : <EOL> last_kwargs = copy . copy ( self . last_kwargs ) <EOL> if ignore_params is not None : <EOL> for key in ignore_params : <EOL> try : <EOL> del last_kwargs [ key ] <EOL> except KeyError : <EOL> pass <EOL> if params != last_kwargs : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> pformat ( params ) , pformat ( last_kwargs ) ) ) <EOL> return stdout , stderr , rc <EOL> def before_parameter_build ( self , params , model , ** kwargs ) : <EOL> self . last_kwargs = params <EOL> self . operations_called . append ( ( model , params ) ) <EOL> def run_cmd ( self , cmd , expected_rc = <NUM_LIT:0> ) : <EOL> logging . debug ( "<STR_LIT>" , cmd ) <EOL> self . patch_make_request ( ) <EOL> self . driver . session . register ( '<STR_LIT>' , self . before_call ) <EOL> self . driver . session . register ( '<STR_LIT>' , <EOL> self . before_parameter_build ) <EOL> if not isinstance ( cmd , list ) : <EOL> cmdlist = cmd . split ( ) <EOL> else : <EOL> cmdlist = cmd <EOL> with capture_output ( ) as captured : <EOL> try : <EOL> rc = self . driver . main ( cmdlist ) <EOL> except SystemExit as e : <EOL> rc = e . code <EOL> stderr = captured . stderr . getvalue ( ) <EOL> stdout = captured . stdout . getvalue ( ) <EOL> self . assertEqual ( <EOL> rc , expected_rc , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> expected_rc , rc , cmd , stdout , stderr ) ) <EOL> return stdout , stderr , rc <EOL> class BaseAWSPreviewCommandParamsTest ( BaseAWSCommandParamsTest ) : <EOL> def setUp ( self ) : <EOL> self . preview_patch = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . preview_patch . start ( ) <EOL> super ( BaseAWSPreviewCommandParamsTest , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> self . preview_patch . stop ( ) <EOL> super ( BaseAWSPreviewCommandParamsTest , self ) . tearDown ( ) <EOL> class FileCreator ( object ) : <EOL> def __init__ ( self ) : <EOL> self . rootdir = tempfile . mkdtemp ( ) <EOL> def remove_all ( self ) : <EOL> shutil . rmtree ( self . rootdir ) <EOL> def create_file ( self , filename , contents , mtime = None , mode = '<STR_LIT:w>' ) : <EOL> """<STR_LIT>""" <EOL> full_path = os . path . join ( self . rootdir , filename ) <EOL> if not os . path . isdir ( os . path . dirname ( full_path ) ) : <EOL> os . makedirs ( os . path . dirname ( full_path ) ) <EOL> with open ( full_path , mode ) as f : <EOL> f . write ( contents ) <EOL> current_time = os . path . getmtime ( full_path ) <EOL> os . utime ( full_path , ( current_time , current_time - <NUM_LIT> ) ) <EOL> if mtime is not None : <EOL> os . utime ( full_path , ( mtime , mtime ) ) <EOL> return full_path <EOL> def append_file ( self , filename , contents ) : <EOL> """<STR_LIT>""" <EOL> full_path = os . path . join ( self . rootdir , filename ) <EOL> if not os . path . isdir ( os . path . dirname ( full_path ) ) : <EOL> os . makedirs ( os . path . dirname ( full_path ) ) <EOL> with open ( full_path , '<STR_LIT:a>' ) as f : <EOL> f . write ( contents ) <EOL> return full_path <EOL> def full_path ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> return os . path . join ( self . rootdir , filename ) <EOL> class ProcessTerminatedError ( Exception ) : <EOL> pass <EOL> class Result ( object ) : <EOL> def __init__ ( self , rc , stdout , stderr , memory_usage = None ) : <EOL> self . rc = rc <EOL> self . stdout = stdout <EOL> self . stderr = stderr <EOL> INTEG_LOG . debug ( "<STR_LIT>" , rc ) <EOL> INTEG_LOG . debug ( "<STR_LIT>" , stdout ) <EOL> INTEG_LOG . debug ( "<STR_LIT>" , stderr ) <EOL> if memory_usage is None : <EOL> memory_usage = [ ] <EOL> self . memory_usage = memory_usage <EOL> @ property <EOL> def json ( self ) : <EOL> return json . loads ( self . stdout ) <EOL> def _escape_quotes ( command ) : <EOL> command = command . replace ( '<STR_LIT:">' , '<STR_LIT>' ) <EOL> command = command . replace ( "<STR_LIT:'>" , '<STR_LIT:">' ) <EOL> return command <EOL> def aws ( command , collect_memory = False , env_vars = None , <EOL> wait_for_finish = True , input_data = None , input_file = None ) : <EOL> """<STR_LIT>""" <EOL> if platform . system ( ) == '<STR_LIT>' : <EOL> command = _escape_quotes ( command ) <EOL> if '<STR_LIT>' in os . environ : <EOL> aws_command = os . environ [ '<STR_LIT>' ] <EOL> else : <EOL> aws_command = '<STR_LIT>' % get_aws_cmd ( ) <EOL> full_command = '<STR_LIT>' % ( aws_command , command ) <EOL> stdout_encoding = get_stdout_encoding ( ) <EOL> if isinstance ( full_command , six . text_type ) and not six . PY3 : <EOL> full_command = full_command . encode ( stdout_encoding ) <EOL> INTEG_LOG . debug ( "<STR_LIT>" , full_command ) <EOL> env = os . environ . copy ( ) <EOL> env [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> if env_vars is not None : <EOL> env = env_vars <EOL> if input_file is None : <EOL> input_file = PIPE <EOL> process = Popen ( full_command , stdout = PIPE , stderr = PIPE , stdin = input_file , <EOL> shell = True , env = env ) <EOL> if not wait_for_finish : <EOL> return process <EOL> memory = None <EOL> if not collect_memory : <EOL> kwargs = { } <EOL> if input_data : <EOL> kwargs = { '<STR_LIT:input>' : input_data } <EOL> stdout , stderr = process . communicate ( ** kwargs ) <EOL> else : <EOL> stdout , stderr , memory = _wait_and_collect_mem ( process ) <EOL> return Result ( process . returncode , <EOL> stdout . decode ( stdout_encoding ) , <EOL> stderr . decode ( stdout_encoding ) , <EOL> memory ) <EOL> def get_stdout_encoding ( ) : <EOL> encoding = getattr ( sys . __stdout__ , '<STR_LIT>' , None ) <EOL> if encoding is None : <EOL> encoding = '<STR_LIT:utf-8>' <EOL> return encoding <EOL> def _wait_and_collect_mem ( process ) : <EOL> if platform . system ( ) == '<STR_LIT>' : <EOL> get_memory = _get_memory_with_ps <EOL> elif platform . system ( ) == '<STR_LIT>' : <EOL> get_memory = _get_memory_with_ps <EOL> else : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> platform . system ( ) ) <EOL> memory = [ ] <EOL> while process . poll ( ) is None : <EOL> try : <EOL> current = get_memory ( process . pid ) <EOL> except ProcessTerminatedError : <EOL> break <EOL> memory . append ( current ) <EOL> stdout , stderr = process . communicate ( ) <EOL> return stdout , stderr , memory <EOL> def _get_memory_with_ps ( pid ) : <EOL> command_list = '<STR_LIT>' . split ( ) <EOL> command_list . append ( str ( pid ) ) <EOL> p = Popen ( command_list , stdout = PIPE ) <EOL> stdout = p . communicate ( ) [ <NUM_LIT:0> ] <EOL> if not p . returncode == <NUM_LIT:0> : <EOL> raise ProcessTerminatedError ( str ( pid ) ) <EOL> else : <EOL> return int ( stdout . splitlines ( ) [ <NUM_LIT:1> ] . split ( ) [ <NUM_LIT:5> ] ) * <NUM_LIT> <EOL> class BaseS3CLICommand ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . files = FileCreator ( ) <EOL> self . session = botocore . session . get_session ( ) <EOL> self . regions = { } <EOL> self . region = '<STR_LIT>' <EOL> self . client = self . session . create_client ( '<STR_LIT>' , region_name = self . region ) <EOL> self . extra_setup ( ) <EOL> def extra_setup ( self ) : <EOL> pass <EOL> def tearDown ( self ) : <EOL> self . files . remove_all ( ) <EOL> self . extra_teardown ( ) <EOL> def extra_teardown ( self ) : <EOL> pass <EOL> def create_client_for_bucket ( self , bucket_name ) : <EOL> region = self . regions . get ( bucket_name , self . region ) <EOL> client = self . session . create_client ( '<STR_LIT>' , region_name = region ) <EOL> return client <EOL> def assert_key_contents_equal ( self , bucket , key , expected_contents ) : <EOL> if isinstance ( expected_contents , six . BytesIO ) : <EOL> expected_contents = expected_contents . getvalue ( ) . decode ( '<STR_LIT:utf-8>' ) <EOL> actual_contents = self . get_key_contents ( bucket , key ) <EOL> self . assertEqual ( len ( actual_contents ) , len ( expected_contents ) ) <EOL> if actual_contents != expected_contents : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( bucket , key ) ) <EOL> def create_bucket ( self , name = None , region = None ) : <EOL> if not region : <EOL> region = self . region <EOL> bucket_name = create_bucket ( self . session , name , region ) <EOL> self . regions [ bucket_name ] = region <EOL> self . addCleanup ( self . delete_bucket , bucket_name ) <EOL> return bucket_name <EOL> def put_object ( self , bucket_name , key_name , contents = '<STR_LIT>' , extra_args = None ) : <EOL> client = self . create_client_for_bucket ( bucket_name ) <EOL> call_args = { <EOL> '<STR_LIT>' : bucket_name , <EOL> '<STR_LIT>' : key_name , '<STR_LIT>' : contents <EOL> } <EOL> if extra_args is not None : <EOL> call_args . update ( extra_args ) <EOL> response = client . put_object ( ** call_args ) <EOL> self . addCleanup ( self . delete_key , bucket_name , key_name ) <EOL> def delete_bucket ( self , bucket_name ) : <EOL> self . remove_all_objects ( bucket_name ) <EOL> client = self . create_client_for_bucket ( bucket_name ) <EOL> response = client . delete_bucket ( Bucket = bucket_name ) <EOL> self . regions . pop ( bucket_name , None ) <EOL> def remove_all_objects ( self , bucket_name ) : <EOL> client = self . create_client_for_bucket ( bucket_name ) <EOL> paginator = client . get_paginator ( '<STR_LIT>' ) <EOL> pages = paginator . paginate ( Bucket = bucket_name ) <EOL> key_names = [ ] <EOL> for page in pages : <EOL> key_names += [ obj [ '<STR_LIT>' ] for obj in page . get ( '<STR_LIT>' , [ ] ) ] <EOL> for key_name in key_names : <EOL> self . delete_key ( bucket_name , key_name ) <EOL> def delete_key ( self , bucket_name , key_name ) : <EOL> client = self . create_client_for_bucket ( bucket_name ) <EOL> response = client . delete_object ( Bucket = bucket_name , Key = key_name ) <EOL> def get_key_contents ( self , bucket_name , key_name ) : <EOL> client = self . create_client_for_bucket ( bucket_name ) <EOL> response = client . get_object ( Bucket = bucket_name , Key = key_name ) <EOL> return response [ '<STR_LIT>' ] . read ( ) . decode ( '<STR_LIT:utf-8>' ) <EOL> def key_exists ( self , bucket_name , key_name ) : <EOL> client = self . create_client_for_bucket ( bucket_name ) <EOL> try : <EOL> client . head_object ( Bucket = bucket_name , Key = key_name ) <EOL> return True <EOL> except ClientError : <EOL> return False <EOL> def list_buckets ( self ) : <EOL> response = self . client . list_buckets ( ) <EOL> return response [ '<STR_LIT>' ] <EOL> def content_type_for_key ( self , bucket_name , key_name ) : <EOL> parsed = self . head_object ( bucket_name , key_name ) <EOL> return parsed [ '<STR_LIT>' ] <EOL> def head_object ( self , bucket_name , key_name ) : <EOL> client = self . create_client_for_bucket ( bucket_name ) <EOL> response = client . head_object ( Bucket = bucket_name , Key = key_name ) <EOL> return response <EOL> def assert_no_errors ( self , p ) : <EOL> self . assertEqual ( <EOL> p . rc , <NUM_LIT:0> , <EOL> "<STR_LIT>" % ( p . rc , p . stdout + p . stderr ) ) <EOL> self . assertNotIn ( "<STR_LIT>" , p . stderr ) <EOL> self . assertNotIn ( "<STR_LIT>" , p . stderr ) <EOL> self . assertNotIn ( "<STR_LIT>" , p . stderr ) <EOL> self . assertNotIn ( "<STR_LIT>" , p . stderr ) <EOL> class StringIOWithFileNo ( StringIO ) : <EOL> def fileno ( self ) : <EOL> return <NUM_LIT:0> </s>
