From 3ef47fa2d5f97f1c34a7fec923e0c7b6b3cb68fb Mon Sep 17 00:00:00 2001 From: Danny Hermes Date: Mon, 13 Apr 2015 10:33:52 -0700 Subject: [PATCH] Allowing custom directory for well-known file. Also raising exception if the directory does not exist. Fixes #151. --- oauth2client/client.py | 36 +++++++++++++++++--------------- tests/test_devshell.py | 11 +++++++--- tests/test_gce.py | 11 ++++++++-- tests/test_oauth2client.py | 42 +++++++++++++++++++++++++++++++++----- 4 files changed, 74 insertions(+), 26 deletions(-) diff --git a/oauth2client/client.py b/oauth2client/client.py index 05f9818..b6fd369 100644 --- a/oauth2client/client.py +++ b/oauth2client/client.py @@ -81,6 +81,8 @@ GOOGLE_APPLICATION_CREDENTIALS = 'GOOGLE_APPLICATION_CREDENTIALS' # The ~/.config subdirectory containing gcloud credentials. Intended # to be swapped out in tests. _CLOUDSDK_CONFIG_DIRECTORY = 'gcloud' +# The environment variable name which can replace ~/.config if set. +_CLOUDSDK_CONFIG_ENV_VAR = 'CLOUDSDK_CONFIG' # The error message we show users when we can't find the Application # Default Credentials. @@ -1268,24 +1270,26 @@ def _get_well_known_file(): WELL_KNOWN_CREDENTIALS_FILE = 'application_default_credentials.json' - if os.name == 'nt': - try: - default_config_path = os.path.join(os.environ['APPDATA'], - _CLOUDSDK_CONFIG_DIRECTORY) - except KeyError: - # This should never happen unless someone is really messing with things. - drive = os.environ.get('SystemDrive', 'C:') - default_config_path = os.path.join(drive, '\\', - _CLOUDSDK_CONFIG_DIRECTORY) - else: - default_config_path = os.path.join(os.path.expanduser('~'), - '.config', - _CLOUDSDK_CONFIG_DIRECTORY) + default_config_dir = os.getenv(_CLOUDSDK_CONFIG_ENV_VAR) + if default_config_dir is None: + if os.name == 'nt': + try: + default_config_dir = os.path.join(os.environ['APPDATA'], + _CLOUDSDK_CONFIG_DIRECTORY) + except KeyError: + # This should never happen unless someone is really messing with things. + drive = os.environ.get('SystemDrive', 'C:') + default_config_dir = os.path.join(drive, '\\', + _CLOUDSDK_CONFIG_DIRECTORY) + else: + default_config_dir = os.path.join(os.path.expanduser('~'), + '.config', + _CLOUDSDK_CONFIG_DIRECTORY) - default_config_path = os.path.join(default_config_path, - WELL_KNOWN_CREDENTIALS_FILE) + if not os.path.isdir(default_config_dir): + raise OSError('Config directory does not exist', default_config_dir) - return default_config_path + return os.path.join(default_config_dir, WELL_KNOWN_CREDENTIALS_FILE) def _get_application_default_credential_from_file(filename): diff --git a/tests/test_devshell.py b/tests/test_devshell.py index 7a13ab4..9da33a7 100644 --- a/tests/test_devshell.py +++ b/tests/test_devshell.py @@ -125,6 +125,11 @@ class DevshellCredentialsTests(unittest.TestCase): self.assertEqual('sometoken', creds.access_token) def test_refuses_to_save_to_well_known_file(self): - with _AuthReferenceServer(): - creds = DevshellCredentials() - self.assertRaises(NotImplementedError, save_to_well_known_file, creds) + ORIGINAL_ISDIR = os.path.isdir + try: + os.path.isdir = lambda path: True + with _AuthReferenceServer(): + creds = DevshellCredentials() + self.assertRaises(NotImplementedError, save_to_well_known_file, creds) + finally: + os.path.isdir = ORIGINAL_ISDIR diff --git a/tests/test_gce.py b/tests/test_gce.py index ef93499..a8d3999 100644 --- a/tests/test_gce.py +++ b/tests/test_gce.py @@ -97,5 +97,12 @@ class AssertionCredentialsTests(unittest.TestCase): 'default/acquire?scope=dummy_scope') def test_save_to_well_known_file(self): - credentials = AppAssertionCredentials([]) - self.assertRaises(NotImplementedError, save_to_well_known_file, credentials) + import os + ORIGINAL_ISDIR = os.path.isdir + try: + os.path.isdir = lambda path: True + credentials = AppAssertionCredentials([]) + self.assertRaises(NotImplementedError, save_to_well_known_file, + credentials) + finally: + os.path.isdir = ORIGINAL_ISDIR diff --git a/tests/test_oauth2client.py b/tests/test_oauth2client.py index 70268c1..66140b7 100644 --- a/tests/test_oauth2client.py +++ b/tests/test_oauth2client.py @@ -247,11 +247,40 @@ class GoogleCredentialsTests(unittest.TestCase): str(error)) def test_get_well_known_file_on_windows(self): - well_known_file = datafile( - os.path.join('gcloud', 'application_default_credentials.json')) - os.name = 'nt' - os.environ['APPDATA'] = DATA_DIR - self.assertEqual(well_known_file, _get_well_known_file()) + ORIGINAL_ISDIR = os.path.isdir + try: + os.path.isdir = lambda path: True + well_known_file = datafile( + os.path.join(client._CLOUDSDK_CONFIG_DIRECTORY, + 'application_default_credentials.json')) + os.name = 'nt' + os.environ['APPDATA'] = DATA_DIR + self.assertEqual(well_known_file, _get_well_known_file()) + finally: + os.path.isdir = ORIGINAL_ISDIR + + def test_get_well_known_file_with_custom_config_dir(self): + ORIGINAL_ENVIRON = os.environ + ORIGINAL_ISDIR = os.path.isdir + CUSTOM_DIR = 'CUSTOM_DIR' + EXPECTED_FILE = os.path.join(CUSTOM_DIR, + 'application_default_credentials.json') + try: + os.environ = {client._CLOUDSDK_CONFIG_ENV_VAR: CUSTOM_DIR} + os.path.isdir = lambda path: True + well_known_file = _get_well_known_file() + self.assertEqual(well_known_file, EXPECTED_FILE) + finally: + os.environ = ORIGINAL_ENVIRON + os.path.isdir = ORIGINAL_ISDIR + + def test_get_well_known_file_with_non_existent_config_dir(self): + ORIGINAL_ISDIR = os.path.isdir + try: + os.path.isdir = lambda path: False + self.assertRaises(OSError, _get_well_known_file) + finally: + os.path.isdir = ORIGINAL_ISDIR def test_get_application_default_credential_from_file_service_account(self): credentials_file = datafile( @@ -410,13 +439,16 @@ class GoogleCredentialsTests(unittest.TestCase): os.environ['APPDATA'] = '' # we can't use self.assertRaisesRegexp() because it is only in Python 2.7+ VALID_CONFIG_DIR = client._CLOUDSDK_CONFIG_DIRECTORY + ORIGINAL_ISDIR = os.path.isdir try: + os.path.isdir = lambda path: True client._CLOUDSDK_CONFIG_DIRECTORY = 'BOGUS_CONFIG_DIR' GoogleCredentials.get_application_default() self.fail('An exception was expected!') except ApplicationDefaultCredentialsError as error: self.assertEqual(ADC_HELP_MSG, str(error)) finally: + os.path.isdir = ORIGINAL_ISDIR client._CLOUDSDK_CONFIG_DIRECTORY = VALID_CONFIG_DIR def test_from_stream_service_account(self):