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
📄 test__client_async.py # Copyright 2020 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 json import urllib import mock import pytest # type: ignore from google.auth import _helpers from google.auth import _jwt_async as jwt from google.auth import exceptions from google.oauth2 import _client as sync_client from google.oauth2 import _client_async as _client from tests.oauth2 import test__client as test_client def make_request(response_data, status=http_client.OK, text=False): response = mock.AsyncMock(spec=["transport.Response"]) response.status = status data = response_data if text else json.dumps(response_data).encode("utf-8") response.data = mock.AsyncMock(spec=["__call__", "read"]) response.data.read = mock.AsyncMock(spec=["__call__"], return_value=data) response.content = mock.AsyncMock(spec=["__call__"], return_value=data) request = mock.AsyncMock(spec=["transport.Request"]) request.return_value = response return request @pytest.mark.asyncio async def test__token_endpoint_request(): request = make_request({"test": "response"}) result = await _client._token_endpoint_request( request, "http://example.com", {"test": "params"} ) # Check request call request.assert_called_with( method="POST", url="http://example.com", headers={"Content-Type": "application/x-www-form-urlencoded"}, body="test=params".encode("utf-8"), ) # Check result assert result == {"test": "response"} @pytest.mark.asyncio async def test__token_endpoint_request_text(): request = make_request("response", text=True) result = await _client._token_endpoint_request( request, "http://example.com", {"test": "params"} ) # Check request call request.assert_called_with( method="POST", url="http://example.com", headers={"Content-Type": "application/x-www-form-urlencoded"}, body="test=params".encode("utf-8"), ) # Check result assert result == "response" @pytest.mark.asyncio async def test__token_endpoint_request_json(): request = make_request({"test": "response"}) access_token = "access_token" result = await _client._token_endpoint_request( request, "http://example.com", {"test": "params"}, access_token=access_token, use_json=True, ) # Check request call request.assert_called_with( method="POST", url="http://example.com", headers={ "Content-Type": "application/json", "Authorization": "Bearer access_token", }, body=b'{"test": "params"}', ) # Check result assert result == {"test": "response"} @pytest.mark.asyncio async def test__token_endpoint_request_error(): request = make_request({}, status=http_client.BAD_REQUEST) with pytest.raises(exceptions.RefreshError) as excinfo: await _client._token_endpoint_request(request, "http://example.com", {}) assert not excinfo.value.retryable @pytest.mark.asyncio async def test__token_endpoint_request_internal_failure_error(): request = make_request( {"error_description": "internal_failure"}, status=http_client.BAD_REQUEST ) with pytest.raises(exceptions.RefreshError) as excinfo: await _client._token_endpoint_request( request, "http://example.com", {"error_description": "internal_failure"} ) assert excinfo.value.retryable request = make_request( {"error": "internal_failure"}, status=http_client.BAD_REQUEST ) with pytest.raises(exceptions.RefreshError): await _client._token_endpoint_request( request, "http://example.com", {"error": "internal_failure"} ) assert excinfo.value.retryable @pytest.mark.asyncio async def test__token_endpoint_request_internal_failure_and_retry_failure_error(): retryable_error = mock.AsyncMock(spec=["transport.Response"]) retryable_error.status = http_client.BAD_REQUEST data = json.dumps({"error_description": "internal_failure"}).encode("utf-8") retryable_error.data = mock.AsyncMock(spec=["__call__", "read"]) retryable_error.data.read = mock.AsyncMock(spec=["__call__"], return_value=data) retryable_error.content = mock.AsyncMock(spec=["__call__"], return_value=data) unretryable_error = mock.AsyncMock(spec=["transport.Response"]) unretryable_error.status = http_client.BAD_REQUEST data = json.dumps({"error_description": "invalid_scope"}).encode("utf-8") unretryable_error.data = mock.AsyncMock(spec=["__call__", "read"]) unretryable_error.data.read = mock.AsyncMock(spec=["__call__"], return_value=data) unretryable_error.content = mock.AsyncMock(spec=["__call__"], return_value=data) request = mock.AsyncMock(spec=["transport.Request"]) request.side_effect = [retryable_error, retryable_error, unretryable_error] with pytest.raises(exceptions.RefreshError): await _client._token_endpoint_request( request, "http://example.com", {"error_description": "invalid_scope"} ) # request should be called three times. Two retryable errors and one # unretryable error to break the retry loop. assert request.call_count == 3 @pytest.mark.asyncio async def test__token_endpoint_request_internal_failure_and_retry_succeeds(): retryable_error = mock.AsyncMock(spec=["transport.Response"]) retryable_error.status = http_client.BAD_REQUEST data = json.dumps({"error_description": "internal_failure"}).encode("utf-8") retryable_error.data = mock.AsyncMock(spec=["__call__", "read"]) retryable_error.data.read = mock.AsyncMock(spec=["__call__"], return_value=data) retryable_error.content = mock.AsyncMock(spec=["__call__"], return_value=data) response = mock.AsyncMock(spec=["transport.Response"]) response.status = http_client.OK data = json.dumps({"hello": "world"}).encode("utf-8") response.data = mock.AsyncMock(spec=["__call__", "read"]) response.data.read = mock.AsyncMock(spec=["__call__"], return_value=data) response.content = mock.AsyncMock(spec=["__call__"], return_value=data) request = mock.AsyncMock(spec=["transport.Request"]) request.side_effect = [retryable_error, response] _ = await _client._token_endpoint_request( request, "http://example.com", {"test": "params"} ) assert request.call_count == 2 def verify_request_params(request, params): request_body = request.call_args[1]["body"].decode("utf-8") request_params = urllib.parse.parse_qs(request_body) for key, value in params.items(): assert request_params[key][0] == value @pytest.mark.asyncio @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) async def test_jwt_grant(utcnow): request = make_request( {"access_token": "token", "expires_in": 500, "extra": "data"} ) token, expiry, extra_data = await _client.jwt_grant( request, "http://example.com", "assertion_value" ) # Check request call verify_request_params( request, {"grant_type": sync_client._JWT_GRANT_TYPE, "assertion": "assertion_value"}, ) # Check result assert token == "token" assert expiry == utcnow() + datetime.timedelta(seconds=500) assert extra_data["extra"] == "data" @pytest.mark.asyncio async def test_jwt_grant_no_access_token(): request = make_request( { # No access token. "expires_in": 500, "extra": "data", } ) with pytest.raises(exceptions.RefreshError) as excinfo: await _client.jwt_grant(request, "http://example.com", "assertion_value") assert not excinfo.value.retryable @pytest.mark.asyncio async def test_id_token_jwt_grant(): now = _helpers.utcnow() id_token_expiry = _helpers.datetime_to_secs(now) id_token = jwt.encode(test_client.SIGNER, {"exp": id_token_expiry}).decode("utf-8") request = make_request({"id_token": id_token, "extra": "data"}) token, expiry, extra_data = await _client.id_token_jwt_grant( request, "http://example.com", "assertion_value" ) # Check request call verify_request_params( request, {"grant_type": sync_client._JWT_GRANT_TYPE, "assertion": "assertion_value"}, ) # Check result assert token == id_token # JWT does not store microseconds now = now.replace(microsecond=0) assert expiry == now assert extra_data["extra"] == "data" @pytest.mark.asyncio async def test_id_token_jwt_grant_no_access_token(): request = make_request( { # No access token. "expires_in": 500, "extra": "data", } ) with pytest.raises(exceptions.RefreshError) as excinfo: await _client.id_token_jwt_grant( request, "http://example.com", "assertion_value" ) assert not excinfo.value.retryable @pytest.mark.asyncio @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) async def test_refresh_grant(unused_utcnow): request = make_request( { "access_token": "token", "refresh_token": "new_refresh_token", "expires_in": 500, "extra": "data", } ) token, refresh_token, expiry, extra_data = await _client.refresh_grant( request, "http://example.com", "refresh_token", "client_id", "client_secret", rapt_token="rapt_token", ) # Check request call verify_request_params( request, { "grant_type": sync_client._REFRESH_GRANT_TYPE, "refresh_token": "refresh_token", "client_id": "client_id", "client_secret": "client_secret", "rapt": "rapt_token", }, ) # Check result assert token == "token" assert refresh_token == "new_refresh_token" assert expiry == datetime.datetime.min + datetime.timedelta(seconds=500) assert extra_data["extra"] == "data" @pytest.mark.asyncio @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) async def test_refresh_grant_with_scopes(unused_utcnow): request = make_request( { "access_token": "token", "refresh_token": "new_refresh_token", "expires_in": 500, "extra": "data", "scope": test_client.SCOPES_AS_STRING, } ) token, refresh_token, expiry, extra_data = await _client.refresh_grant( request, "http://example.com", "refresh_token", "client_id", "client_secret", test_client.SCOPES_AS_LIST, ) # Check request call. verify_request_params( request, { "grant_type": sync_client._REFRESH_GRANT_TYPE, "refresh_token": "refresh_token", "client_id": "client_id", "client_secret": "client_secret", "scope": test_client.SCOPES_AS_STRING, }, ) # Check result. assert token == "token" assert refresh_token == "new_refresh_token" assert expiry == datetime.datetime.min + datetime.timedelta(seconds=500) assert extra_data["extra"] == "data" @pytest.mark.asyncio async def test_refresh_grant_no_access_token(): request = make_request( { # No access token. "refresh_token": "new_refresh_token", "expires_in": 500, "extra": "data", } ) with pytest.raises(exceptions.RefreshError) as excinfo: await _client.refresh_grant( request, "http://example.com", "refresh_token", "client_id", "client_secret" ) assert not excinfo.value.retryable @pytest.mark.asyncio @mock.patch("google.oauth2._client._parse_expiry", return_value=None) @mock.patch.object(_client, "_token_endpoint_request", autospec=True) async def test_jwt_grant_retry_default(mock_token_endpoint_request, mock_expiry): _ = await _client.jwt_grant(mock.Mock(), mock.Mock(), mock.Mock()) mock_token_endpoint_request.assert_called_with( mock.ANY, mock.ANY, mock.ANY, can_retry=True ) @pytest.mark.asyncio @pytest.mark.parametrize("can_retry", [True, False]) @mock.patch("google.oauth2._client._parse_expiry", return_value=None) @mock.patch.object(_client, "_token_endpoint_request", autospec=True) async def test_jwt_grant_retry_with_retry( mock_token_endpoint_request, mock_expiry, can_retry ): _ = await _client.jwt_grant( mock.AsyncMock(), mock.Mock(), mock.Mock(), can_retry=can_retry ) mock_token_endpoint_request.assert_called_with( mock.ANY, mock.ANY, mock.ANY, can_retry=can_retry ) @pytest.mark.asyncio @mock.patch("google.auth.jwt.decode", return_value={"exp": 0}) @mock.patch.object(_client, "_token_endpoint_request", autospec=True) async def test_id_token_jwt_grant_retry_default( mock_token_endpoint_request, mock_jwt_decode ): _ = await _client.id_token_jwt_grant(mock.Mock(), mock.Mock(), mock.Mock()) mock_token_endpoint_request.assert_called_with( mock.ANY, mock.ANY, mock.ANY, can_retry=True ) @pytest.mark.asyncio @pytest.mark.parametrize("can_retry", [True, False]) @mock.patch("google.auth.jwt.decode", return_value={"exp": 0}) @mock.patch.object(_client, "_token_endpoint_request", autospec=True) async def test_id_token_jwt_grant_retry_with_retry( mock_token_endpoint_request, mock_jwt_decode, can_retry ): _ = await _client.id_token_jwt_grant( mock.AsyncMock(), mock.AsyncMock(), mock.AsyncMock(), can_retry=can_retry ) mock_token_endpoint_request.assert_called_with( mock.ANY, mock.ANY, mock.ANY, can_retry=can_retry ) @pytest.mark.asyncio @mock.patch("google.oauth2._client._parse_expiry", return_value=None) @mock.patch.object(_client, "_token_endpoint_request", autospec=True) async def test_refresh_grant_retry_default( mock_token_endpoint_request, mock_parse_expiry ): _ = await _client.refresh_grant( mock.AsyncMock(), mock.AsyncMock(), mock.AsyncMock(), mock.AsyncMock(), mock.AsyncMock(), ) mock_token_endpoint_request.assert_called_with( mock.ANY, mock.ANY, mock.ANY, can_retry=True ) @pytest.mark.asyncio @pytest.mark.parametrize("can_retry", [True, False]) @mock.patch("google.oauth2._client._parse_expiry", return_value=None) @mock.patch.object(_client, "_token_endpoint_request", autospec=True) async def test_refresh_grant_retry_with_retry( mock_token_endpoint_request, mock_parse_expiry, can_retry ): _ = await _client.refresh_grant( mock.AsyncMock(), mock.AsyncMock(), mock.AsyncMock(), mock.AsyncMock(), mock.AsyncMock(), can_retry=can_retry, ) mock_token_endpoint_request.assert_called_with( mock.ANY, mock.ANY, mock.ANY, can_retry=can_retry ) @pytest.mark.asyncio @pytest.mark.parametrize("can_retry", [True, False]) async def test__token_endpoint_request_no_throw_with_retry(can_retry): mock_request = make_request( {"error": "help", "error_description": "I'm alive"}, http_client.INTERNAL_SERVER_ERROR, ) _ = await _client._token_endpoint_request_no_throw( mock_request, mock.AsyncMock(), "body", mock.AsyncMock(), mock.AsyncMock(), can_retry=can_retry, ) if can_retry: assert mock_request.call_count == 3 else: assert mock_request.call_count == 1	15.76 KB	2025-04-14 17:36
📄 test_credentials_async.py # Copyright 2020 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 json import os import pickle import sys import mock import pytest # type: ignore from google.auth import _helpers from google.auth import exceptions from google.oauth2 import _credentials_async as _credentials_async from google.oauth2 import credentials from tests.oauth2 import test_credentials class TestCredentials: TOKEN_URI = "https://example.com/oauth2/token" REFRESH_TOKEN = "refresh_token" CLIENT_ID = "client_id" CLIENT_SECRET = "client_secret" @classmethod def make_credentials(cls): return _credentials_async.Credentials( token=None, refresh_token=cls.REFRESH_TOKEN, token_uri=cls.TOKEN_URI, client_id=cls.CLIENT_ID, client_secret=cls.CLIENT_SECRET, enable_reauth_refresh=True, ) def test_default_state(self): credentials = self.make_credentials() assert not credentials.valid # Expiration hasn't been set yet assert not credentials.expired # Scopes aren't required for these credentials assert not credentials.requires_scopes # Test properties assert credentials.refresh_token == self.REFRESH_TOKEN assert credentials.token_uri == self.TOKEN_URI assert credentials.client_id == self.CLIENT_ID assert credentials.client_secret == self.CLIENT_SECRET @mock.patch("google.oauth2._reauth_async.refresh_grant", autospec=True) @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.min + _helpers.REFRESH_THRESHOLD, ) @pytest.mark.asyncio async def test_refresh_success(self, unused_utcnow, refresh_grant): token = "token" expiry = _helpers.utcnow() + datetime.timedelta(seconds=500) grant_response = {"id_token": mock.sentinel.id_token} rapt_token = "rapt_token" refresh_grant.return_value = ( # Access token token, # New refresh token None, # Expiry, expiry, # Extra data grant_response, # Rapt token rapt_token, ) request = mock.AsyncMock(spec=["transport.Request"]) creds = self.make_credentials() # Refresh credentials await creds.refresh(request) # Check jwt grant call. refresh_grant.assert_called_with( request, self.TOKEN_URI, self.REFRESH_TOKEN, self.CLIENT_ID, self.CLIENT_SECRET, None, None, True, ) # Check that the credentials have the token and expiry assert creds.token == token assert creds.expiry == expiry assert creds.id_token == mock.sentinel.id_token assert creds.rapt_token == rapt_token # Check that the credentials are valid (have a token and are not # expired) assert creds.valid @pytest.mark.asyncio async def test_refresh_no_refresh_token(self): request = mock.AsyncMock(spec=["transport.Request"]) credentials_ = _credentials_async.Credentials(token=None, refresh_token=None) with pytest.raises(exceptions.RefreshError, match="necessary fields"): await credentials_.refresh(request) request.assert_not_called() @mock.patch("google.oauth2._reauth_async.refresh_grant", autospec=True) @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.min + _helpers.REFRESH_THRESHOLD, ) @pytest.mark.asyncio async def test_credentials_with_scopes_requested_refresh_success( self, unused_utcnow, refresh_grant ): scopes = ["email", "profile"] token = "token" expiry = _helpers.utcnow() + datetime.timedelta(seconds=500) grant_response = {"id_token": mock.sentinel.id_token} rapt_token = "rapt_token" refresh_grant.return_value = ( # Access token token, # New refresh token None, # Expiry, expiry, # Extra data grant_response, # Rapt token rapt_token, ) request = mock.AsyncMock(spec=["transport.Request"]) creds = _credentials_async.Credentials( token=None, refresh_token=self.REFRESH_TOKEN, token_uri=self.TOKEN_URI, client_id=self.CLIENT_ID, client_secret=self.CLIENT_SECRET, scopes=scopes, rapt_token="old_rapt_token", ) # Refresh credentials await creds.refresh(request) # Check jwt grant call. refresh_grant.assert_called_with( request, self.TOKEN_URI, self.REFRESH_TOKEN, self.CLIENT_ID, self.CLIENT_SECRET, scopes, "old_rapt_token", False, ) # Check that the credentials have the token and expiry assert creds.token == token assert creds.expiry == expiry assert creds.id_token == mock.sentinel.id_token assert creds.has_scopes(scopes) assert creds.rapt_token == rapt_token # Check that the credentials are valid (have a token and are not # expired.) assert creds.valid @mock.patch("google.oauth2._reauth_async.refresh_grant", autospec=True) @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.min + _helpers.REFRESH_THRESHOLD, ) @pytest.mark.asyncio async def test_credentials_with_scopes_returned_refresh_success( self, unused_utcnow, refresh_grant ): scopes = ["email", "profile"] token = "token" expiry = _helpers.utcnow() + datetime.timedelta(seconds=500) grant_response = {"id_token": mock.sentinel.id_token, "scope": " ".join(scopes)} rapt_token = "rapt_token" refresh_grant.return_value = ( # Access token token, # New refresh token None, # Expiry, expiry, # Extra data grant_response, # Rapt token rapt_token, ) request = mock.AsyncMock(spec=["transport.Request"]) creds = _credentials_async.Credentials( token=None, refresh_token=self.REFRESH_TOKEN, token_uri=self.TOKEN_URI, client_id=self.CLIENT_ID, client_secret=self.CLIENT_SECRET, scopes=scopes, ) # Refresh credentials await creds.refresh(request) # Check jwt grant call. refresh_grant.assert_called_with( request, self.TOKEN_URI, self.REFRESH_TOKEN, self.CLIENT_ID, self.CLIENT_SECRET, scopes, None, False, ) # Check that the credentials have the token and expiry assert creds.token == token assert creds.expiry == expiry assert creds.id_token == mock.sentinel.id_token assert creds.has_scopes(scopes) assert creds.rapt_token == rapt_token # Check that the credentials are valid (have a token and are not # expired.) assert creds.valid @mock.patch("google.oauth2._reauth_async.refresh_grant", autospec=True) @mock.patch( "google.auth._helpers.utcnow", return_value=datetime.datetime.min + _helpers.REFRESH_THRESHOLD, ) @pytest.mark.asyncio async def test_credentials_with_scopes_refresh_failure_raises_refresh_error( self, unused_utcnow, refresh_grant ): scopes = ["email", "profile"] scopes_returned = ["email"] token = "token" expiry = _helpers.utcnow() + datetime.timedelta(seconds=500) grant_response = { "id_token": mock.sentinel.id_token, "scope": " ".join(scopes_returned), } rapt_token = "rapt_token" refresh_grant.return_value = ( # Access token token, # New refresh token None, # Expiry, expiry, # Extra data grant_response, # Rapt token rapt_token, ) request = mock.AsyncMock(spec=["transport.Request"]) creds = _credentials_async.Credentials( token=None, refresh_token=self.REFRESH_TOKEN, token_uri=self.TOKEN_URI, client_id=self.CLIENT_ID, client_secret=self.CLIENT_SECRET, scopes=scopes, rapt_token=None, ) # Refresh credentials with pytest.raises( exceptions.RefreshError, match="Not all requested scopes were granted" ): await creds.refresh(request) # Check jwt grant call. refresh_grant.assert_called_with( request, self.TOKEN_URI, self.REFRESH_TOKEN, self.CLIENT_ID, self.CLIENT_SECRET, scopes, None, False, ) # Check that the credentials have the token and expiry assert creds.token == token assert creds.expiry == expiry assert creds.id_token == mock.sentinel.id_token assert creds.has_scopes(scopes) # Check that the credentials are valid (have a token and are not # expired.) assert creds.valid def test_apply_with_quota_project_id(self): creds = _credentials_async.Credentials( token="token", refresh_token=self.REFRESH_TOKEN, token_uri=self.TOKEN_URI, client_id=self.CLIENT_ID, client_secret=self.CLIENT_SECRET, quota_project_id="quota-project-123", ) headers = {} creds.apply(headers) assert headers["x-goog-user-project"] == "quota-project-123" def test_apply_with_no_quota_project_id(self): creds = _credentials_async.Credentials( token="token", refresh_token=self.REFRESH_TOKEN, token_uri=self.TOKEN_URI, client_id=self.CLIENT_ID, client_secret=self.CLIENT_SECRET, ) headers = {} creds.apply(headers) assert "x-goog-user-project" not in headers def test_with_quota_project(self): creds = _credentials_async.Credentials( token="token", refresh_token=self.REFRESH_TOKEN, token_uri=self.TOKEN_URI, client_id=self.CLIENT_ID, client_secret=self.CLIENT_SECRET, quota_project_id="quota-project-123", ) new_creds = creds.with_quota_project("new-project-456") assert new_creds.quota_project_id == "new-project-456" headers = {} creds.apply(headers) assert "x-goog-user-project" in headers def test_from_authorized_user_info(self): info = test_credentials.AUTH_USER_INFO.copy() creds = _credentials_async.Credentials.from_authorized_user_info(info) assert creds.client_secret == info["client_secret"] assert creds.client_id == info["client_id"] assert creds.refresh_token == info["refresh_token"] assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT assert creds.scopes is None scopes = ["email", "profile"] creds = _credentials_async.Credentials.from_authorized_user_info(info, scopes) assert creds.client_secret == info["client_secret"] assert creds.client_id == info["client_id"] assert creds.refresh_token == info["refresh_token"] assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT assert creds.scopes == scopes def test_from_authorized_user_file(self): info = test_credentials.AUTH_USER_INFO.copy() creds = _credentials_async.Credentials.from_authorized_user_file( test_credentials.AUTH_USER_JSON_FILE ) assert creds.client_secret == info["client_secret"] assert creds.client_id == info["client_id"] assert creds.refresh_token == info["refresh_token"] assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT assert creds.scopes is None scopes = ["email", "profile"] creds = _credentials_async.Credentials.from_authorized_user_file( test_credentials.AUTH_USER_JSON_FILE, scopes ) assert creds.client_secret == info["client_secret"] assert creds.client_id == info["client_id"] assert creds.refresh_token == info["refresh_token"] assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT assert creds.scopes == scopes def test_to_json(self): info = test_credentials.AUTH_USER_INFO.copy() creds = _credentials_async.Credentials.from_authorized_user_info(info) # Test with no `strip` arg json_output = creds.to_json() json_asdict = json.loads(json_output) assert json_asdict.get("token") == creds.token assert json_asdict.get("refresh_token") == creds.refresh_token assert json_asdict.get("token_uri") == creds.token_uri assert json_asdict.get("client_id") == creds.client_id assert json_asdict.get("scopes") == creds.scopes assert json_asdict.get("client_secret") == creds.client_secret # Test with a `strip` arg json_output = creds.to_json(strip=["client_secret"]) json_asdict = json.loads(json_output) assert json_asdict.get("token") == creds.token assert json_asdict.get("refresh_token") == creds.refresh_token assert json_asdict.get("token_uri") == creds.token_uri assert json_asdict.get("client_id") == creds.client_id assert json_asdict.get("scopes") == creds.scopes assert json_asdict.get("client_secret") is None def test_pickle_and_unpickle(self): creds = self.make_credentials() unpickled = pickle.loads(pickle.dumps(creds)) # make sure attributes aren't lost during pickling assert list(creds.__dict__).sort() == list(unpickled.__dict__).sort() for attr in list(creds.__dict__): if attr == "_refresh_worker": assert getattr(unpickled, attr) is None else: assert getattr(creds, attr) == getattr(unpickled, attr) def test_pickle_with_missing_attribute(self): creds = self.make_credentials() # remove an optional attribute before pickling # this mimics a pickle created with a previous class definition with # fewer attributes del creds.__dict__["_quota_project_id"] unpickled = pickle.loads(pickle.dumps(creds)) # Attribute should be initialized by `__setstate__` assert unpickled.quota_project_id is None # pickles are not compatible across versions @pytest.mark.skipif( sys.version_info < (3, 5), reason="pickle file can only be loaded with Python >= 3.5", ) def test_unpickle_old_credentials_pickle(self): # make sure a credentials file pickled with an older # library version (google-auth==1.5.1) can be unpickled with open( os.path.join(test_credentials.DATA_DIR, "old_oauth_credentials_py3.pickle"), "rb", ) as f: credentials = pickle.load(f) assert credentials.quota_project_id is None @mock.patch("google.oauth2._credentials_async.Credentials.apply", autospec=True) @mock.patch("google.oauth2._credentials_async.Credentials.refresh", autospec=True) @pytest.mark.asyncio async def test_before_request(self, refresh, apply): cred = self.make_credentials() assert not cred.valid await cred.before_request(mock.Mock(), "GET", "https://example.com", {}) refresh.assert_called() apply.assert_called() @mock.patch("google.oauth2._credentials_async.Credentials.apply", autospec=True) @mock.patch("google.oauth2._credentials_async.Credentials.refresh", autospec=True) @pytest.mark.asyncio async def test_before_request_no_refresh(self, refresh, apply): cred = self.make_credentials() cred.token = refresh cred.expiry = None assert cred.valid await cred.before_request(mock.Mock(), "GET", "https://example.com", {}) refresh.assert_not_called() apply.assert_called() class TestUserAccessTokenCredentials(object): def test_instance(self): cred = _credentials_async.UserAccessTokenCredentials() assert cred._account is None cred = cred.with_account("account") assert cred._account == "account" @mock.patch("google.auth._cloud_sdk.get_auth_access_token", autospec=True) def test_refresh(self, get_auth_access_token): get_auth_access_token.return_value = "access_token" cred = _credentials_async.UserAccessTokenCredentials() cred.refresh(None) assert cred.token == "access_token" def test_with_quota_project(self): cred = _credentials_async.UserAccessTokenCredentials() quota_project_cred = cred.with_quota_project("project-foo") assert quota_project_cred._quota_project_id == "project-foo" assert quota_project_cred._account == cred._account @mock.patch( "google.oauth2._credentials_async.UserAccessTokenCredentials.apply", autospec=True, ) @mock.patch( "google.oauth2._credentials_async.UserAccessTokenCredentials.refresh", autospec=True, ) def test_before_request(self, refresh, apply): cred = _credentials_async.UserAccessTokenCredentials() cred.before_request(mock.Mock(), "GET", "https://example.com", {}) refresh.assert_called() apply.assert_called()	18.16 KB	2025-04-14 17:36
📄 test_id_token.py # Copyright 2020 Google Inc. # # 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 json import os import mock import pytest # type: ignore from google.auth import environment_vars from google.auth import exceptions import google.auth.compute_engine._metadata from google.oauth2 import _id_token_async as id_token from google.oauth2 import _service_account_async from google.oauth2 import id_token as sync_id_token from tests.oauth2 import test_id_token def make_request(status, data=None): response = mock.AsyncMock(spec=["transport.Response"]) response.status = status if data is not None: response.data = mock.AsyncMock(spec=["__call__", "read"]) response.content = mock.AsyncMock( spec=["__call__"], return_value=json.dumps(data).encode("utf-8") ) request = mock.AsyncMock(spec=["transport.Request"]) request.return_value = response return request @pytest.mark.asyncio async def test__fetch_certs_success(): certs = {"1": "cert"} request = make_request(200, certs) returned_certs = await id_token._fetch_certs(request, mock.sentinel.cert_url) request.assert_called_once_with(mock.sentinel.cert_url, method="GET") assert returned_certs == certs @pytest.mark.asyncio async def test__fetch_certs_failure(): request = make_request(404) with pytest.raises(exceptions.TransportError): await id_token._fetch_certs(request, mock.sentinel.cert_url) request.assert_called_once_with(mock.sentinel.cert_url, method="GET") @mock.patch("google.auth.jwt.decode", autospec=True) @mock.patch("google.oauth2._id_token_async._fetch_certs", autospec=True) @pytest.mark.asyncio async def test_verify_token(_fetch_certs, decode): result = await id_token.verify_token(mock.sentinel.token, mock.sentinel.request) assert result == decode.return_value _fetch_certs.assert_called_once_with( mock.sentinel.request, sync_id_token._GOOGLE_OAUTH2_CERTS_URL ) decode.assert_called_once_with( mock.sentinel.token, certs=_fetch_certs.return_value, audience=None, clock_skew_in_seconds=0, ) @mock.patch("google.auth.jwt.decode", autospec=True) @mock.patch("google.oauth2._id_token_async._fetch_certs", autospec=True) @pytest.mark.asyncio async def test_verify_token_clock_skew(_fetch_certs, decode): result = await id_token.verify_token( mock.sentinel.token, mock.sentinel.request, clock_skew_in_seconds=10 ) assert result == decode.return_value _fetch_certs.assert_called_once_with( mock.sentinel.request, sync_id_token._GOOGLE_OAUTH2_CERTS_URL ) decode.assert_called_once_with( mock.sentinel.token, certs=_fetch_certs.return_value, audience=None, clock_skew_in_seconds=10, ) @mock.patch("google.auth.jwt.decode", autospec=True) @mock.patch("google.oauth2._id_token_async._fetch_certs", autospec=True) @pytest.mark.asyncio async def test_verify_token_args(_fetch_certs, decode): result = await id_token.verify_token( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience, certs_url=mock.sentinel.certs_url, ) assert result == decode.return_value _fetch_certs.assert_called_once_with(mock.sentinel.request, mock.sentinel.certs_url) decode.assert_called_once_with( mock.sentinel.token, certs=_fetch_certs.return_value, audience=mock.sentinel.audience, clock_skew_in_seconds=0, ) @mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) @pytest.mark.asyncio async def test_verify_oauth2_token(verify_token): verify_token.return_value = {"iss": "accounts.google.com"} result = await id_token.verify_oauth2_token( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience ) assert result == verify_token.return_value verify_token.assert_called_once_with( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience, certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL, clock_skew_in_seconds=0, ) @mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) @pytest.mark.asyncio async def test_verify_oauth2_token_clock_skew(verify_token): verify_token.return_value = {"iss": "accounts.google.com"} result = await id_token.verify_oauth2_token( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience, clock_skew_in_seconds=10, ) assert result == verify_token.return_value verify_token.assert_called_once_with( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience, certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL, clock_skew_in_seconds=10, ) @mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) @pytest.mark.asyncio async def test_verify_oauth2_token_invalid_iss(verify_token): verify_token.return_value = {"iss": "invalid_issuer"} with pytest.raises(exceptions.GoogleAuthError): await id_token.verify_oauth2_token( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience ) @mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) @pytest.mark.asyncio async def test_verify_firebase_token(verify_token): result = await id_token.verify_firebase_token( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience ) assert result == verify_token.return_value verify_token.assert_called_once_with( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience, certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL, clock_skew_in_seconds=0, ) @mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) @pytest.mark.asyncio async def test_verify_firebase_token_clock_skew(verify_token): result = await id_token.verify_firebase_token( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience, clock_skew_in_seconds=10, ) assert result == verify_token.return_value verify_token.assert_called_once_with( mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience, certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL, clock_skew_in_seconds=10, ) @pytest.mark.asyncio async def test_fetch_id_token_from_metadata_server(monkeypatch): monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False) def mock_init(self, request, audience, use_metadata_identity_endpoint): assert use_metadata_identity_endpoint self.token = "id_token" with mock.patch("google.auth.compute_engine._metadata.ping", return_value=True): with mock.patch.multiple( google.auth.compute_engine.IDTokenCredentials, __init__=mock_init, refresh=mock.Mock(), ): request = mock.AsyncMock() token = await id_token.fetch_id_token( request, "https://pubsub.googleapis.com" ) assert token == "id_token" @pytest.mark.asyncio async def test_fetch_id_token_from_explicit_cred_json_file(monkeypatch): monkeypatch.setenv(environment_vars.CREDENTIALS, test_id_token.SERVICE_ACCOUNT_FILE) async def mock_refresh(self, request): self.token = "id_token" with mock.patch.object( _service_account_async.IDTokenCredentials, "refresh", mock_refresh ): request = mock.AsyncMock() token = await id_token.fetch_id_token(request, "https://pubsub.googleapis.com") assert token == "id_token" @pytest.mark.asyncio async def test_fetch_id_token_no_cred_exists(monkeypatch): monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False) with mock.patch( "google.auth.compute_engine._metadata.ping", side_effect=exceptions.TransportError(), ): with pytest.raises(exceptions.DefaultCredentialsError) as excinfo: request = mock.AsyncMock() await id_token.fetch_id_token(request, "https://pubsub.googleapis.com") assert excinfo.match( r"Neither metadata server or valid service account credentials are found." ) with mock.patch("google.auth.compute_engine._metadata.ping", return_value=False): with pytest.raises(exceptions.DefaultCredentialsError) as excinfo: request = mock.AsyncMock() await id_token.fetch_id_token(request, "https://pubsub.googleapis.com") assert excinfo.match( r"Neither metadata server or valid service account credentials are found." ) @pytest.mark.asyncio async def test_fetch_id_token_invalid_cred_file(monkeypatch): not_json_file = os.path.join( os.path.dirname(__file__), "../../tests/data/public_cert.pem" ) monkeypatch.setenv(environment_vars.CREDENTIALS, not_json_file) with pytest.raises(exceptions.DefaultCredentialsError) as excinfo: request = mock.AsyncMock() await id_token.fetch_id_token(request, "https://pubsub.googleapis.com") assert excinfo.match( r"GOOGLE_APPLICATION_CREDENTIALS is not valid service account credentials." ) @pytest.mark.asyncio async def test_fetch_id_token_invalid_cred_type(monkeypatch): user_credentials_file = os.path.join( os.path.dirname(__file__), "../../tests/data/authorized_user.json" ) monkeypatch.setenv(environment_vars.CREDENTIALS, user_credentials_file) with mock.patch("google.auth.compute_engine._metadata.ping", return_value=False): with pytest.raises(exceptions.DefaultCredentialsError) as excinfo: request = mock.AsyncMock() await id_token.fetch_id_token(request, "https://pubsub.googleapis.com") assert excinfo.match( r"Neither metadata server or valid service account credentials are found." ) @pytest.mark.asyncio async def test_fetch_id_token_invalid_cred_path(monkeypatch): not_json_file = os.path.join( os.path.dirname(__file__), "../../tests/data/not_exists.json" ) monkeypatch.setenv(environment_vars.CREDENTIALS, not_json_file) with pytest.raises(exceptions.DefaultCredentialsError) as excinfo: request = mock.AsyncMock() await id_token.fetch_id_token(request, "https://pubsub.googleapis.com") assert excinfo.match( r"GOOGLE_APPLICATION_CREDENTIALS path is either not found or invalid." )	10.86 KB	2025-04-14 17:36
📄 test_reauth_async.py # Copyright 2021 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 copy import mock import pytest # type: ignore from google.auth import exceptions from google.oauth2 import _reauth_async from google.oauth2 import reauth MOCK_REQUEST = mock.AsyncMock(spec=["transport.Request"]) CHALLENGES_RESPONSE_TEMPLATE = { "status": "CHALLENGE_REQUIRED", "sessionId": "123", "challenges": [ { "status": "READY", "challengeId": 1, "challengeType": "PASSWORD", "securityKey": {}, } ], } CHALLENGES_RESPONSE_AUTHENTICATED = { "status": "AUTHENTICATED", "sessionId": "123", "encodedProofOfReauthToken": "new_rapt_token", } class MockChallenge(object): def __init__(self, name, locally_eligible, challenge_input): self.name = name self.is_locally_eligible = locally_eligible self.challenge_input = challenge_input def obtain_challenge_input(self, metadata): return self.challenge_input @pytest.mark.asyncio async def test__get_challenges(): with mock.patch( "google.oauth2._client_async._token_endpoint_request" ) as mock_token_endpoint_request: await _reauth_async._get_challenges(MOCK_REQUEST, ["SAML"], "token") mock_token_endpoint_request.assert_called_with( MOCK_REQUEST, reauth._REAUTH_API + ":start", {"supportedChallengeTypes": ["SAML"]}, access_token="token", use_json=True, ) @pytest.mark.asyncio async def test__get_challenges_with_scopes(): with mock.patch( "google.oauth2._client_async._token_endpoint_request" ) as mock_token_endpoint_request: await _reauth_async._get_challenges( MOCK_REQUEST, ["SAML"], "token", requested_scopes=["scope"] ) mock_token_endpoint_request.assert_called_with( MOCK_REQUEST, reauth._REAUTH_API + ":start", { "supportedChallengeTypes": ["SAML"], "oauthScopesForDomainPolicyLookup": ["scope"], }, access_token="token", use_json=True, ) @pytest.mark.asyncio async def test__send_challenge_result(): with mock.patch( "google.oauth2._client_async._token_endpoint_request" ) as mock_token_endpoint_request: await _reauth_async._send_challenge_result( MOCK_REQUEST, "123", "1", {"credential": "password"}, "token" ) mock_token_endpoint_request.assert_called_with( MOCK_REQUEST, reauth._REAUTH_API + "/123:continue", { "sessionId": "123", "challengeId": "1", "action": "RESPOND", "proposalResponse": {"credential": "password"}, }, access_token="token", use_json=True, ) @pytest.mark.asyncio async def test__run_next_challenge_not_ready(): challenges_response = copy.deepcopy(CHALLENGES_RESPONSE_TEMPLATE) challenges_response["challenges"][0]["status"] = "STATUS_UNSPECIFIED" assert ( await _reauth_async._run_next_challenge( challenges_response, MOCK_REQUEST, "token" ) is None ) @pytest.mark.asyncio async def test__run_next_challenge_not_supported(): challenges_response = copy.deepcopy(CHALLENGES_RESPONSE_TEMPLATE) challenges_response["challenges"][0]["challengeType"] = "CHALLENGE_TYPE_UNSPECIFIED" with pytest.raises(exceptions.ReauthFailError) as excinfo: await _reauth_async._run_next_challenge( challenges_response, MOCK_REQUEST, "token" ) assert excinfo.match(r"Unsupported challenge type CHALLENGE_TYPE_UNSPECIFIED") @pytest.mark.asyncio async def test__run_next_challenge_not_locally_eligible(): mock_challenge = MockChallenge("PASSWORD", False, "challenge_input") with mock.patch( "google.oauth2.challenges.AVAILABLE_CHALLENGES", {"PASSWORD": mock_challenge} ): with pytest.raises(exceptions.ReauthFailError) as excinfo: await _reauth_async._run_next_challenge( CHALLENGES_RESPONSE_TEMPLATE, MOCK_REQUEST, "token" ) assert excinfo.match(r"Challenge PASSWORD is not locally eligible") @pytest.mark.asyncio async def test__run_next_challenge_no_challenge_input(): mock_challenge = MockChallenge("PASSWORD", True, None) with mock.patch( "google.oauth2.challenges.AVAILABLE_CHALLENGES", {"PASSWORD": mock_challenge} ): assert ( await _reauth_async._run_next_challenge( CHALLENGES_RESPONSE_TEMPLATE, MOCK_REQUEST, "token" ) is None ) @pytest.mark.asyncio async def test__run_next_challenge_success(): mock_challenge = MockChallenge("PASSWORD", True, {"credential": "password"}) with mock.patch( "google.oauth2.challenges.AVAILABLE_CHALLENGES", {"PASSWORD": mock_challenge} ): with mock.patch( "google.oauth2._reauth_async._send_challenge_result" ) as mock_send_challenge_result: await _reauth_async._run_next_challenge( CHALLENGES_RESPONSE_TEMPLATE, MOCK_REQUEST, "token" ) mock_send_challenge_result.assert_called_with( MOCK_REQUEST, "123", 1, {"credential": "password"}, "token" ) @pytest.mark.asyncio async def test__obtain_rapt_authenticated(): with mock.patch( "google.oauth2._reauth_async._get_challenges", return_value=CHALLENGES_RESPONSE_AUTHENTICATED, ): new_rapt_token = await _reauth_async._obtain_rapt(MOCK_REQUEST, "token", None) assert new_rapt_token == "new_rapt_token" @pytest.mark.asyncio async def test__obtain_rapt_authenticated_after_run_next_challenge(): with mock.patch( "google.oauth2._reauth_async._get_challenges", return_value=CHALLENGES_RESPONSE_TEMPLATE, ): with mock.patch( "google.oauth2._reauth_async._run_next_challenge", side_effect=[ CHALLENGES_RESPONSE_TEMPLATE, CHALLENGES_RESPONSE_AUTHENTICATED, ], ): with mock.patch("google.oauth2.reauth.is_interactive", return_value=True): assert ( await _reauth_async._obtain_rapt(MOCK_REQUEST, "token", None) == "new_rapt_token" ) @pytest.mark.asyncio async def test__obtain_rapt_unsupported_status(): challenges_response = copy.deepcopy(CHALLENGES_RESPONSE_TEMPLATE) challenges_response["status"] = "STATUS_UNSPECIFIED" with mock.patch( "google.oauth2._reauth_async._get_challenges", return_value=challenges_response ): with pytest.raises(exceptions.ReauthFailError) as excinfo: await _reauth_async._obtain_rapt(MOCK_REQUEST, "token", None) assert excinfo.match(r"API error: STATUS_UNSPECIFIED") @pytest.mark.asyncio async def test__obtain_rapt_not_interactive(): with mock.patch( "google.oauth2._reauth_async._get_challenges", return_value=CHALLENGES_RESPONSE_TEMPLATE, ): with mock.patch("google.oauth2.reauth.is_interactive", return_value=False): with pytest.raises(exceptions.ReauthFailError) as excinfo: await _reauth_async._obtain_rapt(MOCK_REQUEST, "token", None) assert excinfo.match(r"not in an interactive session") @pytest.mark.asyncio async def test__obtain_rapt_not_authenticated(): with mock.patch( "google.oauth2._reauth_async._get_challenges", return_value=CHALLENGES_RESPONSE_TEMPLATE, ): with mock.patch("google.oauth2.reauth.RUN_CHALLENGE_RETRY_LIMIT", 0): with pytest.raises(exceptions.ReauthFailError) as excinfo: await _reauth_async._obtain_rapt(MOCK_REQUEST, "token", None) assert excinfo.match(r"Reauthentication failed") @pytest.mark.asyncio async def test_get_rapt_token(): with mock.patch( "google.oauth2._client_async.refresh_grant", return_value=("token", None, None, None), ) as mock_refresh_grant: with mock.patch( "google.oauth2._reauth_async._obtain_rapt", return_value="new_rapt_token" ) as mock_obtain_rapt: assert ( await _reauth_async.get_rapt_token( MOCK_REQUEST, "client_id", "client_secret", "refresh_token", "token_uri", ) == "new_rapt_token" ) mock_refresh_grant.assert_called_with( request=MOCK_REQUEST, client_id="client_id", client_secret="client_secret", refresh_token="refresh_token", token_uri="token_uri", scopes=[reauth._REAUTH_SCOPE], ) mock_obtain_rapt.assert_called_with( MOCK_REQUEST, "token", requested_scopes=None ) @pytest.mark.asyncio async def test_refresh_grant_failed(): with mock.patch( "google.oauth2._client_async._token_endpoint_request_no_throw" ) as mock_token_request: mock_token_request.return_value = (False, {"error": "Bad request"}, True) with pytest.raises(exceptions.RefreshError) as excinfo: await _reauth_async.refresh_grant( MOCK_REQUEST, "token_uri", "refresh_token", "client_id", "client_secret", scopes=["foo", "bar"], rapt_token="rapt_token", ) assert excinfo.match(r"Bad request") assert excinfo.value.retryable mock_token_request.assert_called_with( MOCK_REQUEST, "token_uri", { "grant_type": "refresh_token", "client_id": "client_id", "client_secret": "client_secret", "refresh_token": "refresh_token", "scope": "foo bar", "rapt": "rapt_token", }, ) @pytest.mark.asyncio async def test_refresh_grant_success(): with mock.patch( "google.oauth2._client_async._token_endpoint_request_no_throw" ) as mock_token_request: mock_token_request.side_effect = [ (False, {"error": "invalid_grant", "error_subtype": "rapt_required"}, True), (True, {"access_token": "access_token"}, None), ] with mock.patch( "google.oauth2._reauth_async.get_rapt_token", return_value="new_rapt_token" ): assert await _reauth_async.refresh_grant( MOCK_REQUEST, "token_uri", "refresh_token", "client_id", "client_secret", enable_reauth_refresh=True, ) == ( "access_token", "refresh_token", None, {"access_token": "access_token"}, "new_rapt_token", ) @pytest.mark.asyncio async def test_refresh_grant_reauth_refresh_disabled(): with mock.patch( "google.oauth2._client_async._token_endpoint_request_no_throw" ) as mock_token_request: mock_token_request.side_effect = [ ( False, {"error": "invalid_grant", "error_subtype": "rapt_required"}, False, ), (True, {"access_token": "access_token"}, None), ] with pytest.raises(exceptions.RefreshError) as excinfo: assert await _reauth_async.refresh_grant( MOCK_REQUEST, "token_uri", "refresh_token", "client_id", "client_secret" ) assert excinfo.match(r"Reauthentication is needed") assert not excinfo.value.retryable	12.19 KB	2025-04-14 17:36
📄 test_service_account_async.py # Copyright 2020 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 mock import pytest # type: ignore from google.auth import _helpers from google.auth import crypt from google.auth import jwt from google.auth import transport from google.oauth2 import _service_account_async as service_account from tests.oauth2 import test_service_account class TestCredentials(object): SERVICE_ACCOUNT_EMAIL = "service-account@example.com" TOKEN_URI = "https://example.com/oauth2/token" @classmethod def make_credentials(cls): return service_account.Credentials( test_service_account.SIGNER, cls.SERVICE_ACCOUNT_EMAIL, cls.TOKEN_URI ) def test_from_service_account_info(self): credentials = service_account.Credentials.from_service_account_info( test_service_account.SERVICE_ACCOUNT_INFO ) assert ( credentials._signer.key_id == test_service_account.SERVICE_ACCOUNT_INFO["private_key_id"] ) assert ( credentials.service_account_email == test_service_account.SERVICE_ACCOUNT_INFO["client_email"] ) assert ( credentials._token_uri == test_service_account.SERVICE_ACCOUNT_INFO["token_uri"] ) def test_from_service_account_info_args(self): info = test_service_account.SERVICE_ACCOUNT_INFO.copy() scopes = ["email", "profile"] subject = "subject" additional_claims = {"meta": "data"} credentials = service_account.Credentials.from_service_account_info( info, scopes=scopes, subject=subject, additional_claims=additional_claims ) assert credentials.service_account_email == info["client_email"] assert credentials.project_id == info["project_id"] assert credentials._signer.key_id == info["private_key_id"] assert credentials._token_uri == info["token_uri"] assert credentials._scopes == scopes assert credentials._subject == subject assert credentials._additional_claims == additional_claims def test_from_service_account_file(self): info = test_service_account.SERVICE_ACCOUNT_INFO.copy() credentials = service_account.Credentials.from_service_account_file( test_service_account.SERVICE_ACCOUNT_JSON_FILE ) assert credentials.service_account_email == info["client_email"] assert credentials.project_id == info["project_id"] assert credentials._signer.key_id == info["private_key_id"] assert credentials._token_uri == info["token_uri"] def test_from_service_account_file_args(self): info = test_service_account.SERVICE_ACCOUNT_INFO.copy() scopes = ["email", "profile"] subject = "subject" additional_claims = {"meta": "data"} credentials = service_account.Credentials.from_service_account_file( test_service_account.SERVICE_ACCOUNT_JSON_FILE, subject=subject, scopes=scopes, additional_claims=additional_claims, ) assert credentials.service_account_email == info["client_email"] assert credentials.project_id == info["project_id"] assert credentials._signer.key_id == info["private_key_id"] assert credentials._token_uri == info["token_uri"] assert credentials._scopes == scopes assert credentials._subject == subject assert credentials._additional_claims == additional_claims def test_default_state(self): credentials = self.make_credentials() assert not credentials.valid # Expiration hasn't been set yet assert not credentials.expired # Scopes haven't been specified yet assert credentials.requires_scopes def test_sign_bytes(self): credentials = self.make_credentials() to_sign = b"123" signature = credentials.sign_bytes(to_sign) assert crypt.verify_signature( to_sign, signature, test_service_account.PUBLIC_CERT_BYTES ) def test_signer(self): credentials = self.make_credentials() assert isinstance(credentials.signer, crypt.Signer) def test_signer_email(self): credentials = self.make_credentials() assert credentials.signer_email == self.SERVICE_ACCOUNT_EMAIL def test_create_scoped(self): credentials = self.make_credentials() scopes = ["email", "profile"] credentials = credentials.with_scopes(scopes) assert credentials._scopes == scopes def test_with_claims(self): credentials = self.make_credentials() new_credentials = credentials.with_claims({"meep": "moop"}) assert new_credentials._additional_claims == {"meep": "moop"} def test_with_quota_project(self): credentials = self.make_credentials() new_credentials = credentials.with_quota_project("new-project-456") assert new_credentials.quota_project_id == "new-project-456" hdrs = {} new_credentials.apply(hdrs, token="tok") assert "x-goog-user-project" in hdrs def test__make_authorization_grant_assertion(self): credentials = self.make_credentials() token = credentials._make_authorization_grant_assertion() payload = jwt.decode(token, test_service_account.PUBLIC_CERT_BYTES) assert payload["iss"] == self.SERVICE_ACCOUNT_EMAIL assert ( payload["aud"] == service_account.service_account._GOOGLE_OAUTH2_TOKEN_ENDPOINT ) def test__make_authorization_grant_assertion_scoped(self): credentials = self.make_credentials() scopes = ["email", "profile"] credentials = credentials.with_scopes(scopes) token = credentials._make_authorization_grant_assertion() payload = jwt.decode(token, test_service_account.PUBLIC_CERT_BYTES) assert payload["scope"] == "email profile" def test__make_authorization_grant_assertion_subject(self): credentials = self.make_credentials() subject = "user@example.com" credentials = credentials.with_subject(subject) token = credentials._make_authorization_grant_assertion() payload = jwt.decode(token, test_service_account.PUBLIC_CERT_BYTES) assert payload["sub"] == subject @mock.patch("google.oauth2._client_async.jwt_grant", autospec=True) @pytest.mark.asyncio async def test_refresh_success(self, jwt_grant): credentials = self.make_credentials() token = "token" jwt_grant.return_value = ( token, _helpers.utcnow() + datetime.timedelta(seconds=500), {}, ) request = mock.create_autospec(transport.Request, instance=True) # Refresh credentials await credentials.refresh(request) # Check jwt grant call. assert jwt_grant.called called_request, token_uri, assertion = jwt_grant.call_args[0] assert called_request == request assert token_uri == credentials._token_uri assert jwt.decode(assertion, test_service_account.PUBLIC_CERT_BYTES) # No further assertion done on the token, as there are separate tests # for checking the authorization grant assertion. # Check that the credentials have the token. assert credentials.token == token # Check that the credentials are valid (have a token and are not # expired) assert credentials.valid @mock.patch("google.oauth2._client_async.jwt_grant", autospec=True) @pytest.mark.asyncio async def test_before_request_refreshes(self, jwt_grant): credentials = self.make_credentials() token = "token" jwt_grant.return_value = ( token, _helpers.utcnow() + datetime.timedelta(seconds=500), None, ) request = mock.create_autospec(transport.Request, instance=True) # Credentials should start as invalid assert not credentials.valid # before_request should cause a refresh await credentials.before_request(request, "GET", "http://example.com?a=1#3", {}) # The refresh endpoint should've been called. assert jwt_grant.called # Credentials should now be valid. assert credentials.valid class TestIDTokenCredentials(object): SERVICE_ACCOUNT_EMAIL = "service-account@example.com" TOKEN_URI = "https://example.com/oauth2/token" TARGET_AUDIENCE = "https://example.com" @classmethod def make_credentials(cls): return service_account.IDTokenCredentials( test_service_account.SIGNER, cls.SERVICE_ACCOUNT_EMAIL, cls.TOKEN_URI, cls.TARGET_AUDIENCE, ) def test_from_service_account_info(self): credentials = service_account.IDTokenCredentials.from_service_account_info( test_service_account.SERVICE_ACCOUNT_INFO, target_audience=self.TARGET_AUDIENCE, ) assert ( credentials._signer.key_id == test_service_account.SERVICE_ACCOUNT_INFO["private_key_id"] ) assert ( credentials.service_account_email == test_service_account.SERVICE_ACCOUNT_INFO["client_email"] ) assert ( credentials._token_uri == test_service_account.SERVICE_ACCOUNT_INFO["token_uri"] ) assert credentials._target_audience == self.TARGET_AUDIENCE def test_from_service_account_file(self): info = test_service_account.SERVICE_ACCOUNT_INFO.copy() credentials = service_account.IDTokenCredentials.from_service_account_file( test_service_account.SERVICE_ACCOUNT_JSON_FILE, target_audience=self.TARGET_AUDIENCE, ) assert credentials.service_account_email == info["client_email"] assert credentials._signer.key_id == info["private_key_id"] assert credentials._token_uri == info["token_uri"] assert credentials._target_audience == self.TARGET_AUDIENCE def test_default_state(self): credentials = self.make_credentials() assert not credentials.valid # Expiration hasn't been set yet assert not credentials.expired def test_sign_bytes(self): credentials = self.make_credentials() to_sign = b"123" signature = credentials.sign_bytes(to_sign) assert crypt.verify_signature( to_sign, signature, test_service_account.PUBLIC_CERT_BYTES ) def test_signer(self): credentials = self.make_credentials() assert isinstance(credentials.signer, crypt.Signer) def test_signer_email(self): credentials = self.make_credentials() assert credentials.signer_email == self.SERVICE_ACCOUNT_EMAIL def test_with_target_audience(self): credentials = self.make_credentials() new_credentials = credentials.with_target_audience("https://new.example.com") assert new_credentials._target_audience == "https://new.example.com" def test_with_quota_project(self): credentials = self.make_credentials() new_credentials = credentials.with_quota_project("project-foo") assert new_credentials._quota_project_id == "project-foo" def test__make_authorization_grant_assertion(self): credentials = self.make_credentials() token = credentials._make_authorization_grant_assertion() payload = jwt.decode(token, test_service_account.PUBLIC_CERT_BYTES) assert payload["iss"] == self.SERVICE_ACCOUNT_EMAIL assert ( payload["aud"] == service_account.service_account._GOOGLE_OAUTH2_TOKEN_ENDPOINT ) assert payload["target_audience"] == self.TARGET_AUDIENCE @mock.patch("google.oauth2._client_async.id_token_jwt_grant", autospec=True) @pytest.mark.asyncio async def test_refresh_success(self, id_token_jwt_grant): credentials = self.make_credentials() token = "token" id_token_jwt_grant.return_value = ( token, _helpers.utcnow() + datetime.timedelta(seconds=500), {}, ) request = mock.AsyncMock(spec=["transport.Request"]) # Refresh credentials await credentials.refresh(request) # Check jwt grant call. assert id_token_jwt_grant.called called_request, token_uri, assertion = id_token_jwt_grant.call_args[0] assert called_request == request assert token_uri == credentials._token_uri assert jwt.decode(assertion, test_service_account.PUBLIC_CERT_BYTES) # No further assertion done on the token, as there are separate tests # for checking the authorization grant assertion. # Check that the credentials have the token. assert credentials.token == token # Check that the credentials are valid (have a token and are not # expired) assert credentials.valid @mock.patch("google.oauth2._client_async.id_token_jwt_grant", autospec=True) @pytest.mark.asyncio async def test_before_request_refreshes(self, id_token_jwt_grant): credentials = self.make_credentials() token = "token" id_token_jwt_grant.return_value = ( token, _helpers.utcnow() + datetime.timedelta(seconds=500), None, ) request = mock.AsyncMock(spec=["transport.Request"]) # Credentials should start as invalid assert not credentials.valid # before_request should cause a refresh await credentials.before_request(request, "GET", "http://example.com?a=1#3", {}) # The refresh endpoint should've been called. assert id_token_jwt_grant.called # Credentials should now be valid. assert credentials.valid	14.13 KB	2025-04-14 17:36

Terminal