use of a test util

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 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

@@ -1,26 +1,18 @@
-import boto3
 import pytest
 
-from localstack.sdk.aws import AWSClient
 from localstack.sdk.pods import PodsClient
 from localstack.sdk.state import StateClient
 from localstack.sdk.testing import cloudpods
+from 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():
-    sdk_client = AWSClient()
     pods_client = PodsClient()
-    sqs_client = boto3.client(
-        "sqs",
-        endpoint_url=sdk_client.configuration.host,
-        region_name="us-east-1",
-        aws_access_key_id="test",
-        aws_secret_access_key="test",
-    )
+    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)
@@ -33,11 +25,5 @@ def create_state_and_pod():
 class TestPodsDecorators:
     @cloudpods(name=DECORATOR_POD_NAME)
     def test_pod_load_decorator(self):
-        sqs_client = boto3.client(
-            "sqs",
-            endpoint_url=sdk_client.configuration.host,
-            region_name="us-east-1",
-            aws_access_key_id="test",
-            aws_secret_access_key="test",
-        )
+        sqs_client = boto_client("sqs")
         assert sqs_client.get_queue_url(QueueName=QUEUE_NAME), "state from pod not restored"

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",
+    )