How to use remove_bucket_notification method in localstack

Best Python code snippet using localstack_python

main.py

Source:main.py

...160    if not args.delete:161        bucket_account = get_account_id(s3_notifications_sess)162        add_lambda_permission(attacker_sess, bucket_account, bucket, LAMBDA_NAME)163    if args.delete:164        remove_bucket_notification(s3_notifications_sess, bucket)165    else:166        lambda_account = get_account_id(attacker_sess)167        lambda_arn = f"arn:aws:lambda:{attacker_sess.region_name}:{lambda_account}:function:{LAMBDA_NAME}"168        put_bucket_notification(s3_notifications_sess, bucket, lambda_arn)169    if args.delete:170        msg = "Successfully deleted deployment."171    else:172        msg = f"""173Deployment successful.174After a modified CloudFormation template is successfully deployed in the target account there will be a new role175created with admin privileges that can be assumed from the '{principal}' principal.176This role name is randomly chosen by CloudFormation but will have 'MaintenanceRole' in the name. It is possible to177to explicitly set this name if needed however this is not supported at the moment.178"""179    return msg180def lambda_env(pacu: 'Main', bucket: str, key: 'AWSKey'):181    env = {182        "AWS_ACCESS_KEY_ID": key.access_key_id,183        "AWS_SECRET_ACCESS_KEY": key.secret_access_key,184        "PATH": os.environ["PATH"],185    }186    if key.session_token:187        env["AWS_SESSION_TOKEN"] = key.session_token188    env["AWS_DEFAULT_REGION"] = get_region(bucket, pacu.get_regions('lambda'))189    return env190def deploy_lambda(pacu: 'Main', env: dict, deploy_dir: Path, bucket: str, principal: str, s3_key: 'AWSKey'):191    print = pacu.print192    print(f"Will deploy lambda to {env['AWS_DEFAULT_REGION']}")193    if not deploy_dir.exists():194        shutil.copytree((Path(__file__).parent / 'cfn__resource_injection_lambda'), deploy_dir, dirs_exist_ok=True)195    config_path = deploy_dir / '.chalice' / 'config.json'196    config = json.loads(config_path.read_text())197    config['stages']['dev']['environment_variables']['PRINCIPAL'] = principal198    config['stages']['dev']['environment_variables']['BUCKET'] = bucket199    config['stages']['dev']['environment_variables']['S3_AWS_ACCESS_KEY_ID'] = s3_key.access_key_id200    config['stages']['dev']['environment_variables']['S3_AWS_SECRET_ACCESS_KEY'] = s3_key.secret_access_key201    if s3_key.session_token:202        config['stages']['dev']['environment_variables']['S3_AWS_SESSION_TOKEN'] = s3_key.session_token203    config_path.write_text(json.dumps(config))204    cmd = ['chalice', '--project-dir', str(deploy_dir), 'deploy']205    print(f"Running command: {' '.join(cmd)}")206    try:207        subprocess.check_call(cmd, env=env)208    except subprocess.CalledProcessError as e:209        # The deploy will fail when attempting to add the trigger to the cross account bucket because we're using the210        # keys for the attacker account. This is ok as long as the resource permissions get added to the deployed211        # lambda.212        if e.returncode == 2:213            pass214        else:215            raise e216    (deploy_dir / '.deployed').touch(exist_ok=True)217def delete_lambda(bucket_lambda_dir, env):218    if not bucket_lambda_dir.exists():219        raise UserWarning(f"The directory {str(bucket_lambda_dir)} does not exist.")220    subprocess.check_call(['chalice', '--project-dir', str(bucket_lambda_dir), 'delete'], env=env)221    shutil.rmtree(bucket_lambda_dir)222def get_session_from_key_name(pacu_main: 'Main', key_name: str, region: str = 'us-east-1'):223    key: 'AWSKey' = pacu_main.get_aws_key_by_alias(key_name)224    if not key:225        raise PacuException(f"Did not find the key {key_name} in pacu, make sure to set this with `set_keys` first.")226    return boto3.Session(227        region_name=region,228        aws_access_key_id=key.access_key_id,229        aws_secret_access_key=key.secret_access_key,230        aws_session_token=key.session_token,231    )232def put_bucket_notification(sess: 'boto3.Session', bucket: str, lambda_arn: str):233    s3 = sess.client('s3')234    resp = s3.get_bucket_notification_configuration(Bucket=bucket)235    conf = remove_our_notification(resp)236    conf.setdefault('LambdaFunctionConfigurations', []).append({237        'Id': 'cfn_notifications',238        'LambdaFunctionArn': lambda_arn,239        'Events': [240            's3:ObjectCreated:*',241        ],242    })243    try:244        s3.put_bucket_notification_configuration(245            Bucket=bucket,246            NotificationConfiguration=conf247        )248    except ClientError as e:249        if 'Cannot have overlapping suffixes' in e.response['Error']['Code']:250            raise PacuException(251                "\n\n*****\n"252                "There appears to already be a event configuration set up on this bucket with the same event type."253                "\n*****\n\n"254                "S3 only allows configuring a single notification configuration for each event type. It's possible "255                "that this was left around by a previous run of pacu, if this is the case you can try removing it and "256                "re-running this module.\n"257            )258        else:259            raise e260def remove_bucket_notification(sess: 'boto3.Session', bucket: str):261    s3 = sess.client('s3')262    resp = s3.get_bucket_notification_configuration(Bucket=bucket)263    resp = remove_our_notification(resp)264    s3.put_bucket_notification_configuration(265        Bucket=bucket,266        NotificationConfiguration=resp,267    )268def remove_our_notification(resp: 'NotificationConfigurationResponseMetadataTypeDef') -> \269        'NotificationConfigurationTypeDef':270    del resp['ResponseMetadata']271    resp = cast('NotificationConfigurationTypeDef', resp)272    for conf in resp.get('LambdaFunctionConfigurations', []):273        if conf['Id'] == 'cfn_notifications':274            resp['LambdaFunctionConfigurations'].remove(conf)...

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.