archive
/
ansible
mirror of https://github.com/ansible/ansible.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Removed Ansible API based tests from this PR

Browse Source
reviewable/pr18780/r1
Allen Sanabria 9 years ago
parent 1cc5ea7418
commit e9fcb8b286
1 changed files with 15 additions and 221 deletions
Show Stats Download Patch File Download Diff File

236
test/unit/cloud/amazon/test_kinesis_stream.py
Unescape Escape View File

@ -3,216 +3,10 @@
import boto3
import unittest
from collections import namedtuple
from ansible.parsing.dataloader import DataLoader
from ansible.vars import VariableManager
from ansible.inventory import Inventory
from ansible.playbook.play import Play
from ansible.executor.task_queue_manager import TaskQueueManager
import cloud.amazon.kinesis_stream as kinesis_stream
Options = (
namedtuple(
'Options', [
'connection', 'module_path', 'forks', 'become', 'become_method',
'become_user', 'remote_user', 'private_key_file', 'ssh_common_args',
'sftp_extra_args', 'scp_extra_args', 'ssh_extra_args', 'verbosity',
'check'
]
)
)
# initialize needed objects
variable_manager = VariableManager()
loader = DataLoader()
options = (
Options(
connection='local',
module_path='cloud/amazon',
forks=1, become=None, become_method=None, become_user=None, check=True,
remote_user=None, private_key_file=None, ssh_common_args=None,
sftp_extra_args=None, scp_extra_args=None, ssh_extra_args=None,
verbosity=10
)
)
passwords = dict(vault_pass='')
aws_region = 'us-west-2'
# create inventory and pass to var manager
inventory = Inventory(loader=loader, variable_manager=variable_manager, host_list='localhost')
variable_manager.set_inventory(inventory)
def run(play):
tqm = None
results = None
try:
tqm = TaskQueueManager(
inventory=inventory,
variable_manager=variable_manager,
loader=loader,
options=options,
passwords=passwords,
stdout_callback='default',
)
results = tqm.run(play)
finally:
if tqm is not None:
tqm.cleanup()
return tqm, results
class AnsibleKinesisStreamTasks(unittest.TestCase):
def test_a_create_stream_1(self):
play_source = dict(
name = "Create Kinesis Stream with 10 Shards",
hosts = 'localhost',
gather_facts = 'no',
tasks = [
dict(
action=dict(
module='kinesis_stream',
args=dict(
name='stream-test',
shards=10,
wait='yes',
region=aws_region,
)
),
register='stream',
)
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
tqm, results = run(play)
self.failUnless(tqm._stats.ok['localhost'] == 1)
def test_a_create_stream_2(self):
play_source = dict(
name = "Create Kinesis Stream with 10 Shards and create a tag called environment",
hosts = 'localhost',
gather_facts = 'no',
tasks = [
dict(
action=dict(
module='kinesis_stream',
args=dict(
name='stream-test',
region=aws_region,
shards=10,
tags=dict(
env='development'
),
wait='yes'
)
),
register='stream'
)
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
tqm, results = run(play)
self.failUnless(tqm._stats.ok['localhost'] == 1)
def test_a_create_stream_3(self):
play_source = dict(
name = "Create Kinesis Stream with 10 Shards and create a tag called environment and Change the default retention period from 24 to 48",
hosts = 'localhost',
gather_facts = 'no',
tasks = [
dict(
action=dict(
module='kinesis_stream',
args=dict(
name='stream-test',
retention_period=48,
region=aws_region,
shards=10,
tags=dict(
env='development'
),
wait='yes'
)
),
register='stream'
)
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
tqm, results = run(play)
self.failUnless(tqm._stats.ok['localhost'] == 1)
def test_b_create_stream_1(self):
play_source = dict(
name = "Create Kinesis Stream with out specifying the number of shards",
hosts = 'localhost',
gather_facts = 'no',
tasks = [
dict(
action=dict(
module='kinesis_stream',
args=dict(
name='stream-test',
region=aws_region,
wait='yes'
)
),
register='stream'
)
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
tqm, results = run(play)
self.failUnless(tqm._stats.failures['localhost'] == 1)
def test_b_create_stream_2(self):
play_source = dict(
name = "Create Kinesis Stream with specifying the retention period less than 24 hours",
hosts = 'localhost',
gather_facts = 'no',
tasks = [
dict(
action=dict(
module='kinesis_stream',
args=dict(
name='stream-test',
region=aws_region,
retention_period=23,
shards=10,
wait='yes'
)
),
register='stream'
)
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
tqm, results = run(play)
self.failUnless(tqm._stats.failures['localhost'] == 1)
def test_c_delete_stream_(self):
play_source = dict(
name = "Delete Kinesis Stream test-stream",
hosts = 'localhost',
gather_facts = 'no',
tasks = [
dict(
action=dict(
module='kinesis_stream',
args=dict(
name='stream-test',
region=aws_region,
state='absent',
wait='yes'
)
)
)
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
tqm, results = run(play)
self.failUnless(tqm._stats.ok['localhost'] == 1)
class AnsibleKinesisStreamFunctions(unittest.TestCase):
@ -286,7 +80,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertEqual(aws_tags, should_return)
def test_get_tags(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg, tags = kinesis_stream.get_tags(client, 'test', True)
self.assertTrue(success)
should_return = [
@ -298,7 +92,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertEqual(tags, should_return)
def test_find_stream(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg, stream = (
kinesis_stream.find_stream(client, 'test', check_mode=True)
)
@ -313,7 +107,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertEqual(stream, should_return)
def test_wait_for_status(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg, stream = (
kinesis_stream.wait_for_status(
client, 'test', 'ACTIVE', check_mode=True
@ -330,7 +124,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertEqual(stream, should_return)
def test_tags_action_create(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
tags = {
'env': 'development',
'service': 'web'
@ -343,7 +137,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertTrue(success)
def test_tags_action_delete(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
tags = {
'env': 'development',
'service': 'web'
@ -356,7 +150,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertTrue(success)
def test_tags_action_invalid(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
tags = {
'env': 'development',
'service': 'web'
@ -369,7 +163,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertFalse(success)
def test_update_tags(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
tags = {
'env': 'development',
'service': 'web'
@ -382,7 +176,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertTrue(success)
def test_stream_action_create(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg = (
kinesis_stream.stream_action(
client, 'test', 10, 'create', check_mode=True
@ -391,7 +185,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertTrue(success)
def test_stream_action_delete(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg = (
kinesis_stream.stream_action(
client, 'test', 10, 'delete', check_mode=True
@ -400,7 +194,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertTrue(success)
def test_stream_action_invalid(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg = (
kinesis_stream.stream_action(
client, 'test', 10, 'append', check_mode=True
@ -409,7 +203,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertFalse(success)
def test_retention_action_increase(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg = (
kinesis_stream.retention_action(
client, 'test', 48, 'increase', check_mode=True
@ -418,7 +212,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertTrue(success)
def test_retention_action_decrease(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg = (
kinesis_stream.retention_action(
client, 'test', 24, 'decrease', check_mode=True
@ -427,7 +221,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertTrue(success)
def test_retention_action_invalid(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
success, err_msg = (
kinesis_stream.retention_action(
client, 'test', 24, 'create', check_mode=True
@ -436,7 +230,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertFalse(success)
def test_update(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
current_stream = {
'HasMoreShards': True,
'RetentionPeriodHours': 24,
@ -459,7 +253,7 @@ class AnsibleKinesisStreamFunctions(unittest.TestCase):
self.assertEqual(err_msg, 'Kinesis Stream test updated successfully.')
def test_create_stream(self):
client = boto3.client('kinesis', region_name='us-west-2')
client = boto3.client('kinesis', region_name=aws_region)
tags = {
'env': 'development',
'service': 'web'

Write Preview
Loading…
Cancel
Save