Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use normalize_bucket_name method in localstack

Best Python code snippet using localstack_python

s3.py

Source:s3.py

...38    @staticmethod39    def cloudformation_type():40        return "AWS::S3::Bucket"41    def get_resource_name(self):42        return self.normalize_bucket_name(self.props.get("BucketName"))43    @staticmethod44    def normalize_bucket_name(bucket_name):45        return s3_utils.normalize_bucket_name(bucket_name)46    @staticmethod47    def add_defaults(resource, stack_name: str):48        role_name = resource.get("Properties", {}).get("BucketName")49        if not role_name:50            resource["Properties"]["BucketName"] = s3_listener.normalize_bucket_name(51                generate_default_name(stack_name, resource["LogicalResourceId"])52            )53    @classmethod54    def get_deploy_templates(cls):55        def convert_acl_cf_to_s3(acl):56            """Convert a CloudFormation ACL string (e.g., 'PublicRead') to an S3 ACL string (e.g., 'public-read')"""57            return re.sub("(?<!^)(?=[A-Z])", "-", acl).lower()58        def s3_bucket_notification_config(params, **kwargs):59            notif_config = params.get("NotificationConfiguration")60            if not notif_config:61                return None62            lambda_configs = []63            queue_configs = []64            topic_configs = []65            attr_tuples = (66                (67                    "LambdaConfigurations",68                    lambda_configs,69                    "LambdaFunctionArn",70                    "Function",71                ),72                ("QueueConfigurations", queue_configs, "QueueArn", "Queue"),73                ("TopicConfigurations", topic_configs, "TopicArn", "Topic"),74            )75            # prepare lambda/queue/topic notification configs76            for attrs in attr_tuples:77                for notif_cfg in notif_config.get(attrs[0]) or []:78                    filter_rules = notif_cfg.get("Filter", {}).get("S3Key", {}).get("Rules")79                    entry = {80                        attrs[2]: notif_cfg[attrs[3]],81                        "Events": [notif_cfg["Event"]],82                    }83                    if filter_rules:84                        entry["Filter"] = {"Key": {"FilterRules": filter_rules}}85                    attrs[1].append(entry)86            # construct final result87            result = {88                "Bucket": params.get("BucketName") or PLACEHOLDER_RESOURCE_NAME,89                "NotificationConfiguration": {90                    "LambdaFunctionConfigurations": lambda_configs,91                    "QueueConfigurations": queue_configs,92                    "TopicConfigurations": topic_configs,93                },94            }95            return result96        def get_bucket_location_config(**kwargs):97            region = aws_stack.get_region()98            if region == AWS_REGION_US_EAST_1:99                return None100            return {"LocationConstraint": region}101        def _pre_delete(resource_id, resources, resource_type, func, stack_name):102            s3 = aws_stack.connect_to_service("s3")103            resource = resources[resource_id]104            props = resource["Properties"]105            bucket_name = props.get("BucketName")106            try:107                s3.delete_bucket_policy(Bucket=bucket_name)108            except Exception:109                pass110            s3_listener.remove_bucket_notification(resource["PhysicalResourceId"])111            # TODO: divergence from how AWS deals with bucket deletes (should throw an error)112            try:113                delete_all_s3_objects(bucket_name)114            except Exception as e:115                if "NoSuchBucket" not in str(e):116                    raise117        def _add_bucket_tags(resource_id, resources, resource_type, func, stack_name):118            s3 = aws_stack.connect_to_service("s3")119            resource = resources[resource_id]120            props = resource["Properties"]121            bucket_name = props.get("BucketName")122            tags = props.get("Tags", [])123            if len(tags) > 0:124                s3.put_bucket_tagging(Bucket=bucket_name, Tagging={"TagSet": tags})125        result = {126            "create": [127                {128                    "function": "create_bucket",129                    "parameters": {130                        "Bucket": ["BucketName", PLACEHOLDER_RESOURCE_NAME],131                        "ACL": lambda params, **kwargs: convert_acl_cf_to_s3(132                            params.get("AccessControl", "PublicRead")133                        ),134                        "CreateBucketConfiguration": lambda params, **kwargs: get_bucket_location_config(),135                    },136                },137                {138                    "function": "put_bucket_notification_configuration",139                    "parameters": s3_bucket_notification_config,140                },141                {"function": _add_bucket_tags},142            ],143            "delete": [144                {"function": _pre_delete},145                {"function": "delete_bucket", "parameters": {"Bucket": "BucketName"}},146            ],147        }148        return result149    def fetch_state(self, stack_name, resources):150        props = self.props151        bucket_name = self._get_bucket_name()152        bucket_name = self.resolve_refs_recursively(stack_name, bucket_name, resources)153        bucket_name = self.normalize_bucket_name(bucket_name)154        s3_client = aws_stack.connect_to_service("s3")155        response = s3_client.get_bucket_location(Bucket=bucket_name)156        notifs = props.get("NotificationConfiguration")157        if not response or not notifs:158            return response159        configs = s3_client.get_bucket_notification_configuration(Bucket=bucket_name)160        has_notifs = (161            configs.get("TopicConfigurations")162            or configs.get("QueueConfigurations")163            or configs.get("LambdaFunctionConfigurations")164        )165        if notifs and not has_notifs:166            return None167        return response...

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.