| # Copyright 2015 Google Inc. All rights reserved. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| """Unit tests for oauth2client.multistore_file.""" |
| |
| import datetime |
| import errno |
| import os |
| import stat |
| import tempfile |
| |
| import mock |
| import unittest2 |
| |
| from oauth2client import client |
| from oauth2client import util |
| from oauth2client.contrib import locked_file |
| from oauth2client.contrib import multistore_file |
| |
| _filehandle, FILENAME = tempfile.mkstemp('oauth2client_test.data') |
| os.close(_filehandle) |
| |
| |
| class _MockLockedFile(object): |
| |
| def __init__(self, filename_str, error_class, error_code): |
| self.filename_str = filename_str |
| self.error_class = error_class |
| self.error_code = error_code |
| self.open_and_lock_called = False |
| |
| def open_and_lock(self): |
| self.open_and_lock_called = True |
| raise self.error_class(self.error_code, '') |
| |
| def is_locked(self): |
| return False |
| |
| def filename(self): |
| return self.filename_str |
| |
| |
| class Test__dict_to_tuple_key(unittest2.TestCase): |
| |
| def test_key_conversions(self): |
| key1, val1 = 'somekey', 'some value' |
| key2, val2 = 'another', 'something else' |
| key3, val3 = 'onemore', 'foo' |
| test_dict = { |
| key1: val1, |
| key2: val2, |
| key3: val3, |
| } |
| tuple_key = multistore_file._dict_to_tuple_key(test_dict) |
| |
| # the resulting key should be naturally sorted |
| expected_output = ( |
| (key2, val2), |
| (key3, val3), |
| (key1, val1), |
| ) |
| self.assertTupleEqual(expected_output, tuple_key) |
| # check we get the original dictionary back |
| self.assertDictEqual(test_dict, dict(tuple_key)) |
| |
| |
| class MultistoreFileTests(unittest2.TestCase): |
| |
| def tearDown(self): |
| try: |
| os.unlink(FILENAME) |
| except OSError: |
| pass |
| |
| def setUp(self): |
| try: |
| os.unlink(FILENAME) |
| except OSError: |
| pass |
| |
| def _create_test_credentials(self, client_id='some_client_id', |
| expiration=None): |
| access_token = 'foo' |
| client_secret = 'cOuDdkfjxxnv+' |
| refresh_token = '1/0/a.df219fjls0' |
| token_expiry = expiration or datetime.datetime.utcnow() |
| token_uri = 'https://www.google.com/accounts/o8/oauth2/token' |
| user_agent = 'refresh_checker/1.0' |
| |
| credentials = client.OAuth2Credentials( |
| access_token, client_id, client_secret, |
| refresh_token, token_expiry, token_uri, |
| user_agent) |
| return credentials |
| |
| def test_lock_file_raises_ioerror(self): |
| filehandle, filename = tempfile.mkstemp() |
| os.close(filehandle) |
| |
| try: |
| for error_code in (errno.EDEADLK, errno.ENOSYS, errno.ENOLCK, |
| errno.EACCES): |
| for error_class in (IOError, OSError): |
| multistore = multistore_file._MultiStore(filename) |
| multistore._file = _MockLockedFile( |
| filename, error_class, error_code) |
| # Should not raise though the underlying file class did. |
| multistore._lock() |
| self.assertTrue(multistore._file.open_and_lock_called) |
| finally: |
| os.unlink(filename) |
| |
| def test_lock_file_raise_unexpected_error(self): |
| filehandle, filename = tempfile.mkstemp() |
| os.close(filehandle) |
| |
| try: |
| multistore = multistore_file._MultiStore(filename) |
| multistore._file = _MockLockedFile(filename, IOError, errno.EBUSY) |
| with self.assertRaises(IOError): |
| multistore._lock() |
| self.assertTrue(multistore._file.open_and_lock_called) |
| finally: |
| os.unlink(filename) |
| |
| def test_read_only_file_fail_lock(self): |
| credentials = self._create_test_credentials() |
| |
| open(FILENAME, 'a+b').close() |
| os.chmod(FILENAME, 0o400) |
| |
| store = multistore_file.get_credential_storage( |
| FILENAME, |
| credentials.client_id, |
| credentials.user_agent, |
| ['some-scope', 'some-other-scope']) |
| |
| store.put(credentials) |
| if os.name == 'posix': # pragma: NO COVER |
| self.assertTrue(store._multistore._read_only) |
| os.chmod(FILENAME, 0o600) |
| |
| def test_read_only_file_fail_lock_no_warning(self): |
| open(FILENAME, 'a+b').close() |
| os.chmod(FILENAME, 0o400) |
| |
| multistore = multistore_file._MultiStore(FILENAME) |
| |
| with mock.patch.object(multistore_file.logger, 'warn') as mock_warn: |
| multistore._warn_on_readonly = False |
| multistore._lock() |
| self.assertFalse(mock_warn.called) |
| |
| def test_lock_skip_refresh(self): |
| with open(FILENAME, 'w') as f: |
| f.write('123') |
| os.chmod(FILENAME, 0o400) |
| |
| multistore = multistore_file._MultiStore(FILENAME) |
| |
| refresh_patch = mock.patch.object( |
| multistore, '_refresh_data_cache') |
| |
| with refresh_patch as refresh_mock: |
| multistore._data = {} |
| multistore._lock() |
| self.assertFalse(refresh_mock.called) |
| |
| @unittest2.skipIf(not hasattr(os, 'symlink'), 'No symlink available') |
| def test_multistore_no_symbolic_link_files(self): |
| SYMFILENAME = FILENAME + 'sym' |
| os.symlink(FILENAME, SYMFILENAME) |
| store = multistore_file.get_credential_storage( |
| SYMFILENAME, |
| 'some_client_id', |
| 'user-agent/1.0', |
| ['some-scope', 'some-other-scope']) |
| try: |
| with self.assertRaises( |
| locked_file.CredentialsFileSymbolicLinkError): |
| store.get() |
| finally: |
| os.unlink(SYMFILENAME) |
| |
| def test_multistore_non_existent_file(self): |
| store = multistore_file.get_credential_storage( |
| FILENAME, |
| 'some_client_id', |
| 'user-agent/1.0', |
| ['some-scope', 'some-other-scope']) |
| |
| credentials = store.get() |
| self.assertEquals(None, credentials) |
| |
| def test_multistore_file(self): |
| credentials = self._create_test_credentials() |
| |
| store = multistore_file.get_credential_storage( |
| FILENAME, |
| credentials.client_id, |
| credentials.user_agent, |
| ['some-scope', 'some-other-scope']) |
| |
| # Save credentials |
| store.put(credentials) |
| credentials = store.get() |
| |
| self.assertNotEquals(None, credentials) |
| self.assertEquals('foo', credentials.access_token) |
| |
| # Delete credentials |
| store.delete() |
| credentials = store.get() |
| |
| self.assertEquals(None, credentials) |
| |
| if os.name == 'posix': # pragma: NO COVER |
| self.assertEquals( |
| 0o600, stat.S_IMODE(os.stat(FILENAME).st_mode)) |
| |
| def test_multistore_file_custom_key(self): |
| credentials = self._create_test_credentials() |
| |
| custom_key = {'myapp': 'testing', 'clientid': 'some client'} |
| store = multistore_file.get_credential_storage_custom_key( |
| FILENAME, custom_key) |
| |
| store.put(credentials) |
| stored_credentials = store.get() |
| |
| self.assertNotEquals(None, stored_credentials) |
| self.assertEqual(credentials.access_token, |
| stored_credentials.access_token) |
| |
| store.delete() |
| stored_credentials = store.get() |
| |
| self.assertEquals(None, stored_credentials) |
| |
| def test_multistore_file_custom_string_key(self): |
| credentials = self._create_test_credentials() |
| |
| # store with string key |
| store = multistore_file.get_credential_storage_custom_string_key( |
| FILENAME, 'mykey') |
| |
| store.put(credentials) |
| stored_credentials = store.get() |
| |
| self.assertNotEquals(None, stored_credentials) |
| self.assertEqual(credentials.access_token, |
| stored_credentials.access_token) |
| |
| # try retrieving with a dictionary |
| multistore_file.get_credential_storage_custom_string_key( |
| FILENAME, {'key': 'mykey'}) |
| stored_credentials = store.get() |
| self.assertNotEquals(None, stored_credentials) |
| self.assertEqual(credentials.access_token, |
| stored_credentials.access_token) |
| |
| store.delete() |
| stored_credentials = store.get() |
| |
| self.assertEquals(None, stored_credentials) |
| |
| def test_multistore_file_backwards_compatibility(self): |
| credentials = self._create_test_credentials() |
| scopes = ['scope1', 'scope2'] |
| |
| # store the credentials using the legacy key method |
| store = multistore_file.get_credential_storage( |
| FILENAME, 'client_id', 'user_agent', scopes) |
| store.put(credentials) |
| |
| # retrieve the credentials using a custom key that matches the |
| # legacy key |
| key = {'clientId': 'client_id', 'userAgent': 'user_agent', |
| 'scope': util.scopes_to_string(scopes)} |
| store = multistore_file.get_credential_storage_custom_key( |
| FILENAME, key) |
| stored_credentials = store.get() |
| |
| self.assertEqual(credentials.access_token, |
| stored_credentials.access_token) |
| |
| def test_multistore_file_get_all_keys(self): |
| # start with no keys |
| keys = multistore_file.get_all_credential_keys(FILENAME) |
| self.assertEquals([], keys) |
| |
| # store credentials |
| credentials = self._create_test_credentials(client_id='client1') |
| custom_key = {'myapp': 'testing', 'clientid': 'client1'} |
| store1 = multistore_file.get_credential_storage_custom_key( |
| FILENAME, custom_key) |
| store1.put(credentials) |
| |
| keys = multistore_file.get_all_credential_keys(FILENAME) |
| self.assertEquals([custom_key], keys) |
| |
| # store more credentials |
| credentials = self._create_test_credentials(client_id='client2') |
| string_key = 'string_key' |
| store2 = multistore_file.get_credential_storage_custom_string_key( |
| FILENAME, string_key) |
| store2.put(credentials) |
| |
| keys = multistore_file.get_all_credential_keys(FILENAME) |
| self.assertEquals(2, len(keys)) |
| self.assertTrue(custom_key in keys) |
| self.assertTrue({'key': string_key} in keys) |
| |
| # back to no keys |
| store1.delete() |
| store2.delete() |
| keys = multistore_file.get_all_credential_keys(FILENAME) |
| self.assertEquals([], keys) |
| |
| def _refresh_data_cache_helper(self): |
| multistore = multistore_file._MultiStore(FILENAME) |
| json_patch = mock.patch.object(multistore, '_locked_json_read') |
| |
| return multistore, json_patch |
| |
| def test__refresh_data_cache_bad_json(self): |
| multistore, json_patch = self._refresh_data_cache_helper() |
| |
| with json_patch as json_mock: |
| json_mock.side_effect = ValueError('') |
| multistore._refresh_data_cache() |
| self.assertTrue(json_mock.called) |
| self.assertEqual(multistore._data, {}) |
| |
| def test__refresh_data_cache_bad_version(self): |
| multistore, json_patch = self._refresh_data_cache_helper() |
| |
| with json_patch as json_mock: |
| json_mock.return_value = {} |
| multistore._refresh_data_cache() |
| self.assertTrue(json_mock.called) |
| self.assertEqual(multistore._data, {}) |
| |
| def test__refresh_data_cache_newer_version(self): |
| multistore, json_patch = self._refresh_data_cache_helper() |
| |
| with json_patch as json_mock: |
| json_mock.return_value = {'file_version': 5} |
| with self.assertRaises(multistore_file.NewerCredentialStoreError): |
| multistore._refresh_data_cache() |
| self.assertTrue(json_mock.called) |
| |
| def test__refresh_data_cache_bad_credentials(self): |
| multistore, json_patch = self._refresh_data_cache_helper() |
| |
| with json_patch as json_mock: |
| json_mock.return_value = { |
| 'file_version': 1, |
| 'data': [ |
| {'lol': 'this is a bad credential object.'} |
| ]} |
| multistore._refresh_data_cache() |
| self.assertTrue(json_mock.called) |
| self.assertEqual(multistore._data, {}) |
| |
| def test__delete_credential_nonexistent(self): |
| multistore = multistore_file._MultiStore(FILENAME) |
| |
| with mock.patch.object(multistore, '_write') as write_mock: |
| multistore._data = {} |
| multistore._delete_credential('nonexistent_key') |
| self.assertTrue(write_mock.called) |
