<s> """<STR_LIT>""" <EOL> import datetime <EOL> import errno <EOL> import os <EOL> import stat <EOL> import tempfile <EOL> import unittest2 <EOL> import mock <EOL> from oauth2client import util <EOL> from oauth2client . client import OAuth2Credentials <EOL> from oauth2client . contrib import locked_file <EOL> from oauth2client . contrib import multistore_file <EOL> _filehandle , FILENAME = tempfile . mkstemp ( '<STR_LIT>' ) <EOL> os . close ( _filehandle ) <EOL> class _MockLockedFile ( object ) : <EOL> def __init__ ( self , filename_str , error_code ) : <EOL> self . filename_str = filename_str <EOL> self . error_code = error_code <EOL> self . open_and_lock_called = False <EOL> def open_and_lock ( self ) : <EOL> self . open_and_lock_called = True <EOL> raise IOError ( self . error_code , '<STR_LIT>' ) <EOL> def is_locked ( self ) : <EOL> return False <EOL> def filename ( self ) : <EOL> return self . filename_str <EOL> class Test__dict_to_tuple_key ( unittest2 . TestCase ) : <EOL> def test_key_conversions ( self ) : <EOL> key1 , val1 = '<STR_LIT>' , '<STR_LIT>' <EOL> key2 , val2 = '<STR_LIT>' , '<STR_LIT>' <EOL> key3 , val3 = '<STR_LIT>' , '<STR_LIT:foo>' <EOL> test_dict = { <EOL> key1 : val1 , <EOL> key2 : val2 , <EOL> key3 : val3 , <EOL> } <EOL> tuple_key = multistore_file . _dict_to_tuple_key ( test_dict ) <EOL> expected_output = ( <EOL> ( key2 , val2 ) , <EOL> ( key3 , val3 ) , <EOL> ( key1 , val1 ) , <EOL> ) <EOL> self . assertTupleEqual ( expected_output , tuple_key ) <EOL> self . assertDictEqual ( test_dict , dict ( tuple_key ) ) <EOL> class MultistoreFileTests ( unittest2 . TestCase ) : <EOL> def tearDown ( self ) : <EOL> try : <EOL> os . unlink ( FILENAME ) <EOL> except OSError : <EOL> pass <EOL> def setUp ( self ) : <EOL> try : <EOL> os . unlink ( FILENAME ) <EOL> except OSError : <EOL> pass <EOL> def _create_test_credentials ( self , client_id = '<STR_LIT>' , <EOL> expiration = None ) : <EOL> access_token = '<STR_LIT:foo>' <EOL> client_secret = '<STR_LIT>' <EOL> refresh_token = '<STR_LIT>' <EOL> token_expiry = expiration or datetime . datetime . utcnow ( ) <EOL> token_uri = '<STR_LIT>' <EOL> user_agent = '<STR_LIT>' <EOL> credentials = OAuth2Credentials ( <EOL> access_token , client_id , client_secret , <EOL> refresh_token , token_expiry , token_uri , <EOL> user_agent ) <EOL> return credentials <EOL> def test_lock_file_raises_ioerror ( self ) : <EOL> filehandle , filename = tempfile . mkstemp ( ) <EOL> os . close ( filehandle ) <EOL> try : <EOL> for error_code in ( errno . EDEADLK , errno . ENOSYS , errno . ENOLCK , <EOL> errno . EACCES ) : <EOL> multistore = multistore_file . _MultiStore ( filename ) <EOL> multistore . _file = _MockLockedFile ( filename , error_code ) <EOL> multistore . _lock ( ) <EOL> self . assertTrue ( multistore . _file . open_and_lock_called ) <EOL> finally : <EOL> os . unlink ( filename ) <EOL> def test_lock_file_raise_unexpected_error ( self ) : <EOL> filehandle , filename = tempfile . mkstemp ( ) <EOL> os . close ( filehandle ) <EOL> try : <EOL> multistore = multistore_file . _MultiStore ( filename ) <EOL> multistore . _file = _MockLockedFile ( filename , errno . EBUSY ) <EOL> self . assertRaises ( IOError , multistore . _lock ) <EOL> self . assertTrue ( multistore . _file . open_and_lock_called ) <EOL> finally : <EOL> os . unlink ( filename ) <EOL> def test_read_only_file_fail_lock ( self ) : <EOL> credentials = self . _create_test_credentials ( ) <EOL> open ( FILENAME , '<STR_LIT>' ) . close ( ) <EOL> os . chmod ( FILENAME , <NUM_LIT> ) <EOL> store = multistore_file . get_credential_storage ( <EOL> FILENAME , <EOL> credentials . client_id , <EOL> credentials . user_agent , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> store . put ( credentials ) <EOL> if os . name == '<STR_LIT>' : <EOL> self . assertTrue ( store . _multistore . _read_only ) <EOL> os . chmod ( FILENAME , <NUM_LIT> ) <EOL> def test_read_only_file_fail_lock_no_warning ( self ) : <EOL> open ( FILENAME , '<STR_LIT>' ) . close ( ) <EOL> os . chmod ( FILENAME , <NUM_LIT> ) <EOL> multistore = multistore_file . _MultiStore ( FILENAME ) <EOL> with mock . patch . object ( multistore_file . logger , '<STR_LIT>' ) as mock_warn : <EOL> multistore . _warn_on_readonly = False <EOL> multistore . _lock ( ) <EOL> self . assertFalse ( mock_warn . called ) <EOL> def test_lock_skip_refresh ( self ) : <EOL> with open ( FILENAME , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> os . chmod ( FILENAME , <NUM_LIT> ) <EOL> multistore = multistore_file . _MultiStore ( FILENAME ) <EOL> refresh_patch = mock . patch . object ( <EOL> multistore , '<STR_LIT>' ) <EOL> with refresh_patch as refresh_mock : <EOL> multistore . _data = { } <EOL> multistore . _lock ( ) <EOL> self . assertFalse ( refresh_mock . called ) <EOL> @ unittest2 . skipIf ( not hasattr ( os , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def test_multistore_no_symbolic_link_files ( self ) : <EOL> SYMFILENAME = FILENAME + '<STR_LIT>' <EOL> os . symlink ( FILENAME , SYMFILENAME ) <EOL> store = multistore_file . get_credential_storage ( <EOL> SYMFILENAME , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> try : <EOL> self . assertRaises ( <EOL> locked_file . CredentialsFileSymbolicLinkError , <EOL> store . get ) <EOL> finally : <EOL> os . unlink ( SYMFILENAME ) <EOL> def test_multistore_non_existent_file ( self ) : <EOL> store = multistore_file . get_credential_storage ( <EOL> FILENAME , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> credentials = store . get ( ) <EOL> self . assertEquals ( None , credentials ) <EOL> def test_multistore_file ( self ) : <EOL> credentials = self . _create_test_credentials ( ) <EOL> store = multistore_file . get_credential_storage ( <EOL> FILENAME , <EOL> credentials . client_id , <EOL> credentials . user_agent , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> store . put ( credentials ) <EOL> credentials = store . get ( ) <EOL> self . assertNotEquals ( None , credentials ) <EOL> self . assertEquals ( '<STR_LIT:foo>' , credentials . access_token ) <EOL> store . delete ( ) <EOL> credentials = store . get ( ) <EOL> self . assertEquals ( None , credentials ) <EOL> if os . name == '<STR_LIT>' : <EOL> self . assertEquals ( <EOL> <NUM_LIT> , stat . S_IMODE ( os . stat ( FILENAME ) . st_mode ) ) <EOL> def test_multistore_file_custom_key ( self ) : <EOL> credentials = self . _create_test_credentials ( ) <EOL> custom_key = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> store = multistore_file . get_credential_storage_custom_key ( <EOL> FILENAME , custom_key ) <EOL> store . put ( credentials ) <EOL> stored_credentials = store . get ( ) <EOL> self . assertNotEquals ( None , stored_credentials ) <EOL> self . assertEqual ( credentials . access_token , <EOL> stored_credentials . access_token ) <EOL> store . delete ( ) <EOL> stored_credentials = store . get ( ) <EOL> self . assertEquals ( None , stored_credentials ) <EOL> def test_multistore_file_custom_string_key ( self ) : <EOL> credentials = self . _create_test_credentials ( ) <EOL> store = multistore_file . get_credential_storage_custom_string_key ( <EOL> FILENAME , '<STR_LIT>' ) <EOL> store . put ( credentials ) <EOL> stored_credentials = store . get ( ) <EOL> self . assertNotEquals ( None , stored_credentials ) <EOL> self . assertEqual ( credentials . access_token , <EOL> stored_credentials . access_token ) <EOL> multistore_file . get_credential_storage_custom_string_key ( <EOL> FILENAME , { '<STR_LIT:key>' : '<STR_LIT>' } ) <EOL> stored_credentials = store . get ( ) <EOL> self . assertNotEquals ( None , stored_credentials ) <EOL> self . assertEqual ( credentials . access_token , <EOL> stored_credentials . access_token ) <EOL> store . delete ( ) <EOL> stored_credentials = store . get ( ) <EOL> self . assertEquals ( None , stored_credentials ) <EOL> def test_multistore_file_backwards_compatibility ( self ) : <EOL> credentials = self . _create_test_credentials ( ) <EOL> scopes = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> store = multistore_file . get_credential_storage ( <EOL> FILENAME , '<STR_LIT>' , '<STR_LIT>' , scopes ) <EOL> store . put ( credentials ) <EOL> key = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : util . scopes_to_string ( scopes ) } <EOL> store = multistore_file . get_credential_storage_custom_key ( <EOL> FILENAME , key ) <EOL> stored_credentials = store . get ( ) <EOL> self . assertEqual ( credentials . access_token , <EOL> stored_credentials . access_token ) <EOL> def test_multistore_file_get_all_keys ( self ) : <EOL> keys = multistore_file . get_all_credential_keys ( FILENAME ) <EOL> self . assertEquals ( [ ] , keys ) <EOL> credentials = self . _create_test_credentials ( client_id = '<STR_LIT>' ) <EOL> custom_key = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> store1 = multistore_file . get_credential_storage_custom_key ( <EOL> FILENAME , custom_key ) <EOL> store1 . put ( credentials ) <EOL> keys = multistore_file . get_all_credential_keys ( FILENAME ) <EOL> self . assertEquals ( [ custom_key ] , keys ) <EOL> credentials = self . _create_test_credentials ( client_id = '<STR_LIT>' ) <EOL> string_key = '<STR_LIT>' <EOL> store2 = multistore_file . get_credential_storage_custom_string_key ( <EOL> FILENAME , string_key ) <EOL> store2 . put ( credentials ) <EOL> keys = multistore_file . get_all_credential_keys ( FILENAME ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( keys ) ) <EOL> self . assertTrue ( custom_key in keys ) <EOL> self . assertTrue ( { '<STR_LIT:key>' : string_key } in keys ) <EOL> store1 . delete ( ) <EOL> store2 . delete ( ) <EOL> keys = multistore_file . get_all_credential_keys ( FILENAME ) <EOL> self . assertEquals ( [ ] , keys ) <EOL> def _refresh_data_cache_helper ( self ) : <EOL> multistore = multistore_file . _MultiStore ( FILENAME ) <EOL> json_patch = mock . patch . object ( multistore , '<STR_LIT>' ) <EOL> return multistore , json_patch <EOL> def test__refresh_data_cache_bad_json ( self ) : <EOL> multistore , json_patch = self . _refresh_data_cache_helper ( ) <EOL> with json_patch as json_mock : <EOL> json_mock . side_effect = ValueError ( '<STR_LIT>' ) <EOL> multistore . _refresh_data_cache ( ) <EOL> self . assertTrue ( json_mock . called ) <EOL> self . assertEqual ( multistore . _data , { } ) <EOL> def test__refresh_data_cache_bad_version ( self ) : <EOL> multistore , json_patch = self . _refresh_data_cache_helper ( ) <EOL> with json_patch as json_mock : <EOL> json_mock . return_value = { } <EOL> multistore . _refresh_data_cache ( ) <EOL> self . assertTrue ( json_mock . called ) <EOL> self . assertEqual ( multistore . _data , { } ) <EOL> def test__refresh_data_cache_newer_version ( self ) : <EOL> multistore , json_patch = self . _refresh_data_cache_helper ( ) <EOL> with json_patch as json_mock : <EOL> json_mock . return_value = { '<STR_LIT>' : <NUM_LIT:5> } <EOL> self . assertRaises ( <EOL> multistore_file . NewerCredentialStoreError , <EOL> multistore . _refresh_data_cache ) <EOL> self . assertTrue ( json_mock . called ) <EOL> def test__refresh_data_cache_bad_credentials ( self ) : <EOL> multistore , json_patch = self . _refresh_data_cache_helper ( ) <EOL> with json_patch as json_mock : <EOL> json_mock . return_value = { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT:data>' : [ <EOL> { '<STR_LIT>' : '<STR_LIT>' } <EOL> ] } <EOL> multistore . _refresh_data_cache ( ) <EOL> self . assertTrue ( json_mock . called ) <EOL> self . assertEqual ( multistore . _data , { } ) <EOL> def test__delete_credential_nonexistent ( self ) : <EOL> multistore = multistore_file . _MultiStore ( FILENAME ) <EOL> with mock . patch . object ( multistore , '<STR_LIT>' ) as write_mock : <EOL> multistore . _data = { } <EOL> multistore . _delete_credential ( '<STR_LIT>' ) <EOL> self . assertTrue ( write_mock . called ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest2 . main ( ) </s>
