feat: make load_credentials_from_file a public method (#530)

* feat: make load_credentials_from_file public and alllow scopes

* test: update tests

* fix: raise error for json with no type

* test: fix test names

* refactor: simplify control flow

* fix: raise coverage

* test: update test

Co-authored-by: arithmetic1728 <58957152+arithmetic1728@users.noreply.github.com>
Co-authored-by: Sijun Liu <sijunliu@google.com>

Author: busunkim96

google/auth/__init__.py

@@ -16,10 +16,10 @@
 
 import logging
 
-from google.auth._default import default
+from google.auth._default import default, load_credentials_from_file
 
 
-__all__ = ["default"]
+__all__ = ["default", "load_credentials_from_file"]
 
 
 # Set default logging handler to avoid "No handler found" warnings.

google/auth/_default.py

@@ -69,14 +69,17 @@ def _warn_about_problematic_credentials(credentials):
         warnings.warn(_CLOUD_SDK_CREDENTIALS_WARNING)
 
 
-def _load_credentials_from_file(filename):
-    """Loads credentials from a file.
+def load_credentials_from_file(filename, scopes=None):
+    """Loads Google credentials from a file.
 
     The credentials file must be a service account key or stored authorized
     user credentials.
 
     Args:
         filename (str): The full path to the credentials file.
+        scopes (Optional[Sequence[str]]): The list of scopes for the credentials. If
+            specified, the credentials will automatically be scoped if
+            necessary.
 
     Returns:
         Tuple[google.auth.credentials.Credentials, Optional[str]]: Loaded
@@ -109,7 +112,9 @@ def _load_credentials_from_file(filename):
         from google.oauth2 import credentials
 
         try:
-            credentials = credentials.Credentials.from_authorized_user_info(info)
+            credentials = credentials.Credentials.from_authorized_user_info(
+                info, scopes=scopes
+            )
         except ValueError as caught_exc:
             msg = "Failed to load authorized user credentials from {}".format(filename)
             new_exc = exceptions.DefaultCredentialsError(msg, caught_exc)
@@ -122,7 +127,9 @@ def _load_credentials_from_file(filename):
         from google.oauth2 import service_account
 
         try:
-            credentials = service_account.Credentials.from_service_account_info(info)
+            credentials = service_account.Credentials.from_service_account_info(
+                info, scopes=scopes
+            )
         except ValueError as caught_exc:
             msg = "Failed to load service account credentials from {}".format(filename)
             new_exc = exceptions.DefaultCredentialsError(msg, caught_exc)
@@ -148,7 +155,7 @@ def _get_gcloud_sdk_credentials():
     if not os.path.isfile(credentials_filename):
         return None, None
 
-    credentials, project_id = _load_credentials_from_file(credentials_filename)
+    credentials, project_id = load_credentials_from_file(credentials_filename)
 
     if not project_id:
         project_id = _cloud_sdk.get_project_id()
@@ -162,7 +169,7 @@ def _get_explicit_environ_credentials():
     explicit_file = os.environ.get(environment_vars.CREDENTIALS)
 
     if explicit_file is not None:
-        credentials, project_id = _load_credentials_from_file(
+        credentials, project_id = load_credentials_from_file(
             os.environ[environment_vars.CREDENTIALS]
         )
 

tests/test__default.py

@@ -43,88 +43,120 @@
 
 SERVICE_ACCOUNT_FILE = os.path.join(DATA_DIR, "service_account.json")
 
+CLIENT_SECRETS_FILE = os.path.join(DATA_DIR, "client_secrets.json")
+
 with open(SERVICE_ACCOUNT_FILE) as fh:
     SERVICE_ACCOUNT_FILE_DATA = json.load(fh)
 
 LOAD_FILE_PATCH = mock.patch(
-    "google.auth._default._load_credentials_from_file",
+    "google.auth._default.load_credentials_from_file",
     return_value=(mock.sentinel.credentials, mock.sentinel.project_id),
     autospec=True,
 )
 
 
-def test__load_credentials_from_missing_file():
+def test_load_credentials_from_missing_file():
     with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
-        _default._load_credentials_from_file("")
+        _default.load_credentials_from_file("")
 
     assert excinfo.match(r"not found")
 
 
-def test__load_credentials_from_file_invalid_json(tmpdir):
+def test_load_credentials_from_file_invalid_json(tmpdir):
     jsonfile = tmpdir.join("invalid.json")
     jsonfile.write("{")
 
     with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
-        _default._load_credentials_from_file(str(jsonfile))
+        _default.load_credentials_from_file(str(jsonfile))
 
     assert excinfo.match(r"not a valid json file")
 
 
-def test__load_credentials_from_file_invalid_type(tmpdir):
+def test_load_credentials_from_file_invalid_type(tmpdir):
     jsonfile = tmpdir.join("invalid.json")
     jsonfile.write(json.dumps({"type": "not-a-real-type"}))
 
     with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
-        _default._load_credentials_from_file(str(jsonfile))
+        _default.load_credentials_from_file(str(jsonfile))
 
     assert excinfo.match(r"does not have a valid type")
 
 
-def test__load_credentials_from_file_authorized_user():
-    credentials, project_id = _default._load_credentials_from_file(AUTHORIZED_USER_FILE)
+def test_load_credentials_from_file_authorized_user():
+    credentials, project_id = _default.load_credentials_from_file(AUTHORIZED_USER_FILE)
     assert isinstance(credentials, google.oauth2.credentials.Credentials)
     assert project_id is None
 
 
-def test__load_credentials_from_file_authorized_user_bad_format(tmpdir):
+def test_load_credentials_from_file_no_type(tmpdir):
+    # use the client_secrets.json, which is valid json but not a
+    # loadable credentials type
+    with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
+        _default.load_credentials_from_file(CLIENT_SECRETS_FILE)
+
+    assert excinfo.match(r"does not have a valid type")
+    assert excinfo.match(r"Type is None")
+
+
+def test_load_credentials_from_file_authorized_user_bad_format(tmpdir):
     filename = tmpdir.join("authorized_user_bad.json")
     filename.write(json.dumps({"type": "authorized_user"}))
 
     with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
-        _default._load_credentials_from_file(str(filename))
+        _default.load_credentials_from_file(str(filename))
 
     assert excinfo.match(r"Failed to load authorized user")
     assert excinfo.match(r"missing fields")
 
 
-def test__load_credentials_from_file_authorized_user_cloud_sdk():
+def test_load_credentials_from_file_authorized_user_cloud_sdk():
     with pytest.warns(UserWarning, match="Cloud SDK"):
-        credentials, project_id = _default._load_credentials_from_file(
+        credentials, project_id = _default.load_credentials_from_file(
             AUTHORIZED_USER_CLOUD_SDK_FILE
         )
     assert isinstance(credentials, google.oauth2.credentials.Credentials)
     assert project_id is None
 
     # No warning if the json file has quota project id.
-    credentials, project_id = _default._load_credentials_from_file(
+    credentials, project_id = _default.load_credentials_from_file(
         AUTHORIZED_USER_CLOUD_SDK_WITH_QUOTA_PROJECT_ID_FILE
     )
     assert isinstance(credentials, google.oauth2.credentials.Credentials)
     assert project_id is None
 
 
-def test__load_credentials_from_file_service_account():
-    credentials, project_id = _default._load_credentials_from_file(SERVICE_ACCOUNT_FILE)
+def test_load_credentials_from_file_authorized_user_cloud_sdk_with_scopes():
+    with pytest.warns(UserWarning, match="Cloud SDK"):
+        credentials, project_id = _default.load_credentials_from_file(
+            AUTHORIZED_USER_CLOUD_SDK_FILE,
+            scopes=["https://www.google.com/calendar/feeds"],
+        )
+    assert isinstance(credentials, google.oauth2.credentials.Credentials)
+    assert project_id is None
+    assert credentials.scopes == ["https://www.google.com/calendar/feeds"]
+
+
+def test_load_credentials_from_file_service_account():
+    credentials, project_id = _default.load_credentials_from_file(SERVICE_ACCOUNT_FILE)
+    assert isinstance(credentials, service_account.Credentials)
+    assert project_id == SERVICE_ACCOUNT_FILE_DATA["project_id"]
+
+
+def test_load_credentials_from_file_service_account_with_scopes():
+    credentials, project_id = _default.load_credentials_from_file(
+        SERVICE_ACCOUNT_FILE, scopes=["https://www.google.com/calendar/feeds"]
+    )
     assert isinstance(credentials, service_account.Credentials)
     assert project_id == SERVICE_ACCOUNT_FILE_DATA["project_id"]
+    assert credentials.scopes == ["https://www.google.com/calendar/feeds"]
 
 
-def test__load_credentials_from_file_service_account_bad_format(tmpdir):
+def test_load_credentials_from_file_service_account_bad_format(tmpdir):
     filename = tmpdir.join("serivce_account_bad.json")
     filename.write(json.dumps({"type": "service_account"}))
 
     with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
-        _default._load_credentials_from_file(str(filename))
+        _default.load_credentials_from_file(str(filename))
 
     assert excinfo.match(r"Failed to load service account")
     assert excinfo.match(r"missing fields")