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

How to use describe_clusters method in localstack

Best Python code snippet using localstack_python

redshift.py

Source:redshift.py

...250              'encrypted', 'elastic_ip'):251        if p in module.params:252            params[ p ] = module.params.get( p )253    try:254        redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]255        changed = False256    except boto.exception.JSONResponseError as e:257        try:258            redshift.create_cluster(identifier, node_type, username, password, **params)259        except boto.exception.JSONResponseError as e:260            module.fail_json(msg=str(e))261    try:262        resource = redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]263    except boto.exception.JSONResponseError as e:264        module.fail_json(msg=str(e))265    if wait:266        try:267            wait_timeout = time.time() + wait_timeout268            time.sleep(5)269            while wait_timeout > time.time() and resource['ClusterStatus'] != 'available':270                time.sleep(5)271                if wait_timeout <= time.time():272                    module.fail_json(msg = "Timeout waiting for resource %s" % resource.id)273                resource = redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]274        except boto.exception.JSONResponseError as e:275            module.fail_json(msg=str(e))276    return(changed, _collect_facts(resource))277def describe_cluster(module, redshift):278    """279    Collect data about the cluster.280    module: Ansible module object281    redshift: authenticated redshift connection object282    """283    identifier = module.params.get('identifier')284    try:285        resource = redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]286    except boto.exception.JSONResponseError as e:287        module.fail_json(msg=str(e))288    return(True, _collect_facts(resource))289def delete_cluster(module, redshift):290    """291    Delete a cluster.292    module: Ansible module object293    redshift: authenticated redshift connection object294    """295    identifier   = module.params.get('identifier')296    wait         = module.params.get('wait')297    wait_timeout = module.params.get('wait_timeout')298    try:299        redshift.delete_custer( identifier )300    except boto.exception.JSONResponseError as e:301        module.fail_json(msg=str(e))302    if wait:303        try:304            wait_timeout = time.time() + wait_timeout305            resource = redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]306            while wait_timeout > time.time() and resource['ClusterStatus'] != 'deleting':307                time.sleep(5)308                if wait_timeout <= time.time():309                    module.fail_json(msg = "Timeout waiting for resource %s" % resource.id)310                resource = redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]311        except boto.exception.JSONResponseError as e:312            module.fail_json(msg=str(e))313    return(True, {})314def modify_cluster(module, redshift):315    """316    Modify an existing cluster.317    module: Ansible module object318    redshift: authenticated redshift connection object319    """320    identifier   = module.params.get('identifier')321    wait         = module.params.get('wait')322    wait_timeout = module.params.get('wait_timeout')323    # Package up the optional parameters324    params = {}325    for p in ('cluster_type', 'cluster_security_groups',326              'vpc_security_group_ids', 'cluster_subnet_group_name',327              'availability_zone', 'preferred_maintenance_window',328              'cluster_parameter_group_name',329              'automated_snapshot_retention_period', 'port', 'cluster_version',330              'allow_version_upgrade', 'number_of_nodes', 'new_cluster_identifier'):331        if p in module.params:332            params[p] = module.params.get(p)333    try:334        redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]335    except boto.exception.JSONResponseError as e:336        try:337            redshift.modify_cluster(identifier, **params)338        except boto.exception.JSONResponseError as e:339            module.fail_json(msg=str(e))340    try:341        resource = redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]342    except boto.exception.JSONResponseError as e:343        module.fail_json(msg=str(e))344    if wait:345        try:346            wait_timeout = time.time() + wait_timeout347            time.sleep(5)348            while wait_timeout > time.time() and resource['ClusterStatus'] != 'available':349                time.sleep(5)350                if wait_timeout <= time.time():351                    module.fail_json(msg = "Timeout waiting for resource %s" % resource.id)352                resource = redshift.describe_clusters(identifier)['DescribeClustersResponse']['DescribeClustersResult']['Clusters'][0]353        except boto.exception.JSONResponseError as e:354            # https://github.com/boto/boto/issues/2776 is fixed.355            module.fail_json(msg=str(e))356    return(True, _collect_facts(resource))357def main():358    argument_spec = ec2_argument_spec()359    argument_spec.update(dict(360            command                             = dict(choices=['create', 'facts', 'delete', 'modify'], required=True),361            identifier                          = dict(required=True),362            node_type                           = dict(choices=['ds1.xlarge', 'ds1.8xlarge', 'ds2.xlarge', 'ds2.8xlarge', 'dc1.large', 'dc1.8xlarge', 'dw1.xlarge', 'dw1.8xlarge', 'dw2.large', 'dw2.8xlarge'], required=False),363            username                            = dict(required=False),364            password                            = dict(no_log=True, required=False),365            db_name                             = dict(require=False),366            cluster_type                        = dict(choices=['multi-node', 'single-node', ], default='single-node'),...

test_layer1.py

Source:test_layer1.py

...70        # Per @garnaat, for the sake of suite time, we'll test as much as we71        # can before we teardown.72        # Test a non-existent cluster ID.73        with self.assertRaises(ClusterNotFoundFault):74            self.api.describe_clusters('badpipelineid')75        # Now create the cluster & move on.76        cluster_id = self.create_cluster()77        # Test never resized.78        with self.assertRaises(ResizeNotFoundFault):79            self.api.describe_resize(cluster_id)80        # The cluster shows up in describe_clusters81        clusters = self.api.describe_clusters()['DescribeClustersResponse']\82                                               ['DescribeClustersResult']\83                                               ['Clusters']84        cluster_ids = [c['ClusterIdentifier'] for c in clusters]85        self.assertIn(cluster_id, cluster_ids)86        # The cluster shows up in describe_clusters w/ id87        response = self.api.describe_clusters(cluster_id)88        self.assertEqual(response['DescribeClustersResponse']\89                         ['DescribeClustersResult']['Clusters'][0]\90                         ['ClusterIdentifier'], cluster_id)91        snapshot_id = "snap-%s" % cluster_id92        # Test creating a snapshot.93        response = self.api.create_cluster_snapshot(snapshot_id, cluster_id)94        self.assertEqual(response['CreateClusterSnapshotResponse']\95                         ['CreateClusterSnapshotResult']['Snapshot']\96                         ['SnapshotIdentifier'], snapshot_id)97        self.assertEqual(response['CreateClusterSnapshotResponse']\98                         ['CreateClusterSnapshotResult']['Snapshot']\99                         ['Status'], 'creating')100        self.addCleanup(self.api.delete_cluster_snapshot, snapshot_id)101        # More waiting. :(...

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.