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>
diff --git a/google/auth/__init__.py b/google/auth/__init__.py
index 6b4b78b..5ca20a3 100644
--- a/google/auth/__init__.py
+++ b/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.
diff --git a/google/auth/_default.py b/google/auth/_default.py
index 559695e..d3bbbda 100644
--- a/google/auth/_default.py
+++ b/google/auth/_default.py
@@ -69,14 +69,17 @@
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 @@
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 @@
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 @@
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 @@
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]
)
diff --git a/tests/test__default.py b/tests/test__default.py
index b769fc7..3c87b35 100644
--- a/tests/test__default.py
+++ b/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_bad_format(tmpdir):
+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):
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")