Added tests to ensure that the JWT token based authentication is working as expected

Author: mridang

.github/workflows/test.yml

@@ -35,3 +35,4 @@ jobs:
         env:
           BASE_URL: ${{ secrets.BASE_URL }}
           AUTH_TOKEN: ${{ secrets.AUTH_TOKEN }}
+          JWT_KEY: ${{ secrets.JWT_KEY }}

spec/conftest.py

@@ -1,6 +1,7 @@
 import pytest
 from dotenv import load_dotenv
 
+
 @pytest.fixture(scope="session", autouse=True)
 def load_env():
     """Load the .env file for the entire test session."""

spec/sdk_test_using_personal_access_token_authentication_spec.py

@@ -1,25 +1,31 @@
-import pytest
-import uuid
 import os
+import uuid
+
+import pytest
+
 import zitadel_client as zitadel
 from zitadel_client.auth.personal_access_token_authenticator import PersonalAccessTokenAuthenticator
 from zitadel_client.exceptions import UnauthorizedException
 
+
 @pytest.fixture
 def valid_token():
     """Fixture to return a valid personal access token."""
     return os.getenv("AUTH_TOKEN")
 
+
 @pytest.fixture
 def invalid_token():
     """Fixture to return an invalid token."""
     return "whoops"
 
+
 @pytest.fixture
 def base_url():
     """Fixture to return the base URL."""
     return os.getenv("BASE_URL")
 
+
 @pytest.fixture
 def user_id(valid_token, base_url):
     """Fixture to create a user and return their ID."""
@@ -37,6 +43,7 @@ def user_id(valid_token, base_url):
         except Exception as e:
             pytest.fail(f"Exception while creating user: {e}")
 
+
 def test_should_deactivate_and_reactivate_user_with_valid_token(user_id, valid_token, base_url):
     """Test to (de)activate the user with a valid token."""
     with zitadel.Zitadel(PersonalAccessTokenAuthenticator(base_url, valid_token)) as client:
@@ -51,6 +58,7 @@ def test_should_deactivate_and_reactivate_user_with_valid_token(user_id, valid_t
         except Exception as e:
             pytest.fail(f"Exception when calling deactivate_user or reactivate_user with valid token: {e}")
 
+
 def test_should_not_deactivate_or_reactivate_user_with_invalid_token(user_id, invalid_token, base_url):
     """Test to attempt (de)activating the user with an invalid token."""
     with zitadel.Zitadel(PersonalAccessTokenAuthenticator(base_url, invalid_token)) as client:

spec/sdk_test_using_web_token_authentication_spec.py

@@ -0,0 +1,57 @@
+import os
+import tempfile
+import uuid
+
+import pytest
+
+import zitadel_client as zitadel
+from zitadel_client.auth.web_token_authenticator import WebTokenAuthenticator
+
+
+@pytest.fixture
+def key_file():
+  jwt_key = os.getenv("JWT_KEY")
+  if jwt_key is None:
+    pytest.fail("JWT_KEY is not set in the environment")
+  with tempfile.NamedTemporaryFile(delete=False, mode="w") as tf:
+    tf.write(jwt_key)
+    return tf.name
+
+
+@pytest.fixture
+def base_url():
+  """Fixture to return the base URL."""
+  return os.getenv("BASE_URL")
+
+
+@pytest.fixture
+def user_id(key_file, base_url):
+  """Fixture to create a user and return their ID."""
+  with zitadel.Zitadel(WebTokenAuthenticator.from_json(base_url, key_file)) as client:
+    try:
+      response = client.users.add_human_user(
+        body=zitadel.models.V2AddHumanUserRequest(
+          username=uuid.uuid4().hex,
+          profile=zitadel.models.V2SetHumanProfile(given_name="John", family_name="Doe"),
+          email=zitadel.models.V2SetHumanEmail(email=f"johndoe{uuid.uuid4().hex}@caos.ag")
+        )
+      )
+      print("User created:", response)
+      return response.user_id
+    except Exception as e:
+      pytest.fail(f"Exception while creating user: {e}")
+
+
+def test_should_deactivate_and_reactivate_user_with_valid_token(user_id, key_file, base_url):
+  """Test to (de)activate the user with a valid token."""
+  with zitadel.Zitadel(WebTokenAuthenticator.from_json(base_url, key_file)) as client:
+    try:
+      deactivate_response = client.users.deactivate_user(user_id=user_id)
+      print("User deactivated:", deactivate_response)
+
+      reactivate_response = client.users.reactivate_user(user_id=user_id)
+      print("User reactivated:", reactivate_response)
+      # Adjust based on actual response format
+      # assert reactivate_response["status"] == "success"
+    except Exception as e:
+      pytest.fail(f"Exception when calling deactivate_user or reactivate_user with valid token: {e}")

zitadel_client/auth/authenticator.py

@@ -6,95 +6,95 @@
 
 
 class Authenticator(ABC):
-  """
-  Abstract base class for authenticators.
-
-  This class defines the basic structure for any authenticator by requiring the implementation
-  of a method to retrieve authentication headers, and provides a way to store and retrieve the host.
-  """
-
-  def __init__(self, host: str):
     """
-    Initializes the Authenticator with the specified host.
+    Abstract base class for authenticators.
 
-    :param host: The base URL or endpoint for the service.
+    This class defines the basic structure for any authenticator by requiring the implementation
+    of a method to retrieve authentication headers, and provides a way to store and retrieve the host.
     """
-    self.host = host
-
-  @abstractmethod
-  def get_auth_headers(self) -> Dict[str, str]:
-    """
-    Retrieves the authentication headers to be sent with requests.
-
-    Subclasses must override this method to return the appropriate headers.
 
-    :return: A dictionary mapping header names to their values.
-    """
-    pass
+    def __init__(self, host: str):
+        """
+        Initializes the Authenticator with the specified host.
 
-  def get_host(self) -> str:
-    """
-    Returns the stored host.
+        :param host: The base URL or endpoint for the service.
+        """
+        self.host = host
 
-    :return: The host as a string.
-    """
-    return self.host
+    @abstractmethod
+    def get_auth_headers(self) -> Dict[str, str]:
+        """
+        Retrieves the authentication headers to be sent with requests.
 
+        Subclasses must override this method to return the appropriate headers.
 
-class Token:
-  def __init__(self, access_token: str, expires_at: datetime):
-    """
-    Initializes a new Token instance.
+        :return: A dictionary mapping header names to their values.
+        """
+        pass
 
-    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
+    def get_host(self) -> str:
+        """
+        Returns the stored host.
 
-    # 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
+        :return: The host as a string.
+        """
+        return self.host
 
-  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
+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
+
+        # 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
 
 
 T = TypeVar("T", bound="OAuthAuthenticatorBuilder")
 
 
 class OAuthAuthenticatorBuilder(ABC, Generic[T]):
-  """
-  Abstract builder class for constructing OAuth authenticator instances.
-
-  This builder provides common configuration options such as the OpenId instance and authentication scopes.
-  """
-
-  def __init__(self, host: str):
     """
-    Initializes the OAuthAuthenticatorBuilder with a given host.
+    Abstract builder class for constructing OAuth authenticator instances.
 
-    :param host: The base URL for the OAuth provider.
+    This builder provides common configuration options such as the OpenId instance and authentication scopes.
     """
-    self.open_id = OpenId(host)
-    self.auth_scopes = {"openid", "urn:zitadel:iam:org:project:id:zitadel:aud"}
 
-  def scopes(self: T, *auth_scopes: str) -> T:
-    """
-    Sets the authentication scopes for the OAuth authenticator.
+    def __init__(self, host: str):
+        """
+        Initializes the OAuthAuthenticatorBuilder with a given host.
 
-    :param auth_scopes: A variable number of scope strings.
-    :return: The builder instance to allow for method chaining.
-    """
-    self.auth_scopes = set(auth_scopes)
-    return self
+        :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"}
+
+    def scopes(self: T, *auth_scopes: str) -> T:
+        """
+        Sets the authentication scopes for the OAuth authenticator.
+
+        :param auth_scopes: A variable number of scope strings.
+        :return: The builder instance to allow for method chaining.
+        """
+        self.auth_scopes = set(auth_scopes)
+        return self