salt/tests/integration/modules/test_boto_sns.py

# -*- coding: utf-8 -*-
"""
Validate the boto_sns module
"""

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

import re

# Import Salt Testing libs
from tests.support.case import ModuleCase
from tests.support.unit import skipIf

# Import 3rd-party libs
try:
    import boto

    NO_BOTO_MODULE = False
except ImportError:
    NO_BOTO_MODULE = True


@skipIf(
    NO_BOTO_MODULE,
    "Please install the boto library before running boto integration tests.",
)
class BotoSNSTest(ModuleCase):
    def setUp(self):
        try:
            boto.connect_iam()
        except boto.exception.NoAuthHandlerFound:
            self.skipTest(
                "Please setup boto AWS credentials before running boto integration tests."
            )
        # The name of the topic you want to create.
        # Constraints: Topic names must be made up of only uppercase and
        # lowercase ASCII letters, numbers, underscores, and hyphens,
        # and must be between 1 and 256 characters long.
        # http://docs.aws.amazon.com/sns/latest/api/API_CreateTopic.html
        self.topic_name = re.sub(r"[^a-zA-Z_-]", "_", self.id())[0:256]
        self.topic_names = [self.topic_name]
        self.run_function("boto_sns.delete", name=self.topic_name)

    def tearDown(self):
        for topic in self.topic_names:
            self.run_function("boto_sns.delete", name=topic)

    def test_exists_non_existing(self):
        ret = self.run_function("boto_sns.exists", ["nonexistent"])
        self.assertSaltModuleFalseReturn(ret)

    def test_exists_existing(self):
        self.run_function("boto_sns.create", [self.topic_name])
        ret = self.run_function("boto_sns.exists", [self.topic_name])
        self.assertSaltModuleTrueReturn(ret)

    def test_create(self):
        ret = self.run_function("boto_sns.create", [self.topic_name])
        self.assertSaltModuleTrueReturn(ret)

        ret = self.run_function("boto_sns.get_all_topics")
        self.assertIn(self.topic_name, list(ret.keys()))
        self.assertIn(self._get_arn(self.topic_name), list(ret.values()))

    def test_delete_non_existing(self):
        ret = self.run_function("boto_sns.delete", [self.topic_name])
        self.assertSaltModuleTrueReturn(ret)

    def test_delete_existing(self):
        self.run_function("boto_sns.create", [self.topic_name])
        ret = self.run_function("boto_sns.delete", [self.topic_name])
        self.assertSaltModuleTrueReturn(ret)

        ret = self.run_function("boto_sns.get_all_topics")
        self.assertNotIn(self.topic_name, list(ret.keys()))
        self.assertNotIn(self._get_arn(self.topic_name), list(ret.values()))

    def test_get_all_topics(self):
        self.topic_names.append(self.topic_name + "-2")
        for topic in self.topic_names:
            self.run_function("boto_sns.create", [topic])

        ret = self.run_function("boto_sns.get_all_topics")

        for topic in self.topic_names:
            self.assertIn(topic, list(ret.keys()))
            self.assertIn(self._get_arn(topic), list(ret.values()))

    def test_subscribe_and_get_all_subscriptions_by_topic(self):
        topic_name = self.topic_name
        ret = self.run_function("boto_sns.create", [topic_name])

        ret = self.run_function(
            "boto_sns.subscribe",
            [topic_name, "https", "https://www.example.com/sns/endpoint"],
        )
        self.assertSaltModuleTrueReturn(ret)

        ret = self.run_function("boto_sns.get_all_subscriptions_by_topic", [topic_name])
        self.assertDictContainsSubset(
            {"Protocol": "https", "Endpoint": "https://www.example.com/sns/endpoint"},
            ret[0],
        )

    def _get_arn(self, name):
        return "arn:aws:sns:us-east-1:{0}:{1}".format(self.account_id, name)

    @property
    def account_id(self):
        if not hasattr(self, "_account_id"):
            account_id = self.run_function("boto_iam.get_account_id")
            setattr(self, "_account_id", account_id)
        return self._account_id

    def assertSaltModuleTrueReturn(self, ret):
        self.assertIsInstance(ret, bool)
        self.assertTrue(ret)

    def assertSaltModuleFalseReturn(self, ret):
        self.assertIsInstance(ret, bool)
        self.assertFalse(ret)