100% coverage for oauth2client.contrib.multistore_file

oauth2client/contrib/multistore_file.py

@@ -108,7 +108,7 @@ def get_credential_storage(filename, client_id, user_agent, scope,
     key = {'clientId': client_id, 'userAgent': user_agent,
            'scope': util.scopes_to_string(scope)}
     return get_credential_storage_custom_key(
-      filename, key, warn_on_readonly=warn_on_readonly)
+        filename, key, warn_on_readonly=warn_on_readonly)
 
 
 @util.positional(2)
@@ -131,7 +131,7 @@ def get_credential_storage_custom_string_key(filename, key_string,
     # Create a key dictionary that can be used
     key_dict = {'key': key_string}
     return get_credential_storage_custom_key(
-      filename, key_dict, warn_on_readonly=warn_on_readonly)
+        filename, key_dict, warn_on_readonly=warn_on_readonly)
 
 
 @util.positional(2)
@@ -319,6 +319,7 @@ def _lock(self):
                             'Opening in read-only mode. Any refreshed '
                             'credentials will only be '
                             'valid for this run.', self._file.filename())
+
         if os.path.getsize(self._file.filename()) == 0:
             logger.debug('Initializing empty multistore file')
             # The multistore is empty so write out an empty file.

tests/contrib/test_multistore_file.py

@@ -21,6 +21,8 @@
 import tempfile
 import unittest2
 
+import mock
+
 from oauth2client import util
 from oauth2client.client import OAuth2Credentials
 from oauth2client.contrib import locked_file
@@ -116,6 +118,18 @@ def test_lock_file_raises_ioerror(self):
         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, errno.EBUSY)
+            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()
 
@@ -133,6 +147,32 @@ def test_read_only_file_fail_lock(self):
             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'
@@ -168,12 +208,14 @@ def test_multistore_file(self):
             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()
 
@@ -282,6 +324,61 @@ def test_multistore_file_get_all_keys(self):
         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}
+            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)
+
 
 if __name__ == '__main__':  # pragma: NO COVER
     unittest2.main()