Restructure tests: split TransportOptions unit tests from Zitadel integration tests

Author: mridang

test/fixtures/mappings/settings.json

@@ -8,6 +8,9 @@
     "headers": {
       "Content-Type": "application/json"
     },
-    "jsonBody": {}
+    "jsonBody": {
+      "defaultLanguage": "{{request.scheme}}",
+      "defaultOrgId": "{{request.headers.X-Custom-Header}}"
+    }
   }
 }

test/test_transport_options.py

@@ -1,148 +1,40 @@
-import json
-import os
 import unittest
-import urllib.request
-from typing import Optional
-
-from testcontainers.core.container import DockerContainer
-from testcontainers.core.network import Network
-from testcontainers.core.wait_strategies import PortWaitStrategy
-from testcontainers.core.waiting_utils import wait_container_is_ready
 
 from zitadel_client.transport_options import TransportOptions
-from zitadel_client.zitadel import Zitadel
-
-FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
-
-
-@wait_container_is_ready()
-def _wait_for_wiremock(host: str, port: str) -> None:
-    url = f"http://{host}:{port}/__admin/mappings"
-    with urllib.request.urlopen(url, timeout=5) as resp:  # noqa: S310
-        if resp.status != 200:
-            raise ConnectionError(f"WireMock not ready: {resp.status}")
 
 
 class TransportOptionsTest(unittest.TestCase):
-    host: Optional[str] = None
-    http_port: Optional[str] = None
-    https_port: Optional[str] = None
-    proxy_port: Optional[str] = None
-    ca_cert_path: Optional[str] = None
-    wiremock: DockerContainer = None
-    proxy: DockerContainer = None
-    network: Network = None
-
-    @classmethod
-    def setup_class(cls) -> None:
-        cls.ca_cert_path = os.path.join(FIXTURES_DIR, "ca.pem")
-        keystore_path = os.path.join(FIXTURES_DIR, "keystore.p12")
-        squid_conf = os.path.join(FIXTURES_DIR, "squid.conf")
-
-        cls.network = Network().create()
-
-        cls.wiremock = (
-            DockerContainer("wiremock/wiremock:3.12.1")
-            .with_network(cls.network)
-            .with_network_aliases("wiremock")
-            .with_exposed_ports(8080, 8443)
-            .with_volume_mapping(keystore_path, "/home/wiremock/keystore.p12", mode="ro")
-            .with_volume_mapping(
-                os.path.join(FIXTURES_DIR, "mappings"), "/home/wiremock/mappings", mode="ro"
-            )
-            .with_command(
-                "--https-port 8443"
-                " --https-keystore /home/wiremock/keystore.p12"
-                " --keystore-password password"
-                " --keystore-type PKCS12"
-                " --global-response-templating"
-            )
-        )
-        cls.wiremock.start()
-
-        cls.proxy = (
-            DockerContainer("ubuntu/squid:6.10-24.10_beta")
-            .with_network(cls.network)
-            .with_exposed_ports(3128)
-            .with_volume_mapping(squid_conf, "/etc/squid/squid.conf", mode="ro")
-            .waiting_for(PortWaitStrategy(3128))
-        )
-        cls.proxy.start()
 
-        cls.host = cls.wiremock.get_container_host_ip()
-        cls.http_port = cls.wiremock.get_exposed_port(8080)
-        cls.https_port = cls.wiremock.get_exposed_port(8443)
-        cls.proxy_port = cls.proxy.get_exposed_port(3128)
+    def test_defaults_returns_empty(self) -> None:
+        self.assertEqual({}, TransportOptions.defaults().to_session_kwargs())
 
-        _wait_for_wiremock(cls.host, cls.http_port)
+    def test_insecure_sets_verify_false(self) -> None:
+        opts = TransportOptions(insecure=True)
+        self.assertEqual({"verify": False}, opts.to_session_kwargs())
 
-    @classmethod
-    def teardown_class(cls) -> None:
-        if cls.proxy is not None:
-            cls.proxy.stop()
-        if cls.wiremock is not None:
-            cls.wiremock.stop()
-        if cls.network is not None:
-            cls.network.remove()
-
-    def test_custom_ca_cert(self) -> None:
-        zitadel = Zitadel.with_client_credentials(
-            f"https://{self.host}:{self.https_port}",
-            "dummy-client",
-            "dummy-secret",
-            transport_options=TransportOptions(ca_cert_path=self.ca_cert_path),
-        )
-        self.assertIsNotNone(zitadel)
-
-    def test_insecure_mode(self) -> None:
-        zitadel = Zitadel.with_client_credentials(
-            f"https://{self.host}:{self.https_port}",
-            "dummy-client",
-            "dummy-secret",
-            transport_options=TransportOptions(insecure=True),
-        )
-        self.assertIsNotNone(zitadel)
-
-    def test_default_headers(self) -> None:
-        zitadel = Zitadel.with_client_credentials(
-            f"http://{self.host}:{self.http_port}",
-            "dummy-client",
-            "dummy-secret",
-            transport_options=TransportOptions(default_headers={"X-Custom-Header": "test-value"}),
-        )
-        self.assertIsNotNone(zitadel)
-
-        zitadel.settings.get_general_settings({})
-
-        verify_body = json.dumps(
-            {
-                "url": "/zitadel.settings.v2.SettingsService/GetGeneralSettings",
-                "headers": {"X-Custom-Header": {"equalTo": "test-value"}},
-            }
-        ).encode()
-        req = urllib.request.Request(
-            f"http://{self.host}:{self.http_port}/__admin/requests/count",
-            data=verify_body,
-            headers={"Content-Type": "application/json"},
-            method="POST",
-        )
-        with urllib.request.urlopen(req) as resp:  # noqa: S310
-            result = json.loads(resp.read().decode())
-        self.assertGreaterEqual(result["count"], 1, "Custom header should be present on API call")
+    def test_ca_cert_path_sets_verify(self) -> None:
+        opts = TransportOptions(ca_cert_path="/path/to/ca.pem")
+        self.assertEqual({"verify": "/path/to/ca.pem"}, opts.to_session_kwargs())
 
-    def test_proxy_url(self) -> None:
-        zitadel = Zitadel.with_access_token(
-            "http://wiremock:8080",
-            "test-token",
-            transport_options=TransportOptions(proxy_url=f"http://{self.host}:{self.proxy_port}"),
+    def test_proxy_url_sets_proxies(self) -> None:
+        opts = TransportOptions(proxy_url="http://proxy:3128")
+        self.assertEqual(
+            {"proxies": {"http": "http://proxy:3128", "https": "http://proxy:3128"}},
+            opts.to_session_kwargs(),
         )
-        self.assertIsNotNone(zitadel)
-        zitadel.settings.get_general_settings({})
 
-    def test_no_ca_cert_fails(self) -> None:
-        with self.assertRaises(Exception):  # noqa: B017
-            Zitadel.with_client_credentials(
-                f"https://{self.host}:{self.https_port}",
-                "dummy-client",
-                "dummy-secret",
-            )
+    def test_insecure_takes_precedence_over_ca_cert(self) -> None:
+        opts = TransportOptions(insecure=True, ca_cert_path="/path/to/ca.pem")
+        self.assertEqual({"verify": False}, opts.to_session_kwargs())
+
+    def test_immutability(self) -> None:
+        opts = TransportOptions.defaults()
+        with self.assertRaises(AttributeError):
+            opts.insecure = True  # type: ignore[misc]
+
+    def test_defaults_factory(self) -> None:
+        opts = TransportOptions.defaults()
+        self.assertEqual({}, dict(opts.default_headers))
+        self.assertIsNone(opts.ca_cert_path)
+        self.assertFalse(opts.insecure)
+        self.assertIsNone(opts.proxy_url)

test/test_zitadel.py

@@ -1,17 +1,32 @@
 import importlib
 import inspect
+import os
 import pkgutil
 import unittest
+import urllib.request
+from typing import Optional
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.network import Network
+from testcontainers.core.wait_strategies import PortWaitStrategy
+from testcontainers.core.waiting_utils import wait_container_is_ready
 
 from zitadel_client.auth.no_auth_authenticator import NoAuthAuthenticator
+from zitadel_client.transport_options import TransportOptions
 from zitadel_client.zitadel import Zitadel
 
+FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
+
+
+@wait_container_is_ready()
+def _wait_for_wiremock(host: str, port: str) -> None:
+    url = f"http://{host}:{port}/__admin/mappings"
+    with urllib.request.urlopen(url, timeout=5) as resp:  # noqa: S310
+        if resp.status != 200:
+            raise ConnectionError(f"WireMock not ready: {resp.status}")
+
 
 class ZitadelServicesTest(unittest.TestCase):
-    """
-    Test to verify that all API service classes defined in the "zitadel_client.api" namespace
-    are registered as attributes in the Zitadel class.
-    """
 
     def test_services_dynamic(self) -> None:
         expected = set()
@@ -30,3 +45,114 @@ def test_services_dynamic(self) -> None:
             and getattr(zitadel, attr).__class__.__module__.startswith("zitadel_client.api")
         }
         self.assertEqual(expected, actual)
+
+
+class ZitadelTransportTest(unittest.TestCase):
+    host: Optional[str] = None
+    http_port: Optional[str] = None
+    https_port: Optional[str] = None
+    proxy_port: Optional[str] = None
+    ca_cert_path: Optional[str] = None
+    wiremock: DockerContainer = None
+    proxy: DockerContainer = None
+    network: Network = None
+
+    @classmethod
+    def setup_class(cls) -> None:
+        cls.ca_cert_path = os.path.join(FIXTURES_DIR, "ca.pem")
+        keystore_path = os.path.join(FIXTURES_DIR, "keystore.p12")
+        squid_conf = os.path.join(FIXTURES_DIR, "squid.conf")
+
+        cls.network = Network().create()
+
+        cls.wiremock = (
+            DockerContainer("wiremock/wiremock:3.12.1")
+            .with_network(cls.network)
+            .with_network_aliases("wiremock")
+            .with_exposed_ports(8080, 8443)
+            .with_volume_mapping(keystore_path, "/home/wiremock/keystore.p12", mode="ro")
+            .with_volume_mapping(
+                os.path.join(FIXTURES_DIR, "mappings"), "/home/wiremock/mappings", mode="ro"
+            )
+            .with_command(
+                "--https-port 8443"
+                " --https-keystore /home/wiremock/keystore.p12"
+                " --keystore-password password"
+                " --keystore-type PKCS12"
+                " --global-response-templating"
+            )
+        )
+        cls.wiremock.start()
+
+        cls.proxy = (
+            DockerContainer("ubuntu/squid:6.10-24.10_beta")
+            .with_network(cls.network)
+            .with_exposed_ports(3128)
+            .with_volume_mapping(squid_conf, "/etc/squid/squid.conf", mode="ro")
+            .waiting_for(PortWaitStrategy(3128))
+        )
+        cls.proxy.start()
+
+        cls.host = cls.wiremock.get_container_host_ip()
+        cls.http_port = cls.wiremock.get_exposed_port(8080)
+        cls.https_port = cls.wiremock.get_exposed_port(8443)
+        cls.proxy_port = cls.proxy.get_exposed_port(3128)
+
+        _wait_for_wiremock(cls.host, cls.http_port)
+
+    @classmethod
+    def teardown_class(cls) -> None:
+        if cls.proxy is not None:
+            cls.proxy.stop()
+        if cls.wiremock is not None:
+            cls.wiremock.stop()
+        if cls.network is not None:
+            cls.network.remove()
+
+    def test_custom_ca_cert(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"https://{self.host}:{self.https_port}",
+            "dummy-client",
+            "dummy-secret",
+            transport_options=TransportOptions(ca_cert_path=self.ca_cert_path),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("https", response.default_language)
+
+    def test_insecure_mode(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"https://{self.host}:{self.https_port}",
+            "dummy-client",
+            "dummy-secret",
+            transport_options=TransportOptions(insecure=True),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("https", response.default_language)
+
+    def test_default_headers(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"http://{self.host}:{self.http_port}",
+            "dummy-client",
+            "dummy-secret",
+            transport_options=TransportOptions(default_headers={"X-Custom-Header": "test-value"}),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("http", response.default_language)
+        self.assertEqual("test-value", response.default_org_id)
+
+    def test_proxy_url(self) -> None:
+        zitadel = Zitadel.with_access_token(
+            "http://wiremock:8080",
+            "test-token",
+            transport_options=TransportOptions(proxy_url=f"http://{self.host}:{self.proxy_port}"),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("http", response.default_language)
+
+    def test_no_ca_cert_fails(self) -> None:
+        with self.assertRaises(Exception):  # noqa: B017
+            Zitadel.with_client_credentials(
+                f"https://{self.host}:{self.https_port}",
+                "dummy-client",
+                "dummy-secret",
+            )