diff --git a/localstack-sdk-python/localstack/sdk/testing/__init__.py b/localstack-sdk-python/localstack/sdk/testing/__init__.py new file mode 100644 index 0000000..ec4930c --- /dev/null +++ b/localstack-sdk-python/localstack/sdk/testing/__init__.py @@ -0,0 +1,3 @@ +from localstack.sdk.testing.decorators import cloudpods + +__all__ = ["cloudpods"] diff --git a/localstack-sdk-python/localstack/sdk/testing/decorators.py b/localstack-sdk-python/localstack/sdk/testing/decorators.py new file mode 100644 index 0000000..adb625a --- /dev/null +++ b/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 diff --git a/localstack-sdk-python/localstack/sdk/testing/pytest/__init__.py b/localstack-sdk-python/localstack/sdk/testing/pytest/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/localstack-sdk-python/localstack/sdk/testing/pytest/plugins.py b/localstack-sdk-python/localstack/sdk/testing/pytest/plugins.py new file mode 100644 index 0000000..6881b32 --- /dev/null +++ b/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() diff --git a/pyproject.toml b/pyproject.toml index f3ca690..e7fc89e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,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" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..c6481d5 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1 @@ +pytest_plugins = ["pytester"] diff --git a/tests/integration/test_aws.py b/tests/integration/test_aws.py index b1df4b0..9ef0232 100644 --- a/tests/integration/test_aws.py +++ b/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)}", diff --git a/tests/integration/test_state.py b/tests/integration/test_state.py index 8400177..d6a5700 100644 --- a/tests/integration/test_state.py +++ b/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 diff --git a/tests/testing/__init__.py b/tests/testing/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/testing/test_decorators.py b/tests/testing/test_decorators.py new file mode 100644 index 0000000..85e3762 --- /dev/null +++ b/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" diff --git a/tests/testing/test_fixtures.py b/tests/testing/test_fixtures.py new file mode 100644 index 0000000..59a613f --- /dev/null +++ b/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) diff --git a/tests/utils.py b/tests/utils.py index 4b4fc3d..35dc7aa 100644 --- a/tests/utils.py +++ b/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", + )