2020-01-13 09:49:45 -08:00
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2020-04-22 17:36:41 -07:00
# SPDX-License-Identifier: Apache-2.0
2020-01-13 09:49:45 -08:00
"""
2020-04-24 11:19:31 -07:00
Contains common test fixtures used to run Amazon SQS tests.
2020-01-13 09:49:45 -08:00
"""
2020-04-22 17:36:41 -07:00
import sys
2022-02-03 16:28:27 -08:00
import pytest
2023-10-18 10:35:05 -07:00
2020-04-22 17:36:41 -07:00
# This is needed so Python can find test_tools on the path.
2023-10-18 10:35:05 -07:00
sys . path . append ( " ../.. " )
2020-04-22 17:36:41 -07:00
from test_tools . fixtures . common import *
2020-01-13 09:49:45 -08:00
2023-10-18 10:35:05 -07:00
@pytest.fixture ( name = " make_queue " )
2020-04-22 17:36:41 -07:00
def fixture_make_queue ( request , make_unique_name ) :
2020-01-13 09:49:45 -08:00
"""
2020-04-22 17:36:41 -07:00
Return a factory function that can be used to make a queue for testing.
2020-01-13 09:49:45 -08:00
2020-04-22 17:36:41 -07:00
:param request: The Pytest request object that contains configuration data.
:param make_unique_name: A fixture that returns a unique name.
:return: The factory function to make a test queue.
2020-01-13 09:49:45 -08:00
"""
2023-10-18 10:35:05 -07:00
2020-04-22 17:36:41 -07:00
def _make_queue ( sqs_stubber , sqs_resource ) :
"""
Make a queue that can be used for testing. When stubbing is used, a stubbed
queue is created. When AWS services are used, the queue is deleted after
the test completes.
:param sqs_stubber: The SqsStubber object, configured for stubbing or AWS.
:param sqs_resource: The SQS resource, used to create the queue.
:return: The test queue.
"""
2023-10-18 10:35:05 -07:00
queue_name = make_unique_name ( " queue " )
2020-04-22 17:36:41 -07:00
sqs_stubber . add_response (
2023-10-18 10:35:05 -07:00
" create_queue " ,
expected_params = { " QueueName " : queue_name , " Attributes " : { } } ,
service_response = { " QueueUrl " : " url- " + queue_name } ,
2020-04-22 17:36:41 -07:00
)
queue = sqs_resource . create_queue ( QueueName = queue_name , Attributes = { } )
def fin ( ) :
if not sqs_stubber . use_stubs :
queue . delete ( )
2023-10-18 10:35:05 -07:00
2020-04-22 17:36:41 -07:00
request . addfinalizer ( fin )
return queue
return _make_queue
