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

How to use _receive_assert_delete method in localstack

Best Python code snippet using localstack_python

test_notifications.py

Source:test_notifications.py

...37            # make sure we receive the correct topic ARN in notifications38            assertions.append({'TopicArn': topic_info['TopicArn']})39            # make sure the notification contains message attributes40            assertions.append({'Value': test_value})41            self._receive_assert_delete(queue_url, assertions, sqs_client)42        retry(assert_message, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)43    def test_bucket_notifications(self):44        s3_resource = aws_stack.connect_to_resource('s3')45        s3_client = aws_stack.connect_to_service('s3')46        sqs_client = aws_stack.connect_to_service('sqs')47        # create test bucket and queue48        s3_resource.create_bucket(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)49        queue_info = sqs_client.create_queue(QueueName=TEST_QUEUE_NAME_FOR_S3)50        # create notification on bucket51        queue_url = queue_info['QueueUrl']52        queue_arn = aws_stack.sqs_queue_arn(TEST_QUEUE_NAME_FOR_S3)53        events = ['s3:ObjectCreated:*', 's3:ObjectRemoved:Delete']54        filter_rules = {55            'FilterRules': [{56                'Name': 'prefix',57                'Value': 'testupload/'58            }, {59                'Name': 'suffix',60                'Value': 'testfile.txt'61            }]62        }63        s3_client.put_bucket_notification_configuration(64            Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,65            NotificationConfiguration={66                'QueueConfigurations': [{67                    'Id': 'id0001',68                    'QueueArn': queue_arn,69                    'Events': events,70                    'Filter': {71                        'Key': filter_rules72                    }73                }, {74                    # Add second dummy config to fix https://github.com/localstack/localstack/issues/45075                    'Id': 'id0002',76                    'QueueArn': queue_arn,77                    'Events': [],78                    'Filter': {79                        'Key': filter_rules80                    }81                }]82            }83        )84        # retrieve and check notification config85        config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)86        self.assertEqual(len(config['QueueConfigurations']), 2)87        config = [c for c in config['QueueConfigurations'] if c['Events']][0]88        self.assertEqual(events, config['Events'])89        self.assertEqual(filter_rules, config['Filter']['Key'])90        # upload file to S3 (this should NOT trigger a notification)91        test_key1 = '/testdata'92        test_data1 = b'{"test": "bucket_notification1"}'93        s3_client.upload_fileobj(BytesIO(test_data1), TEST_BUCKET_NAME_WITH_NOTIFICATIONS, test_key1)94        # upload file to S3 (this should trigger a notification)95        test_key2 = 'testupload/dir1/testfile.txt'96        test_data2 = b'{"test": "bucket_notification2"}'97        s3_client.upload_fileobj(BytesIO(test_data2), TEST_BUCKET_NAME_WITH_NOTIFICATIONS, test_key2)98        # receive, assert, and delete message from SQS99        self._receive_assert_delete(queue_url,100            [{'key': test_key2}, {'name': TEST_BUCKET_NAME_WITH_NOTIFICATIONS}], sqs_client)101        # delete notification config102        self._delete_notification_config()103        # put notification config with single event type104        event = 's3:ObjectCreated:*'105        s3_client.put_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,106            NotificationConfiguration={107                'QueueConfigurations': [{108                    'Id': 'id123456',109                    'QueueArn': queue_arn,110                    'Events': [event]111                }]112            }113        )114        config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)115        config = config['QueueConfigurations'][0]116        self.assertEqual(config['Events'], [event])117        # put notification config with single event type118        event = 's3:ObjectCreated:*'119        filter_rules = {120            'FilterRules': [{121                'Name': 'prefix',122                'Value': 'testupload/'123            }]124        }125        s3_client.put_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,126            NotificationConfiguration={127                'QueueConfigurations': [{128                    'Id': 'id123456',129                    'QueueArn': queue_arn,130                    'Events': [event],131                    'Filter': {132                        'Key': filter_rules133                    }134                }]135            }136        )137        config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)138        config = config['QueueConfigurations'][0]139        self.assertEqual(config['Events'], [event])140        self.assertEqual(filter_rules, config['Filter']['Key'])141        # upload file to S3 (this should trigger a notification)142        test_key2 = 'testupload/dir1/testfile.txt'143        test_data2 = b'{"test": "bucket_notification2"}'144        s3_client.upload_fileobj(BytesIO(test_data2), TEST_BUCKET_NAME_WITH_NOTIFICATIONS, test_key2)145        # receive, assert, and delete message from SQS146        self._receive_assert_delete(queue_url,147            [{'key': test_key2}, {'name': TEST_BUCKET_NAME_WITH_NOTIFICATIONS}], sqs_client)148        # delete notification config149        self._delete_notification_config()150        #151        # Tests s3->sns->sqs notifications152        #153        sns_client = aws_stack.connect_to_service('sns')154        topic_info = sns_client.create_topic(Name=TEST_S3_TOPIC_NAME)155        s3_client.put_bucket_notification_configuration(156            Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,157            NotificationConfiguration={158                'TopicConfigurations': [159                    {160                        'Id': 'id123',161                        'Events': ['s3:ObjectCreated:*'],162                        'TopicArn': topic_info['TopicArn']163                    }164                ]165            })166        sns_client.subscribe(TopicArn=topic_info['TopicArn'], Protocol='sqs', Endpoint=queue_arn)167        test_key2 = 'testupload/dir1/testfile.txt'168        test_data2 = b'{"test": "bucket_notification2"}'169        s3_client.upload_fileobj(BytesIO(test_data2), TEST_BUCKET_NAME_WITH_NOTIFICATIONS, test_key2)170        # verify subject and records171        def verify():172            response = sqs_client.receive_message(QueueUrl=queue_url)173            for message in response['Messages']:174                snsObj = json.loads(message['Body'])175                testutil.assert_object({'Subject': 'Amazon S3 Notification'}, snsObj)176                notificationObj = json.loads(snsObj['Message'])177                testutil.assert_objects(178                    [179                        {'key': test_key2},180                        {'name': TEST_BUCKET_NAME_WITH_NOTIFICATIONS}181                    ], notificationObj['Records'])182                sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=message['ReceiptHandle'])183        retry(verify, retries=PUBLICATION_RETRIES, sleep=PUBLICATION_TIMEOUT)184        self._delete_notification_config()185    def _delete_notification_config(self):186        s3_client = aws_stack.connect_to_service('s3')187        s3_client.put_bucket_notification_configuration(188            Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS, NotificationConfiguration={})189        config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)190        self.assertFalse(config.get('QueueConfigurations'))191        self.assertFalse(config.get('TopicConfiguration'))192    def _receive_assert_delete(self, queue_url, assertions, sqs_client=None, required_subject=None):193        if not sqs_client:194            sqs_client = aws_stack.connect_to_service('sqs')195        response = sqs_client.receive_message(QueueUrl=queue_url)196        messages = [json.loads(to_str(m['Body'])) for m in response['Messages']]197        testutil.assert_objects(assertions, messages)198        for message in response['Messages']:...

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.