123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876 |
- # 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 json
- import os
- import pickle
- import sys
- import mock
- import pytest
- from google.auth import _helpers
- from google.auth import exceptions
- from google.auth import transport
- from google.oauth2 import credentials
- import yatest.common
- DATA_DIR = os.path.join(yatest.common.test_source_path(), "data")
- AUTH_USER_JSON_FILE = os.path.join(DATA_DIR, "authorized_user.json")
- with open(AUTH_USER_JSON_FILE, "r") as fh:
- AUTH_USER_INFO = json.load(fh)
- class TestCredentials(object):
- TOKEN_URI = "https://example.com/oauth2/token"
- REFRESH_TOKEN = "refresh_token"
- RAPT_TOKEN = "rapt_token"
- CLIENT_ID = "client_id"
- CLIENT_SECRET = "client_secret"
- @classmethod
- def make_credentials(cls):
- return credentials.Credentials(
- token=None,
- refresh_token=cls.REFRESH_TOKEN,
- token_uri=cls.TOKEN_URI,
- client_id=cls.CLIENT_ID,
- client_secret=cls.CLIENT_SECRET,
- rapt_token=cls.RAPT_TOKEN,
- )
- 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
- assert credentials.rapt_token == self.RAPT_TOKEN
- assert credentials.refresh_handler is None
- def test_refresh_handler_setter_and_getter(self):
- scopes = ["email", "profile"]
- original_refresh_handler = mock.Mock(return_value=("ACCESS_TOKEN_1", None))
- updated_refresh_handler = mock.Mock(return_value=("ACCESS_TOKEN_2", None))
- creds = credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- scopes=scopes,
- default_scopes=None,
- refresh_handler=original_refresh_handler,
- )
- assert creds.refresh_handler is original_refresh_handler
- creds.refresh_handler = updated_refresh_handler
- assert creds.refresh_handler is updated_refresh_handler
- creds.refresh_handler = None
- assert creds.refresh_handler is None
- def test_invalid_refresh_handler(self):
- scopes = ["email", "profile"]
- with pytest.raises(TypeError) as excinfo:
- credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- scopes=scopes,
- default_scopes=None,
- refresh_handler=object(),
- )
- assert excinfo.match("The provided refresh_handler is not a callable or None.")
- @mock.patch("google.oauth2.reauth.refresh_grant", autospec=True)
- @mock.patch(
- "google.auth._helpers.utcnow",
- return_value=datetime.datetime.min + _helpers.CLOCK_SKEW,
- )
- def test_refresh_success(self, unused_utcnow, refresh_grant):
- token = "token"
- new_rapt_token = "new_rapt_token"
- expiry = _helpers.utcnow() + datetime.timedelta(seconds=500)
- grant_response = {"id_token": mock.sentinel.id_token}
- refresh_grant.return_value = (
- # Access token
- token,
- # New refresh token
- None,
- # Expiry,
- expiry,
- # Extra data
- grant_response,
- # rapt_token
- new_rapt_token,
- )
- request = mock.create_autospec(transport.Request)
- credentials = self.make_credentials()
- # Refresh credentials
- credentials.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,
- self.RAPT_TOKEN,
- )
- # Check that the credentials have the token and expiry
- assert credentials.token == token
- assert credentials.expiry == expiry
- assert credentials.id_token == mock.sentinel.id_token
- assert credentials.rapt_token == new_rapt_token
- # Check that the credentials are valid (have a token and are not
- # expired)
- assert credentials.valid
- def test_refresh_no_refresh_token(self):
- request = mock.create_autospec(transport.Request)
- credentials_ = credentials.Credentials(token=None, refresh_token=None)
- with pytest.raises(exceptions.RefreshError, match="necessary fields"):
- credentials_.refresh(request)
- request.assert_not_called()
- @mock.patch("google.oauth2.reauth.refresh_grant", autospec=True)
- @mock.patch(
- "google.auth._helpers.utcnow",
- return_value=datetime.datetime.min + _helpers.CLOCK_SKEW,
- )
- def test_refresh_with_refresh_token_and_refresh_handler(
- self, unused_utcnow, refresh_grant
- ):
- token = "token"
- new_rapt_token = "new_rapt_token"
- expiry = _helpers.utcnow() + datetime.timedelta(seconds=500)
- grant_response = {"id_token": mock.sentinel.id_token}
- refresh_grant.return_value = (
- # Access token
- token,
- # New refresh token
- None,
- # Expiry,
- expiry,
- # Extra data
- grant_response,
- # rapt_token
- new_rapt_token,
- )
- refresh_handler = mock.Mock()
- request = mock.create_autospec(transport.Request)
- creds = credentials.Credentials(
- token=None,
- refresh_token=self.REFRESH_TOKEN,
- token_uri=self.TOKEN_URI,
- client_id=self.CLIENT_ID,
- client_secret=self.CLIENT_SECRET,
- rapt_token=self.RAPT_TOKEN,
- refresh_handler=refresh_handler,
- )
- # Refresh credentials
- 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,
- self.RAPT_TOKEN,
- )
- # 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 == new_rapt_token
- # Check that the credentials are valid (have a token and are not
- # expired)
- assert creds.valid
- # Assert refresh handler not called as the refresh token has
- # higher priority.
- refresh_handler.assert_not_called()
- @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min)
- def test_refresh_with_refresh_handler_success_scopes(self, unused_utcnow):
- expected_expiry = datetime.datetime.min + datetime.timedelta(seconds=2800)
- refresh_handler = mock.Mock(return_value=("ACCESS_TOKEN", expected_expiry))
- scopes = ["email", "profile"]
- default_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
- request = mock.create_autospec(transport.Request)
- creds = credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- scopes=scopes,
- default_scopes=default_scopes,
- refresh_handler=refresh_handler,
- )
- creds.refresh(request)
- assert creds.token == "ACCESS_TOKEN"
- assert creds.expiry == expected_expiry
- assert creds.valid
- assert not creds.expired
- # Confirm refresh handler called with the expected arguments.
- refresh_handler.assert_called_with(request, scopes=scopes)
- @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min)
- def test_refresh_with_refresh_handler_success_default_scopes(self, unused_utcnow):
- expected_expiry = datetime.datetime.min + datetime.timedelta(seconds=2800)
- original_refresh_handler = mock.Mock(
- return_value=("UNUSED_TOKEN", expected_expiry)
- )
- refresh_handler = mock.Mock(return_value=("ACCESS_TOKEN", expected_expiry))
- default_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
- request = mock.create_autospec(transport.Request)
- creds = credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- scopes=None,
- default_scopes=default_scopes,
- refresh_handler=original_refresh_handler,
- )
- # Test newly set refresh_handler is used instead of the original one.
- creds.refresh_handler = refresh_handler
- creds.refresh(request)
- assert creds.token == "ACCESS_TOKEN"
- assert creds.expiry == expected_expiry
- assert creds.valid
- assert not creds.expired
- # default_scopes should be used since no developer provided scopes
- # are provided.
- refresh_handler.assert_called_with(request, scopes=default_scopes)
- @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min)
- def test_refresh_with_refresh_handler_invalid_token(self, unused_utcnow):
- expected_expiry = datetime.datetime.min + datetime.timedelta(seconds=2800)
- # Simulate refresh handler does not return a valid token.
- refresh_handler = mock.Mock(return_value=(None, expected_expiry))
- scopes = ["email", "profile"]
- default_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
- request = mock.create_autospec(transport.Request)
- creds = credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- scopes=scopes,
- default_scopes=default_scopes,
- refresh_handler=refresh_handler,
- )
- with pytest.raises(
- exceptions.RefreshError, match="returned token is not a string"
- ):
- creds.refresh(request)
- assert creds.token is None
- assert creds.expiry is None
- assert not creds.valid
- # Confirm refresh handler called with the expected arguments.
- refresh_handler.assert_called_with(request, scopes=scopes)
- def test_refresh_with_refresh_handler_invalid_expiry(self):
- # Simulate refresh handler returns expiration time in an invalid unit.
- refresh_handler = mock.Mock(return_value=("TOKEN", 2800))
- scopes = ["email", "profile"]
- default_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
- request = mock.create_autospec(transport.Request)
- creds = credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- scopes=scopes,
- default_scopes=default_scopes,
- refresh_handler=refresh_handler,
- )
- with pytest.raises(
- exceptions.RefreshError, match="returned expiry is not a datetime object"
- ):
- creds.refresh(request)
- assert creds.token is None
- assert creds.expiry is None
- assert not creds.valid
- # Confirm refresh handler called with the expected arguments.
- refresh_handler.assert_called_with(request, scopes=scopes)
- @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min)
- def test_refresh_with_refresh_handler_expired_token(self, unused_utcnow):
- expected_expiry = datetime.datetime.min + _helpers.CLOCK_SKEW
- # Simulate refresh handler returns an expired token.
- refresh_handler = mock.Mock(return_value=("TOKEN", expected_expiry))
- scopes = ["email", "profile"]
- default_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
- request = mock.create_autospec(transport.Request)
- creds = credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- scopes=scopes,
- default_scopes=default_scopes,
- refresh_handler=refresh_handler,
- )
- with pytest.raises(exceptions.RefreshError, match="already expired"):
- creds.refresh(request)
- assert creds.token is None
- assert creds.expiry is None
- assert not creds.valid
- # Confirm refresh handler called with the expected arguments.
- refresh_handler.assert_called_with(request, scopes=scopes)
- @mock.patch("google.oauth2.reauth.refresh_grant", autospec=True)
- @mock.patch(
- "google.auth._helpers.utcnow",
- return_value=datetime.datetime.min + _helpers.CLOCK_SKEW,
- )
- def test_credentials_with_scopes_requested_refresh_success(
- self, unused_utcnow, refresh_grant
- ):
- scopes = ["email", "profile"]
- default_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
- token = "token"
- new_rapt_token = "new_rapt_token"
- expiry = _helpers.utcnow() + datetime.timedelta(seconds=500)
- grant_response = {"id_token": mock.sentinel.id_token, "scope": "email profile"}
- refresh_grant.return_value = (
- # Access token
- token,
- # New refresh token
- None,
- # Expiry,
- expiry,
- # Extra data
- grant_response,
- # rapt token
- new_rapt_token,
- )
- request = mock.create_autospec(transport.Request)
- creds = credentials.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,
- default_scopes=default_scopes,
- rapt_token=self.RAPT_TOKEN,
- )
- # Refresh credentials
- 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,
- self.RAPT_TOKEN,
- )
- # 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 == new_rapt_token
- # Check that the credentials are valid (have a token and are not
- # expired.)
- assert creds.valid
- @mock.patch("google.oauth2.reauth.refresh_grant", autospec=True)
- @mock.patch(
- "google.auth._helpers.utcnow",
- return_value=datetime.datetime.min + _helpers.CLOCK_SKEW,
- )
- def test_credentials_with_only_default_scopes_requested(
- self, unused_utcnow, refresh_grant
- ):
- default_scopes = ["email", "profile"]
- token = "token"
- new_rapt_token = "new_rapt_token"
- expiry = _helpers.utcnow() + datetime.timedelta(seconds=500)
- grant_response = {"id_token": mock.sentinel.id_token}
- refresh_grant.return_value = (
- # Access token
- token,
- # New refresh token
- None,
- # Expiry,
- expiry,
- # Extra data
- grant_response,
- # rapt token
- new_rapt_token,
- )
- request = mock.create_autospec(transport.Request)
- creds = credentials.Credentials(
- token=None,
- refresh_token=self.REFRESH_TOKEN,
- token_uri=self.TOKEN_URI,
- client_id=self.CLIENT_ID,
- client_secret=self.CLIENT_SECRET,
- default_scopes=default_scopes,
- rapt_token=self.RAPT_TOKEN,
- )
- # Refresh credentials
- 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,
- default_scopes,
- self.RAPT_TOKEN,
- )
- # 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(default_scopes)
- assert creds.rapt_token == new_rapt_token
- # Check that the credentials are valid (have a token and are not
- # expired.)
- assert creds.valid
- @mock.patch("google.oauth2.reauth.refresh_grant", autospec=True)
- @mock.patch(
- "google.auth._helpers.utcnow",
- return_value=datetime.datetime.min + _helpers.CLOCK_SKEW,
- )
- def test_credentials_with_scopes_returned_refresh_success(
- self, unused_utcnow, refresh_grant
- ):
- scopes = ["email", "profile"]
- token = "token"
- new_rapt_token = "new_rapt_token"
- expiry = _helpers.utcnow() + datetime.timedelta(seconds=500)
- grant_response = {
- "id_token": mock.sentinel.id_token,
- "scopes": " ".join(scopes),
- }
- refresh_grant.return_value = (
- # Access token
- token,
- # New refresh token
- None,
- # Expiry,
- expiry,
- # Extra data
- grant_response,
- # rapt token
- new_rapt_token,
- )
- request = mock.create_autospec(transport.Request)
- creds = credentials.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=self.RAPT_TOKEN,
- )
- # Refresh credentials
- 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,
- self.RAPT_TOKEN,
- )
- # 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 == new_rapt_token
- # Check that the credentials are valid (have a token and are not
- # expired.)
- assert creds.valid
- @mock.patch("google.oauth2.reauth.refresh_grant", autospec=True)
- @mock.patch(
- "google.auth._helpers.utcnow",
- return_value=datetime.datetime.min + _helpers.CLOCK_SKEW,
- )
- def test_credentials_with_scopes_refresh_failure_raises_refresh_error(
- self, unused_utcnow, refresh_grant
- ):
- scopes = ["email", "profile"]
- scopes_returned = ["email"]
- token = "token"
- new_rapt_token = "new_rapt_token"
- expiry = _helpers.utcnow() + datetime.timedelta(seconds=500)
- grant_response = {
- "id_token": mock.sentinel.id_token,
- "scope": " ".join(scopes_returned),
- }
- refresh_grant.return_value = (
- # Access token
- token,
- # New refresh token
- None,
- # Expiry,
- expiry,
- # Extra data
- grant_response,
- # rapt token
- new_rapt_token,
- )
- request = mock.create_autospec(transport.Request)
- creds = credentials.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=self.RAPT_TOKEN,
- )
- # Refresh credentials
- with pytest.raises(
- exceptions.RefreshError, match="Not all requested scopes were granted"
- ):
- 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,
- self.RAPT_TOKEN,
- )
- # 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 == new_rapt_token
- # 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.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"
- assert "token" in headers["authorization"]
- def test_apply_with_no_quota_project_id(self):
- creds = credentials.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
- assert "token" in headers["authorization"]
- def test_with_quota_project(self):
- creds = credentials.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 = AUTH_USER_INFO.copy()
- creds = credentials.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.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
- info["scopes"] = "email" # single non-array scope from file
- creds = credentials.Credentials.from_authorized_user_info(info)
- assert creds.scopes == [info["scopes"]]
- info["scopes"] = ["email", "profile"] # array scope from file
- creds = credentials.Credentials.from_authorized_user_info(info)
- assert creds.scopes == info["scopes"]
- expiry = datetime.datetime(2020, 8, 14, 15, 54, 1)
- info["expiry"] = expiry.isoformat() + "Z"
- creds = credentials.Credentials.from_authorized_user_info(info)
- assert creds.expiry == expiry
- assert creds.expired
- def test_from_authorized_user_file(self):
- info = AUTH_USER_INFO.copy()
- creds = credentials.Credentials.from_authorized_user_file(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.Credentials.from_authorized_user_file(
- 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 = AUTH_USER_INFO.copy()
- expiry = datetime.datetime(2020, 8, 14, 15, 54, 1)
- info["expiry"] = expiry.isoformat() + "Z"
- creds = credentials.Credentials.from_authorized_user_info(info)
- assert creds.expiry == expiry
- # 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
- assert json_asdict.get("expiry") == info["expiry"]
- # 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
- # Test with no expiry
- creds.expiry = None
- json_output = creds.to_json()
- json_asdict = json.loads(json_output)
- assert json_asdict.get("expiry") 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__):
- assert getattr(creds, attr) == getattr(unpickled, attr)
- def test_pickle_and_unpickle_with_refresh_handler(self):
- expected_expiry = _helpers.utcnow() + datetime.timedelta(seconds=2800)
- refresh_handler = mock.Mock(return_value=("TOKEN", expected_expiry))
- creds = credentials.Credentials(
- token=None,
- refresh_token=None,
- token_uri=None,
- client_id=None,
- client_secret=None,
- rapt_token=None,
- refresh_handler=refresh_handler,
- )
- 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__):
- # For the _refresh_handler property, the unpickled creds should be
- # set to None.
- if attr == "_refresh_handler":
- 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(DATA_DIR, "old_oauth_credentials_py3.pickle"), "rb"
- ) as f:
- credentials = pickle.load(f)
- assert credentials.quota_project_id is None
- class TestUserAccessTokenCredentials(object):
- def test_instance(self):
- cred = credentials.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.UserAccessTokenCredentials()
- cred.refresh(None)
- assert cred.token == "access_token"
- def test_with_quota_project(self):
- cred = credentials.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.UserAccessTokenCredentials.apply", autospec=True
- )
- @mock.patch(
- "google.oauth2.credentials.UserAccessTokenCredentials.refresh", autospec=True
- )
- def test_before_request(self, refresh, apply):
- cred = credentials.UserAccessTokenCredentials()
- cred.before_request(mock.Mock(), "GET", "https://example.com", {})
- refresh.assert_called()
- apply.assert_called()
|