Added some tests to ensure that the auth works

Author: mridang

.openapi-generator-ignore

@@ -17,3 +17,4 @@ poetry.lock
 requirements.txt
 test-requirements.txt
 tox.ini
+test/*.py

poetry.lock

@@ -36,6 +36,7 @@ tox = ">= 3.9.0"
 flake8 = ">= 4.0.0"
 types-python-dateutil = ">= 2.8.19.14"
 mypy = ">= 1.5"
+testcontainers = "3.7.1"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

pyproject.toml

@@ -0,0 +1,29 @@
+import time
+from datetime import datetime, timezone
+
+from test.auth.test_oauth_authenticator import OAuthAuthenticatorTest
+from zitadel_client.auth.client_credentials_authenticator import ClientCredentialsAuthenticator
+
+
+class ClientCredentialsAuthenticatorTest(OAuthAuthenticatorTest):
+  """
+  Test for ClientCredentialsAuthenticator to verify token refresh functionality.
+  Extends the base OAuthAuthenticatorTest class.
+  """
+
+  def test_refresh_token(self):
+    time.sleep(20)
+
+    authenticator = ClientCredentialsAuthenticator.builder(self.oauth_host, "dummy-client", "dummy-secret") \
+      .scopes({"openid", "foo"}) \
+      .build()
+
+    self.assertTrue(authenticator.get_auth_token(), "Access token should not be empty")
+    token = authenticator.refresh_token()
+    self.assertEqual({"Authorization": "Bearer " + token.access_token}, authenticator.get_auth_headers())
+    self.assertTrue(token.access_token, "Access token should not be null")
+    self.assertTrue(token.expires_at > datetime.now(timezone.utc), "Token expiry should be in the future")
+    self.assertEqual(token.access_token, authenticator.get_auth_token())
+    self.assertEqual(self.oauth_host, authenticator.get_host())
+    self.assertNotEqual(authenticator.refresh_token().access_token, authenticator.refresh_token().access_token,
+                        "Two refreshToken calls should produce different tokens")

test/auth/__init__.py

@@ -0,0 +1,15 @@
+import unittest
+
+from zitadel_client.auth.no_auth_authenticator import NoAuthAuthenticator
+
+
+class NoAuthAuthenticatorTest(unittest.TestCase):
+  def test_returns_empty_headers_and_default_host(self):
+    auth = NoAuthAuthenticator()
+    self.assertEqual({}, auth.get_auth_headers())
+    self.assertEqual("http://localhost", auth.get_host())
+
+  def test_returns_empty_headers_and_custom_host(self):
+    auth = NoAuthAuthenticator("https://custom-host")
+    self.assertEqual({}, auth.get_auth_headers())
+    self.assertEqual("https://custom-host", auth.get_host())

test/auth/test_client_credentials_authenticator.py

@@ -0,0 +1,32 @@
+import unittest
+
+from testcontainers.core.container import DockerContainer
+
+
+class OAuthAuthenticatorTest(unittest.TestCase):
+  """
+  Base test class for OAuth authenticators.
+
+  This class starts a Docker container running the mock OAuth2 server
+  (ghcr.io/navikt/mock-oauth2-server:2.1.10) before any tests run and stops it after all tests.
+  It sets the class variable `oauth_host` to the container’s accessible URL.
+
+  The container is configured to wait for an HTTP response from the "/" endpoint
+  with a status code of 405, using HttpWaitStrategy.
+  """
+  oauth_host: str = None
+  mock_oauth2_server: DockerContainer = None
+
+  @classmethod
+  def setUpClass(cls):
+    cls.mock_oauth2_server = DockerContainer("ghcr.io/navikt/mock-oauth2-server:2.1.10") \
+      .with_exposed_ports(8080)
+    cls.mock_oauth2_server.start()
+    host = cls.mock_oauth2_server.get_container_host_ip()
+    port = cls.mock_oauth2_server.get_exposed_port(8080)
+    cls.oauth_host = f"http://{host}:{port}"
+
+  @classmethod
+  def tearDownClass(cls):
+    if cls.mock_oauth2_server is not None:
+      cls.mock_oauth2_server.stop()

test/auth/test_no_auth_authenticator.py

@@ -0,0 +1,10 @@
+import unittest
+
+from zitadel_client.auth.personal_access_token_authenticator import PersonalAccessTokenAuthenticator
+
+
+class PersonalAccessTokenAuthenticatorTest(unittest.TestCase):
+  def test_returns_expected_headers_and_host(self):
+    auth = PersonalAccessTokenAuthenticator("https://api.example.com", "my-secret-token")
+    self.assertEqual({"Authorization": "Bearer my-secret-token"}, auth.get_auth_headers())
+    self.assertEqual("https://api.example.com", auth.get_host())

test/auth/test_oauth_authenticator.py

@@ -0,0 +1,38 @@
+import time
+from datetime import datetime, timezone
+
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, NoEncryption
+
+from test.auth.test_oauth_authenticator import OAuthAuthenticatorTest
+from zitadel_client.auth.web_token_authenticator import WebTokenAuthenticator
+
+
+class WebTokenAuthenticatorTest(OAuthAuthenticatorTest):
+  """
+  Test for WebTokenAuthenticator to verify JWT token refresh functionality using the builder.
+  """
+
+  def test_refresh_token_using_builder(self):
+    time.sleep(20)
+
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    private_key_pem = key.private_bytes(
+      encoding=Encoding.PEM,
+      format=PrivateFormat.PKCS8,
+      encryption_algorithm=NoEncryption()
+    ).decode('utf-8')
+
+    authenticator = WebTokenAuthenticator.builder(self.oauth_host, "dummy-client", private_key_pem) \
+      .token_lifetime_seconds(3600) \
+      .build()
+
+    self.assertTrue(authenticator.get_auth_token(), "Access token should not be empty")
+    token = authenticator.refresh_token()
+    self.assertEqual({"Authorization": "Bearer " + token.access_token}, authenticator.get_auth_headers())
+    self.assertTrue(token.access_token, "Access token should not be null")
+    self.assertTrue(token.expires_at > datetime.now(timezone.utc), "Token expiry should be in the future")
+    self.assertEqual(token.access_token, authenticator.get_auth_token())
+    self.assertEqual(self.oauth_host, authenticator.get_host())
+    self.assertNotEqual(authenticator.refresh_token().access_token, authenticator.refresh_token().access_token,
+                        "Two refreshToken calls should produce different tokens")