salt/tests/unit/states/test_boto_cloudwatch_event.py

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

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

import logging
import random
import string

import pytest

# Import Salt libs
import salt.config
import salt.loader
import salt.states.boto_cloudwatch_event as boto_cloudwatch_event
from salt.ext.six.moves import range

# 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
from tests.unit.modules.test_boto_cloudwatch_event import (
    BotoCloudWatchEventTestCaseMixin,
)

# pylint: disable=unused-import
# Import 3rd-party libs
try:
    import boto3
    from botocore.exceptions import ClientError

    HAS_BOTO = True
except ImportError:
    HAS_BOTO = False
# pylint: enable=unused-import


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

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"
)
error_content = {"Error": {"Code": 101, "Message": "Test-defined error"}}
if HAS_BOTO:
    not_found_error = ClientError(
        {
            "Error": {
                "Code": "ResourceNotFoundException",
                "Message": "Test-defined error",
            }
        },
        "msg",
    )
rule_name = "test_thing_type"
rule_desc = "test_thing_type_desc"
rule_sched = "rate(20 min)"
rule_arn = "arn:::::rule/arn"
rule_ret = dict(
    Arn=rule_arn,
    Description=rule_desc,
    EventPattern=None,
    Name=rule_name,
    RoleArn=None,
    ScheduleExpression=rule_sched,
    State="ENABLED",
)


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
    else:
        return True


class BotoCloudWatchEventStateTestCaseBase(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_cloudwatch_event"]
        )
        self.salt_states = salt.loader.states(
            opts=self.opts,
            functions=funcs,
            utils=utils,
            whitelist=["boto_cloudwatch_event"],
            serializers=serializers,
        )
        return {
            boto_cloudwatch_event: {
                "__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

    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.")
class BotoCloudWatchEventTestCase(
    BotoCloudWatchEventStateTestCaseBase, BotoCloudWatchEventTestCaseMixin
):
    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_failing_to_describe_rule(self):
        """
        Tests exceptions when checking rule existence
        """
        self.conn.list_rules.side_effect = ClientError(
            error_content, "error on list rules"
        )
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("error on list rules" in result.get("comment", {}))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_failing_to_create_a_new_rule(self):
        """
        Tests present on a rule name that doesn't exist and
        an error is thrown on creation.
        """
        self.conn.list_rules.return_value = {"Rules": []}
        self.conn.put_rule.side_effect = ClientError(error_content, "put_rule")
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("put_rule" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_failing_to_describe_the_new_rule(self):
        """
        Tests present on a rule name that doesn't exist and
        an error is thrown when adding targets.
        """
        self.conn.list_rules.return_value = {"Rules": []}
        self.conn.put_rule.return_value = rule_ret
        self.conn.describe_rule.side_effect = ClientError(
            error_content, "describe_rule"
        )
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("describe_rule" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_failing_to_create_a_new_rules_targets(self):
        """
        Tests present on a rule name that doesn't exist and
        an error is thrown when adding targets.
        """
        self.conn.list_rules.return_value = {"Rules": []}
        self.conn.put_rule.return_value = rule_ret
        self.conn.describe_rule.return_value = rule_ret
        self.conn.put_targets.side_effect = ClientError(error_content, "put_targets")
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("put_targets" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_rule_does_not_exist(self):
        """
        Tests the successful case of creating a new rule, and updating its
        targets
        """
        self.conn.list_rules.return_value = {"Rules": []}
        self.conn.put_rule.return_value = rule_ret
        self.conn.describe_rule.return_value = rule_ret
        self.conn.put_targets.return_value = {"FailedEntryCount": 0}
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), True)

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_failing_to_update_an_existing_rule(self):
        """
        Tests present on an existing rule where an error is thrown on updating the pool properties.
        """
        self.conn.list_rules.return_value = {"Rules": [rule_ret]}
        self.conn.describe_rule.side_effect = ClientError(
            error_content, "describe_rule"
        )
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("describe_rule" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_failing_to_get_targets(self):
        """
        Tests present on an existing rule where put_rule succeeded, but an error
        is thrown on getting targets
        """
        self.conn.list_rules.return_value = {"Rules": [rule_ret]}
        self.conn.put_rule.return_value = rule_ret
        self.conn.describe_rule.return_value = rule_ret
        self.conn.list_targets_by_rule.side_effect = ClientError(
            error_content, "list_targets"
        )
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("list_targets" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_failing_to_put_targets(self):
        """
        Tests present on an existing rule where put_rule succeeded, but an error
        is thrown on putting targets
        """
        self.conn.list_rules.return_value = {"Rules": []}
        self.conn.put_rule.return_value = rule_ret
        self.conn.describe_rule.return_value = rule_ret
        self.conn.list_targets.return_value = {"Targets": []}
        self.conn.put_targets.side_effect = ClientError(error_content, "put_targets")
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("put_targets" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_putting_targets(self):
        """
        Tests present on an existing rule where put_rule succeeded, and targets
        must be added
        """
        self.conn.list_rules.return_value = {"Rules": []}
        self.conn.put_rule.return_value = rule_ret
        self.conn.describe_rule.return_value = rule_ret
        self.conn.list_targets.return_value = {"Targets": []}
        self.conn.put_targets.return_value = {"FailedEntryCount": 0}
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), True)

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_present_when_removing_targets(self):
        """
        Tests present on an existing rule where put_rule succeeded, and targets
        must be removed
        """
        self.conn.list_rules.return_value = {"Rules": []}
        self.conn.put_rule.return_value = rule_ret
        self.conn.describe_rule.return_value = rule_ret
        self.conn.list_targets.return_value = {
            "Targets": [{"Id": "target1"}, {"Id": "target2"}]
        }
        self.conn.put_targets.return_value = {"FailedEntryCount": 0}
        result = self.salt_states["boto_cloudwatch_event.present"](
            name="test present",
            Name=rule_name,
            Description=rule_desc,
            ScheduleExpression=rule_sched,
            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],
            **conn_parameters
        )
        self.assertEqual(result.get("result"), True)

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_absent_when_failing_to_describe_rule(self):
        """
        Tests exceptions when checking rule existence
        """
        self.conn.list_rules.side_effect = ClientError(
            error_content, "error on list rules"
        )
        result = self.salt_states["boto_cloudwatch_event.absent"](
            name="test present", Name=rule_name, **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("error on list rules" in result.get("comment", {}))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_absent_when_rule_does_not_exist(self):
        """
        Tests absent on an non-existing rule
        """
        self.conn.list_rules.return_value = {"Rules": []}
        result = self.salt_states["boto_cloudwatch_event.absent"](
            name="test absent", Name=rule_name, **conn_parameters
        )
        self.assertEqual(result.get("result"), True)
        self.assertEqual(result["changes"], {})

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_absent_when_failing_to_list_targets(self):
        """
        Tests absent on an rule when the list_targets call fails
        """
        self.conn.list_rules.return_value = {"Rules": [rule_ret]}
        self.conn.list_targets_by_rule.side_effect = ClientError(
            error_content, "list_targets"
        )
        result = self.salt_states["boto_cloudwatch_event.absent"](
            name="test absent", Name=rule_name, **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("list_targets" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_absent_when_failing_to_remove_targets_exception(self):
        """
        Tests absent on an rule when the remove_targets call fails
        """
        self.conn.list_rules.return_value = {"Rules": [rule_ret]}
        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}
        self.conn.remove_targets.side_effect = ClientError(
            error_content, "remove_targets"
        )
        result = self.salt_states["boto_cloudwatch_event.absent"](
            name="test absent", Name=rule_name, **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("remove_targets" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_absent_when_failing_to_remove_targets_nonexception(self):
        """
        Tests absent on an rule when the remove_targets call fails
        """
        self.conn.list_rules.return_value = {"Rules": [rule_ret]}
        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}
        self.conn.remove_targets.return_value = {"FailedEntryCount": 1}
        result = self.salt_states["boto_cloudwatch_event.absent"](
            name="test absent", Name=rule_name, **conn_parameters
        )
        self.assertEqual(result.get("result"), False)

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_absent_when_failing_to_delete_rule(self):
        """
        Tests absent on an rule when the delete_rule call fails
        """
        self.conn.list_rules.return_value = {"Rules": [rule_ret]}
        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}
        self.conn.remove_targets.return_value = {"FailedEntryCount": 0}
        self.conn.delete_rule.side_effect = ClientError(error_content, "delete_rule")
        result = self.salt_states["boto_cloudwatch_event.absent"](
            name="test absent", Name=rule_name, **conn_parameters
        )
        self.assertEqual(result.get("result"), False)
        self.assertTrue("delete_rule" in result.get("comment", ""))

    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_absent(self):
        """
        Tests absent on an rule
        """
        self.conn.list_rules.return_value = {"Rules": [rule_ret]}
        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}
        self.conn.remove_targets.return_value = {"FailedEntryCount": 0}
        result = self.salt_states["boto_cloudwatch_event.absent"](
            name="test absent", Name=rule_name, **conn_parameters
        )
        self.assertEqual(result.get("result"), True)