Switch to container auth token for pods client (#10)

Author: giograno

localstack-sdk-python/localstack/clients.py

@@ -1,5 +1,3 @@
-import os
-
 from localstack.sdk.api_client import ApiClient
 from localstack.sdk.configuration import Configuration
 
@@ -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)

localstack-sdk-python/localstack/sdk/pods/client.py

@@ -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:
         """

tests/integration/test_pods.py

@@ -1,3 +1,5 @@
+import pytest
+
 from localstack.sdk.pods import PodsClient
 
 POD_NAME = "ls-sdk-integration"
@@ -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")