Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _delete_notification_config method in localstack

Best Python code snippet using localstack_python

test_notifications.py

Source:test_notifications.py

...41    assertions.append({'TopicArn': topic_info['TopicArn']})42    # make sure the notification contains message attributes43    assertions.append({'Value': test_value})44    receive_assert_delete(queue_url, assertions, sqs_client)45def _delete_notification_config():46    s3_client = aws_stack.connect_to_service('s3')47    s3_client.put_bucket_notification_configuration(48        Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS, NotificationConfiguration={})49    config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)50    assert not config.get('QueueConfigurations')51    assert not config.get('TopicConfiguration')52def test_bucket_notifications():53    s3_resource = aws_stack.connect_to_resource('s3')54    s3_client = aws_stack.connect_to_service('s3')55    sqs_client = aws_stack.connect_to_service('sqs')56    # create test bucket and queue57    s3_resource.create_bucket(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)58    queue_info = sqs_client.create_queue(QueueName=TEST_QUEUE_NAME_FOR_S3)59    # create notification on bucket60    queue_url = queue_info['QueueUrl']61    queue_arn = aws_stack.sqs_queue_arn(TEST_QUEUE_NAME_FOR_S3)62    events = ['s3:ObjectCreated:*', 's3:ObjectRemoved:Delete']63    filter_rules = {64        'FilterRules': [{65            'Name': 'prefix',66            'Value': 'testupload/'67        }, {68            'Name': 'suffix',69            'Value': 'testfile.txt'70        }]71    }72    s3_client.put_bucket_notification_configuration(73        Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,74        NotificationConfiguration={75            'QueueConfigurations': [{76                'Id': 'id123456',77                'QueueArn': queue_arn,78                'Events': events,79                'Filter': {80                    'Key': filter_rules81                }82            }]83        }84    )85    # retrieve and check notification config86    config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)87    config = config['QueueConfigurations'][0]88    assert events == config['Events']89    assert 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    receive_assert_delete(queue_url, [{'key': test_key2}, {'name': TEST_BUCKET_NAME_WITH_NOTIFICATIONS}], sqs_client)100    # delete notification config101    _delete_notification_config()102    # put notification config with single event type103    event = 's3:ObjectCreated:*'104    s3_client.put_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,105        NotificationConfiguration={106            'QueueConfigurations': [{107                'Id': 'id123456',108                'QueueArn': queue_arn,109                'Events': [event]110            }]111        }112    )113    config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)114    config = config['QueueConfigurations'][0]115    assert config['Events'] == [event]116    # put notification config with single event type117    event = 's3:ObjectCreated:*'118    filter_rules = {119        'FilterRules': [{120            'Name': 'prefix',121            'Value': 'testupload/'122        }]123    }124    s3_client.put_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,125        NotificationConfiguration={126            'QueueConfigurations': [{127                'Id': 'id123456',128                'QueueArn': queue_arn,129                'Events': [event],130                'Filter': {131                    'Key': filter_rules132                }133            }]134        }135    )136    config = s3_client.get_bucket_notification_configuration(Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS)137    config = config['QueueConfigurations'][0]138    assert config['Events'] == [event]139    assert filter_rules == config['Filter']['Key']140    # upload file to S3 (this should trigger a notification)141    test_key2 = 'testupload/dir1/testfile.txt'142    test_data2 = b'{"test": "bucket_notification2"}'143    s3_client.upload_fileobj(BytesIO(test_data2), TEST_BUCKET_NAME_WITH_NOTIFICATIONS, test_key2)144    # receive, assert, and delete message from SQS145    receive_assert_delete(queue_url, [{'key': test_key2}, {'name': TEST_BUCKET_NAME_WITH_NOTIFICATIONS}], sqs_client)146    # delete notification config147    _delete_notification_config()148    #149    # Tests s3->sns->sqs notifications150    #151    sns_client = aws_stack.connect_to_service('sns')152    topic_info = sns_client.create_topic(Name=TEST_S3_TOPIC_NAME)153    s3_client.put_bucket_notification_configuration(154        Bucket=TEST_BUCKET_NAME_WITH_NOTIFICATIONS,155        NotificationConfiguration={156            'TopicConfigurations': [157                {158                    'Id': 'id123',159                    'Events': ['s3:ObjectCreated:*'],160                    'TopicArn': topic_info['TopicArn']161                }162            ]163        })164    sns_client.subscribe(TopicArn=topic_info['TopicArn'], Protocol='sqs', Endpoint=queue_arn)165    test_key2 = 'testupload/dir1/testfile.txt'166    test_data2 = b'{"test": "bucket_notification2"}'167    s3_client.upload_fileobj(BytesIO(test_data2), TEST_BUCKET_NAME_WITH_NOTIFICATIONS, test_key2)168    # verify subject and records169    response = sqs_client.receive_message(QueueUrl=queue_url)170    for message in response['Messages']:171        snsObj = json.loads(message['Body'])172        testutil.assert_object({'Subject': 'Amazon S3 Notification'}, snsObj)173        notificationObj = json.loads(snsObj['Message'])174#        notificationRecs = [ json.loads( rec_text ) for rec_text in notification175        testutil.assert_objects(176            [177                {'key': test_key2},178                {'name': TEST_BUCKET_NAME_WITH_NOTIFICATIONS}179            ], notificationObj['Records'])180        sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=message['ReceiptHandle'])...

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.