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

How to use _create_cluster method in localstack

Best Python code snippet using localstack_python

cluster_t.py

Source:cluster_t.py

...19from couchbase.cluster import Cluster, ClassicAuthenticator,\20    PasswordAuthenticator, NoBucketError, MixedAuthError21import gc22class ClusterTest(CouchbaseTestCase):23    def _create_cluster(self):24        connargs = self.make_connargs()25        connstr = ConnectionString.parse(str(connargs.pop('connection_string')))26        connstr.clear_option('username')27        bucket = connstr.bucket28        connstr.bucket = None29        password = connargs.get('password', '')30        # Can I open a new bucket via open_bucket?31        cluster = Cluster(connstr, bucket_class=self.factory)32        cluster.authenticate(ClassicAuthenticator(buckets={bucket: password}))33        return cluster, bucket34    def test_cluster(self):35        cluster, bucket_name = self._create_cluster()36        cb = cluster.open_bucket(bucket_name)37        key = self.gen_key('cluster_test')38        cb.upsert(key, 'cluster test')39    def test_query(self):40        self.skipUnlessMock()41        cluster, bucket_name = self._create_cluster()42        # Should fail if no bucket is active yet43        self.assertRaises(NoBucketError, cluster.n1ql_query, "select mockrow")44        # Open a bucket45        cb = cluster.open_bucket(bucket_name)46        row = cluster.n1ql_query('select mockrow').get_single_result()47        # Should fail again once the bucket has been GC'd48        del cb49        gc.collect()50        self.assertRaises(NoBucketError, cluster.n1ql_query, 'select mockrow')51    def test_no_mixed_auth(self):52        cluster, bucket_name = self._create_cluster()53        auther = PasswordAuthenticator(bucket_name,54                                       self.cluster_info.bucket_password)55        cluster.authenticate(auther)56        cb1 = cluster.open_bucket(bucket_name)57        self.assertRaises(MixedAuthError, cluster.open_bucket, bucket_name,58                          password=self.cluster_info.bucket_password)59        cluster2, bucket_name = self._create_cluster()60        cb2 = cluster2.open_bucket(bucket_name,61                                   password=self.cluster_info.bucket_password)62    def test_pathless_connstr(self):63        # Not strictly a cluster test, but relevant64        connstr = ConnectionString.parse('couchbase://localhost?opt1=val1&opt2=val2')65        self.assertTrue('opt1' in connstr.options)66        self.assertTrue('opt2' in connstr.options)67    def test_validate_authenticate(self):68        cluster, bucket_name = self._create_cluster()69        self.assertRaises(ValueError, cluster.authenticate, username=None, password=None)70        self.assertRaises(ValueError, cluster.authenticate, username='', password='')71        self.assertRaises(ValueError, cluster.authenticate, username='username', password=None)72        self.assertRaises(ValueError, cluster.authenticate, username='username', password='')73        self.assertRaises(ValueError, cluster.authenticate, username=None, password='password')74        self.assertRaises(ValueError, cluster.authenticate, username='', password='password')75    def test_can_authenticate_with_username_password(self):76        cluster, bucket_name = self._create_cluster()77        cluster.authenticate(username='Administrator', password='password')78        bucket = cluster.open_bucket(bucket_name)...

test_aws_redshift_cluster_sensor.py

Source:test_aws_redshift_cluster_sensor.py

...25except ImportError:26    mock_redshift = None27class TestAwsRedshiftClusterSensor(unittest.TestCase):28    @staticmethod29    def _create_cluster():30        client = boto3.client('redshift', region_name='us-east-1')31        client.create_cluster(32            ClusterIdentifier='test_cluster',33            NodeType='dc1.large',34            MasterUsername='admin',35            MasterUserPassword='mock_password'36        )37        if len(client.describe_clusters()['Clusters']) == 0:38            raise ValueError('AWS not properly mocked')39    @unittest.skipIf(mock_redshift is None, 'mock_redshift package not present')40    @mock_redshift41    def test_poke(self):42        self._create_cluster()43        op = AwsRedshiftClusterSensor(task_id='test_cluster_sensor',44                                      poke_interval=1,45                                      timeout=5,46                                      aws_conn_id='aws_default',47                                      cluster_identifier='test_cluster',48                                      target_status='available')49        self.assertTrue(op.poke(None))50    @unittest.skipIf(mock_redshift is None, 'mock_redshift package not present')51    @mock_redshift52    def test_poke_false(self):53        self._create_cluster()54        op = AwsRedshiftClusterSensor(task_id='test_cluster_sensor',55                                      poke_interval=1,56                                      timeout=5,57                                      aws_conn_id='aws_default',58                                      cluster_identifier='test_cluster_not_found',59                                      target_status='available')60        self.assertFalse(op.poke(None))61    @unittest.skipIf(mock_redshift is None, 'mock_redshift package not present')62    @mock_redshift63    def test_poke_cluster_not_found(self):64        self._create_cluster()65        op = AwsRedshiftClusterSensor(task_id='test_cluster_sensor',66                                      poke_interval=1,67                                      timeout=5,68                                      aws_conn_id='aws_default',69                                      cluster_identifier='test_cluster_not_found',70                                      target_status='cluster_not_found')71        self.assertTrue(op.poke(None))72if __name__ == '__main__':...

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.