How to use get_sqs_policy method in localstack

Best Python code snippet using localstack_python

create_sqs.py

Source:create_sqs.py

...48        queue_name = f"{environment_prefix}-sqs-s3-{service_name}"49    else:50        queue_name = f"{environment_prefix}-sqs-s3-{service_name}_{branch_name}"51    return queue_name52def get_sqs_policy(queue_name):53    return """{54        "Version": "2012-10-17",55        "Statement": [ {56            "Effect": "Allow",57            "Principal": "*",58            "Action": "sqs:SendMessage",59            "Resource": "%s%s",60            "Condition":61                {62                "ArnEquals":63                    {64                    "aws:SourceArn":65                        "arn:aws:s3:::%s"66                    }67                }68            } ]69    }""" % (SQS_ARN_PREFIX, queue_name, S3_BUCKET_NAME)70def create_janus_queue(environment, service_name, branch_name):71    environment_prefix = environment_map[environment]["alias"]72    queue_name = get_resource_name(environment_prefix, service_name,73                                   branch_name)74    response = {}75    tags = {76        "Environment": environment,77        "Group": service_name,78        "Branch": branch_name,79        "Role": "janus-queue",80        "Service": service_name,81        # "Tribe": get_service_tribe_mapping(service_name),82    }83    queue_policy = get_sqs_policy(queue_name)84    logger.info(f"Creating queue {queue_name}, tags={tags}.")85    # Create queue86    queue = sqs_resources.create_queue(QueueName=queue_name,87                                       Attributes={"Policy": queue_policy},88                                       tags=tags)89    sqs_url = queue.url90    sqs_arn = queue.attributes.get("QueueArn")91    response["sqs_url"] = sqs_url92    response["sqs_arn"] = sqs_arn93    response["status"] = "Created"94    return response95def create_folder_in_s3(environment, service_name, branch_name):96    environment_prefix = environment_map[environment]["alias"]97    folder_name = get_resource_name(environment_prefix, service_name,...

awsmanager.py

Source:awsmanager.py

...27    def create_user_key_pair(self, username=None):28        return self.aws_manager.IAM.create_user_key_pair(username=username)29    def get_s3_policy(self, bucket=None, folder=None):30        return self.aws_manager.S3.get_s3_policy(bucket=bucket, folder=folder)31    def get_sqs_policy(self, queue_name=None, full_access=None):32        return self.aws_manager.SQS.get_sqs_policy(queue_name=queue_name, full_access=full_access)33    def allow_sqs_queue_for_user(self, username=None, queue_name=None, full_access=None):34        policy = self.get_sqs_policy(queue_name=queue_name, full_access=full_access)35        self.put_user_policy(username=username, policyname='sqs_policy_%s_%s'%(username, queue_name),36            policydocument=policy)37    def create_s3_for_user(self, username=None, bucket=None, region=None):38        return self.aws_manager.S3.create_s3_for_user(username=username,39            bucket=bucket, region=region)40    def allow_s3_folder_for_user(self, username=None, bucket=None, folder=None):41        return self.aws_manager.S3.allow_s3_folder_for_user(username=username,42            bucket=bucket, folder=folder)43    def get_hosted_zones(self):44        return self.aws_manager.Route53.get_hosted_zones()45class ServiceParameters(BaseComponent):46    def service_parameters(self,pane,datapath=None,**kwargs):47        fb = pane.formbuilder(datapath=datapath)48        fb.textbox(value='^.aws_access_key_id',lbl='Aws Access Key Id')...

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.