Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-04-14 17:36
📄 __init__.py	0 B	2025-04-14 17:36
📁 data	-	2025-04-14 17:36
📄 test__metadata.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import datetime import http.client as http_client import importlib import json import os import mock import pytest # type: ignore from google.auth import _helpers from google.auth import environment_vars from google.auth import exceptions from google.auth import transport from google.auth.compute_engine import _metadata PATH = "instance/service-accounts/default" DATA_DIR = os.path.join(os.path.dirname(__file__), "data") SMBIOS_PRODUCT_NAME_FILE = os.path.join(DATA_DIR, "smbios_product_name") SMBIOS_PRODUCT_NAME_NONEXISTENT_FILE = os.path.join( DATA_DIR, "smbios_product_name_nonexistent" ) SMBIOS_PRODUCT_NAME_NON_GOOGLE = os.path.join( DATA_DIR, "smbios_product_name_non_google" ) ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE = ( "gl-python/3.7 auth/1.1 auth-request-type/at cred-type/mds" ) MDS_PING_METRICS_HEADER_VALUE = "gl-python/3.7 auth/1.1 auth-request-type/mds" MDS_PING_REQUEST_HEADER = { "metadata-flavor": "Google", "x-goog-api-client": MDS_PING_METRICS_HEADER_VALUE, } def make_request(data, status=http_client.OK, headers=None, retry=False): response = mock.create_autospec(transport.Response, instance=True) response.status = status response.data = _helpers.to_bytes(data) response.headers = headers or {} request = mock.create_autospec(transport.Request) if retry: request.side_effect = [exceptions.TransportError(), response] else: request.return_value = response return request def test_detect_gce_residency_linux_success(): _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_FILE assert _metadata.detect_gce_residency_linux() def test_detect_gce_residency_linux_non_google(): _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_NON_GOOGLE assert not _metadata.detect_gce_residency_linux() def test_detect_gce_residency_linux_nonexistent(): _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_NONEXISTENT_FILE assert not _metadata.detect_gce_residency_linux() def test_is_on_gce_ping_success(): request = make_request("", headers=_metadata._METADATA_HEADERS) assert _metadata.is_on_gce(request) @mock.patch("os.name", new="nt") def test_is_on_gce_windows_success(): request = make_request("", headers={_metadata._METADATA_FLAVOR_HEADER: "meep"}) assert not _metadata.is_on_gce(request) @mock.patch("os.name", new="posix") def test_is_on_gce_linux_success(): request = make_request("", headers={_metadata._METADATA_FLAVOR_HEADER: "meep"}) _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_FILE assert _metadata.is_on_gce(request) @mock.patch("google.auth.metrics.mds_ping", return_value=MDS_PING_METRICS_HEADER_VALUE) def test_ping_success(mock_metrics_header_value): request = make_request("", headers=_metadata._METADATA_HEADERS) assert _metadata.ping(request) request.assert_called_once_with( method="GET", url=_metadata._METADATA_IP_ROOT, headers=MDS_PING_REQUEST_HEADER, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) @mock.patch("google.auth.metrics.mds_ping", return_value=MDS_PING_METRICS_HEADER_VALUE) def test_ping_success_retry(mock_metrics_header_value): request = make_request("", headers=_metadata._METADATA_HEADERS, retry=True) assert _metadata.ping(request) request.assert_called_with( method="GET", url=_metadata._METADATA_IP_ROOT, headers=MDS_PING_REQUEST_HEADER, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert request.call_count == 2 @mock.patch("time.sleep", return_value=None) def test_ping_failure_bad_flavor(mock_sleep): request = make_request("", headers={_metadata._METADATA_FLAVOR_HEADER: "meep"}) assert not _metadata.ping(request) @mock.patch("time.sleep", return_value=None) def test_ping_failure_connection_failed(mock_sleep): request = make_request("") request.side_effect = exceptions.TransportError() assert not _metadata.ping(request) @mock.patch("google.auth.metrics.mds_ping", return_value=MDS_PING_METRICS_HEADER_VALUE) def test_ping_success_custom_root(mock_metrics_header_value): request = make_request("", headers=_metadata._METADATA_HEADERS) fake_ip = "1.2.3.4" os.environ[environment_vars.GCE_METADATA_IP] = fake_ip importlib.reload(_metadata) try: assert _metadata.ping(request) finally: del os.environ[environment_vars.GCE_METADATA_IP] importlib.reload(_metadata) request.assert_called_once_with( method="GET", url="http://" + fake_ip, headers=MDS_PING_REQUEST_HEADER, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) def test_get_success_json(): key, value = "foo", "bar" data = json.dumps({key: value}) request = make_request(data, headers={"content-type": "application/json"}) result = _metadata.get(request, PATH) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert result[key] == value def test_get_success_json_content_type_charset(): key, value = "foo", "bar" data = json.dumps({key: value}) request = make_request( data, headers={"content-type": "application/json; charset=UTF-8"} ) result = _metadata.get(request, PATH) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert result[key] == value @mock.patch("time.sleep", return_value=None) def test_get_success_retry(mock_sleep): key, value = "foo", "bar" data = json.dumps({key: value}) request = make_request( data, headers={"content-type": "application/json"}, retry=True ) result = _metadata.get(request, PATH) request.assert_called_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert request.call_count == 2 assert result[key] == value def test_get_success_text(): data = "foobar" request = make_request(data, headers={"content-type": "text/plain"}) result = _metadata.get(request, PATH) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert result == data def test_get_success_params(): data = "foobar" request = make_request(data, headers={"content-type": "text/plain"}) params = {"recursive": "true"} result = _metadata.get(request, PATH, params=params) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH + "?recursive=true", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert result == data def test_get_success_recursive_and_params(): data = "foobar" request = make_request(data, headers={"content-type": "text/plain"}) params = {"recursive": "false"} result = _metadata.get(request, PATH, recursive=True, params=params) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH + "?recursive=true", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert result == data def test_get_success_recursive(): data = "foobar" request = make_request(data, headers={"content-type": "text/plain"}) result = _metadata.get(request, PATH, recursive=True) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH + "?recursive=true", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert result == data def test_get_success_custom_root_new_variable(): request = make_request("{}", headers={"content-type": "application/json"}) fake_root = "another.metadata.service" os.environ[environment_vars.GCE_METADATA_HOST] = fake_root importlib.reload(_metadata) try: _metadata.get(request, PATH) finally: del os.environ[environment_vars.GCE_METADATA_HOST] importlib.reload(_metadata) request.assert_called_once_with( method="GET", url="http://{}/computeMetadata/v1/{}".format(fake_root, PATH), headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) def test_get_success_custom_root_old_variable(): request = make_request("{}", headers={"content-type": "application/json"}) fake_root = "another.metadata.service" os.environ[environment_vars.GCE_METADATA_ROOT] = fake_root importlib.reload(_metadata) try: _metadata.get(request, PATH) finally: del os.environ[environment_vars.GCE_METADATA_ROOT] importlib.reload(_metadata) request.assert_called_once_with( method="GET", url="http://{}/computeMetadata/v1/{}".format(fake_root, PATH), headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) @mock.patch("time.sleep", return_value=None) def test_get_failure(mock_sleep): request = make_request("Metadata error", status=http_client.NOT_FOUND) with pytest.raises(exceptions.TransportError) as excinfo: _metadata.get(request, PATH) assert excinfo.match(r"Metadata error") request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) def test_get_return_none_for_not_found_error(): request = make_request("Metadata error", status=http_client.NOT_FOUND) assert _metadata.get(request, PATH, return_none_for_not_found_error=True) is None request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) @mock.patch("time.sleep", return_value=None) def test_get_failure_connection_failed(mock_sleep): request = make_request("") request.side_effect = exceptions.TransportError("failure message") with pytest.raises(exceptions.TransportError) as excinfo: _metadata.get(request, PATH) assert excinfo.match( r"Compute Engine Metadata server unavailable due to failure message" ) request.assert_called_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert request.call_count == 5 def test_get_too_many_requests_retryable_error_failure(): request = make_request("too many requests", status=http_client.TOO_MANY_REQUESTS) with pytest.raises(exceptions.TransportError) as excinfo: _metadata.get(request, PATH) assert excinfo.match( r"Compute Engine Metadata server unavailable due to too many requests" ) request.assert_called_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert request.call_count == 5 def test_get_failure_bad_json(): request = make_request("{", headers={"content-type": "application/json"}) with pytest.raises(exceptions.TransportError) as excinfo: _metadata.get(request, PATH) assert excinfo.match(r"invalid JSON") request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH, headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) def test_get_project_id(): project = "example-project" request = make_request(project, headers={"content-type": "text/plain"}) project_id = _metadata.get_project_id(request) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + "project/project-id", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert project_id == project def test_get_universe_domain_success(): request = make_request( "fake_universe_domain", headers={"content-type": "text/plain"} ) universe_domain = _metadata.get_universe_domain(request) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + "universe/universe-domain", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert universe_domain == "fake_universe_domain" def test_get_universe_domain_success_empty_response(): request = make_request("", headers={"content-type": "text/plain"}) universe_domain = _metadata.get_universe_domain(request) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + "universe/universe-domain", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert universe_domain == "googleapis.com" def test_get_universe_domain_not_found(): # Test that if the universe domain endpoint returns 404 error, we should # use googleapis.com as the universe domain request = make_request("not found", status=http_client.NOT_FOUND) universe_domain = _metadata.get_universe_domain(request) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + "universe/universe-domain", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert universe_domain == "googleapis.com" def test_get_universe_domain_retryable_error_failure(): # Test that if the universe domain endpoint returns a retryable error # we should retry. # # In this case, the error persists, and we still fail after retrying. request = make_request("too many requests", status=http_client.TOO_MANY_REQUESTS) with pytest.raises(exceptions.TransportError) as excinfo: _metadata.get_universe_domain(request) assert excinfo.match(r"Compute Engine Metadata server unavailable") request.assert_called_with( method="GET", url=_metadata._METADATA_ROOT + "universe/universe-domain", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert request.call_count == 5 def test_get_universe_domain_retryable_error_success(): # Test that if the universe domain endpoint returns a retryable error # we should retry. # # In this case, the error is temporary, and we succeed after retrying. request_error = make_request( "too many requests", status=http_client.TOO_MANY_REQUESTS ) request_ok = make_request( "fake_universe_domain", headers={"content-type": "text/plain"} ) class _RequestErrorOnce: """This class forwards the request parameters to `request_error` once. All subsequent calls are forwarded to `request_ok`. """ def __init__(self, request_error, request_ok): self._request_error = request_error self._request_ok = request_ok self._call_index = 0 def request(self, args, kwargs): if self._call_index == 0: self._call_index += 1 return self._request_error(args, *kwargs) return self._request_ok(args, **kwargs) request = _RequestErrorOnce(request_error, request_ok).request universe_domain = _metadata.get_universe_domain(request) request_error.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + "universe/universe-domain", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) request_ok.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + "universe/universe-domain", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert universe_domain == "fake_universe_domain" def test_get_universe_domain_other_error(): # Test that if the universe domain endpoint returns an error other than 404 # we should throw the error request = make_request("unauthorized", status=http_client.UNAUTHORIZED) with pytest.raises(exceptions.TransportError) as excinfo: _metadata.get_universe_domain(request) assert excinfo.match(r"unauthorized") request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + "universe/universe-domain", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) @mock.patch( "google.auth.metrics.token_request_access_token_mds", return_value=ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) def test_get_service_account_token(utcnow, mock_metrics_header_value): ttl = 500 request = make_request( json.dumps({"access_token": "token", "expires_in": ttl}), headers={"content-type": "application/json"}, ) token, expiry = _metadata.get_service_account_token(request) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH + "/token", headers={ "metadata-flavor": "Google", "x-goog-api-client": ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, }, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert token == "token" assert expiry == utcnow() + datetime.timedelta(seconds=ttl) @mock.patch( "google.auth.metrics.token_request_access_token_mds", return_value=ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) def test_get_service_account_token_with_scopes_list(utcnow, mock_metrics_header_value): ttl = 500 request = make_request( json.dumps({"access_token": "token", "expires_in": ttl}), headers={"content-type": "application/json"}, ) token, expiry = _metadata.get_service_account_token(request, scopes=["foo", "bar"]) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH + "/token" + "?scopes=foo%2Cbar", headers={ "metadata-flavor": "Google", "x-goog-api-client": ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, }, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert token == "token" assert expiry == utcnow() + datetime.timedelta(seconds=ttl) @mock.patch( "google.auth.metrics.token_request_access_token_mds", return_value=ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) def test_get_service_account_token_with_scopes_string( utcnow, mock_metrics_header_value ): ttl = 500 request = make_request( json.dumps({"access_token": "token", "expires_in": ttl}), headers={"content-type": "application/json"}, ) token, expiry = _metadata.get_service_account_token(request, scopes="foo,bar") request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH + "/token" + "?scopes=foo%2Cbar", headers={ "metadata-flavor": "Google", "x-goog-api-client": ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, }, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert token == "token" assert expiry == utcnow() + datetime.timedelta(seconds=ttl) def test_get_service_account_info(): key, value = "foo", "bar" request = make_request( json.dumps({key: value}), headers={"content-type": "application/json"} ) info = _metadata.get_service_account_info(request) request.assert_called_once_with( method="GET", url=_metadata._METADATA_ROOT + PATH + "/?recursive=true", headers=_metadata._METADATA_HEADERS, timeout=_metadata._METADATA_DEFAULT_TIMEOUT, ) assert info[key] == value	20.58 KB	2025-04-14 17:36
📄 test_credentials.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import base64 import datetime import mock import pytest # type: ignore import responses # type: ignore from google.auth import _helpers from google.auth import exceptions from google.auth import jwt from google.auth import transport from google.auth.compute_engine import credentials from google.auth.transport import requests SAMPLE_ID_TOKEN_EXP = 1584393400 # header: {"alg": "RS256", "typ": "JWT", "kid": "1"} # payload: {"iss": "issuer", "iat": 1584393348, "sub": "subject", # "exp": 1584393400,"aud": "audience"} SAMPLE_ID_TOKEN = ( b"eyJhbGciOiAiUlMyNTYiLCAidHlwIjogIkpXVCIsICJraWQiOiAiMSJ9." b"eyJpc3MiOiAiaXNzdWVyIiwgImlhdCI6IDE1ODQzOTMzNDgsICJzdWIiO" b"iAic3ViamVjdCIsICJleHAiOiAxNTg0MzkzNDAwLCAiYXVkIjogImF1ZG" b"llbmNlIn0." b"OquNjHKhTmlgCk361omRo18F_uY-7y0f_AmLbzW062Q1Zr61HAwHYP5FM" b"316CK4_0cH8MUNGASsvZc3VqXAqub6PUTfhemH8pFEwBdAdG0LhrNkU0H" b"WN1YpT55IiQ31esLdL5q-qDsOPpNZJUti1y1lAreM5nIn2srdWzGXGs4i" b"TRQsn0XkNUCL4RErpciXmjfhMrPkcAjKA-mXQm2fa4jmTlEZFqFmUlym1" b"ozJ0yf5grjN6AslN4OGvAv1pS-_Ko_pGBS6IQtSBC6vVKCUuBfaqNjykg" b"bsxbLa6Fp0SYeYwO8ifEnkRvasVpc1WTQqfRB2JCj5pTBDzJpIpFCMmnQ" ) ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE = ( "gl-python/3.7 auth/1.1 auth-request-type/at cred-type/mds" ) ID_TOKEN_REQUEST_METRICS_HEADER_VALUE = ( "gl-python/3.7 auth/1.1 auth-request-type/it cred-type/mds" ) FAKE_SERVICE_ACCOUNT_EMAIL = "foo@bar.com" FAKE_QUOTA_PROJECT_ID = "fake-quota-project" FAKE_SCOPES = ["scope1", "scope2"] FAKE_DEFAULT_SCOPES = ["scope3", "scope4"] FAKE_UNIVERSE_DOMAIN = "fake-universe-domain" class TestCredentials(object): credentials = None credentials_with_all_fields = None @pytest.fixture(autouse=True) def credentials_fixture(self): self.credentials = credentials.Credentials() self.credentials_with_all_fields = credentials.Credentials( service_account_email=FAKE_SERVICE_ACCOUNT_EMAIL, quota_project_id=FAKE_QUOTA_PROJECT_ID, scopes=FAKE_SCOPES, default_scopes=FAKE_DEFAULT_SCOPES, universe_domain=FAKE_UNIVERSE_DOMAIN, ) def test_get_cred_info(self): assert self.credentials.get_cred_info() == { "credential_source": "metadata server", "credential_type": "VM credentials", "principal": "default", } def test_default_state(self): assert not self.credentials.valid # Expiration hasn't been set yet assert not self.credentials.expired # Scopes are needed assert self.credentials.requires_scopes # Service account email hasn't been populated assert self.credentials.service_account_email == "default" # No quota project assert not self.credentials._quota_project_id # Universe domain is the default and not cached assert self.credentials._universe_domain == "googleapis.com" assert not self.credentials._universe_domain_cached @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.min + _helpers.REFRESH_THRESHOLD, ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_refresh_success(self, get, utcnow): get.side_effect = [ { # First request is for sevice account info. "email": "service-account@example.com", "scopes": ["one", "two"], }, { # Second request is for the token. "access_token": "token", "expires_in": 500, }, ] # Refresh credentials self.credentials.refresh(None) # Check that the credentials have the token and proper expiration assert self.credentials.token == "token" assert self.credentials.expiry == (utcnow() + datetime.timedelta(seconds=500)) # Check the credential info assert self.credentials.service_account_email == "service-account@example.com" assert self.credentials._scopes == ["one", "two"] # Check that the credentials are valid (have a token and are not # expired) assert self.credentials.valid @mock.patch( "google.auth.metrics.token_request_access_token_mds", return_value=ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.min + _helpers.REFRESH_THRESHOLD, ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_refresh_success_with_scopes(self, get, utcnow, mock_metrics_header_value): get.side_effect = [ { # First request is for sevice account info. "email": "service-account@example.com", "scopes": ["one", "two"], }, { # Second request is for the token. "access_token": "token", "expires_in": 500, }, ] # Refresh credentials scopes = ["three", "four"] self.credentials = self.credentials.with_scopes(scopes) self.credentials.refresh(None) # Check that the credentials have the token and proper expiration assert self.credentials.token == "token" assert self.credentials.expiry == (utcnow() + datetime.timedelta(seconds=500)) # Check the credential info assert self.credentials.service_account_email == "service-account@example.com" assert self.credentials._scopes == scopes # Check that the credentials are valid (have a token and are not # expired) assert self.credentials.valid kwargs = get.call_args[1] assert kwargs["params"] == {"scopes": "three,four"} assert kwargs["headers"] == { "x-goog-api-client": ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE } @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_refresh_error(self, get): get.side_effect = exceptions.TransportError("http error") with pytest.raises(exceptions.RefreshError) as excinfo: self.credentials.refresh(None) assert excinfo.match(r"http error") @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_before_request_refreshes(self, get): get.side_effect = [ { # First request is for sevice account info. "email": "service-account@example.com", "scopes": "one two", }, { # Second request is for the token. "access_token": "token", "expires_in": 500, }, ] # Credentials should start as invalid assert not self.credentials.valid # before_request should cause a refresh request = mock.create_autospec(transport.Request, instance=True) self.credentials.before_request(request, "GET", "http://example.com?a=1#3", {}) # The refresh endpoint should've been called. assert get.called # Credentials should now be valid. assert self.credentials.valid def test_with_quota_project(self): creds = self.credentials_with_all_fields.with_quota_project("project-foo") assert creds._quota_project_id == "project-foo" assert creds._service_account_email == FAKE_SERVICE_ACCOUNT_EMAIL assert creds._scopes == FAKE_SCOPES assert creds._default_scopes == FAKE_DEFAULT_SCOPES assert creds.universe_domain == FAKE_UNIVERSE_DOMAIN assert creds._universe_domain_cached def test_with_scopes(self): scopes = ["one", "two"] creds = self.credentials_with_all_fields.with_scopes(scopes) assert creds._scopes == scopes assert creds._quota_project_id == FAKE_QUOTA_PROJECT_ID assert creds._service_account_email == FAKE_SERVICE_ACCOUNT_EMAIL assert creds._default_scopes is None assert creds.universe_domain == FAKE_UNIVERSE_DOMAIN assert creds._universe_domain_cached def test_with_universe_domain(self): creds = self.credentials_with_all_fields.with_universe_domain("universe_domain") assert creds._scopes == FAKE_SCOPES assert creds._quota_project_id == FAKE_QUOTA_PROJECT_ID assert creds._service_account_email == FAKE_SERVICE_ACCOUNT_EMAIL assert creds._default_scopes == FAKE_DEFAULT_SCOPES assert creds.universe_domain == "universe_domain" assert creds._universe_domain_cached def test_token_usage_metrics(self): self.credentials.token = "token" self.credentials.expiry = None headers = {} self.credentials.before_request(mock.Mock(), None, None, headers) assert headers["authorization"] == "Bearer token" assert headers["x-goog-api-client"] == "cred-type/mds" @mock.patch( "google.auth.compute_engine._metadata.get_universe_domain", return_value="fake_universe_domain", ) def test_universe_domain(self, get_universe_domain): # Check the default state assert not self.credentials._universe_domain_cached assert self.credentials._universe_domain == "googleapis.com" # calling the universe_domain property should trigger a call to # get_universe_domain to fetch the value. The value should be cached. assert self.credentials.universe_domain == "fake_universe_domain" assert self.credentials._universe_domain == "fake_universe_domain" assert self.credentials._universe_domain_cached get_universe_domain.assert_called_once() # calling the universe_domain property the second time should use the # cached value instead of calling get_universe_domain assert self.credentials.universe_domain == "fake_universe_domain" get_universe_domain.assert_called_once() @mock.patch("google.auth.compute_engine._metadata.get_universe_domain") def test_user_provided_universe_domain(self, get_universe_domain): assert self.credentials_with_all_fields.universe_domain == FAKE_UNIVERSE_DOMAIN assert self.credentials_with_all_fields._universe_domain_cached # Since user provided universe_domain, we will not call the universe # domain endpoint. get_universe_domain.assert_not_called() class TestIDTokenCredentials(object): credentials = None @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_default_state(self, get): get.side_effect = [ {"email": "service-account@example.com", "scope": ["one", "two"]} ] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://example.com" ) assert not self.credentials.valid # Expiration hasn't been set yet assert not self.credentials.expired # Service account email hasn't been populated assert self.credentials.service_account_email == "service-account@example.com" # Signer is initialized assert self.credentials.signer assert self.credentials.signer_email == "service-account@example.com" # No quota project assert not self.credentials._quota_project_id @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_make_authorization_grant_assertion(self, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com" ) # Generate authorization grant: token = self.credentials._make_authorization_grant_assertion() payload = jwt.decode(token, verify=False) # The JWT token signature is 'signature' encoded in base 64: assert token.endswith(b".c2lnbmF0dXJl") # Check that the credentials have the token and proper expiration assert payload == { "aud": "https://www.googleapis.com/oauth2/v4/token", "exp": 3600, "iat": 0, "iss": "service-account@example.com", "target_audience": "https://audience.com", } @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_with_service_account(self, sign, get, utcnow): sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com", service_account_email="service-account@other.com", ) # Generate authorization grant: token = self.credentials._make_authorization_grant_assertion() payload = jwt.decode(token, verify=False) # The JWT token signature is 'signature' encoded in base 64: assert token.endswith(b".c2lnbmF0dXJl") # Check that the credentials have the token and proper expiration assert payload == { "aud": "https://www.googleapis.com/oauth2/v4/token", "exp": 3600, "iat": 0, "iss": "service-account@other.com", "target_audience": "https://audience.com", } @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_additional_claims(self, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com", additional_claims={"foo": "bar"}, ) # Generate authorization grant: token = self.credentials._make_authorization_grant_assertion() payload = jwt.decode(token, verify=False) # The JWT token signature is 'signature' encoded in base 64: assert token.endswith(b".c2lnbmF0dXJl") # Check that the credentials have the token and proper expiration assert payload == { "aud": "https://www.googleapis.com/oauth2/v4/token", "exp": 3600, "iat": 0, "iss": "service-account@example.com", "target_audience": "https://audience.com", "foo": "bar", } def test_token_uri(self): request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, signer=mock.Mock(), service_account_email="foo@example.com", target_audience="https://audience.com", ) assert self.credentials._token_uri == credentials._DEFAULT_TOKEN_URI self.credentials = credentials.IDTokenCredentials( request=request, signer=mock.Mock(), service_account_email="foo@example.com", target_audience="https://audience.com", token_uri="https://example.com/token", ) assert self.credentials._token_uri == "https://example.com/token" @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_with_target_audience(self, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com" ) self.credentials = self.credentials.with_target_audience("https://actually.not") # Generate authorization grant: token = self.credentials._make_authorization_grant_assertion() payload = jwt.decode(token, verify=False) # The JWT token signature is 'signature' encoded in base 64: assert token.endswith(b".c2lnbmF0dXJl") # Check that the credentials have the token and proper expiration assert payload == { "aud": "https://www.googleapis.com/oauth2/v4/token", "exp": 3600, "iat": 0, "iss": "service-account@example.com", "target_audience": "https://actually.not", } # Check that the signer have been initialized with a Request object assert isinstance(self.credentials._signer._request, transport.Request) @responses.activate def test_with_target_audience_integration(self): """ Test that it is possible to refresh credentials generated from `with_target_audience`. Instead of mocking the methods, the HTTP responses have been mocked. """ # mock information about credentials responses.add( responses.GET, "http://metadata.google.internal/computeMetadata/v1/instance/" "service-accounts/default/?recursive=true", status=200, content_type="application/json", json={ "scopes": "email", "email": "service-account@example.com", "aliases": ["default"], }, ) # mock information about universe_domain responses.add( responses.GET, "http://metadata.google.internal/computeMetadata/v1/universe/" "universe-domain", status=200, content_type="application/json", json={}, ) # mock token for credentials responses.add( responses.GET, "http://metadata.google.internal/computeMetadata/v1/instance/" "service-accounts/service-account@example.com/token", status=200, content_type="application/json", json={ "access_token": "some-token", "expires_in": 3210, "token_type": "Bearer", }, ) # mock sign blob endpoint signature = base64.b64encode(b"some-signature").decode("utf-8") responses.add( responses.POST, "https://iamcredentials.googleapis.com/v1/projects/-/" "serviceAccounts/service-account@example.com:signBlob", status=200, content_type="application/json", json={"keyId": "some-key-id", "signedBlob": signature}, ) id_token = "{}.{}.{}".format( base64.b64encode(b'{"some":"some"}').decode("utf-8"), base64.b64encode(b'{"exp": 3210}').decode("utf-8"), base64.b64encode(b"token").decode("utf-8"), ) # mock id token endpoint responses.add( responses.POST, "https://www.googleapis.com/oauth2/v4/token", status=200, content_type="application/json", json={"id_token": id_token, "expiry": 3210}, ) self.credentials = credentials.IDTokenCredentials( request=requests.Request(), service_account_email="service-account@example.com", target_audience="https://audience.com", ) self.credentials = self.credentials.with_target_audience("https://actually.not") self.credentials.refresh(requests.Request()) assert self.credentials.token is not None @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_with_quota_project(self, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com" ) self.credentials = self.credentials.with_quota_project("project-foo") assert self.credentials._quota_project_id == "project-foo" # Generate authorization grant: token = self.credentials._make_authorization_grant_assertion() payload = jwt.decode(token, verify=False) # The JWT token signature is 'signature' encoded in base 64: assert token.endswith(b".c2lnbmF0dXJl") # Check that the credentials have the token and proper expiration assert payload == { "aud": "https://www.googleapis.com/oauth2/v4/token", "exp": 3600, "iat": 0, "iss": "service-account@example.com", "target_audience": "https://audience.com", } # Check that the signer have been initialized with a Request object assert isinstance(self.credentials._signer._request, transport.Request) @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_with_token_uri(self, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com", token_uri="http://xyz.com", ) assert self.credentials._token_uri == "http://xyz.com" creds_with_token_uri = self.credentials.with_token_uri("http://example.com") assert creds_with_token_uri._token_uri == "http://example.com" @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_with_token_uri_exception(self, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com", use_metadata_identity_endpoint=True, ) assert self.credentials._token_uri is None with pytest.raises(ValueError): self.credentials.with_token_uri("http://example.com") @responses.activate def test_with_quota_project_integration(self): """ Test that it is possible to refresh credentials generated from `with_quota_project`. Instead of mocking the methods, the HTTP responses have been mocked. """ # mock information about credentials responses.add( responses.GET, "http://metadata.google.internal/computeMetadata/v1/instance/" "service-accounts/default/?recursive=true", status=200, content_type="application/json", json={ "scopes": "email", "email": "service-account@example.com", "aliases": ["default"], }, ) # mock token for credentials responses.add( responses.GET, "http://metadata.google.internal/computeMetadata/v1/instance/" "service-accounts/service-account@example.com/token", status=200, content_type="application/json", json={ "access_token": "some-token", "expires_in": 3210, "token_type": "Bearer", }, ) # stubby response about universe_domain responses.add( responses.GET, "http://metadata.google.internal/computeMetadata/v1/universe/" "universe-domain", status=200, content_type="application/json", json={}, ) # mock sign blob endpoint signature = base64.b64encode(b"some-signature").decode("utf-8") responses.add( responses.POST, "https://iamcredentials.googleapis.com/v1/projects/-/" "serviceAccounts/service-account@example.com:signBlob", status=200, content_type="application/json", json={"keyId": "some-key-id", "signedBlob": signature}, ) id_token = "{}.{}.{}".format( base64.b64encode(b'{"some":"some"}').decode("utf-8"), base64.b64encode(b'{"exp": 3210}').decode("utf-8"), base64.b64encode(b"token").decode("utf-8"), ) # mock id token endpoint responses.add( responses.POST, "https://www.googleapis.com/oauth2/v4/token", status=200, content_type="application/json", json={"id_token": id_token, "expiry": 3210}, ) self.credentials = credentials.IDTokenCredentials( request=requests.Request(), service_account_email="service-account@example.com", target_audience="https://audience.com", ) self.credentials = self.credentials.with_quota_project("project-foo") self.credentials.refresh(requests.Request()) assert self.credentials.token is not None assert self.credentials._quota_project_id == "project-foo" @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) @mock.patch("google.oauth2._client.id_token_jwt_grant", autospec=True) def test_refresh_success(self, id_token_jwt_grant, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] id_token_jwt_grant.side_effect = [ ("idtoken", datetime.datetime.utcfromtimestamp(3600), {}) ] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com" ) # Refresh credentials self.credentials.refresh(None) # Check that the credentials have the token and proper expiration assert self.credentials.token == "idtoken" assert self.credentials.expiry == (datetime.datetime.utcfromtimestamp(3600)) # Check the credential info assert self.credentials.service_account_email == "service-account@example.com" # Check that the credentials are valid (have a token and are not # expired) assert self.credentials.valid @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_refresh_error(self, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) response = mock.Mock() response.data = b'{"error": "http error"}' response.status = 404 # Throw a 404 so the request is not retried. request.side_effect = [response] self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com" ) with pytest.raises(exceptions.RefreshError) as excinfo: self.credentials.refresh(request) assert excinfo.match(r"http error") @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.utcfromtimestamp(0), ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) @mock.patch("google.oauth2._client.id_token_jwt_grant", autospec=True) def test_before_request_refreshes(self, id_token_jwt_grant, sign, get, utcnow): get.side_effect = [ {"email": "service-account@example.com", "scopes": "one two"} ] sign.side_effect = [b"signature"] id_token_jwt_grant.side_effect = [ ("idtoken", datetime.datetime.utcfromtimestamp(3600), {}) ] request = mock.create_autospec(transport.Request, instance=True) self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com" ) # Credentials should start as invalid assert not self.credentials.valid # before_request should cause a refresh request = mock.create_autospec(transport.Request, instance=True) self.credentials.before_request(request, "GET", "http://example.com?a=1#3", {}) # The refresh endpoint should've been called. assert get.called # Credentials should now be valid. assert self.credentials.valid @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) @mock.patch("google.auth.iam.Signer.sign", autospec=True) def test_sign_bytes(self, sign, get): get.side_effect = [ {"email": "service-account@example.com", "scopes": ["one", "two"]} ] sign.side_effect = [b"signature"] request = mock.create_autospec(transport.Request, instance=True) response = mock.Mock() response.data = b'{"signature": "c2lnbmF0dXJl"}' response.status = 200 request.side_effect = [response] self.credentials = credentials.IDTokenCredentials( request=request, target_audience="https://audience.com" ) # Generate authorization grant: signature = self.credentials.sign_bytes(b"some bytes") # The JWT token signature is 'signature' encoded in base 64: assert signature == b"signature" @mock.patch( "google.auth.metrics.token_request_id_token_mds", return_value=ID_TOKEN_REQUEST_METRICS_HEADER_VALUE, ) @mock.patch( "google.auth.compute_engine._metadata.get_service_account_info", autospec=True ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_get_id_token_from_metadata( self, get, get_service_account_info, mock_metrics_header_value ): get.return_value = SAMPLE_ID_TOKEN get_service_account_info.return_value = {"email": "foo@example.com"} cred = credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True ) cred.refresh(request=mock.Mock()) assert get.call_args.kwargs["headers"] == { "x-goog-api-client": ID_TOKEN_REQUEST_METRICS_HEADER_VALUE } assert cred.token == SAMPLE_ID_TOKEN assert cred.expiry == datetime.datetime.utcfromtimestamp(SAMPLE_ID_TOKEN_EXP) assert cred._use_metadata_identity_endpoint assert cred._signer is None assert cred._token_uri is None assert cred._service_account_email == "foo@example.com" assert cred._target_audience == "audience" with pytest.raises(ValueError): cred.sign_bytes(b"bytes") @mock.patch( "google.auth.compute_engine._metadata.get_service_account_info", autospec=True ) def test_with_target_audience_for_metadata(self, get_service_account_info): get_service_account_info.return_value = {"email": "foo@example.com"} cred = credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True ) cred = cred.with_target_audience("new_audience") assert cred._target_audience == "new_audience" assert cred._use_metadata_identity_endpoint assert cred._signer is None assert cred._token_uri is None assert cred._service_account_email == "foo@example.com" @mock.patch( "google.auth.compute_engine._metadata.get_service_account_info", autospec=True ) def test_id_token_with_quota_project(self, get_service_account_info): get_service_account_info.return_value = {"email": "foo@example.com"} cred = credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True ) cred = cred.with_quota_project("project-foo") assert cred._quota_project_id == "project-foo" assert cred._use_metadata_identity_endpoint assert cred._signer is None assert cred._token_uri is None assert cred._service_account_email == "foo@example.com" @mock.patch( "google.auth.compute_engine._metadata.get_service_account_info", autospec=True ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_invalid_id_token_from_metadata(self, get, get_service_account_info): get.return_value = "invalid_id_token" get_service_account_info.return_value = {"email": "foo@example.com"} cred = credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True ) with pytest.raises(ValueError): cred.refresh(request=mock.Mock()) @mock.patch( "google.auth.compute_engine._metadata.get_service_account_info", autospec=True ) @mock.patch("google.auth.compute_engine._metadata.get", autospec=True) def test_transport_error_from_metadata(self, get, get_service_account_info): get.side_effect = exceptions.TransportError("transport error") get_service_account_info.return_value = {"email": "foo@example.com"} cred = credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True ) with pytest.raises(exceptions.RefreshError) as excinfo: cred.refresh(request=mock.Mock()) assert excinfo.match(r"transport error") def test_get_id_token_from_metadata_constructor(self): with pytest.raises(ValueError): credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True, token_uri="token_uri", ) with pytest.raises(ValueError): credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True, signer=mock.Mock(), ) with pytest.raises(ValueError): credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True, additional_claims={"key", "value"}, ) with pytest.raises(ValueError): credentials.IDTokenCredentials( mock.Mock(), "audience", use_metadata_identity_endpoint=True, service_account_email="foo@example.com", )	36.66 KB	2025-04-14 17:36

Terminal