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

How to use create_cluster_security_group method in localstack

Best Python code snippet using localstack_python

create_cluster.py

Source:create_cluster.py

...43        if cluster_info['ClusterStatus'] == 'available':44            break45        time.sleep(60)46    return cluster_info47def create_cluster_security_group():48    response = ec2_client.describe_vpcs()49    vpc_id = response.get('Vpcs', [{}])[0].get('VpcId', '')50    try:51        response = ec2_client.create_security_group(GroupName='myredshiftsg', Description='Redshift security group',52                                                    VpcId=vpc_id)53        security_group_id = response['GroupId']54        print('Security Group Created %s in vpc %s.' % (security_group_id, vpc_id))55        data = ec2_client.authorize_security_group_ingress(56            GroupId=security_group_id,57            IpPermissions=[58                {'IpProtocol': 'tcp',59                 'FromPort': 80,60                 'ToPort': 80,61                 'IpRanges': [{'CidrIp': '0.0.0.0/0'}]},62                {'IpProtocol': 'tcp',63                 'FromPort': 5439,64                 'ToPort': 5439,65                 'IpRanges': [{'CidrIp': '0.0.0.0/0'}]}66            ])67        return security_group_id68    except ClientError as e:69        print(e)70def create_iam_role(config):71    role = iam_client.create_role(72        RoleName=config.get('SECURITY', 'ROLE_NAME'),73        Description='Allows Redshift to call AWS services on your behalf',74        AssumeRolePolicyDocument=json.dumps({75            'Version': '2012-10-17',76            'Statement': [{77                'Action': 'sts:AssumeRole',78                'Effect': 'Allow',79                'Principal': {'Service': 'redshift.amazonaws.com'}80            }]81        })82    )83    iam_client.attach_role_policy(84        RoleName=config.get('SECURITY', 'ROLE_NAME'),85        PolicyArn='arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess'86    )87    return role88def main():89    """Initiate and wait for redshift cluster deletion"""90    config = configparser.ConfigParser()91    config.read('../dwh.cfg')92    cluster_sg_id = create_cluster_security_group()93    iam_role = create_iam_role(config)94    cluster_info = create_redshift_cluster(config, iam_role['Role']['Arn'], cluster_sg_id)95    if cluster_info is not None:96        print(f'Creating cluster: {cluster_info["ClusterIdentifier"]}')97        print(f'Cluster status: {cluster_info["ClusterStatus"]}')98        print(f'Database name: {cluster_info["DBName"]}')99        print('Waiting for cluster to be created...')100        cluster_info = wait_for_cluster_creation(cluster_info['ClusterIdentifier'])101        print(f'Cluster created.')102        print(f"Endpoint={cluster_info['Endpoint']['Address']}")103        print(f"Role_ARN={iam_role['Role']['Arn']}")104        print(f"Security_Group={cluster_sg_id}")105if __name__ == '__main__':106    main()

test_redshift.py

Source:test_redshift.py

...25    def test_cluster_security_groups(self, redshift_client, snapshot):26        # Note: AWS parity testing not easily possible with our account, due to error message27        #  "VPC-by-Default customers cannot use cluster security groups"28        group_name = f"g-{short_uid()}"29        redshift_client.create_cluster_security_group(30            ClusterSecurityGroupName=group_name, Description="test 123"31        )32        cidr_ip = "192.168.100.101/32"33        redshift_client.authorize_cluster_security_group_ingress(34            ClusterSecurityGroupName=group_name, CIDRIP=cidr_ip35        )36        result = redshift_client.describe_cluster_security_groups(37            ClusterSecurityGroupName=group_name38        )39        groups = result.get("ClusterSecurityGroups", [])40        assert len(groups) == 141        assert groups[0].get("IPRanges")42        assert groups[0]["IPRanges"][0]["Status"] == "authorized"43        assert groups[0]["IPRanges"][0]["CIDRIP"] == cidr_ip

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.