salt/tests/unit/states/test_boto_elasticsearch_domain.py

# -*- coding: utf-8 -*-

# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals

import logging
import random
import string

import salt.loader
import salt.states.boto_elasticsearch_domain as boto_elasticsearch_domain

# Import Salt libs
from salt.ext import six

# Import 3rd-party libs
from salt.ext.six.moves import range  # pylint: disable=import-error,redefined-builtin
from salt.utils.versions import LooseVersion

# Import Salt Testing libs
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
from tests.support.unit import TestCase, skipIf

# pylint: disable=import-error,no-name-in-module,unused-import
from tests.unit.modules.test_boto_elasticsearch_domain import (
    BotoElasticsearchDomainTestCaseMixin,
)

# Import test suite libs


try:
    import boto
    import boto3
    from botocore.exceptions import ClientError

    HAS_BOTO = True
except ImportError:
    HAS_BOTO = False

# pylint: enable=import-error,no-name-in-module,unused-import

# the boto_elasticsearch_domain module relies on the connect_to_region() method
# which was added in boto 2.8.0
# https://github.com/boto/boto/commit/33ac26b416fbb48a60602542b4ce15dcc7029f12
required_boto3_version = "1.2.1"

log = logging.getLogger(__name__)


def _has_required_boto():
    """
    Returns True/False boolean depending on if Boto is installed and correct
    version.
    """
    if not HAS_BOTO:
        return False
    elif LooseVersion(boto3.__version__) < LooseVersion(required_boto3_version):
        return False
    else:
        return True


if _has_required_boto():
    region = "us-east-1"
    access_key = "GKTADJGHEIQSXMKKRBJ08H"
    secret_key = "askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs"
    conn_parameters = {
        "region": region,
        "key": access_key,
        "keyid": secret_key,
        "profile": {},
    }
    error_message = (
        "An error occurred (101) when calling the {0} operation: Test-defined error"
    )
    not_found_error = ClientError(
        {
            "Error": {
                "Code": "ResourceNotFoundException",
                "Message": "Test-defined error",
            }
        },
        "msg",
    )
    error_content = {"Error": {"Code": 101, "Message": "Test-defined error"}}
    domain_ret = dict(
        DomainName="testdomain",
        ElasticsearchClusterConfig={},
        EBSOptions={},
        AccessPolicies={},
        SnapshotOptions={},
        AdvancedOptions={},
        ElasticsearchVersion="1.5",
    )


class BotoElasticsearchDomainStateTestCaseBase(TestCase, LoaderModuleMockMixin):
    conn = None

    def setup_loader_modules(self):
        ctx = {}
        utils = salt.loader.utils(
            self.opts,
            whitelist=["boto3", "args", "systemd", "path", "platform"],
            context=ctx,
        )
        serializers = salt.loader.serializers(self.opts)
        self.funcs = funcs = salt.loader.minion_mods(
            self.opts, context=ctx, utils=utils, whitelist=["boto_elasticsearch_domain"]
        )
        self.salt_states = salt.loader.states(
            opts=self.opts,
            functions=funcs,
            utils=utils,
            whitelist=["boto_elasticsearch_domain"],
            serializers=serializers,
        )
        return {
            boto_elasticsearch_domain: {
                "__opts__": self.opts,
                "__salt__": funcs,
                "__utils__": utils,
                "__states__": self.salt_states,
                "__serializers__": serializers,
            }
        }

    @classmethod
    def setUpClass(cls):
        cls.opts = salt.config.DEFAULT_MINION_OPTS.copy()
        cls.opts["grains"] = salt.loader.grains(cls.opts)

    @classmethod
    def tearDownClass(cls):
        del cls.opts

    # Set up MagicMock to replace the boto3 session
    def setUp(self):
        self.addCleanup(delattr, self, "funcs")
        self.addCleanup(delattr, self, "salt_states")
        # Set up MagicMock to replace the boto3 session
        # connections keep getting cached from prior tests, can't find the
        # correct context object to clear it. So randomize the cache key, to prevent any
        # cache hits
        conn_parameters["key"] = "".join(
            random.choice(string.ascii_lowercase + string.digits) for _ in range(50)
        )

        self.patcher = patch("boto3.session.Session")
        self.addCleanup(self.patcher.stop)
        self.addCleanup(delattr, self, "patcher")
        mock_session = self.patcher.start()

        session_instance = mock_session.return_value
        self.conn = MagicMock()
        self.addCleanup(delattr, self, "conn")
        session_instance.client.return_value = self.conn


@skipIf(HAS_BOTO is False, "The boto module must be installed.")
@skipIf(
    _has_required_boto() is False,
    "The boto3 module must be greater than"
    " or equal to version {0}".format(required_boto3_version),
)
class BotoElasticsearchDomainTestCase(
    BotoElasticsearchDomainStateTestCaseBase, BotoElasticsearchDomainTestCaseMixin
):
    """
    TestCase for salt.modules.boto_elasticsearch_domain state.module
    """

    def test_present_when_domain_does_not_exist(self):
        """
        Tests present on a domain that does not exist.
        """
        self.conn.describe_elasticsearch_domain.side_effect = not_found_error
        self.conn.describe_elasticsearch_domain_config.return_value = {
            "DomainConfig": domain_ret
        }
        self.conn.create_elasticsearch_domain.return_value = {
            "DomainStatus": domain_ret
        }
        result = self.salt_states["boto_elasticsearch_domain.present"](
            "domain present", **domain_ret
        )

        self.assertTrue(result["result"])
        self.assertEqual(
            result["changes"]["new"]["domain"]["ElasticsearchClusterConfig"], None
        )

    def test_present_when_domain_exists(self):
        self.conn.describe_elasticsearch_domain.return_value = {
            "DomainStatus": domain_ret
        }
        cfg = {}
        for k, v in six.iteritems(domain_ret):
            cfg[k] = {"Options": v}
        cfg["AccessPolicies"] = {"Options": '{"a": "b"}'}
        self.conn.describe_elasticsearch_domain_config.return_value = {
            "DomainConfig": cfg
        }
        self.conn.update_elasticsearch_domain_config.return_value = {
            "DomainConfig": cfg
        }
        result = self.salt_states["boto_elasticsearch_domain.present"](
            "domain present", **domain_ret
        )
        self.assertTrue(result["result"])
        self.assertEqual(
            result["changes"],
            {"new": {"AccessPolicies": {}}, "old": {"AccessPolicies": {"a": "b"}}},
        )

    def test_present_with_failure(self):
        self.conn.describe_elasticsearch_domain.side_effect = not_found_error
        self.conn.describe_elasticsearch_domain_config.return_value = {
            "DomainConfig": domain_ret
        }
        self.conn.create_elasticsearch_domain.side_effect = ClientError(
            error_content, "create_domain"
        )
        result = self.salt_states["boto_elasticsearch_domain.present"](
            "domain present", **domain_ret
        )
        self.assertFalse(result["result"])
        self.assertTrue("An error occurred" in result["comment"])

    def test_absent_when_domain_does_not_exist(self):
        """
        Tests absent on a domain that does not exist.
        """
        self.conn.describe_elasticsearch_domain.side_effect = not_found_error
        result = self.salt_states["boto_elasticsearch_domain.absent"](
            "test", "mydomain"
        )
        self.assertTrue(result["result"])
        self.assertEqual(result["changes"], {})

    def test_absent_when_domain_exists(self):
        self.conn.describe_elasticsearch_domain.return_value = {
            "DomainStatus": domain_ret
        }
        self.conn.describe_elasticsearch_domain_config.return_value = {
            "DomainConfig": domain_ret
        }
        result = self.salt_states["boto_elasticsearch_domain.absent"](
            "test", domain_ret["DomainName"]
        )
        self.assertTrue(result["result"])
        self.assertEqual(result["changes"]["new"]["domain"], None)

    def test_absent_with_failure(self):
        self.conn.describe_elasticsearch_domain.return_value = {
            "DomainStatus": domain_ret
        }
        self.conn.describe_elasticsearch_domain_config.return_value = {
            "DomainConfig": domain_ret
        }
        self.conn.delete_elasticsearch_domain.side_effect = ClientError(
            error_content, "delete_domain"
        )
        result = self.salt_states["boto_elasticsearch_domain.absent"](
            "test", domain_ret["DomainName"]
        )
        self.assertFalse(result["result"])
        self.assertTrue("An error occurred" in result["comment"])