Nov 15, 2024 · Nov 15, 2024
diff --git a/localstack-sdk-python/localstack/clients.py b/localstack-sdk-python/localstack/clients.py
 import os

 from localstack.sdk.api_client import ApiClient
 from localstack.sdk.configuration import Configuration

    auth_token: str | None

    def __init__(self, host: str | None = None, auth_token: str | None = None, **kwargs) -> None:
        """
        Initialize a base client to interact with LocalStack developer endpoint.
        :param host: the host, http://localhost.localstack.cloud:4566 by default.
        :param auth_token: if provided, this token would be used for authentication against platform. It not, the
            LocalStack runtime will use the one used to start the container. The token used determines the Cloud
            Pods identity, i.e., which pods are available.
        """
        _host = host or "http://localhost.localstack.cloud:4566"
        self.auth_token = auth_token or os.getenv("LOCALSTACK_AUTH_TOKEN", "").strip("'\" ")
        self.auth_token = auth_token
        self.configuration = Configuration(host=_host)
        self._api_client = ApiClient(configuration=self.configuration)
diff --git a/localstack-sdk-python/localstack/sdk/pods/client.py b/localstack-sdk-python/localstack/sdk/pods/client.py
    def __init__(self, **args) -> None:
        super().__init__(**args)
        self._client = PodsApi(self._api_client)
        # https://github.com/localstack/localstack-ext/pull/3469 could be avoided after this
        assert self.auth_token
        auth_header = get_platform_auth_header(self.auth_token)
        self._api_client.set_default_header("Authorization", auth_header["Authorization"])
        if self.auth_token:
            # If an auth token is provided, it will be used to authenticate platform calls for Cloud Pods.
            #   Only the pods tied to this token will be visible. If not provided, the token will be fetched from the
            #   container. This allows to separate container identity for caller identity, if needed.
            auth_header = get_platform_auth_header(self.auth_token)
            self._api_client.set_default_header("Authorization", auth_header["Authorization"])

    def save_pod(self, pod_name: str) -> None:
        """
diff --git a/tests/integration/test_pods.py b/tests/integration/test_pods.py
 import pytest

 from localstack.sdk.pods import PodsClient

 POD_NAME = "ls-sdk-integration"
        self.client.save_pod(pod_name=POD_NAME)
        self.client.load_pod(pod_name=POD_NAME)
        self.client.delete_pod(pod_name=POD_NAME)

    def test_not_existing_pod(self):
        with pytest.raises(Exception):
            self.client.load_pod(pod_name="i-do-not-exists")
Original file line number	Diff line number	Diff line change
		@@ -1,5 +1,3 @@
		import os

		from localstack.sdk.api_client import ApiClient
		from localstack.sdk.configuration import Configuration

Expand All		@@ -12,7 +10,14 @@ class BaseClient:
		auth_token: str \| None

		def __init__(self, host: str \| None = None, auth_token: str \| None = None, **kwargs) -> None:
		"""
		Initialize a base client to interact with LocalStack developer endpoint.
		:param host: the host, http://localhost.localstack.cloud:4566 by default.
		:param auth_token: if provided, this token would be used for authentication against platform. It not, the
		LocalStack runtime will use the one used to start the container. The token used determines the Cloud
		Pods identity, i.e., which pods are available.
		"""
		_host = host or "http://localhost.localstack.cloud:4566"
		self.auth_token = auth_token or os.getenv("LOCALSTACK_AUTH_TOKEN", "").strip("'\" ")
		self.auth_token = auth_token
		self.configuration = Configuration(host=_host)
		self._api_client = ApiClient(configuration=self.configuration)
Original file line number	Diff line number	Diff line change
Expand Up		@@ -24,10 +24,12 @@ class PodsClient(BaseClient):
		def __init__(self, **args) -> None:
		super().__init__(**args)
		self._client = PodsApi(self._api_client)
		# https://github.com/localstack/localstack-ext/pull/3469 could be avoided after this
		assert self.auth_token
		auth_header = get_platform_auth_header(self.auth_token)
		self._api_client.set_default_header("Authorization", auth_header["Authorization"])
		if self.auth_token:
		# If an auth token is provided, it will be used to authenticate platform calls for Cloud Pods.
		# Only the pods tied to this token will be visible. If not provided, the token will be fetched from the
		# container. This allows to separate container identity for caller identity, if needed.
		auth_header = get_platform_auth_header(self.auth_token)
		self._api_client.set_default_header("Authorization", auth_header["Authorization"])

		def save_pod(self, pod_name: str) -> None:
		"""
Expand Down
Original file line number	Diff line number	Diff line change
		@@ -1,3 +1,5 @@
		import pytest

		from localstack.sdk.pods import PodsClient

		POD_NAME = "ls-sdk-integration"
Expand All		@@ -14,3 +16,7 @@ def test_pod_crud(self):
		self.client.save_pod(pod_name=POD_NAME)
		self.client.load_pod(pod_name=POD_NAME)
		self.client.delete_pod(pod_name=POD_NAME)

		def test_not_existing_pod(self):
		with pytest.raises(Exception):
		self.client.load_pod(pod_name="i-do-not-exists")