diff --git a/auth/gcloud/aio/auth/token.py b/auth/gcloud/aio/auth/token.py index 405134c4a..f476abf04 100644 --- a/auth/gcloud/aio/auth/token.py +++ b/auth/gcloud/aio/auth/token.py @@ -6,6 +6,8 @@ import json import os import time +from pathlib import Path +from typing import cast from typing import Any from typing import AnyStr from typing import Dict @@ -27,13 +29,6 @@ # where plumbing this error through will require several changes to otherwise- # good error handling. -# Handle differences in exceptions -try: - # TODO: Type[Exception] should work here, no? - CustomFileError: Any = FileNotFoundError -except NameError: - CustomFileError = IOError - # Selectively load libraries based on the package if BUILD_GCLOUD_REST: @@ -52,6 +47,7 @@ '/default/token?recursive=true') GCLOUD_TOKEN_DURATION = 3600 REFRESH_HEADERS = {'Content-Type': 'application/x-www-form-urlencoded'} +ServiceFile = Optional[Union[str, IO[AnyStr], Path]] class Type(enum.Enum): @@ -60,40 +56,41 @@ class Type(enum.Enum): SERVICE_ACCOUNT = 'service_account' -def get_service_data( - service: Optional[Union[str, IO[AnyStr]]]) -> Dict[str, Any]: - service = service or os.environ.get('GOOGLE_APPLICATION_CREDENTIALS') +def get_service_data(service: ServiceFile) -> Dict[str, str]: + # if a stream passed explicitly, read it + if hasattr(service, 'read'): + return json.loads(service.read()) # type: ignore + service = cast(Union[None, str, Path], service) + + set_explicitly = True + if not service: + service = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS') if not service: - cloudsdk_config = os.environ.get('CLOUDSDK_CONFIG') - sdkpath = (cloudsdk_config - or os.path.join(os.path.expanduser('~'), '.config', - 'gcloud')) - service = os.path.join(sdkpath, 'application_default_credentials.json') - set_explicitly = bool(cloudsdk_config) - else: - set_explicitly = True + service = os.environ.get('CLOUDSDK_CONFIG') + if not service: + service = Path.home() / '.config' / 'gcloud' + set_explicitly = False + + service_path = Path(service) + if service_path.is_dir(): + service_path = service_path / 'application_default_credentials.json' + + # if not an existing file, try to read as a raw content + if isinstance(service, str) and not service_path.exists(): + return json.loads(service) try: - try: - with open(service) as f: # type: ignore[arg-type] - data: Dict[str, Any] = json.loads(f.read()) - return data - except TypeError: - data = json.loads(service.read()) # type: ignore[union-attr] - return data - except CustomFileError: + with service_path.open('r', encoding='utf8') as stream: + return json.load(stream) + except Exception: # pylint: disable=broad-except if set_explicitly: - # only warn users if they have explicitly set the service_file path raise - - return {} - except Exception: # pylint: disable=broad-except return {} class Token: # pylint: disable=too-many-instance-attributes - def __init__(self, service_file: Optional[Union[str, IO[AnyStr]]] = None, + def __init__(self, service_file: ServiceFile = None, session: Optional[Session] = None, scopes: Optional[List[str]] = None) -> None: self.service_data = get_service_data(service_file) diff --git a/auth/tests/unit/token_test.py b/auth/tests/unit/token_test.py index 3d63505a0..4e40cbae0 100644 --- a/auth/tests/unit/token_test.py +++ b/auth/tests/unit/token_test.py @@ -1,5 +1,7 @@ import io +import os import json +from pathlib import Path import gcloud.aio.auth.token as token import pytest @@ -32,3 +34,73 @@ async def test_service_as_io(): assert t.token_type == token.Type.SERVICE_ACCOUNT assert t.token_uri == 'https://oauth2.googleapis.com/token' assert await t.get_project() == 'random-project-123' + + +@pytest.fixture +def chdir(tmp_path): + old_dir = os.curdir + os.chdir(str(tmp_path)) + try: + yield + finally: + os.chdir(old_dir) + + +@pytest.fixture +def clean_environ(): + old_environ = os.environ.copy() + os.environ.clear() + try: + yield + finally: + os.environ.update(old_environ) + + +@pytest.mark.parametrize('given, expected', [ + ('{"name": "aragorn"}', {'name': 'aragorn'}), + (io.StringIO('{"name": "aragorn"}'), {'name': 'aragorn'}), + ('key.json', {'hello': 'world'}), + (Path('key.json'), {'hello': 'world'}), +]) +def test_get_service_data__explicit(tmp_path: Path, chdir, given, expected): + (tmp_path / 'key.json').write_text('{"hello": "world"}') + assert token.get_service_data(given) == expected + + +@pytest.mark.parametrize('given, expected', [ + ('something', json.JSONDecodeError), + (io.StringIO('something'), json.JSONDecodeError), + (Path('something'), FileNotFoundError), +]) +def test_get_service_data__explicit__raise(given, expected): + with pytest.raises(expected): + token.get_service_data(given) + + +@pytest.mark.parametrize('given, expected', [ + ({'GOOGLE_APPLICATION_CREDENTIALS': 'key.json'}, {'hello': 'world'}), + ({'GOOGLE_APPLICATION_CREDENTIALS': '{"name": "aragorn"}'}, {'name': 'aragorn'}), + ({'CLOUDSDK_CONFIG': '.'}, {'hi': 'mark'}), + ({'CLOUDSDK_CONFIG': '{"name": "aragorn"}'}, {'name': 'aragorn'}), +]) +def test_get_service_data__explicit_env_var( + tmp_path: Path, chdir, clean_environ, given, expected, +): + (tmp_path / 'key.json').write_text('{"hello": "world"}') + (tmp_path / 'application_default_credentials.json').write_text('{"hi": "mark"}') + os.environ.update(given) + assert token.get_service_data(None) == expected + + +def test_get_service_data__explicit_env_var__raises(clean_environ): + os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = 'garbage' + with pytest.raises(json.JSONDecodeError): + token.get_service_data(None) + + +SDK_CONFIG = Path.home() / '.config' / 'gcloud' / 'application_default_credentials.json' + + +@pytest.mark.skipif(not SDK_CONFIG.exists(), reason='no default credentials installed') +def test_get_service_data__implicit_sdk_config(clean_environ): + assert 'client_id' in token.get_service_data(None)