KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use waitForCreation method in pyatom

Best Python code snippet using pyatom_python

RedshiftUtility.py

Source:RedshiftUtility.py

1#!/usr/bin/python2import boto33from botocore.client import ClientError4import yaml5"""6RedshiftUtility.py - 03.14.16 Kate Drogaieva7The class in this module allows to create a Redshift claster standalone or in VPC8"""9#---------------------------------------------------------------------------------------------10class RedshiftUtility(object):11    """12    RedshiftUtility class can be used to create a Redshift cluster standalone or in VPC13    The configuration parameters must be provided in a YAML resource file14    RedshiftCluster:15      Endpoint:16      DBName: "supportdw"17      ClusterIdentifier: "Support"18      ClusterType: "single-node"19      NodeType: "dc1.large"20      MasterUsername: "masteruser"21      MasterUserPassword: "Kaktus2015"22      Port: "5439"23      PubliclyAccessible: "False"24      Encrypted: "False"25      26      Optional section to create a cluster in a VPC27      28      SubnetGroup:29        Name: "testvpcprivate"30        Description: "Private Subnets in Test VPC"31        VPC: "Test"32        SecurityGroup: "RedshiftAll"33    """34    def __str__(self):35        return self.endpoint+":"+str(self.port)36#---------------------------------------------------------------------------------------------37    def __init__(self,resource):38        """39            Init Redshift cluster parameters using resource file in YAML format40        """41        try:42            with open(resource, "r") as f:43                self.res = yaml.load(f)44            DBName=self.res["RedshiftCluster"]["DBName"],45            self.ClusterIdentifier=self.res["RedshiftCluster"]["ClusterIdentifier"]46        except KeyError or IOError:47            sys.exit("Wrong Cluster Parameters")48        self.client=boto3.client("redshift",self.res["Region"])49        self.endpoint=""50        self.port=""51        self.GetEndpoint()52        return53#---------------------------------------------------------------------------------------------54    def CreateClusterSubnetGroup(self,SubnetIds):55        """56            Creates a cluster Subnet Group if ["RedshiftCluster"]["SubnetGroup"]  exists57            in YAML configuration file58            SubnetIds is a list of public or private subnets from a VPC59            return Subgroup name or default if VPC does not exists60        """61        SubnetGroup="default"62        response = self.client.create_cluster_subnet_group(63                   ClusterSubnetGroupName=self.res["RedshiftCluster"]["SubnetGroup"]["Name"],64                   Description=self.res["RedshiftCluster"]["SubnetGroup"]["Description"],65                   SubnetIds=SubnetIds)66        SubnetGroup=response["ClusterSubnetGroup"]["ClusterSubnetGroupName"]67        return SubnetGroup68#---------------------------------------------------------------------------------------------69#---------------------------------------------------------------------------------------------70    def CreateCluster(self,SubnetIds=[],SecurityGroupId=[]):71        """72        Create a new cluster acoording to the configuration parameters in YAML resource file73        If there is ["RedshiftCluster"]["SubnetGroup"] in the resource file 74        a cluster Subnet Group is created first based on SubnetIds (list) from a VPC75        SecurityGroupId (list) can be used to create the cluster in VPC76        """77        SubnetGroup="default"78        try:79            if self.res["RedshiftCluster"]["SubnetGroup"]:80                try: #test if a group already exists81                    response = self.client.describe_cluster_subnet_groups(ClusterSubnetGroupName=self.res["RedshiftCluster"]["SubnetGroup"]["Name"])82                    SubnetGroup=self.res["RedshiftCluster"]["SubnetGroup"]["Name"]83                except ClientError: #group does not exist - create84                    SubnetGroup=self.CreateClusterSubnetGroup(SubnetIds)85        except KeyError:86            pass87        PubliclyAccessible=True88        if self.res["RedshiftCluster"]["PubliclyAccessible"]=="False":89            PubliclyAccessible=False90        Encrypted=True91        if self.res["RedshiftCluster"]["Encrypted"]=="False":92            Encrypted=False93        response = self.client.create_cluster(94                DBName=self.res["RedshiftCluster"]["DBName"],95                ClusterIdentifier=self.ClusterIdentifier,96                ClusterType=self.res["RedshiftCluster"]["ClusterType"],97                NodeType=self.res["RedshiftCluster"]["NodeType"],98                MasterUsername=self.res["RedshiftCluster"]["MasterUsername"],99                MasterUserPassword=self.res["RedshiftCluster"]["MasterUserPassword"],100                Port=int(self.res["RedshiftCluster"]["Port"]),101                PubliclyAccessible=PubliclyAccessible,102                Encrypted=Encrypted,103                ClusterSubnetGroupName=SubnetGroup,104                VpcSecurityGroupIds=SecurityGroupId105                )106        return107#---------------------------------------------------------------------------------------------108    def DeleteCluster(self):109        """110            The function deletes the cluster and do not create a final cluster snapshot111        """112        response = self.client.delete_cluster(113            ClusterIdentifier=self.ClusterIdentifier,114            SkipFinalClusterSnapshot=True115            )116        return117#---------------------------------------------------------------------------------------------118    def DeleteClusterSubgroup(self):119        """120            The function deletes the cluster Subnet Group if it exists121        """122        try: #to delete a group if  exists123                response = self.client.delete_cluster_subnet_group(ClusterSubnetGroupName=self.res["RedshiftCluster"]["SubnetGroup"]["Name"])124        except (ClientError,KeyError) as e: #group does not exist or not configured in the resource file - pass125            pass126        return127#---------------------------------------------------------------------------------------------128    def GetEndpoint(self):129        """130            The function set endpoint and port variable and returns 0 if no errors131            or 1 if the cluster is not available132        """133        try:134            response = self.client.describe_clusters(135            ClusterIdentifier=self.ClusterIdentifier136            )137            try:138                self.endpoint=response["Clusters"][0]["Endpoint"]["Address"]139                self.port=response["Clusters"][0]["Endpoint"]["Port"]140                return 0141            except KeyError:142                return 1143        except ClientError: #Cluster does not exists144            return 1145#---------------------------------------------------------------------------------------------146    def CheckStatus(self):147        """148            The same as GetEndpoint149        """150        return self.GetEndpoint()151#---------------------------------------------------------------------------------------------152    def WaitForCreation(self):153        """154            The function waits while the cluster is creating It can take significant amount of time155        """156        waiter = self.client.get_waiter("cluster_available")157        try:158            waiter.wait(ClusterIdentifier=self.ClusterIdentifier)159            self.GetEndpoint160            #Cluster is available161            result=0162        except ClientError: #Cluster does not exists163            result=1164        return result165#---------------------------------------------------------------------------------------------166    def WaitForDeletion(self):167        """168            The function waits while the cluster is deleting It can take significant amount of time169        """170        waiter = self.client.get_waiter("cluster_deleted")171        try:172            waiter.wait(ClusterIdentifier=self.ClusterIdentifier)173            #Cluster is available174            result=0175        except ClientError: #Cluster does not exists176            result=1177        return result178#---------------------------------------------------------------------------------------------179#=============================================================================================180#============= Usage example==================================================================181if __name__ == "__main__":182    import sys183    import VPCUtility184    def initVPC():185        MyVPC=VPCUtility.VPCUtility("ProjectResources.yml")186        print MyVPC.VpcName187        if not MyVPC.Vpc:188            print "Creating a new VPC..."189            MyVPC.create_vpc()190        else:191            print "VPC exists"192        return MyVPC193    def init(VPC=None,isPublic=True):194        c=RedshiftUtility("ProjectResources.yml")195        if not(c.endpoint):196            print "Creating cluster. It will take several minutes..."197            if not(VPC):198            #Create standalone cluster199                c.CreateCluster()200            else:201            #Create Cluster in VPC202                if isPublic:203                    SubnetId=MyVPC.GetPublicSubnets()[0]204                else:205                    SubnetId=MyVPC.GetPrivateSubnets()[0]206                SecurityGroupId=MyVPC.GetSecurityGroupId("RedshiftAll")207                c.CreateCluster([SubnetId],SecurityGroupId)208            c.WaitForCreation()209            c.CheckStatus()210            if c.CheckStatus()==0:211                print "Cluster is available: %s" %c212            else:213                print "Cluster is not available..."214        else:215            print "Cluster is available: %s" %c216        return c217        218    Action = sys.argv[1]219    220    if Action=="Create":221        c=init()222    elif Action=="CreateVPC":223        MyVPC=initVPC()224    elif Action=="CreateInVPCPublic":225        MyVPC=initVPC()226        c=init(VPC=MyVPC,isPublic=True)227    elif Action=="CreateInVPCPrivate":228        MyVPC=initVPC()229        c=init(VPC=MyVPC,isPublic=False)230    elif Action=="RedshiftAll":231        MyVPC=initVPC()232        print "Creating Redshift SecurityGroup..."233	RedshiftSecGrp=MyVPC.create_security_group("RedshiftAll")234        print "Scurity group created with ID= %s" %RedshiftSecGrp235    elif Action=="DeleteVPC":236        MyVPC=initVPC()237        print "Deleting VPC..."238        MyVPC.delete_vpc()239    elif Action=="Delete":240        c=init()241        print "Deleting cluster. It will take several minutes..."242        c.DeleteCluster()243        c.WaitForDeletion()244        c.DeleteClusterSubgroup()245    elif Action=="Status":246        c=init()247        if c.CheckStatus()==0:248            print c249        else:...

RDSUtility.py

Source:RDSUtility.py

1#!/usr/bin/python2import boto33from botocore.client import ClientError4import yaml5"""6RDSUtility.py - 03.19.16 Kate Drogaieva7The class in this module allows to create an RDS database8"""9#---------------------------------------------------------------------------------------------10class RDSUtility(object):11    """12        RDSUtility class can be used to create MySQL instance13        The configuration parameters must be provided in a YAML resource file14    """15    def __str__(self):16        return self.endpoint+":"+str(self.port)17    def __init__(self,resource):18        """19            Init MySQL parameters using resource file in YAML format20        """21        try:22            with open(resource, "r") as f:23                self.res = yaml.load(f)24            self.DBName=self.res["RDS"]["DBName"],25            self.DBInstanceIdentifier=self.res["RDS"]["DBInstanceIdentifier"]26        except KeyError or IOError:27            sys.exit("Wrong MySQL Parameters")28        self.client=boto3.client("rds",self.res["Region"])29        self.endpoint=""30        self.port=""31        self.GetEndpoint()32        return33#---------------------------------------------------------------------------------------------34    def CreateDBSubnetGroup(self,SubnetIds):35        """36            Creates an instance Subnet Group if ["RDS"]["SubnetGroup"]  exists37            in YAML configuration file38            SubnetIds is a list of public or private subnets from a VPC39            return Subgroup name or default if VPC does not exists40        """41        SubnetGroup="default"42        response = self.client.create_db_subnet_group(43                   DBSubnetGroupName=self.res["RDS"]["SubnetGroup"]["Name"],44                   DBSubnetGroupDescription=self.res["RDS"]["SubnetGroup"]["Description"],45                   SubnetIds=SubnetIds)46        SubnetGroup=response["DBSubnetGroup"]["DBSubnetGroupName"]47        return SubnetGroup48#---------------------------------------------------------------------------------------------49    def CreateDBInstance(self,SubnetIds=[],SecurityGroupId=[]):50        """51            Create a new DB instance acoording to the configuration parameters in YAML resource file52            If there is ["RDS"]["SubnetGroup"] in the resource file53            an instance Subnet Group is created first based on SubnetIds (list) from a VPC54            SecurityGroupId (list) can be used to create the cluster in VPC55        """56        SubnetGroup="default"57        try:58            if self.res["RDS"]["SubnetGroup"]:59                try: #test if a group already exists60                    response = self.client.describe_db_subnet_groups(DBSubnetGroupName=self.res["RDS"]["SubnetGroup"]["Name"])61                    SubnetGroup=self.res["RDS"]["SubnetGroup"]["Name"]62                except ClientError: #group does not exist - create63                    SubnetGroup=self.CreateDBSubnetGroup(SubnetIds)64        except KeyError:65            pass66        PubliclyAccessible=True67        if self.res["RDS"]["PubliclyAccessible"]=="False":68            PubliclyAccessible=False69        response = self.client.create_db_instance(70    DBName=self.res["RDS"]["DBName"],71    DBInstanceIdentifier=self.res["RDS"]["DBInstanceIdentifier"],72    AllocatedStorage=int(self.res["RDS"]["AllocatedStorage"]),73    DBInstanceClass=self.res["RDS"]["DBInstanceClass"],74    Engine=self.res["RDS"]["Engine"],75    MasterUsername=self.res["RDS"]["MasterUsername"],76    MasterUserPassword=self.res["RDS"]["MasterUserPassword"],77    VpcSecurityGroupIds=SecurityGroupId,78    Port=int(self.res["RDS"]["Port"]),79    MultiAZ=True,80    DBSubnetGroupName=SubnetGroup,81    EngineVersion=self.res["RDS"]["EngineVersion"],82    LicenseModel=self.res["RDS"]["License"],83    PubliclyAccessible=PubliclyAccessible,84    StorageType="standard",85    BackupRetentionPeriod=0)86#---------------------------------------------------------------------------------------------87    def DeleteDBInstance(self):88        """89            The function deletes the DB Instance and do not create a final snapshot90        """91        response = self.client.delete_db_instance(92            DBInstanceIdentifier=self.DBInstanceIdentifier,93            SkipFinalSnapshot=True94            )95        return96#---------------------------------------------------------------------------------------------97    def DeleteDBSubgroup(self):98        """99            The function deletes the DB Subnet Group if it exists100        """101        try: #to delete a group if  exists102                response = self.client.delete_db_subnet_group(DBSubnetGroupName=self.res["RDS"]["SubnetGroup"]["Name"])103        except (ClientError,KeyError) as e: #group does not exist or not configured in the resource file - pass104            pass105        return106#---------------------------------------------------------------------------------------------107    def GetEndpoint(self):108        """109            The function set endpoint and port variable and returns 0 if no errors110            or 1 if the DB is not available111        """112        try:113            response = self.client.describe_db_instances(114            DBInstanceIdentifier=self.DBInstanceIdentifier115            )116            try:117                self.endpoint=response["DBInstances"][0]["Endpoint"]["Address"]118                self.port=response["DBInstances"][0]["Endpoint"]["Port"]119                return 0120            except KeyError:121                return 1122        except ClientError: #Cluster does not exists123            return 1124#---------------------------------------------------------------------------------------------125    def CheckStatus(self):126        """127            The same as GetEndpoint128        """129        return self.GetEndpoint()130#---------------------------------------------------------------------------------------------131    def WaitForCreation(self):132        """133            The function waits while the DB instance is creating It can take significant amount of time134        """135        waiter = self.client.get_waiter("db_instance_available")136        try:137            waiter.wait(DBInstanceIdentifier=self.DBInstanceIdentifier)138            self.GetEndpoint139            #DB Instance is available140            result=0141        except ClientError: #DB Instance does not exists142            result=1143        return result144#---------------------------------------------------------------------------------------------145    def WaitForDeletion(self):146        """147            The function waits while the DB Instance is deleting It can take significant amount of time148        """149        waiter = self.client.get_waiter("db_instance_deleted")150        try:151            waiter.wait(DBInstanceIdentifier=self.DBInstanceIdentifier)152            result=0153        except ClientError: #DB Instance does not exists154            result=1155        return result156#---------------------------------------------------------------------------------------------157#=============================================================================================158#============= Usage example==================================================================159if __name__ == "__main__":160    import sys161    import VPCUtility162    ResourceFile="ProjectResources.yml"163    164    def initVPC():165        MyVPC=VPCUtility.VPCUtility(ResourceFile)166        print MyVPC.VpcName167        if not MyVPC.Vpc:168            print "Creating a new VPC..."169            MyVPC.create_vpc()170            MyVPC.Vpc.modify_attribute(171             EnableDnsHostnames={"Value":True}172              )173        else:174            print "VPC exists"175        return MyVPC176    def init(VPC=None,isPublic=True):177        c=RDSUtility(ResourceFile)178        if not(c.endpoint):179            print "Creating DB instance. It will take several minutes..."180            if not(VPC):181            #Create standalone cluster182                c.CreateDBInstance()183            else:184            #Create Cluster in VPC185                if isPublic:186                    SubnetId=MyVPC.GetPublicSubnets()187                else:188                    SubnetId=MyVPC.GetPrivateSubnets()189                SecurityGroupId=MyVPC.GetSecurityGroupId("RDSAll")190                c.CreateDBInstance(SubnetId,SecurityGroupId)191            c.WaitForCreation()192            c.CheckStatus()193            if c.CheckStatus()==0:194                print "DB Instance is available: %s" %c195            else:196                print "DB Instance is not available..."197        else:198            print "DB Instance is available: %s" %c199        return c200    try:201        ResourceFile=sys.argv[1]202    except:203        ResourceFile ="ProjectResources.yml"204    Action = sys.argv[2]205        206    if Action=="Create":207        c=init()208    elif Action=="CreateVPC":209        MyVPC=initVPC()210    elif Action=="CreateInVPCPublic":211        MyVPC=initVPC()212        c=init(VPC=MyVPC,isPublic=True)213    elif Action=="CreateInVPCPrivate":214        MyVPC=initVPC()215        c=init(VPC=MyVPC,isPublic=False)216    elif Action=="RDSAll":217        MyVPC=initVPC()218        print "Creating RDS SecurityGroup..."219        RDSSecGrp=MyVPC.create_security_group("RDSAll")220        print "Security group created with ID= %s" %RDSSecGrp221    elif Action=="DeleteVPC":222        MyVPC=initVPC()223        print "Deleting VPC..."224        MyVPC.delete_vpc()225    elif Action=="Delete":226        c=init()227        print "Deleting DB Instance. It will take several minutes..."228        c.DeleteDBInstance()229        c.WaitForDeletion()230        c.DeleteDBSubgroup()231    elif Action=="Status":232        c=init()233        if c.CheckStatus()==0:234            print c235        else:...

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.