mirror of
https://github.com/valitydev/salt.git
synced 2024-11-07 17:09:03 +00:00
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
# -*- 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)
|