<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import logging <EOL> import os <EOL> import sys <EOL> import tempfile <EOL> import boto <EOL> from gslib import wildcard_iterator <EOL> from gslib . boto_translation import BotoTranslation <EOL> from gslib . cloud_api_delegator import CloudApiDelegator <EOL> from gslib . command_runner import CommandRunner <EOL> from gslib . cs_api_map import ApiMapConstants <EOL> from gslib . cs_api_map import ApiSelector <EOL> from gslib . tests . mock_logging_handler import MockLoggingHandler <EOL> from gslib . tests . testcase import base <EOL> import gslib . tests . util as util <EOL> from gslib . tests . util import unittest <EOL> from gslib . tests . util import WorkingDirectory <EOL> from gslib . util import DiscardMessagesQueue <EOL> from gslib . util import GsutilStreamHandler <EOL> class GsutilApiUnitTestClassMapFactory ( object ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def GetClassMap ( cls ) : <EOL> """<STR_LIT>""" <EOL> gs_class_map = { <EOL> ApiSelector . XML : BotoTranslation , <EOL> ApiSelector . JSON : BotoTranslation <EOL> } <EOL> s3_class_map = { <EOL> ApiSelector . XML : BotoTranslation <EOL> } <EOL> class_map = { <EOL> '<STR_LIT>' : gs_class_map , <EOL> '<STR_LIT>' : s3_class_map <EOL> } <EOL> return class_map <EOL> @ unittest . skipUnless ( util . RUN_UNIT_TESTS , <EOL> '<STR_LIT>' ) <EOL> class GsUtilUnitTestCase ( base . GsUtilTestCase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> base . GsUtilTestCase . setUpClass ( ) <EOL> cls . mock_bucket_storage_uri = util . GSMockBucketStorageUri <EOL> cls . mock_gsutil_api_class_map_factory = GsutilApiUnitTestClassMapFactory <EOL> cls . logger = logging . getLogger ( ) <EOL> cls . command_runner = CommandRunner ( <EOL> bucket_storage_uri_class = cls . mock_bucket_storage_uri , <EOL> gsutil_api_class_map_factory = cls . mock_gsutil_api_class_map_factory ) <EOL> def setUp ( self ) : <EOL> super ( GsUtilUnitTestCase , self ) . setUp ( ) <EOL> self . bucket_uris = [ ] <EOL> self . stdout_save = sys . stdout <EOL> self . stderr_save = sys . stderr <EOL> fd , self . stdout_file = tempfile . mkstemp ( ) <EOL> sys . stdout = os . fdopen ( fd , '<STR_LIT>' ) <EOL> fd , self . stderr_file = tempfile . mkstemp ( ) <EOL> sys . stderr = os . fdopen ( fd , '<STR_LIT>' ) <EOL> self . accumulated_stdout = [ ] <EOL> self . accumulated_stderr = [ ] <EOL> self . root_logger = logging . getLogger ( ) <EOL> self . is_debugging = self . root_logger . isEnabledFor ( logging . DEBUG ) <EOL> self . log_handlers_save = self . root_logger . handlers <EOL> fd , self . log_handler_file = tempfile . mkstemp ( ) <EOL> self . log_handler_stream = os . fdopen ( fd , '<STR_LIT>' ) <EOL> self . temp_log_handler = GsutilStreamHandler ( self . log_handler_stream ) <EOL> self . root_logger . handlers = [ self . temp_log_handler ] <EOL> def tearDown ( self ) : <EOL> super ( GsUtilUnitTestCase , self ) . tearDown ( ) <EOL> self . root_logger . handlers = self . log_handlers_save <EOL> self . temp_log_handler . flush ( ) <EOL> self . temp_log_handler . close ( ) <EOL> self . log_handler_stream . seek ( <NUM_LIT:0> ) <EOL> log_output = self . log_handler_stream . read ( ) <EOL> self . log_handler_stream . close ( ) <EOL> os . unlink ( self . log_handler_file ) <EOL> sys . stdout . seek ( <NUM_LIT:0> ) <EOL> sys . stderr . seek ( <NUM_LIT:0> ) <EOL> stdout = sys . stdout . read ( ) <EOL> stderr = sys . stderr . read ( ) <EOL> stdout += '<STR_LIT>' . join ( self . accumulated_stdout ) <EOL> stderr += '<STR_LIT>' . join ( self . accumulated_stderr ) <EOL> sys . stdout . close ( ) <EOL> sys . stderr . close ( ) <EOL> sys . stdout = self . stdout_save <EOL> sys . stderr = self . stderr_save <EOL> os . unlink ( self . stdout_file ) <EOL> os . unlink ( self . stderr_file ) <EOL> if self . is_debugging and stdout : <EOL> sys . stderr . write ( '<STR_LIT>' % self . id ( ) ) <EOL> sys . stderr . write ( stdout ) <EOL> sys . stderr . write ( '<STR_LIT>' ) <EOL> if self . is_debugging and stderr : <EOL> sys . stderr . write ( '<STR_LIT>' % self . id ( ) ) <EOL> sys . stderr . write ( stderr ) <EOL> sys . stderr . write ( '<STR_LIT>' ) <EOL> if self . is_debugging and log_output : <EOL> sys . stderr . write ( '<STR_LIT>' % self . id ( ) ) <EOL> sys . stderr . write ( log_output ) <EOL> sys . stderr . write ( '<STR_LIT>' ) <EOL> def RunCommand ( self , command_name , args = None , headers = None , debug = <NUM_LIT:0> , <EOL> return_stdout = False , return_stderr = False , <EOL> return_log_handler = False , cwd = None ) : <EOL> """<STR_LIT>""" <EOL> args = args or [ ] <EOL> command_line = '<STR_LIT:U+0020>' . join ( [ command_name ] + args ) <EOL> if self . is_debugging : <EOL> self . stderr_save . write ( '<STR_LIT>' % command_line ) <EOL> sys . stdout . seek ( <NUM_LIT:0> ) <EOL> sys . stderr . seek ( <NUM_LIT:0> ) <EOL> stdout = sys . stdout . read ( ) <EOL> stderr = sys . stderr . read ( ) <EOL> if stdout : <EOL> self . accumulated_stdout . append ( stdout ) <EOL> if stderr : <EOL> self . accumulated_stderr . append ( stderr ) <EOL> sys . stdout . seek ( <NUM_LIT:0> ) <EOL> sys . stderr . seek ( <NUM_LIT:0> ) <EOL> sys . stdout . truncate ( ) <EOL> sys . stderr . truncate ( ) <EOL> mock_log_handler = MockLoggingHandler ( ) <EOL> logging . getLogger ( command_name ) . addHandler ( mock_log_handler ) <EOL> try : <EOL> with WorkingDirectory ( cwd ) : <EOL> self . command_runner . RunNamedCommand ( <EOL> command_name , args = args , headers = headers , debug = debug , <EOL> parallel_operations = False , do_shutdown = False ) <EOL> finally : <EOL> sys . stdout . seek ( <NUM_LIT:0> ) <EOL> stdout = sys . stdout . read ( ) <EOL> sys . stderr . seek ( <NUM_LIT:0> ) <EOL> stderr = sys . stderr . read ( ) <EOL> logging . getLogger ( command_name ) . removeHandler ( mock_log_handler ) <EOL> mock_log_handler . close ( ) <EOL> log_output = '<STR_LIT:\n>' . join ( <EOL> '<STR_LIT>' % level + '<STR_LIT>' . join ( records ) <EOL> for level , records in mock_log_handler . messages . iteritems ( ) <EOL> if records ) <EOL> if self . is_debugging and log_output : <EOL> self . stderr_save . write ( <EOL> '<STR_LIT>' % ( self . id ( ) , command_line ) ) <EOL> self . stderr_save . write ( log_output ) <EOL> self . stderr_save . write ( '<STR_LIT>' ) <EOL> if self . is_debugging and stdout : <EOL> self . stderr_save . write ( <EOL> '<STR_LIT>' % ( self . id ( ) , command_line ) ) <EOL> self . stderr_save . write ( stdout ) <EOL> self . stderr_save . write ( '<STR_LIT>' ) <EOL> if self . is_debugging and stderr : <EOL> self . stderr_save . write ( <EOL> '<STR_LIT>' % ( self . id ( ) , command_line ) ) <EOL> self . stderr_save . write ( stderr ) <EOL> self . stderr_save . write ( '<STR_LIT>' ) <EOL> sys . stdout . seek ( <NUM_LIT:0> ) <EOL> sys . stderr . seek ( <NUM_LIT:0> ) <EOL> sys . stdout . truncate ( ) <EOL> sys . stderr . truncate ( ) <EOL> to_return = [ ] <EOL> if return_stdout : <EOL> to_return . append ( stdout ) <EOL> if return_stderr : <EOL> to_return . append ( stderr ) <EOL> if return_log_handler : <EOL> to_return . append ( mock_log_handler ) <EOL> if len ( to_return ) == <NUM_LIT:1> : <EOL> return to_return [ <NUM_LIT:0> ] <EOL> return tuple ( to_return ) <EOL> @ classmethod <EOL> def MakeGsUtilApi ( cls , debug = <NUM_LIT:0> ) : <EOL> gsutil_api_map = { <EOL> ApiMapConstants . API_MAP : ( <EOL> cls . mock_gsutil_api_class_map_factory . GetClassMap ( ) ) , <EOL> ApiMapConstants . SUPPORT_MAP : { <EOL> '<STR_LIT>' : [ ApiSelector . XML , ApiSelector . JSON ] , <EOL> '<STR_LIT>' : [ ApiSelector . XML ] <EOL> } , <EOL> ApiMapConstants . DEFAULT_MAP : { <EOL> '<STR_LIT>' : ApiSelector . JSON , <EOL> '<STR_LIT>' : ApiSelector . XML <EOL> } <EOL> } <EOL> return CloudApiDelegator ( <EOL> cls . mock_bucket_storage_uri , gsutil_api_map , cls . logger , <EOL> DiscardMessagesQueue ( ) , debug = debug ) <EOL> @ classmethod <EOL> def _test_wildcard_iterator ( cls , uri_or_str , debug = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> uri_string = uri_or_str <EOL> if hasattr ( uri_or_str , '<STR_LIT>' ) : <EOL> uri_string = uri_or_str . uri <EOL> return wildcard_iterator . CreateWildcardIterator ( <EOL> uri_string , cls . MakeGsUtilApi ( ) ) <EOL> @ staticmethod <EOL> def _test_storage_uri ( uri_str , default_scheme = '<STR_LIT:file>' , debug = <NUM_LIT:0> , <EOL> validate = True ) : <EOL> """<STR_LIT>""" <EOL> return boto . storage_uri ( uri_str , default_scheme , debug , validate , <EOL> util . GSMockBucketStorageUri ) <EOL> def CreateBucket ( self , bucket_name = None , test_objects = <NUM_LIT:0> , storage_class = None , <EOL> provider = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> bucket_name = bucket_name or self . MakeTempName ( '<STR_LIT>' ) <EOL> bucket_uri = boto . storage_uri ( <EOL> '<STR_LIT>' % ( provider , bucket_name . lower ( ) ) , <EOL> suppress_consec_slashes = False , <EOL> bucket_storage_uri_class = util . GSMockBucketStorageUri ) <EOL> bucket_uri . create_bucket ( storage_class = storage_class ) <EOL> self . bucket_uris . append ( bucket_uri ) <EOL> try : <EOL> iter ( test_objects ) <EOL> except TypeError : <EOL> test_objects = [ self . MakeTempName ( '<STR_LIT>' ) for _ in range ( test_objects ) ] <EOL> for i , name in enumerate ( test_objects ) : <EOL> self . CreateObject ( bucket_uri = bucket_uri , object_name = name , <EOL> contents = '<STR_LIT>' % i ) <EOL> return bucket_uri <EOL> def CreateObject ( self , bucket_uri = None , object_name = None , contents = None ) : <EOL> """<STR_LIT>""" <EOL> bucket_uri = bucket_uri or self . CreateBucket ( ) <EOL> object_name = object_name or self . MakeTempName ( '<STR_LIT>' ) <EOL> key_uri = bucket_uri . clone_replace_name ( object_name ) <EOL> if contents is not None : <EOL> key_uri . set_contents_from_string ( contents ) <EOL> return key_uri </s>
