KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_sqs_queue_url method in localstack

Best Python code snippet using localstack_python

test_check_and_cache_sqs_queues_lengths.py

Source:test_check_and_cache_sqs_queues_lengths.py

...9_faker = faker.Faker()10_base_sqs_url = "http://localhost/sqs/"11class UnexpectedException(Exception):12    """Dummy exception for testing."""13def get_sqs_queue_url(queue_name, *args, **kwargs):14    """Override behavior of sqs.get_sqs_queue_url."""15    return f"{_base_sqs_url}{queue_name}"16class CheckAndCacheSqsQueueLengthsTest(TestCase):17    """tasks.maintenance.check_and_cache_sqs_queues_lengths test case."""18    def setUp(self):19        """Set up common variables for tests."""20        self.houndigrade_queue_name = f"houndigrade_{_faker.slug()}"21        self.houndigrade_dlq_name = aws.get_sqs_queue_dlq_name(22            self.houndigrade_queue_name23        )24        self.houndigrade_queue_url = get_sqs_queue_url(self.houndigrade_queue_name)25        self.houndigrade_dlq_url = get_sqs_queue_url(self.houndigrade_dlq_name)26        self.cloudtrail_queue_name = f"cloudtrail_{_faker.slug()}"27        self.cloudtrail_dlq_name = aws.get_sqs_queue_dlq_name(28            self.cloudtrail_queue_name29        )30        self.cloudtrail_queue_url = get_sqs_queue_url(self.cloudtrail_queue_name)31        self.cloudtrail_dlq_url = get_sqs_queue_url(self.cloudtrail_dlq_name)32        # It's very important to clear cache between these test runs because different33        # tests expect values *not* to be set whereas other tests may set them.34        cache.clear()35    def test_check_and_cache_sqs_queues_lengths(self):36        """37        Test happy path for check_and_cache_sqs_queues_lengths.38        We expect to get valid integers for all queues and to cache those values.39        """40        expected_counts = {41            "houndigrade_results": _faker.random_int(),42            "houndigrade_results_dlq": _faker.random_int(),43            "cloudtrail_notifications": _faker.random_int(),44            "cloudtrail_notifications_dlq": _faker.random_int(),45        }...

services.py

Source:services.py

...47        ),48    )49    response = lambda_client.invoke(FunctionName="main", InvocationType="Event", Payload=json.dumps(data))50    return response51def get_sqs_queue_url(file_size: int) -> str:52    """Return queue url based on input file size"""53    if file_size < settings.BASE_QUEUE_LIMIT:54        return settings.SQS_WORKER_QUEUE_URL55    elif file_size < settings.EXT_QUEUE_LIMIT:56        return settings.SQS_WORKER_EXT_QUEUE_URL57    else:58        return settings.SQS_WORKER_MAX_QUEUE_URL59def schedule_worker_with(data: dict, source_file_size: int):60    if settings.LOCAL_LAMBDA:61        return local_lambda_invoke(data)62    queue_url = get_sqs_queue_url(file_size=source_file_size)63    sqs_response = sqs.send_message(QueueUrl=queue_url, MessageBody=json.dumps(data))64    return sqs_response["MessageId"]65def schedule_object_meta_processing(obj, source_file_size, copy_steps, auto_refresh):66    data = {67        "type": obj.meta_file_processing_type,68        "data": obj.meta_file_serialization(),69        "copy_steps": copy_steps,70        "auto_refresh": auto_refresh,71    }72    try:73        message_id = schedule_worker_with(data=data, source_file_size=source_file_size)74    except Exception as e:75        obj.meta_data.status = constants.ProcessingState.FAILED76        obj.meta_data.save(update_fields=["status"])...

aws_services.py

Source:aws_services.py

...33def recive_messages():34    message_dict = {}35    try:36        response = aws_connect().receive_message(37            QueueUrl=RuntimeConfig.get_sqs_queue_url(),38            AttributeNames=["All"],39            WaitTimeSeconds=20,40        )41        if not response.get("Messages"):42            return False43        receipt_handler = response.get("Messages")[0].get("ReceiptHandle")44        message = response.get("Messages")[0].get("Body")45        message_dict["message"] = {"messageKey": receipt_handler, "body": message}46        return message_dict47    except Exception as e:48        raise e49def delete_message(reciptHandler):50    try:51        response = aws_connect().delete_message(52            QueueUrl=RuntimeConfig.get_sqs_queue_url(), ReceiptHandle=reciptHandler53        )54        return response55    except Exception as e:56        raise e57def sqs_list_queues(aws_conn, prefix_name):58    response = aws_conn.list_queues(QueueNamePrefix=prefix_name)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.