<s> from __future__ import absolute_import <EOL> from contextlib import contextmanager <EOL> import functools <EOL> import os <EOL> import pkgutil <EOL> import posixpath <EOL> import re <EOL> import tempfile <EOL> import unittest <EOL> import urlparse <EOL> import boto <EOL> import crcmod <EOL> from gslib . encryption_helper import Base64Sha256FromBase64EncryptionKey <EOL> import gslib . tests as gslib_tests <EOL> from gslib . util import UsingCrcmodExtension <EOL> if not hasattr ( unittest . TestCase , '<STR_LIT>' ) : <EOL> import unittest2 as unittest <EOL> TEST_ENCRYPTION_KEY1 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_KEY1_SHA256_B64 = Base64Sha256FromBase64EncryptionKey ( <EOL> TEST_ENCRYPTION_KEY1 ) <EOL> TEST_ENCRYPTION_KEY2 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_KEY2_SHA256_B64 = Base64Sha256FromBase64EncryptionKey ( <EOL> TEST_ENCRYPTION_KEY2 ) <EOL> TEST_ENCRYPTION_KEY3 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_KEY3_SHA256_B64 = Base64Sha256FromBase64EncryptionKey ( <EOL> TEST_ENCRYPTION_KEY3 ) <EOL> TEST_ENCRYPTION_KEY4 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_KEY4_SHA256_B64 = Base64Sha256FromBase64EncryptionKey ( <EOL> TEST_ENCRYPTION_KEY4 ) <EOL> TEST_ENCRYPTION_CONTENT1 = '<STR_LIT:bar>' <EOL> TEST_ENCRYPTION_CONTENT1_MD5 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT1_CRC32C = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT2 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT2_MD5 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT2_CRC32C = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT3 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT3_MD5 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT3_CRC32C = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT4 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT4_MD5 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT4_CRC32C = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT5 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT5_MD5 = '<STR_LIT>' <EOL> TEST_ENCRYPTION_CONTENT5_CRC32C = '<STR_LIT>' <EOL> RUN_INTEGRATION_TESTS = True <EOL> RUN_UNIT_TESTS = True <EOL> RUN_S3_TESTS = False <EOL> PARALLEL_COMPOSITE_UPLOAD_TEST_CONFIG = '<STR_LIT>' <EOL> def _HasS3Credentials ( ) : <EOL> return ( boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , None ) and <EOL> boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , None ) ) <EOL> HAS_S3_CREDS = _HasS3Credentials ( ) <EOL> def _HasGSHost ( ) : <EOL> return boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , None ) is not None <EOL> HAS_GS_HOST = _HasGSHost ( ) <EOL> def _HasGSPort ( ) : <EOL> return boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , None ) is not None <EOL> HAS_GS_PORT = _HasGSPort ( ) <EOL> def _UsingJSONApi ( ) : <EOL> return boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) . upper ( ) != '<STR_LIT>' <EOL> USING_JSON_API = _UsingJSONApi ( ) <EOL> def _ArgcompleteAvailable ( ) : <EOL> argcomplete = None <EOL> try : <EOL> import argcomplete <EOL> except ImportError : <EOL> pass <EOL> return argcomplete is not None <EOL> ARGCOMPLETE_AVAILABLE = _ArgcompleteAvailable ( ) <EOL> def _NormalizeURI ( uri ) : <EOL> """<STR_LIT>""" <EOL> uri = uri . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> parsed = list ( urlparse . urlparse ( uri ) ) <EOL> parsed [ <NUM_LIT:2> ] = posixpath . normpath ( parsed [ <NUM_LIT:2> ] ) <EOL> if parsed [ <NUM_LIT:2> ] . startswith ( '<STR_LIT>' ) : <EOL> parsed [ <NUM_LIT:2> ] = parsed [ <NUM_LIT:2> ] [ <NUM_LIT:1> : ] <EOL> unparsed = urlparse . urlunparse ( parsed ) <EOL> unparsed = unparsed . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> return unparsed <EOL> def GenerationFromURI ( uri ) : <EOL> """<STR_LIT>""" <EOL> if not ( uri . generation or uri . version_id ) : <EOL> if uri . scheme == '<STR_LIT>' : return '<STR_LIT:null>' <EOL> return uri . generation or uri . version_id <EOL> def ObjectToURI ( obj , * suffixes ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( obj , file ) : <EOL> return '<STR_LIT>' % os . path . abspath ( os . path . join ( obj . name , * suffixes ) ) <EOL> if isinstance ( obj , basestring ) : <EOL> return '<STR_LIT>' % os . path . join ( obj , * suffixes ) <EOL> uri = obj . uri <EOL> if suffixes : <EOL> uri = _NormalizeURI ( '<STR_LIT:/>' . join ( [ uri ] + list ( suffixes ) ) ) <EOL> if uri . endswith ( '<STR_LIT:/>' ) : <EOL> uri = uri [ : - <NUM_LIT:1> ] <EOL> return uri <EOL> try : <EOL> from gslib . tests import mock_storage_service <EOL> except ImportError : <EOL> try : <EOL> from boto . tests . integration . s3 import mock_storage_service <EOL> except ImportError : <EOL> try : <EOL> from tests . integration . s3 import mock_storage_service <EOL> except ImportError : <EOL> import mock_storage_service <EOL> class GSMockConnection ( mock_storage_service . MockConnection ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . debug = <NUM_LIT:0> <EOL> super ( GSMockConnection , self ) . __init__ ( * args , ** kwargs ) <EOL> mock_connection = GSMockConnection ( ) <EOL> class GSMockBucketStorageUri ( mock_storage_service . MockBucketStorageUri ) : <EOL> def connect ( self , access_key_id = None , secret_access_key = None ) : <EOL> return mock_connection <EOL> def compose ( self , components , headers = None ) : <EOL> """<STR_LIT>""" <EOL> return self . new_key ( ) <EOL> TEST_BOTO_REMOVE_SECTION = '<STR_LIT>' <EOL> def _SetBotoConfig ( section , name , value , revert_list ) : <EOL> """<STR_LIT>""" <EOL> prev_value = boto . config . get ( section , name , None ) <EOL> if not boto . config . has_section ( section ) : <EOL> revert_list . append ( ( section , TEST_BOTO_REMOVE_SECTION , None ) ) <EOL> boto . config . add_section ( section ) <EOL> revert_list . append ( ( section , name , prev_value ) ) <EOL> if value is None : <EOL> boto . config . remove_option ( section , name ) <EOL> else : <EOL> boto . config . set ( section , name , value ) <EOL> def _RevertBotoConfig ( revert_list ) : <EOL> """<STR_LIT>""" <EOL> sections_to_remove = [ ] <EOL> for section , name , value in revert_list : <EOL> if value is None : <EOL> if name == TEST_BOTO_REMOVE_SECTION : <EOL> sections_to_remove . append ( section ) <EOL> else : <EOL> boto . config . remove_option ( section , name ) <EOL> else : <EOL> boto . config . set ( section , name , value ) <EOL> for section in sections_to_remove : <EOL> boto . config . remove_section ( section ) <EOL> def SequentialAndParallelTransfer ( func ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( func ) <EOL> def Wrapper ( * args , ** kwargs ) : <EOL> func ( * args , ** kwargs ) <EOL> if not RUN_S3_TESTS and UsingCrcmodExtension ( crcmod ) : <EOL> with SetBotoConfigForTest ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:1>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:1>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:3>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ] ) : <EOL> func ( * args , ** kwargs ) <EOL> return Wrapper <EOL> def _SectionDictFromConfigList ( boto_config_list ) : <EOL> """<STR_LIT>""" <EOL> sections = { } <EOL> for config_entry in boto_config_list : <EOL> section , key , value = ( config_entry [ <NUM_LIT:0> ] , config_entry [ <NUM_LIT:1> ] , config_entry [ <NUM_LIT:2> ] ) <EOL> if section not in sections : <EOL> sections [ section ] = { } <EOL> if value is not None : <EOL> sections [ section ] [ key ] = value <EOL> return sections <EOL> def _WriteSectionDictToFile ( section_dict , tmp_filename ) : <EOL> """<STR_LIT>""" <EOL> with open ( tmp_filename , '<STR_LIT:w>' ) as tmp_file : <EOL> for section , key_value_pairs in section_dict . iteritems ( ) : <EOL> tmp_file . write ( '<STR_LIT>' % section ) <EOL> for key , value in key_value_pairs . iteritems ( ) : <EOL> tmp_file . write ( '<STR_LIT>' % ( key , value ) ) <EOL> @ contextmanager <EOL> def SetBotoConfigForTest ( boto_config_list , use_existing_config = True ) : <EOL> """<STR_LIT>""" <EOL> revert_configs = [ ] <EOL> tmp_filename = None <EOL> try : <EOL> tmp_fd , tmp_filename = tempfile . mkstemp ( prefix = '<STR_LIT>' ) <EOL> os . close ( tmp_fd ) <EOL> if use_existing_config : <EOL> for boto_config in boto_config_list : <EOL> _SetBotoConfig ( boto_config [ <NUM_LIT:0> ] , boto_config [ <NUM_LIT:1> ] , boto_config [ <NUM_LIT:2> ] , <EOL> revert_configs ) <EOL> with open ( tmp_filename , '<STR_LIT:w>' ) as tmp_file : <EOL> boto . config . write ( tmp_file ) <EOL> else : <EOL> _WriteSectionDictToFile ( _SectionDictFromConfigList ( boto_config_list ) , <EOL> tmp_filename ) <EOL> with _SetBotoConfigFileForTest ( tmp_filename ) : <EOL> yield <EOL> finally : <EOL> _RevertBotoConfig ( revert_configs ) <EOL> if tmp_filename : <EOL> try : <EOL> os . remove ( tmp_filename ) <EOL> except OSError : <EOL> pass <EOL> @ contextmanager <EOL> def SetEnvironmentForTest ( env_variable_dict ) : <EOL> """<STR_LIT>""" <EOL> def _ApplyDictToEnvironment ( dict_to_apply ) : <EOL> for k , v in dict_to_apply . iteritems ( ) : <EOL> old_values [ k ] = os . environ . get ( k ) <EOL> if v is not None : <EOL> os . environ [ k ] = v <EOL> elif k in os . environ : <EOL> del os . environ [ k ] <EOL> old_values = { } <EOL> for k in env_variable_dict : <EOL> old_values [ k ] = os . environ . get ( k ) <EOL> try : <EOL> _ApplyDictToEnvironment ( env_variable_dict ) <EOL> yield <EOL> finally : <EOL> _ApplyDictToEnvironment ( old_values ) <EOL> @ contextmanager <EOL> def _SetBotoConfigFileForTest ( boto_config_path ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> old_boto_config_env_variable = os . environ [ '<STR_LIT>' ] <EOL> boto_config_was_set = True <EOL> except KeyError : <EOL> boto_config_was_set = False <EOL> os . environ [ '<STR_LIT>' ] = boto_config_path <EOL> try : <EOL> yield <EOL> finally : <EOL> if boto_config_was_set : <EOL> os . environ [ '<STR_LIT>' ] = old_boto_config_env_variable <EOL> else : <EOL> os . environ . pop ( '<STR_LIT>' , None ) <EOL> def GetTestNames ( ) : <EOL> """<STR_LIT>""" <EOL> matcher = re . compile ( r'<STR_LIT>' ) <EOL> names = [ ] <EOL> for _ , modname , _ in pkgutil . iter_modules ( gslib_tests . __path__ ) : <EOL> m = matcher . match ( modname ) <EOL> if m : <EOL> names . append ( m . group ( '<STR_LIT:name>' ) ) <EOL> return names <EOL> @ contextmanager <EOL> def WorkingDirectory ( new_working_directory ) : <EOL> """<STR_LIT>""" <EOL> prev_working_directory = None <EOL> try : <EOL> prev_working_directory = os . getcwd ( ) <EOL> except OSError : <EOL> pass <EOL> if new_working_directory : <EOL> os . chdir ( new_working_directory ) <EOL> try : <EOL> yield <EOL> finally : <EOL> if new_working_directory and prev_working_directory : <EOL> os . chdir ( prev_working_directory ) </s>
