From fc5d945abac25e601a84406ffee22bf2a16f214e Mon Sep 17 00:00:00 2001 From: BryanFauble <17128019+BryanFauble@users.noreply.github.com> Date: Mon, 23 Oct 2023 17:29:16 -0700 Subject: [PATCH] [ETL-553] Allow merging of notification configs (#87) * Allow merging of notification configs --- src/lambda_function/s3_event_config/app.py | 297 ++++++++++++-- tests/test_s3_event_config_lambda.py | 435 +++++++++++++++++++-- 2 files changed, 671 insertions(+), 61 deletions(-) diff --git a/src/lambda_function/s3_event_config/app.py b/src/lambda_function/s3_event_config/app.py index 307b2ce2..ff01a9a4 100644 --- a/src/lambda_function/s3_event_config/app.py +++ b/src/lambda_function/s3_event_config/app.py @@ -4,11 +4,14 @@ input data S3 bucket set by the environment variable `S3_SOURCE_BUCKET_NAME`. This Lambda app also has the option of deleting the notification configuration -from an S3 bucket +from an S3 bucket. + +Only certain notification configurations work. Only `QueueConfigurations` are expected. """ import os import json import logging +import typing import boto3 @@ -23,7 +26,12 @@ def lambda_handler(event, context): logger.info(f"Received event: {json.dumps(event, indent=2)}") if event["RequestType"] == "Delete": logger.info(f'Request Type:{event["RequestType"]}') - delete_notification(s3, bucket=os.environ["S3_SOURCE_BUCKET_NAME"]) + delete_notification( + s3, + bucket=os.environ["S3_SOURCE_BUCKET_NAME"], + destination_type=os.environ["S3_TO_GLUE_DESTINATION_TYPE"], + destination_arn=os.environ["S3_TO_GLUE_DESTINATION_ARN"], + ) logger.info("Sending response to custom resource after Delete") elif event["RequestType"] in ["Update", "Create"]: logger.info(f'Request Type: {event["RequestType"]}') @@ -40,14 +48,164 @@ def lambda_handler(event, context): raise KeyError(err_msg) +class ExistingNotificationConfiguration(typing.NamedTuple): + existing_bucket_notification_configuration: dict + existing_notification_configurations_for_type: list + + +def get_existing_bucket_notification_configuration_and_type( + s3_client: boto3.client, bucket: str, destination_type: str +) -> ExistingNotificationConfiguration: + """ + Gets the existing bucket notification configuration and the existing notification + configurations for a specific destination type. + + Arguments: + s3_client (boto3.client) : s3 client to use for s3 event config + bucket (str): bucket name of the s3 bucket to delete the config in + destination_type (str): String name of the destination type for the configuration + + Returns: + ExistingNotificationConfiguration: A bucket notifiction configuration, + and the notification configurations for a specific destination type + """ + existing_bucket_notification_configuration = ( + s3_client.get_bucket_notification_configuration(Bucket=bucket) + ) + + # Remove ResponseMetadata because we don't want to persist it + existing_bucket_notification_configuration.pop("ResponseMetadata", None) + + existing_notification_configurations_for_type = ( + existing_bucket_notification_configuration.get( + f"{destination_type}Configurations", [] + ) + ) + + # Initialize this with an empty list to have consistent handling if it's present + # or missing. + if not existing_notification_configurations_for_type: + existing_bucket_notification_configuration[ + f"{destination_type}Configurations" + ] = [] + + return ExistingNotificationConfiguration( + existing_bucket_notification_configuration=existing_bucket_notification_configuration, + existing_notification_configurations_for_type=existing_notification_configurations_for_type, + ) + + +class MatchingNotificationConfiguration(typing.NamedTuple): + index_of_matching_arn: typing.Union[int, None] + matching_notification_configuration: typing.Union[dict, None] + + +def get_matching_notification_configuration( + destination_type_arn: str, + existing_notification_configurations_for_type: list, + destination_arn: str, +) -> MatchingNotificationConfiguration: + """ + Search through the list of existing notifications and find the one that has a key of + `destination_type_arn` and a value of `destination_arn`. + + Arguments: + destination_type_arn (str): Key value for the destination type arn + existing_notification_configurations_for_type (list): The existing notification configs + destination_arn (str): Arn of the destination's s3 event config + + Returns: + MatchingNotificationConfiguration: The index of the matching notification + configuration and the matching notification configuration + or None, None if no match is found + """ + for index, existing_notification_configuration_for_type in enumerate( + existing_notification_configurations_for_type + ): + if ( + destination_type_arn in existing_notification_configuration_for_type + and existing_notification_configuration_for_type[destination_type_arn] + == destination_arn + ): + return MatchingNotificationConfiguration( + index_of_matching_arn=index, + matching_notification_configuration=existing_notification_configuration_for_type, + ) + return MatchingNotificationConfiguration(None, None) + + +def create_formatted_message( + bucket: str, destination_type: str, destination_arn: str +) -> str: + """Creates a formatted message for logging purposes. + + Arguments: + bucket (str): bucket name of the s3 bucket + destination_type (str): String name of the destination type for the configuration + destination_arn (str): Arn of the destination's s3 event config + + Returns: + str: A formatted message + """ + return f"Bucket: {bucket}, DestinationType: {destination_type}, DestinationArn: {destination_arn}" + + +def notification_configuration_matches( + matching_notification_configuration: dict, new_notification_configuration: dict +) -> bool: + """Determines if the Events and Filter.key.FilterRules match. + + Arguments: + matching_notification_configuration (dict): The existing notification config + new_notification_configuration (dict): The new notification config + + Returns: + bool: True if the Events and Filter.key.FilterRules match. + """ + # Check if the Events match + if matching_notification_configuration.get( + "Events" + ) != new_notification_configuration.get("Events"): + return False + + # Check if the Filter.key.FilterRules match + matching_filter_rules = ( + matching_notification_configuration.get("Filter", {}) + .get("Key", {}) + .get("FilterRules", []) + ) + new_filter_rules = ( + new_notification_configuration.get("Filter", {}) + .get("Key", {}) + .get("FilterRules", []) + ) + if len(matching_filter_rules) != len(new_filter_rules): + return False + for i in range(len(matching_filter_rules)): + if matching_filter_rules[i] != new_filter_rules[i]: + return False + + # All checks passed, the notification configurations match + return True + + def add_notification( s3_client: boto3.client, - destination_type : str, + destination_type: str, destination_arn: str, bucket: str, bucket_key_prefix: str, -): - """Adds the S3 notification configuration to an existing bucket +) -> None: + """Adds the S3 notification configuration to an existing bucket. + + Use cases: + 1) If a bucket has no `NotificationConfiguration` then create the config + 2) If a bucket has a `NotificationConfiguration` but no matching + "{destination_arn}" for the "{destination_type}" then add the config + 3) If a bucket has a `NotificationConfiguration` and a matching + "{destination_arn}" for the "{destination_type}": + 3a) If the config is the same then do nothing + 3b) If the config is different then overwrite the matching config Args: s3_client (boto3.client) : s3 client to use for s3 event config @@ -56,35 +214,120 @@ def add_notification( bucket (str): bucket name of the s3 bucket to add the config to bucket_key_prefix (str): bucket key prefix for where to look for s3 object notifications """ - s3_client.put_bucket_notification_configuration( - Bucket=bucket, - NotificationConfiguration={ - f"{destination_type}Configurations": [ - { - f"{destination_type}Arn": destination_arn, - "Events": ["s3:ObjectCreated:*"], - "Filter": { - "Key": { - "FilterRules": [ - {"Name": "prefix", "Value": f"{bucket_key_prefix}/"} - ] - } - }, - } - ] + update_required = False + destination_type_arn = f"{destination_type}Arn" + new_notification_configuration = { + destination_type_arn: destination_arn, + "Events": ["s3:ObjectCreated:*"], + "Filter": { + "Key": { + "FilterRules": [{"Name": "prefix", "Value": f"{bucket_key_prefix}/"}] + } }, + } + + ( + existing_bucket_notification_configuration, + existing_notification_configurations_for_type, + ) = get_existing_bucket_notification_configuration_and_type( + s3_client, bucket, destination_type + ) + + ( + index_of_matching_arn, + matching_notification_configuration, + ) = get_matching_notification_configuration( + destination_type_arn, + existing_notification_configurations_for_type, + destination_arn, ) - logger.info("Put request completed....") + + if ( + index_of_matching_arn is not None + and matching_notification_configuration is not None + ): + if not notification_configuration_matches( + matching_notification_configuration, new_notification_configuration + ): + existing_notification_configurations_for_type[ + index_of_matching_arn + ] = new_notification_configuration + update_required = True + else: + existing_notification_configurations_for_type.append( + new_notification_configuration + ) + update_required = True + + if update_required: + existing_bucket_notification_configuration[ + f"{destination_type}Configurations" + ] = existing_notification_configurations_for_type + s3_client.put_bucket_notification_configuration( + Bucket=bucket, + NotificationConfiguration=existing_bucket_notification_configuration, + ) + logger.info( + f"Put request completed to add a NotificationConfiguration for" + + create_formatted_message(bucket, destination_type, destination_arn) + ) + else: + logger.info( + f"Put not required as an existing NotificationConfiguration already exists for" + + create_formatted_message(bucket, destination_type, destination_arn) + ) -def delete_notification(s3_client: boto3.client, bucket: str): - """Deletes the S3 notification configuration from an existing bucket +def delete_notification( + s3_client: boto3.client, bucket: str, destination_type: str, destination_arn: str +) -> None: + """Deletes the S3 notification configuration from an existing bucket for a specific destination type. Args: s3_client (boto3.client) : s3 client to use for s3 event config bucket (str): bucket name of the s3 bucket to delete the config in + destination_type (str): String name of the destination type for the configuration """ - s3_client.put_bucket_notification_configuration( - Bucket=bucket, NotificationConfiguration={} + destination_type_arn = f"{destination_type}Arn" + + ( + existing_bucket_notification_configuration, + existing_notification_configurations_for_type, + ) = get_existing_bucket_notification_configuration_and_type( + s3_client, bucket, destination_type ) - logger.info("Delete request completed....") + + ( + index_of_matching_arn, + _matching_notification_configuration, + ) = get_matching_notification_configuration( + destination_type_arn, + existing_notification_configurations_for_type, + destination_arn, + ) + + if index_of_matching_arn is not None: + del existing_notification_configurations_for_type[index_of_matching_arn] + + if existing_notification_configurations_for_type: + existing_bucket_notification_configuration[ + f"{destination_type}Configurations" + ] = existing_notification_configurations_for_type + else: + del existing_bucket_notification_configuration[ + f"{destination_type}Configurations" + ] + + s3_client.put_bucket_notification_configuration( + Bucket=bucket, + NotificationConfiguration=existing_bucket_notification_configuration, + ) + logger.info( + f"Delete request completed to remove a NotificationConfiguration for" + + create_formatted_message(bucket, destination_type, destination_arn) + ) + else: + logger.info( + f"Delete not required as no NotificationConfiguration exists for" + + create_formatted_message(bucket, destination_type, destination_arn) + ) diff --git a/tests/test_s3_event_config_lambda.py b/tests/test_s3_event_config_lambda.py index cbe1032e..2f8d7c8f 100644 --- a/tests/test_s3_event_config_lambda.py +++ b/tests/test_s3_event_config_lambda.py @@ -1,5 +1,4 @@ -import zipfile -import io +from unittest import mock import boto3 from moto import mock_s3, mock_lambda, mock_iam, mock_sqs import pytest @@ -43,15 +42,20 @@ def mock_sqs_queue(mock_aws_credentials): @mock_s3 -def test_that_add_notification_adds_expected_settings_for_lambda(s3, mock_lambda_function): +def test_add_notification_adds_expected_settings_for_lambda(s3, mock_lambda_function): s3.create_bucket(Bucket="some_bucket") - set_config = app.add_notification( + with mock.patch.object( s3, - "LambdaFunction", - mock_lambda_function["Configuration"]["FunctionArn"], - "some_bucket", - "test_folder", - ) + "get_bucket_notification_configuration", + return_value={}, + ): + app.add_notification( + s3, + "LambdaFunction", + mock_lambda_function["Configuration"]["FunctionArn"], + "some_bucket", + "test_folder", + ) get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") assert ( get_config["LambdaFunctionConfigurations"][0]["LambdaFunctionArn"] @@ -66,53 +70,416 @@ def test_that_add_notification_adds_expected_settings_for_lambda(s3, mock_lambda @mock_s3 -def test_that_delete_notification_is_successful_for_lambda(s3, mock_lambda_function): +def test_add_notification_adds_expected_settings_for_sqs(s3, mock_sqs_queue): s3.create_bucket(Bucket="some_bucket") - app.add_notification( + with mock.patch.object( s3, - "LambdaFunction", - mock_lambda_function["Configuration"]["FunctionArn"], - "some_bucket", - "test_folder", - ) - app.delete_notification(s3, "some_bucket") + "get_bucket_notification_configuration", + return_value={}, + ): + app.add_notification( + s3, + "Queue", + mock_sqs_queue["Attributes"]["QueueArn"], + "some_bucket", + "test_folder", + ) get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") - assert "LambdaFunctionConfigurations" not in get_config + assert ( + get_config["QueueConfigurations"][0]["QueueArn"] + == mock_sqs_queue["Attributes"]["QueueArn"] + ) + assert get_config["QueueConfigurations"][0]["Events"] == ["s3:ObjectCreated:*"] + assert get_config["QueueConfigurations"][0]["Filter"] == { + "Key": {"FilterRules": [{"Name": "prefix", "Value": "test_folder/"}]} + } @mock_s3 -def test_that_add_notification_adds_expected_settings_for_sqs(s3, mock_sqs_queue): +def test_add_notification_replace_if_filter_rule_different(s3, mock_sqs_queue): + # GIVEN an S3 bucket s3.create_bucket(Bucket="some_bucket") - set_config = app.add_notification( + + # AND the bucket has an existing `QueueConfigurations` that is different to what we want + # but matches the ARN. + with mock.patch.object( s3, - "Queue", - mock_sqs_queue['Attributes']['QueueArn'], - "some_bucket", - "test_folder", + "get_bucket_notification_configuration", + return_value={ + f"QueueConfigurations": [ + { + "QueueArn": mock_sqs_queue["Attributes"]["QueueArn"], + "Events": ["s3:ObjectCreated:*"], + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"some_other_folder/"} + ] + } + }, + } + ] + }, + ): + app.add_notification( + s3, + "Queue", + mock_sqs_queue["Attributes"]["QueueArn"], + "some_bucket", + "test_folder", + ) + get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") + assert ( + get_config["QueueConfigurations"][0]["QueueArn"] + == mock_sqs_queue["Attributes"]["QueueArn"] ) + assert get_config["QueueConfigurations"][0]["Events"] == ["s3:ObjectCreated:*"] + assert get_config["QueueConfigurations"][0]["Filter"] == { + "Key": {"FilterRules": [{"Name": "prefix", "Value": "test_folder/"}]} + } + +@mock_s3 +def test_add_notification_replace_if_events_different(s3, mock_sqs_queue): + # GIVEN an S3 bucket + s3.create_bucket(Bucket="some_bucket") + + # AND the bucket has an existing `QueueConfigurations` that is different to what we want + # but matches the ARN. + with mock.patch.object( + s3, + "get_bucket_notification_configuration", + return_value={ + f"QueueConfigurations": [ + { + "QueueArn": mock_sqs_queue["Attributes"]["QueueArn"], + "Events": ["s3:ASDF:*"], + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"test_folder/"} + ] + } + }, + } + ] + }, + ): + app.add_notification( + s3, + "Queue", + mock_sqs_queue["Attributes"]["QueueArn"], + "some_bucket", + "test_folder", + ) get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") assert ( get_config["QueueConfigurations"][0]["QueueArn"] - == mock_sqs_queue['Attributes']['QueueArn'] + == mock_sqs_queue["Attributes"]["QueueArn"] + ) + assert get_config["QueueConfigurations"][0]["Events"] == ["s3:ObjectCreated:*"] + assert get_config["QueueConfigurations"][0]["Filter"] == { + "Key": {"FilterRules": [{"Name": "prefix", "Value": "test_folder/"}]} + } + +@mock_s3 +def test_add_notification_does_nothing_if_notification_already_exists( + s3, mock_lambda_function +): + # GIVEN an S3 bucket + s3.create_bucket(Bucket="some_bucket") + + # AND the bucket has an existing `LambdaFunctionConfigurations` that matches the one we will add + with mock.patch.object( + s3, + "get_bucket_notification_configuration", + return_value={ + f"LambdaFunctionConfigurations": [ + { + f"LambdaFunctionArn": mock_lambda_function["Configuration"][ + "FunctionArn" + ], + "Events": ["s3:ObjectCreated:*"], + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"test_folder/"} + ] + } + }, + } + ] + }, + ), mock.patch.object(s3, "put_bucket_notification_configuration") as put_config: + # WHEN I add the existing matching `LambdaFunction` configuration + app.add_notification( + s3, + "LambdaFunction", + mock_lambda_function["Configuration"]["FunctionArn"], + "some_bucket", + "test_folder", + ) + + # AND I get the notification configuration + get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") + + # THEN I expect nothing to have been saved in our mocked environment + assert not put_config.called + + +@mock_s3 +def test_add_notification_does_nothing_if_notification_already_exists_even_in_different_dict_order( + s3, mock_lambda_function +): + # GIVEN an S3 bucket + s3.create_bucket(Bucket="some_bucket") + + # AND the bucket has an existing `LambdaFunctionConfigurations` that matches + # content of the one we will add - But differs in order + with mock.patch.object( + s3, + "get_bucket_notification_configuration", + return_value={ + f"LambdaFunctionConfigurations": [ + { + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"test_folder/"} + ] + } + }, + "Events": ["s3:ObjectCreated:*"], + f"LambdaFunctionArn": mock_lambda_function["Configuration"][ + "FunctionArn" + ], + } + ] + }, + ), mock.patch.object(s3, "put_bucket_notification_configuration") as put_config: + # WHEN I add the existing matching `LambdaFunction` configuration + app.add_notification( + s3, + "LambdaFunction", + mock_lambda_function["Configuration"]["FunctionArn"], + "some_bucket", + "test_folder", + ) + + # AND I get the notification configuration + get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") + + # THEN I expect nothing to have been saved in our mocked environment + assert not put_config.called + + +@mock_s3 +def test_add_notification_adds_config_if_requested_notification_does_not_exist( + s3, mock_lambda_function, mock_sqs_queue +): + # GIVEN an S3 bucket + s3.create_bucket(Bucket="some_bucket") + + # AND the bucket has an existing `QueueConfigurations` + with mock.patch.object( + s3, + "get_bucket_notification_configuration", + return_value={ + f"QueueConfigurations": [ + { + "Id": "123", + "QueueArn": mock_sqs_queue["Attributes"]["QueueArn"], + "Events": ["s3:ObjectCreated:*"], + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"test_folder/"} + ] + } + }, + } + ] + }, + ): + # WHEN I add a new `LambdaFunction` configuration + app.add_notification( + s3, + "LambdaFunction", + mock_lambda_function["Configuration"]["FunctionArn"], + "some_bucket", + "test_folder", + ) + + # AND I get the notification configuration + get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") + + # THEN I expect to see a new `LambdaFunction` configuration + assert ( + get_config["LambdaFunctionConfigurations"][0]["LambdaFunctionArn"] + == mock_lambda_function["Configuration"]["FunctionArn"] ) - assert get_config["QueueConfigurations"][0]["Events"] == [ + assert get_config["LambdaFunctionConfigurations"][0]["Events"] == [ "s3:ObjectCreated:*" ] + assert get_config["LambdaFunctionConfigurations"][0]["Filter"] == { + "Key": {"FilterRules": [{"Name": "prefix", "Value": "test_folder/"}]} + } + + # AND I expect the `QueueConfigurations` to be unchanged + assert ( + get_config["QueueConfigurations"][0]["QueueArn"] + == mock_sqs_queue["Attributes"]["QueueArn"] + ) + assert get_config["QueueConfigurations"][0]["Events"] == ["s3:ObjectCreated:*"] assert get_config["QueueConfigurations"][0]["Filter"] == { "Key": {"FilterRules": [{"Name": "prefix", "Value": "test_folder/"}]} } @mock_s3 -def test_that_delete_notification_is_successful_for_sqs(s3, mock_sqs_queue): +def test_add_notification_adds_config_if_existing_config_does_not_match( + s3, mock_lambda_function +): + # GIVEN an S3 bucket s3.create_bucket(Bucket="some_bucket") - app.add_notification( + + # AND the bucket has an existing `LambdaFunctionConfigurations` that does not match the one we are adding + with mock.patch.object( s3, - "Queue", - mock_sqs_queue['Attributes']['QueueArn'], - "some_bucket", - "test_folder", + "get_bucket_notification_configuration", + return_value={ + f"LambdaFunctionConfigurations": [ + { + f"LambdaFunctionArn": mock_lambda_function["Configuration"][ + "FunctionArn" + ] + + "asdf", + "Events": ["s3:ObjectCreated:*"], + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"test_folder/"} + ] + } + }, + } + ] + }, + ): + # WHEN I add the `LambdaFunction` configuration + app.add_notification( + s3, + "LambdaFunction", + mock_lambda_function["Configuration"]["FunctionArn"], + "some_bucket", + "test_folder", + ) + + # AND I get the notification configuration + get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") + + # THEN I expect to see the updated `LambdaFunction` configuration for the existing one + assert ( + get_config["LambdaFunctionConfigurations"][0]["LambdaFunctionArn"] + == mock_lambda_function["Configuration"]["FunctionArn"] + "asdf" ) - app.delete_notification(s3, "some_bucket") + assert get_config["LambdaFunctionConfigurations"][0]["Events"] == [ + "s3:ObjectCreated:*" + ] + + assert get_config["LambdaFunctionConfigurations"][0]["Filter"] == { + "Key": {"FilterRules": [{"Name": "prefix", "Value": "test_folder/"}]} + } + + # AND I expect to see the updated `LambdaFunction` configuration for the new one + assert ( + get_config["LambdaFunctionConfigurations"][1]["LambdaFunctionArn"] + == mock_lambda_function["Configuration"]["FunctionArn"] + ) + assert get_config["LambdaFunctionConfigurations"][1]["Events"] == [ + "s3:ObjectCreated:*" + ] + assert get_config["LambdaFunctionConfigurations"][1]["Filter"] == { + "Key": {"FilterRules": [{"Name": "prefix", "Value": "test_folder/"}]} + } + + +@mock_s3 +def test_delete_notification_is_successful_for_configuration_that_exists( + s3, mock_lambda_function +): + # GIVEN an S3 bucket + s3.create_bucket(Bucket="some_bucket") + + # AND a configuration exists + with mock.patch.object( + s3, + "get_bucket_notification_configuration", + return_value={ + f"LambdaFunctionConfigurations": [ + { + f"LambdaFunctionArn": mock_lambda_function["Configuration"][ + "FunctionArn" + ], + "Events": ["s3:ObjectCreated:*"], + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"test_folder/"} + ] + } + }, + } + ] + }, + ): + # WHEN I delete the notification + app.delete_notification( + s3_client=s3, + bucket="some_bucket", + destination_type="LambdaFunction", + destination_arn=mock_lambda_function["Configuration"]["FunctionArn"], + ) + # THEN the notification should be deleted get_config = s3.get_bucket_notification_configuration(Bucket="some_bucket") - assert "QueueConfigurations" not in get_config + assert "LambdaFunctionConfigurations" not in get_config + + +@mock_s3 +def test_delete_notification_does_nothing_when_deleting_configuration_that_does_not_exist( + s3, mock_lambda_function +): + # GIVEN an S3 bucket + s3.create_bucket(Bucket="some_bucket") + + # AND a configuration exists for a different notification type + with mock.patch.object( + s3, + "get_bucket_notification_configuration", + return_value={ + f"LambdaFunctionConfigurations": [ + { + f"LambdaFunctionArn": mock_lambda_function["Configuration"][ + "FunctionArn" + ], + "Events": ["s3:ObjectCreated:*"], + "Filter": { + "Key": { + "FilterRules": [ + {"Name": "prefix", "Value": f"test_folder/"} + ] + } + }, + } + ] + }, + # AND a mock for the put_bucket_notification_configuration method + ), mock.patch.object(s3, "put_bucket_notification_configuration") as put_config: + # WHEN I delete a notification that does not exist + app.delete_notification( + s3_client=s3, + bucket="some_bucket", + destination_type="doesNotExist", + destination_arn=mock_lambda_function["Configuration"]["FunctionArn"], + ) + # THEN nothing should have been called + assert not put_config.called