chore: blacken (#375)
diff --git a/tests/test_jwt.py b/tests/test_jwt.py
index 4a64717..b0c6e48 100644
--- a/tests/test_jwt.py
+++ b/tests/test_jwt.py
@@ -26,41 +26,45 @@
from google.auth import jwt
-DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
+DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
-with open(os.path.join(DATA_DIR, 'privatekey.pem'), 'rb') as fh:
+with open(os.path.join(DATA_DIR, "privatekey.pem"), "rb") as fh:
PRIVATE_KEY_BYTES = fh.read()
-with open(os.path.join(DATA_DIR, 'public_cert.pem'), 'rb') as fh:
+with open(os.path.join(DATA_DIR, "public_cert.pem"), "rb") as fh:
PUBLIC_CERT_BYTES = fh.read()
-with open(os.path.join(DATA_DIR, 'other_cert.pem'), 'rb') as fh:
+with open(os.path.join(DATA_DIR, "other_cert.pem"), "rb") as fh:
OTHER_CERT_BYTES = fh.read()
-SERVICE_ACCOUNT_JSON_FILE = os.path.join(DATA_DIR, 'service_account.json')
+SERVICE_ACCOUNT_JSON_FILE = os.path.join(DATA_DIR, "service_account.json")
-with open(SERVICE_ACCOUNT_JSON_FILE, 'r') as fh:
+with open(SERVICE_ACCOUNT_JSON_FILE, "r") as fh:
SERVICE_ACCOUNT_INFO = json.load(fh)
@pytest.fixture
def signer():
- return crypt.RSASigner.from_string(PRIVATE_KEY_BYTES, '1')
+ return crypt.RSASigner.from_string(PRIVATE_KEY_BYTES, "1")
def test_encode_basic(signer):
- test_payload = {'test': 'value'}
+ test_payload = {"test": "value"}
encoded = jwt.encode(signer, test_payload)
header, payload, _, _ = jwt._unverified_decode(encoded)
assert payload == test_payload
- assert header == {'typ': 'JWT', 'alg': 'RS256', 'kid': signer.key_id}
+ assert header == {"typ": "JWT", "alg": "RS256", "kid": signer.key_id}
def test_encode_extra_headers(signer):
- encoded = jwt.encode(signer, {}, header={'extra': 'value'})
+ encoded = jwt.encode(signer, {}, header={"extra": "value"})
header = jwt.decode_header(encoded)
assert header == {
- 'typ': 'JWT', 'alg': 'RS256', 'kid': signer.key_id, 'extra': 'value'}
+ "typ": "JWT",
+ "alg": "RS256",
+ "kid": signer.key_id,
+ "extra": "value",
+ }
@pytest.fixture
@@ -68,11 +72,11 @@
def factory(claims=None, key_id=None):
now = _helpers.datetime_to_secs(_helpers.utcnow())
payload = {
- 'aud': 'audience@example.com',
- 'iat': now,
- 'exp': now + 300,
- 'user': 'billy bob',
- 'metadata': {'meta': 'data'}
+ "aud": "audience@example.com",
+ "iat": now,
+ "exp": now + 300,
+ "user": "billy bob",
+ "metadata": {"meta": "data"},
}
payload.update(claims or {})
@@ -83,154 +87,168 @@
key_id = None
return jwt.encode(signer, payload, key_id=key_id)
+
return factory
def test_decode_valid(token_factory):
payload = jwt.decode(token_factory(), certs=PUBLIC_CERT_BYTES)
- assert payload['aud'] == 'audience@example.com'
- assert payload['user'] == 'billy bob'
- assert payload['metadata']['meta'] == 'data'
+ assert payload["aud"] == "audience@example.com"
+ assert payload["user"] == "billy bob"
+ assert payload["metadata"]["meta"] == "data"
def test_decode_valid_with_audience(token_factory):
payload = jwt.decode(
- token_factory(), certs=PUBLIC_CERT_BYTES,
- audience='audience@example.com')
- assert payload['aud'] == 'audience@example.com'
- assert payload['user'] == 'billy bob'
- assert payload['metadata']['meta'] == 'data'
+ token_factory(), certs=PUBLIC_CERT_BYTES, audience="audience@example.com"
+ )
+ assert payload["aud"] == "audience@example.com"
+ assert payload["user"] == "billy bob"
+ assert payload["metadata"]["meta"] == "data"
def test_decode_valid_unverified(token_factory):
payload = jwt.decode(token_factory(), certs=OTHER_CERT_BYTES, verify=False)
- assert payload['aud'] == 'audience@example.com'
- assert payload['user'] == 'billy bob'
- assert payload['metadata']['meta'] == 'data'
+ assert payload["aud"] == "audience@example.com"
+ assert payload["user"] == "billy bob"
+ assert payload["metadata"]["meta"] == "data"
def test_decode_bad_token_wrong_number_of_segments():
with pytest.raises(ValueError) as excinfo:
- jwt.decode('1.2', PUBLIC_CERT_BYTES)
- assert excinfo.match(r'Wrong number of segments')
+ jwt.decode("1.2", PUBLIC_CERT_BYTES)
+ assert excinfo.match(r"Wrong number of segments")
def test_decode_bad_token_not_base64():
with pytest.raises((ValueError, TypeError)) as excinfo:
- jwt.decode('1.2.3', PUBLIC_CERT_BYTES)
- assert excinfo.match(r'Incorrect padding|more than a multiple of 4')
+ jwt.decode("1.2.3", PUBLIC_CERT_BYTES)
+ assert excinfo.match(r"Incorrect padding|more than a multiple of 4")
def test_decode_bad_token_not_json():
- token = b'.'.join([base64.urlsafe_b64encode(b'123!')] * 3)
+ token = b".".join([base64.urlsafe_b64encode(b"123!")] * 3)
with pytest.raises(ValueError) as excinfo:
jwt.decode(token, PUBLIC_CERT_BYTES)
- assert excinfo.match(r'Can\'t parse segment')
+ assert excinfo.match(r"Can\'t parse segment")
def test_decode_bad_token_no_iat_or_exp(signer):
- token = jwt.encode(signer, {'test': 'value'})
+ token = jwt.encode(signer, {"test": "value"})
with pytest.raises(ValueError) as excinfo:
jwt.decode(token, PUBLIC_CERT_BYTES)
- assert excinfo.match(r'Token does not contain required claim')
+ assert excinfo.match(r"Token does not contain required claim")
def test_decode_bad_token_too_early(token_factory):
- token = token_factory(claims={
- 'iat': _helpers.datetime_to_secs(
- _helpers.utcnow() + datetime.timedelta(hours=1))
- })
+ token = token_factory(
+ claims={
+ "iat": _helpers.datetime_to_secs(
+ _helpers.utcnow() + datetime.timedelta(hours=1)
+ )
+ }
+ )
with pytest.raises(ValueError) as excinfo:
jwt.decode(token, PUBLIC_CERT_BYTES)
- assert excinfo.match(r'Token used too early')
+ assert excinfo.match(r"Token used too early")
def test_decode_bad_token_expired(token_factory):
- token = token_factory(claims={
- 'exp': _helpers.datetime_to_secs(
- _helpers.utcnow() - datetime.timedelta(hours=1))
- })
+ token = token_factory(
+ claims={
+ "exp": _helpers.datetime_to_secs(
+ _helpers.utcnow() - datetime.timedelta(hours=1)
+ )
+ }
+ )
with pytest.raises(ValueError) as excinfo:
jwt.decode(token, PUBLIC_CERT_BYTES)
- assert excinfo.match(r'Token expired')
+ assert excinfo.match(r"Token expired")
def test_decode_bad_token_wrong_audience(token_factory):
token = token_factory()
- audience = 'audience2@example.com'
+ audience = "audience2@example.com"
with pytest.raises(ValueError) as excinfo:
jwt.decode(token, PUBLIC_CERT_BYTES, audience=audience)
- assert excinfo.match(r'Token has wrong audience')
+ assert excinfo.match(r"Token has wrong audience")
def test_decode_wrong_cert(token_factory):
with pytest.raises(ValueError) as excinfo:
jwt.decode(token_factory(), OTHER_CERT_BYTES)
- assert excinfo.match(r'Could not verify token signature')
+ assert excinfo.match(r"Could not verify token signature")
def test_decode_multicert_bad_cert(token_factory):
- certs = {'1': OTHER_CERT_BYTES, '2': PUBLIC_CERT_BYTES}
+ certs = {"1": OTHER_CERT_BYTES, "2": PUBLIC_CERT_BYTES}
with pytest.raises(ValueError) as excinfo:
jwt.decode(token_factory(), certs)
- assert excinfo.match(r'Could not verify token signature')
+ assert excinfo.match(r"Could not verify token signature")
def test_decode_no_cert(token_factory):
- certs = {'2': PUBLIC_CERT_BYTES}
+ certs = {"2": PUBLIC_CERT_BYTES}
with pytest.raises(ValueError) as excinfo:
jwt.decode(token_factory(), certs)
- assert excinfo.match(r'Certificate for key id 1 not found')
+ assert excinfo.match(r"Certificate for key id 1 not found")
def test_decode_no_key_id(token_factory):
token = token_factory(key_id=False)
- certs = {'2': PUBLIC_CERT_BYTES}
+ certs = {"2": PUBLIC_CERT_BYTES}
payload = jwt.decode(token, certs)
- assert payload['user'] == 'billy bob'
+ assert payload["user"] == "billy bob"
def test_roundtrip_explicit_key_id(token_factory):
- token = token_factory(key_id='3')
- certs = {'2': OTHER_CERT_BYTES, '3': PUBLIC_CERT_BYTES}
+ token = token_factory(key_id="3")
+ certs = {"2": OTHER_CERT_BYTES, "3": PUBLIC_CERT_BYTES}
payload = jwt.decode(token, certs)
- assert payload['user'] == 'billy bob'
+ assert payload["user"] == "billy bob"
class TestCredentials(object):
- SERVICE_ACCOUNT_EMAIL = 'service-account@example.com'
- SUBJECT = 'subject'
- AUDIENCE = 'audience'
- ADDITIONAL_CLAIMS = {'meta': 'data'}
+ SERVICE_ACCOUNT_EMAIL = "service-account@example.com"
+ SUBJECT = "subject"
+ AUDIENCE = "audience"
+ ADDITIONAL_CLAIMS = {"meta": "data"}
credentials = None
@pytest.fixture(autouse=True)
def credentials_fixture(self, signer):
self.credentials = jwt.Credentials(
- signer, self.SERVICE_ACCOUNT_EMAIL, self.SERVICE_ACCOUNT_EMAIL,
- self.AUDIENCE)
+ signer,
+ self.SERVICE_ACCOUNT_EMAIL,
+ self.SERVICE_ACCOUNT_EMAIL,
+ self.AUDIENCE,
+ )
def test_from_service_account_info(self):
- with open(SERVICE_ACCOUNT_JSON_FILE, 'r') as fh:
+ with open(SERVICE_ACCOUNT_JSON_FILE, "r") as fh:
info = json.load(fh)
credentials = jwt.Credentials.from_service_account_info(
- info, audience=self.AUDIENCE)
+ info, audience=self.AUDIENCE
+ )
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
- assert credentials._subject == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
+ assert credentials._subject == info["client_email"]
assert credentials._audience == self.AUDIENCE
def test_from_service_account_info_args(self):
info = SERVICE_ACCOUNT_INFO.copy()
credentials = jwt.Credentials.from_service_account_info(
- info, subject=self.SUBJECT, audience=self.AUDIENCE,
- additional_claims=self.ADDITIONAL_CLAIMS)
+ info,
+ subject=self.SUBJECT,
+ audience=self.AUDIENCE,
+ additional_claims=self.ADDITIONAL_CLAIMS,
+ )
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
assert credentials._subject == self.SUBJECT
assert credentials._audience == self.AUDIENCE
assert credentials._additional_claims == self.ADDITIONAL_CLAIMS
@@ -239,33 +257,37 @@
info = SERVICE_ACCOUNT_INFO.copy()
credentials = jwt.Credentials.from_service_account_file(
- SERVICE_ACCOUNT_JSON_FILE, audience=self.AUDIENCE)
+ SERVICE_ACCOUNT_JSON_FILE, audience=self.AUDIENCE
+ )
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
- assert credentials._subject == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
+ assert credentials._subject == info["client_email"]
assert credentials._audience == self.AUDIENCE
def test_from_service_account_file_args(self):
info = SERVICE_ACCOUNT_INFO.copy()
credentials = jwt.Credentials.from_service_account_file(
- SERVICE_ACCOUNT_JSON_FILE, subject=self.SUBJECT,
- audience=self.AUDIENCE, additional_claims=self.ADDITIONAL_CLAIMS)
+ SERVICE_ACCOUNT_JSON_FILE,
+ subject=self.SUBJECT,
+ audience=self.AUDIENCE,
+ additional_claims=self.ADDITIONAL_CLAIMS,
+ )
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
assert credentials._subject == self.SUBJECT
assert credentials._audience == self.AUDIENCE
assert credentials._additional_claims == self.ADDITIONAL_CLAIMS
def test_from_signing_credentials(self):
jwt_from_signing = self.credentials.from_signing_credentials(
- self.credentials,
- audience=mock.sentinel.new_audience)
+ self.credentials, audience=mock.sentinel.new_audience
+ )
jwt_from_info = jwt.Credentials.from_service_account_info(
- SERVICE_ACCOUNT_INFO,
- audience=mock.sentinel.new_audience)
+ SERVICE_ACCOUNT_INFO, audience=mock.sentinel.new_audience
+ )
assert isinstance(jwt_from_signing, jwt.Credentials)
assert jwt_from_signing._signer.key_id == jwt_from_info._signer.key_id
@@ -279,19 +301,17 @@
assert not self.credentials.expired
def test_with_claims(self):
- new_audience = 'new_audience'
- new_credentials = self.credentials.with_claims(
- audience=new_audience)
+ new_audience = "new_audience"
+ new_credentials = self.credentials.with_claims(audience=new_audience)
assert new_credentials._signer == self.credentials._signer
assert new_credentials._issuer == self.credentials._issuer
assert new_credentials._subject == self.credentials._subject
assert new_credentials._audience == new_audience
- assert (new_credentials._additional_claims ==
- self.credentials._additional_claims)
+ assert new_credentials._additional_claims == self.credentials._additional_claims
def test_sign_bytes(self):
- to_sign = b'123'
+ to_sign = b"123"
signature = self.credentials.sign_bytes(to_sign)
assert crypt.verify_signature(to_sign, signature, PUBLIC_CERT_BYTES)
@@ -299,12 +319,11 @@
assert isinstance(self.credentials.signer, crypt.RSASigner)
def test_signer_email(self):
- assert (self.credentials.signer_email ==
- SERVICE_ACCOUNT_INFO['client_email'])
+ assert self.credentials.signer_email == SERVICE_ACCOUNT_INFO["client_email"]
def _verify_token(self, token):
payload = jwt.decode(token, PUBLIC_CERT_BYTES)
- assert payload['iss'] == self.SERVICE_ACCOUNT_EMAIL
+ assert payload["iss"] == self.SERVICE_ACCOUNT_EMAIL
return payload
def test_refresh(self):
@@ -318,7 +337,7 @@
self.credentials.refresh(None)
assert not self.credentials.expired
- with mock.patch('google.auth._helpers.utcnow') as now:
+ with mock.patch("google.auth._helpers.utcnow") as now:
one_day = datetime.timedelta(days=1)
now.return_value = self.credentials.expiry + one_day
assert self.credentials.expired
@@ -328,55 +347,58 @@
self.credentials.refresh(None)
self.credentials.before_request(
- None, 'GET', 'http://example.com?a=1#3', headers)
+ None, "GET", "http://example.com?a=1#3", headers
+ )
- header_value = headers['authorization']
- _, token = header_value.split(' ')
+ header_value = headers["authorization"]
+ _, token = header_value.split(" ")
# Since the audience is set, it should use the existing token.
- assert token.encode('utf-8') == self.credentials.token
+ assert token.encode("utf-8") == self.credentials.token
payload = self._verify_token(token)
- assert payload['aud'] == self.AUDIENCE
+ assert payload["aud"] == self.AUDIENCE
def test_before_request_refreshes(self):
assert not self.credentials.valid
- self.credentials.before_request(
- None, 'GET', 'http://example.com?a=1#3', {})
+ self.credentials.before_request(None, "GET", "http://example.com?a=1#3", {})
assert self.credentials.valid
class TestOnDemandCredentials(object):
- SERVICE_ACCOUNT_EMAIL = 'service-account@example.com'
- SUBJECT = 'subject'
- ADDITIONAL_CLAIMS = {'meta': 'data'}
+ SERVICE_ACCOUNT_EMAIL = "service-account@example.com"
+ SUBJECT = "subject"
+ ADDITIONAL_CLAIMS = {"meta": "data"}
credentials = None
@pytest.fixture(autouse=True)
def credentials_fixture(self, signer):
self.credentials = jwt.OnDemandCredentials(
- signer, self.SERVICE_ACCOUNT_EMAIL, self.SERVICE_ACCOUNT_EMAIL,
- max_cache_size=2)
+ signer,
+ self.SERVICE_ACCOUNT_EMAIL,
+ self.SERVICE_ACCOUNT_EMAIL,
+ max_cache_size=2,
+ )
def test_from_service_account_info(self):
- with open(SERVICE_ACCOUNT_JSON_FILE, 'r') as fh:
+ with open(SERVICE_ACCOUNT_JSON_FILE, "r") as fh:
info = json.load(fh)
credentials = jwt.OnDemandCredentials.from_service_account_info(info)
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
- assert credentials._subject == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
+ assert credentials._subject == info["client_email"]
def test_from_service_account_info_args(self):
info = SERVICE_ACCOUNT_INFO.copy()
credentials = jwt.OnDemandCredentials.from_service_account_info(
- info, subject=self.SUBJECT,
- additional_claims=self.ADDITIONAL_CLAIMS)
+ info, subject=self.SUBJECT, additional_claims=self.ADDITIONAL_CLAIMS
+ )
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
assert credentials._subject == self.SUBJECT
assert credentials._additional_claims == self.ADDITIONAL_CLAIMS
@@ -384,29 +406,32 @@
info = SERVICE_ACCOUNT_INFO.copy()
credentials = jwt.OnDemandCredentials.from_service_account_file(
- SERVICE_ACCOUNT_JSON_FILE)
+ SERVICE_ACCOUNT_JSON_FILE
+ )
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
- assert credentials._subject == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
+ assert credentials._subject == info["client_email"]
def test_from_service_account_file_args(self):
info = SERVICE_ACCOUNT_INFO.copy()
credentials = jwt.OnDemandCredentials.from_service_account_file(
- SERVICE_ACCOUNT_JSON_FILE, subject=self.SUBJECT,
- additional_claims=self.ADDITIONAL_CLAIMS)
+ SERVICE_ACCOUNT_JSON_FILE,
+ subject=self.SUBJECT,
+ additional_claims=self.ADDITIONAL_CLAIMS,
+ )
- assert credentials._signer.key_id == info['private_key_id']
- assert credentials._issuer == info['client_email']
+ assert credentials._signer.key_id == info["private_key_id"]
+ assert credentials._issuer == info["client_email"]
assert credentials._subject == self.SUBJECT
assert credentials._additional_claims == self.ADDITIONAL_CLAIMS
def test_from_signing_credentials(self):
- jwt_from_signing = self.credentials.from_signing_credentials(
- self.credentials)
+ jwt_from_signing = self.credentials.from_signing_credentials(self.credentials)
jwt_from_info = jwt.OnDemandCredentials.from_service_account_info(
- SERVICE_ACCOUNT_INFO)
+ SERVICE_ACCOUNT_INFO
+ )
assert isinstance(jwt_from_signing, jwt.OnDemandCredentials)
assert jwt_from_signing._signer.key_id == jwt_from_info._signer.key_id
@@ -420,9 +445,8 @@
assert not self.credentials.expired
def test_with_claims(self):
- new_claims = {'meep': 'moop'}
- new_credentials = self.credentials.with_claims(
- additional_claims=new_claims)
+ new_claims = {"meep": "moop"}
+ new_credentials = self.credentials.with_claims(additional_claims=new_claims)
assert new_credentials._signer == self.credentials._signer
assert new_credentials._issuer == self.credentials._issuer
@@ -430,7 +454,7 @@
assert new_credentials._additional_claims == new_claims
def test_sign_bytes(self):
- to_sign = b'123'
+ to_sign = b"123"
signature = self.credentials.sign_bytes(to_sign)
assert crypt.verify_signature(to_sign, signature, PUBLIC_CERT_BYTES)
@@ -438,12 +462,11 @@
assert isinstance(self.credentials.signer, crypt.RSASigner)
def test_signer_email(self):
- assert (self.credentials.signer_email ==
- SERVICE_ACCOUNT_INFO['client_email'])
+ assert self.credentials.signer_email == SERVICE_ACCOUNT_INFO["client_email"]
def _verify_token(self, token):
payload = jwt.decode(token, PUBLIC_CERT_BYTES)
- assert payload['iss'] == self.SERVICE_ACCOUNT_EMAIL
+ assert payload["iss"] == self.SERVICE_ACCOUNT_EMAIL
return payload
def test_refresh(self):
@@ -454,25 +477,27 @@
headers = {}
self.credentials.before_request(
- None, 'GET', 'http://example.com?a=1#3', headers)
+ None, "GET", "http://example.com?a=1#3", headers
+ )
- _, token = headers['authorization'].split(' ')
+ _, token = headers["authorization"].split(" ")
payload = self._verify_token(token)
- assert payload['aud'] == 'http://example.com'
+ assert payload["aud"] == "http://example.com"
# Making another request should re-use the same token.
- self.credentials.before_request(
- None, 'GET', 'http://example.com?b=2', headers)
+ self.credentials.before_request(None, "GET", "http://example.com?b=2", headers)
- _, new_token = headers['authorization'].split(' ')
+ _, new_token = headers["authorization"].split(" ")
assert new_token == token
def test_expired_token(self):
- self.credentials._cache['audience'] = (
- mock.sentinel.token, datetime.datetime.min)
+ self.credentials._cache["audience"] = (
+ mock.sentinel.token,
+ datetime.datetime.min,
+ )
- token = self.credentials._get_jwt_for_audience('audience')
+ token = self.credentials._get_jwt_for_audience("audience")
assert token != mock.sentinel.token