Hardened the test suite to ensure that all code paths are covered

Author: mridang

test/auth/test_client_credentials_authenticator.py

@@ -15,7 +15,7 @@ def test_refresh_token(self):
     time.sleep(20)
 
     authenticator = ClientCredentialsAuthenticator.builder(self.oauth_host, "dummy-client", "dummy-secret") \
-      .scopes({"openid", "foo"}) \
+      .scopes("openid", "foo") \
       .build()
 
     self.assertTrue(authenticator.get_auth_token(), "Access token should not be empty")

zitadel_client/auth/authenticator.py

@@ -43,10 +43,30 @@ def get_host(self) -> str:
 
 class Token:
   def __init__(self, access_token: str, expires_at: datetime):
+    """
+    Initializes a new Token instance.
+
+    Parameters:
+    - access_token (str): The JWT or OAuth token.
+    - expires_at (datetime): The expiration time of the token. It should be timezone-aware.
+      If a naive datetime is provided, it will be converted to an aware datetime in UTC.
+    """
     self.access_token = access_token
-    self.expires_at = expires_at
+
+    # Ensure expires_at is timezone-aware. If naive, assume UTC.
+    if expires_at.tzinfo is None:
+      self.expires_at = expires_at.replace(tzinfo=timezone.utc)
+    else:
+      self.expires_at = expires_at
 
   def is_expired(self) -> bool:
+    """
+    Checks if the token is expired by comparing the current UTC time
+    with the token's expiration time.
+
+    Returns:
+    - bool: True if expired, False otherwise.
+    """
     return datetime.now(timezone.utc) >= self.expires_at
 
 
@@ -64,14 +84,14 @@ def __init__(self, host: str):
     :param host: The base URL for the OAuth provider.
     """
     self.open_id = OpenId(host)
-    self.auth_scopes = "openid urn:zitadel:iam:org:project:id:zitadel:aud"
+    self.auth_scopes = {"openid", "urn:zitadel:iam:org:project:id:zitadel:aud"}
 
-  def scopes(self, auth_scopes: set) -> "OAuthAuthenticatorBuilder":
+  def scopes(self, *auth_scopes: str) -> "OAuthAuthenticatorBuilder":
     """
     Sets the authentication scopes for the OAuth authenticator.
 
-    :param auth_scopes: A set of scope strings.
+    :param auth_scopes: A variable number of scope strings.
     :return: The builder instance to allow for method chaining.
     """
-    self.auth_scopes = " ".join(auth_scopes)
+    self.auth_scopes = set(auth_scopes)
     return self

zitadel_client/auth/client_credentials_authenticator.py

@@ -1,4 +1,4 @@
-from typing import override
+from typing import override, Set
 
 from authlib.integrations.requests_client import OAuth2Session
 
@@ -14,7 +14,7 @@ class ClientCredentialsAuthenticator(OAuthAuthenticator):
   Uses client_id and client_secret to obtain an access token from the OAuth2 token endpoint.
   """
 
-  def __init__(self, open_id: OpenId, client_id: str, client_secret: str, auth_scopes: str):
+  def __init__(self, open_id: OpenId, client_id: str, client_secret: str, auth_scopes: Set[str]):
     """
     Constructs a ClientCredentialsAuthenticator.
 
@@ -23,7 +23,7 @@ def __init__(self, open_id: OpenId, client_id: str, client_secret: str, auth_sco
     :param client_secret: The OAuth client secret.
     :param auth_scopes: The scope(s) for the token request.
     """
-    super().__init__(open_id, OAuth2Session(client_id=client_id, client_secret=client_secret, scope=auth_scopes))
+    super().__init__(open_id, OAuth2Session(client_id=client_id, client_secret=client_secret, scope=" ".join(auth_scopes)))
 
   @override
   def get_grant(self) -> dict:
@@ -73,9 +73,4 @@ def build(self) -> ClientCredentialsAuthenticator:
 
     :return: A configured ClientCredentialsAuthenticator.
     """
-    return ClientCredentialsAuthenticator(
-      self.open_id,  # OpenId instance with endpoint info (constructed from host)
-      self.client_id,  # OAuth client identifier
-      self.client_secret,  # OAuth client secret
-      self.auth_scopes  # Authentication scopes configured in the builder
-    )
+    return ClientCredentialsAuthenticator(self.open_id, self.client_id, self.client_secret, self.auth_scopes)

zitadel_client/auth/web_token_authenticator.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timedelta, timezone
+from typing import Set
 
 from authlib.integrations.requests_client import OAuth2Session
 from authlib.jose import jwt, JoseError
@@ -15,7 +16,7 @@ class WebTokenAuthenticator(OAuthAuthenticator):
   This implementation builds a JWT assertion dynamically in get_grant().
   """
 
-  def __init__(self, open_id: OpenId, auth_scopes: str,
+  def __init__(self, open_id: OpenId, auth_scopes: Set[str],
                jwt_issuer: str, jwt_subject: str, jwt_audience: str,
                private_key: str, jwt_lifetime: timedelta = timedelta(hours=1), jwt_algorithm: str = "RS256"):
     """
@@ -30,7 +31,7 @@ def __init__(self, open_id: OpenId, auth_scopes: str,
     :param jwt_lifetime: Lifetime of the JWT in seconds.
     :param jwt_algorithm: The JWT signing algorithm (default "RS256").
     """
-    super().__init__(open_id, OAuth2Session(scope=auth_scopes))
+    super().__init__(open_id, OAuth2Session(scope=" ".join(auth_scopes)))
     self.jwt_issuer = jwt_issuer
     self.jwt_subject = jwt_subject
     self.jwt_audience = jwt_audience
@@ -109,16 +110,6 @@ def token_lifetime_seconds(self, seconds: int) -> "JWTAuthenticatorBuilder":
     self.jwt_lifetime = timedelta(seconds=seconds)
     return self
 
-  def scopes(self, scopes: set) -> "JWTAuthenticatorBuilder":
-    """
-    Overrides the default scopes for the JWTAuthenticator.
-
-    :param scopes: A set of scope strings.
-    :return: The builder instance.
-    """
-    self.auth_scopes = " ".join(scopes)
-    return self
-
   def build(self) -> WebTokenAuthenticator:
     """
     Builds and returns a new JWTAuthenticator instance using the configured parameters.