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

How to use describe_elasticsearch_domain_config method in localstack

Best Python code snippet using localstack_python

boto_elasticsearch_domain_test.py

Source:boto_elasticsearch_domain_test.py

1# -*- coding: utf-8 -*-2# Import Python libs3from __future__ import absolute_import4from distutils.version import LooseVersion  # pylint: disable=import-error,no-name-in-module5import logging6import random7import string8# Import Salt Testing libs9from salttesting.unit import skipIf, TestCase10from salttesting.mock import (11    MagicMock,12    NO_MOCK,13    NO_MOCK_REASON,14    patch15)16from salttesting.helpers import ensure_in_syspath17ensure_in_syspath('../../')18# Import Salt libs19import salt.ext.six as six20import salt.loader21from salt.ext.six.moves import range  # pylint: disable=import-error,redefined-builtin22# pylint: disable=import-error,no-name-in-module,unused-import23from unit.modules.boto_elasticsearch_domain_test import BotoElasticsearchDomainTestCaseMixin24# Import 3rd-party libs25try:26    import boto27    import boto328    from botocore.exceptions import ClientError29    HAS_BOTO = True30except ImportError:31    HAS_BOTO = False32# pylint: enable=import-error,no-name-in-module,unused-import33# the boto_elasticsearch_domain module relies on the connect_to_region() method34# which was added in boto 2.8.035# https://github.com/boto/boto/commit/33ac26b416fbb48a60602542b4ce15dcc7029f1236required_boto3_version = '1.2.1'37log = logging.getLogger(__name__)38opts = salt.config.DEFAULT_MINION_OPTS39context = {}40utils = salt.loader.utils(opts, whitelist=['boto3'], context=context)41serializers = salt.loader.serializers(opts)42funcs = salt.loader.minion_mods(opts, context=context, utils=utils, whitelist=['boto_elasticsearch_domain'])43salt_states = salt.loader.states(opts=opts, functions=funcs, utils=utils, whitelist=['boto_elasticsearch_domain'], serializers=serializers)44def _has_required_boto():45    '''46    Returns True/False boolean depending on if Boto is installed and correct47    version.48    '''49    if not HAS_BOTO:50        return False51    elif LooseVersion(boto3.__version__) < LooseVersion(required_boto3_version):52        return False53    else:54        return True55if _has_required_boto():56    region = 'us-east-1'57    access_key = 'GKTADJGHEIQSXMKKRBJ08H'58    secret_key = 'askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs'59    conn_parameters = {'region': region, 'key': access_key, 'keyid': secret_key, 'profile': {}}60    error_message = 'An error occurred (101) when calling the {0} operation: Test-defined error'61    not_found_error = ClientError({62        'Error': {63            'Code': 'ResourceNotFoundException',64            'Message': "Test-defined error"65        }66    }, 'msg')67    error_content = {68      'Error': {69        'Code': 101,70        'Message': "Test-defined error"71      }72    }73    domain_ret = dict(DomainName='testdomain',74                  ElasticsearchClusterConfig={},75                  EBSOptions={},76                  AccessPolicies={},77                  SnapshotOptions={},78                  AdvancedOptions={})79class BotoElasticsearchDomainStateTestCaseBase(TestCase):80    conn = None81    # Set up MagicMock to replace the boto3 session82    def setUp(self):83        context.clear()84        # connections keep getting cached from prior tests, can't find the85        # correct context object to clear it. So randomize the cache key, to prevent any86        # cache hits87        conn_parameters['key'] = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(50))88        self.patcher = patch('boto3.session.Session')89        self.addCleanup(self.patcher.stop)90        mock_session = self.patcher.start()91        session_instance = mock_session.return_value92        self.conn = MagicMock()93        session_instance.client.return_value = self.conn94@skipIf(HAS_BOTO is False, 'The boto module must be installed.')95@skipIf(_has_required_boto() is False, 'The boto3 module must be greater than'96                                       ' or equal to version {0}'97        .format(required_boto3_version))98@skipIf(NO_MOCK, NO_MOCK_REASON)99class BotoElasticsearchDomainTestCase(BotoElasticsearchDomainStateTestCaseBase, BotoElasticsearchDomainTestCaseMixin):100    '''101    TestCase for salt.modules.boto_elasticsearch_domain state.module102    '''103    def test_present_when_domain_does_not_exist(self):104        '''105        Tests present on a domain that does not exist.106        '''107        self.conn.describe_elasticsearch_domain.side_effect = not_found_error108        self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}109        self.conn.create_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}110        result = salt_states['boto_elasticsearch_domain.present'](111                         'domain present',112                         **domain_ret)113        self.assertTrue(result['result'])114        self.assertEqual(result['changes']['new']['domain']['ElasticsearchClusterConfig'], None)115    def test_present_when_domain_exists(self):116        self.conn.describe_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}117        cfg = {}118        for k, v in six.iteritems(domain_ret):119            cfg[k] = {'Options': v}120        cfg['AccessPolicies'] = {'Options': '{"a": "b"}'}121        self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': cfg}122        self.conn.update_elasticsearch_domain_config.return_value = {'DomainConfig': cfg}123        result = salt_states['boto_elasticsearch_domain.present'](124                         'domain present',125                         **domain_ret)126        self.assertTrue(result['result'])127        self.assertEqual(result['changes'], {'new': {'AccessPolicies': {}}, 'old': {'AccessPolicies': {u'a': u'b'}}})128    def test_present_with_failure(self):129        self.conn.describe_elasticsearch_domain.side_effect = not_found_error130        self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}131        self.conn.create_elasticsearch_domain.side_effect = ClientError(error_content, 'create_domain')132        result = salt_states['boto_elasticsearch_domain.present'](133                         'domain present',134                         **domain_ret)135        self.assertFalse(result['result'])136        self.assertTrue('An error occurred' in result['comment'])137    def test_absent_when_domain_does_not_exist(self):138        '''139        Tests absent on a domain that does not exist.140        '''141        self.conn.describe_elasticsearch_domain.side_effect = not_found_error142        result = salt_states['boto_elasticsearch_domain.absent']('test', 'mydomain')143        self.assertTrue(result['result'])144        self.assertEqual(result['changes'], {})145    def test_absent_when_domain_exists(self):146        self.conn.describe_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}147        self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}148        result = salt_states['boto_elasticsearch_domain.absent']('test', domain_ret['DomainName'])149        self.assertTrue(result['result'])150        self.assertEqual(result['changes']['new']['domain'], None)151    def test_absent_with_failure(self):152        self.conn.describe_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}153        self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}154        self.conn.delete_elasticsearch_domain.side_effect = ClientError(error_content, 'delete_domain')155        result = salt_states['boto_elasticsearch_domain.absent']('test', domain_ret['DomainName'])156        self.assertFalse(result['result'])...

aws_es_info.py

Source:aws_es_info.py

...98            return client.describe_elasticsearch_domain(99                DomainName=module.params['name'],100            ), False101        elif module.params['describe_elasticsearch_domain_config']:102            return client.describe_elasticsearch_domain_config(103                DomainName=module.params['name'],104            ), False105        else:106            if client.can_paginate('list_domain_names'):107                paginator = client.get_paginator('list_domain_names')108                return paginator.paginate(), True109            else:110                return client.list_domain_names(), False111    except (BotoCoreError, ClientError) as e:112        module.fail_json_aws(e, msg='Failed to fetch Amazon ES details')113def main():114    argument_spec = dict(115        name=dict(required=False),116        list_packages_for_domain=dict(required=False, type=bool),...

test_boto_elasticsearch_domain.py

Source:test_boto_elasticsearch_domain.py

1# -*- coding: utf-8 -*-2# Import Python libs3from __future__ import absolute_import, print_function, unicode_literals4import logging5import random6import string7# Import Salt libs8from salt.ext import six9import salt.loader10# Import Fractus Libs11import fractus.cloudstates.boto_elasticsearch_domain as boto_elasticsearch_domain12# Import Testing libs13import pytest14pytestmark = pytest.mark.skip('all tests still WIP')15# the boto_elasticsearch_domain module relies on the connect_to_region() method16# which was added in boto 2.8.017# https://github.com/boto/boto/commit/33ac26b416fbb48a60602542b4ce15dcc7029f1218required_boto3_version = '1.2.1'19log = logging.getLogger(__name__)20def setup_loader_modules(self):21    ctx = {}22    utils = salt.loader.utils(self.opts, whitelist=['boto3'], context=ctx)23    serializers = salt.loader.serializers(self.opts)24    self.funcs = funcs = salt.loader.minion_mods(self.opts, context=ctx, utils=utils, whitelist=['boto_elasticsearch_domain'])25    self.salt_states = salt.loader.states(opts=self.opts, functions=funcs, utils=utils, whitelist=['boto_elasticsearch_domain'],26                                          serializers=serializers)27    return {28        boto_elasticsearch_domain: {29            '__opts__': self.opts,30            '__salt__': funcs,31            '__utils__': utils,32            '__states__': self.salt_states,33            '__serializers__': serializers,34        }35    }36def test_present_when_domain_does_not_exist():37    '''38    Tests present on a domain that does not exist.39    '''40    self.conn.describe_elasticsearch_domain.side_effect = not_found_error41    self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}42    self.conn.create_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}43    result = self.salt_states['boto_elasticsearch_domain.present'](44                     'domain present',45                     **domain_ret)46    self.assertTrue(result['result'])47    self.assertEqual(result['changes']['new']['domain']['ElasticsearchClusterConfig'], None)48def test_present_when_domain_exists():49    self.conn.describe_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}50    cfg = {}51    for k, v in six.iteritems(domain_ret):52        cfg[k] = {'Options': v}53    cfg['AccessPolicies'] = {'Options': '{"a": "b"}'}54    self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': cfg}55    self.conn.update_elasticsearch_domain_config.return_value = {'DomainConfig': cfg}56    result = self.salt_states['boto_elasticsearch_domain.present'](57                     'domain present',58                     **domain_ret)59    self.assertTrue(result['result'])60    self.assertEqual(result['changes'], {'new': {'AccessPolicies': {}}, 'old': {'AccessPolicies': {'a': 'b'}}})61def test_present_with_failure():62    self.conn.describe_elasticsearch_domain.side_effect = not_found_error63    self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}64    self.conn.create_elasticsearch_domain.side_effect = ClientError(error_content, 'create_domain')65    result = self.salt_states['boto_elasticsearch_domain.present'](66                     'domain present',67                     **domain_ret)68    self.assertFalse(result['result'])69    self.assertTrue('An error occurred' in result['comment'])70def test_absent_when_domain_does_not_exist():71    '''72    Tests absent on a domain that does not exist.73    '''74    self.conn.describe_elasticsearch_domain.side_effect = not_found_error75    result = self.salt_states['boto_elasticsearch_domain.absent']('test', 'mydomain')76    self.assertTrue(result['result'])77    self.assertEqual(result['changes'], {})78def test_absent_when_domain_exists():79    self.conn.describe_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}80    self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}81    result = self.salt_states['boto_elasticsearch_domain.absent']('test', domain_ret['DomainName'])82    self.assertTrue(result['result'])83    self.assertEqual(result['changes']['new']['domain'], None)84def test_absent_with_failure():85    self.conn.describe_elasticsearch_domain.return_value = {'DomainStatus': domain_ret}86    self.conn.describe_elasticsearch_domain_config.return_value = {'DomainConfig': domain_ret}87    self.conn.delete_elasticsearch_domain.side_effect = ClientError(error_content, 'delete_domain')88    result = self.salt_states['boto_elasticsearch_domain.absent']('test', domain_ret['DomainName'])89    self.assertFalse(result['result'])...

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.