auth: allow to pass credentials as text

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)

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)