Initial set of test utilities (#11)

localstack-sdk-python/localstack/sdk/testing/__init__.py

@@ -0,0 +1,3 @@
+from localstack.sdk.testing.decorators import cloudpods
+
+__all__ = ["cloudpods"]

localstack-sdk-python/localstack/sdk/testing/decorators.py

@@ -0,0 +1,31 @@
+import logging
+from functools import wraps
+
+from localstack.sdk.pods import PodsClient
+from localstack.sdk.state import StateClient
+
+LOG = logging.getLogger(__name__)
+
+
+def cloudpods(*args, **kwargs):
+    """This is a decorator that loads a cloud pod before a test and resets the state afterward."""
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*test_args, **test_kwargs):
+            if not (pod_name := kwargs.get("name")):
+                raise Exception("Specify a Cloud Pod name in the `name` arg")
+            pods_client = PodsClient()
+            LOG.debug("Loading %s", pod_name)
+            pods_client.load_pod(pod_name=pod_name)
+            try:
+                result = func(*test_args, **test_kwargs)
+            finally:
+                LOG.debug("Reset state of the container")
+                state_client = StateClient()
+                state_client.reset_state()
+            return result
+
+        return wrapper
+
+    return decorator

localstack-sdk-python/localstack/sdk/testing/pytest/plugins.py

@@ -0,0 +1,11 @@
+import pytest
+
+from localstack.sdk.state import StateClient
+
+
+@pytest.fixture
+def reset_state():
+    """This fixture is used to completely reset the state of LocalStack after a test runs."""
+    yield
+    state_client = StateClient()
+    state_client.reset_state()

pyproject.toml

@@ -47,6 +47,9 @@ where = ["localstack-sdk-python/"]
 include = ["localstack*"]
 exclude = ["tests*"]
 
+[project.entry-points.pytest11]
+localstack = "localstack.sdk.testing.pytest.plugins"
+
 [tool.ruff]
 # Always generate Python 3.8-compatible code.
 target-version = "py38"

tests/conftest.py

@@ -0,0 +1 @@
+pytest_plugins = ["pytester"]

tests/integration/test_aws.py

@@ -1,10 +1,8 @@
 import json
 import random
 
-import boto3
-
 import localstack.sdk.aws
-from tests.utils import retry, short_uid
+from tests.utils import boto_client, retry, short_uid
 
 SAMPLE_SIMPLE_EMAIL = {
     "Subject": {
@@ -25,13 +23,7 @@ class TestLocalStackAWS:
     client = localstack.sdk.aws.AWSClient()
 
     def test_list_sqs_messages(self):
-        sqs_client = boto3.client(
-            "sqs",
-            endpoint_url=self.client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
+        sqs_client = boto_client("sqs")
         queue_name = f"queue-{short_uid()}"
         sqs_client.create_queue(QueueName=queue_name)
         queue_url = sqs_client.get_queue_url(QueueName=queue_name)["QueueUrl"]
@@ -49,18 +41,12 @@ def test_list_sqs_messages(self):
         assert len(messages) == 5
 
     def test_list_sqs_messages_from_account_region(self):
-        sqs_client_us = boto3.client(
-            "sqs",
-            endpoint_url=self.client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
+        sqs_client = boto_client("sqs")
         queue_name = f"queue-{short_uid()}"
-        sqs_client_us.create_queue(QueueName=queue_name)
-        queue_url = sqs_client_us.get_queue_url(QueueName=queue_name)["QueueUrl"]
+        sqs_client.create_queue(QueueName=queue_name)
+        queue_url = sqs_client.get_queue_url(QueueName=queue_name)["QueueUrl"]
 
-        send_result = sqs_client_us.send_message(
+        send_result = sqs_client.send_message(
             QueueUrl=queue_url,
             MessageBody=json.dumps({"event": "random-event", "message": "random-message"}),
         )
@@ -72,13 +58,7 @@ def test_list_sqs_messages_from_account_region(self):
         assert messages[0].message_id == send_result["MessageId"]
 
     def test_empty_queue(self):
-        sqs_client = boto3.client(
-            "sqs",
-            endpoint_url=self.client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
+        sqs_client = boto_client("sqs")
         queue_name = f"queue-{short_uid()}"
         sqs_client.create_queue(QueueName=queue_name)
         messages = self.client.list_sqs_messages(
@@ -87,14 +67,7 @@ def test_empty_queue(self):
         assert messages == []
 
     def test_get_and_discard_ses_messages(self):
-        aws_client = boto3.client(
-            "ses",
-            endpoint_url=self.client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
-
+        aws_client = boto_client("ses")
         email = f"user-{short_uid()}@example.com"
         aws_client.verify_email_address(EmailAddress=email)
 
@@ -143,14 +116,7 @@ def test_get_and_discard_ses_messages(self):
         assert not self.client.get_ses_messages()
 
     def test_sns_platform_endpoint_messages(self):
-        client = boto3.client(
-            "sns",
-            endpoint_url=self.client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
-
+        client = boto_client("sns")
         # create a topic
         topic_name = f"topic-{short_uid()}"
         topic_arn = client.create_topic(Name=topic_name)["TopicArn"]
@@ -219,14 +185,7 @@ def test_sns_platform_endpoint_messages(self):
         ], "platform messages not cleared"
 
     def test_sns_messages(self):
-        client = boto3.client(
-            "sns",
-            endpoint_url=self.client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
-
+        client = boto_client("sns")
         numbers = [
             f"+{random.randint(100000000, 9999999999)}",
             f"+{random.randint(100000000, 9999999999)}",

tests/integration/test_state.py

@@ -1,20 +1,14 @@
-import boto3
 import pytest
 
 from localstack.sdk.state import StateClient
+from tests.utils import boto_client
 
 
 class TestStateClient:
     client = StateClient()
 
     def test_reset_state(self):
-        sqs_client = boto3.client(
-            "sqs",
-            endpoint_url=self.client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
+        sqs_client = boto_client("sqs")
         sqs_client.create_queue(QueueName="test-queue")
         url = sqs_client.get_queue_url(QueueName="test-queue")["QueueUrl"]
         assert url

tests/testing/test_decorators.py

@@ -0,0 +1,29 @@
+import pytest
+
+from localstack.sdk.pods import PodsClient
+from localstack.sdk.state import StateClient
+from localstack.sdk.testing import cloudpods
+from tests.utils import boto_client
+
+DECORATOR_POD_NAME = "ls-sdk-pod-decorator"
+QUEUE_NAME = "ls-decorator-queue"
+
+
+@pytest.fixture(scope="class", autouse=True)
+def create_state_and_pod():
+    pods_client = PodsClient()
+    sqs_client = boto_client("sqs")
+    queue_url = sqs_client.create_queue(QueueName=QUEUE_NAME)["QueueUrl"]
+    pods_client.save_pod(DECORATOR_POD_NAME)
+    sqs_client.delete_queue(QueueUrl=queue_url)
+    yield
+    state_client = StateClient()
+    state_client.reset_state()
+    pods_client.delete_pod(DECORATOR_POD_NAME)
+
+
+class TestPodsDecorators:
+    @cloudpods(name=DECORATOR_POD_NAME)
+    def test_pod_load_decorator(self):
+        sqs_client = boto_client("sqs")
+        assert sqs_client.get_queue_url(QueueName=QUEUE_NAME), "state from pod not restored"

tests/testing/test_fixtures.py

@@ -0,0 +1,12 @@
+def test_reset_state(pytester):
+    """Smoke test for the reset_state fixture"""
+    pytester.makeconftest("")
+    pytester.makepyfile(
+        """
+        def test_hello_default(reset_state):
+            pass
+    """
+    )
+
+    result = pytester.runpytest()
+    result.assert_outcomes(passed=1)

tests/utils.py

@@ -2,6 +2,8 @@
 import uuid
 from typing import Callable, TypeVar
 
+import boto3
+
 
 def short_uid():
     return str(uuid.uuid4())[:8]
@@ -22,3 +24,13 @@ def retry(function: Callable[..., T], retries=3, sleep=1.0, sleep_before=0, **kw
             raise_error = error
             time.sleep(sleep)
     raise raise_error
+
+
+def boto_client(service: str):
+    return boto3.client(
+        service,
+        endpoint_url="http://localhost.localstack.cloud:4566",
+        region_name="us-east-1",
+        aws_access_key_id="test",
+        aws_secret_access_key="test",
+    )