flame/flameclient/tests/test_flame.py

2560 lines
88 KiB
Python

# -*- coding: utf-8 -*-
# This software is released under the MIT License.
#
# Copyright (c) 2014 Cloudwatt
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import mock
from flameclient import client as flame_client
from flameclient import flame
from flameclient.tests import base
class FakeBase(object):
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
class FakeVolume(FakeBase):
id = 1234
size = 1
source_volid = None
bootable = 'false'
snapshot_id = None
display_name = 'vol1'
display_description = 'Description'
volume_type = 'fast'
metadata = None
class FakeUnnamedVolume(FakeBase):
id = 1234
size = 1
source_volid = None
bootable = 'false'
snapshot_id = None
volume_type = 'fast'
metadata = None
class FakeServer(FakeBase):
id = '1234'
name = 'server1'
config_drive = None
flavor = {'id': '2'}
image = {'id': '3333',
'links': [{'href': 'http://p/7777/images/3333',
'rel': 'bookmark'}]}
key_name = 'testkey'
addresses = []
metadata = None
def __init__(self, **kwargs):
kwargs.setdefault('OS-DCF:diskConfig', 'MANUAL')
kwargs.setdefault('os-extended-volumes:volumes_attached', [])
super(FakeServer, self).__init__(**kwargs)
class FakeFlavor(FakeBase):
name = 'm1.tiny'
id = '1'
class FakeKeypair(FakeBase):
name = 'key'
id = 'key'
public_key = 'ssh-rsa AAAAB3NzaC'
class FakeSecurityGroup(FakeBase):
id = '1'
name = 'name'
class FakeServerGroup(FakeBase):
name = 'policy_group'
id = '1234'
policies = 'affinity'
members = ['12345']
class FakeNeutronManager(object):
def __init__(self):
self.groups = []
self.routers = [{'name': 'myrouter',
'id': '1234',
'admin_state_up': 'true',
'external_gateway_info': None}, ]
self.ports = []
self.subnets = []
self.networks = [{'status': 'ACTIVE',
'subnets': ['1111'],
'name': 'mynetwork',
'router:external': False,
'admin_state_up': True,
'shared': False,
'id': '2222'}, ]
self.floatingips = []
def subnet_list(self):
return self.subnets
def network_list(self):
return self.networks
def port_list(self):
return self.ports
def router_list(self):
return self.routers
def router_interfaces_list(self, router):
return self.ports
def secgroup_list(self):
return self.groups
def floatingip_list(self):
return self.floatingips
class FakeNovaManager(object):
def __init__(self):
self.servers = [FakeServer()]
self.flavors = [FakeFlavor(id='2', name='m1.small')]
self.groups = {}
self.keypairs = [FakeKeypair(name='testkey',
public_key='ssh-rsa XXXX')]
self.servergroups = [FakeServerGroup()]
def keypair_list(self):
return self.keypairs
def flavor_list(self):
return self.flavors
def server_list(self):
return self.servers
def server_security_group_list(self, server):
return self.groups.get(server.name, [])
def servergroup_list(self):
return self.servergroups
class FakeCinderManager(object):
def __init__(self):
self.volumes = [FakeVolume(), ]
def volume_list(self):
return self.volumes
class ResourceTestCase(base.TestCase):
def test_template_resource(self):
resource = flame.Resource('my-name',
'my-type',
properties='my-properties')
expected = {
'my-name': {
'type': 'my-type',
'properties': 'my-properties',
}
}
self.assertEqual(expected, resource.template_resource)
def test_stack_resource(self):
resource = flame.Resource('my-name', 'my-type', id='my-id')
expected = {
'my-name': {
'status': 'COMPLETE',
'name': 'my-name',
'resource_data': {},
'resource_id': 'my-id',
'type': 'my-type',
'metadata': {},
'action': 'CREATE'
}
}
self.assertEqual(expected, resource.stack_resource)
def test_empty_stack_resource(self):
resource = flame.Resource('my-name', 'my-type')
self.assertEqual({}, resource.stack_resource)
def test_add_parameter(self):
resource = flame.Resource('my-name', 'my-type')
resource.add_parameter('my-parameter', 'my-description')
expected = {
'my-parameter': {
'type': 'string',
'description': 'my-description'
}
}
self.assertEqual(expected, resource.template_parameter)
def test_add_parameter_with_type(self):
resource = flame.Resource('my-name', 'my-type')
resource.add_parameter('my-parameter', 'my-description',
parameter_type='my-type')
expected = {
'my-parameter': {
'type': 'my-type',
'description': 'my-description'
}
}
self.assertEqual(expected, resource.template_parameter)
def test_add_parameter_with_default(self):
resource = flame.Resource('my-name', 'my-type')
resource.add_parameter('my-parameter', 'my-description',
default='my-default')
expected = {
'my-parameter': {
'type': 'string',
'description': 'my-description',
'default': 'my-default'
}
}
self.assertEqual(expected, resource.template_parameter)
class BaseTestCase(base.TestCase):
def setUp(self):
super(BaseTestCase, self).setUp()
self.patch_neutron = mock.patch('flameclient.managers.NeutronManager')
self.mock_neutron = self.patch_neutron.start()
self.patch_nova = mock.patch('flameclient.managers.NovaManager')
self.mock_nova = self.patch_nova.start()
self.patch_cinder = mock.patch('flameclient.managers.CinderManager')
self.mock_cinder = self.patch_cinder.start()
self.patch_keystone = mock.patch(
'flameclient.managers.KeystoneManager'
)
self.mock_keystone = self.patch_keystone.start()
def tearDown(self):
self.mock_neutron.stop()
self.mock_nova.stop()
self.mock_cinder.stop()
self.mock_keystone.stop()
super(BaseTestCase, self).tearDown()
def get_generator(self, exclude_servers, exclude_volumes,
exclude_keypairs, generate_data):
generator = flame.TemplateGenerator('x', 'x', 'x', 'x', True,
'publicURL')
generator.extract_vm_details(exclude_servers, exclude_volumes,
exclude_keypairs, generate_data)
return generator
def check_stackdata(self, resources, expected_resources):
merged_resources = {}
for resource in resources:
merged_resources.update(resource.stack_resource)
self.assertEqual(expected_resources, merged_resources)
def check_template(self, resources, expected_resources,
expected_parameters=None):
expected_parameters = expected_parameters or {}
merged_resources = {}
merged_parameters = {}
for resource in resources:
merged_resources.update(resource.template_resource)
merged_parameters.update(resource.template_parameter)
self.assertEqual(expected_resources, merged_resources)
self.assertEqual(expected_parameters, merged_parameters)
class ClientTest(BaseTestCase):
def setUp(self):
super(ClientTest, self).setUp()
self.c = flame_client.Client('username', 'password', 'tenant_name',
'authUrl', 'auth_token')
def test_generate(self):
out = self.c.generate(False, False, False, True)
self.assertIsInstance(out, tuple)
self.assertIsNotNone(out[1])
def test_generate_no_stack_data(self):
out = self.c.generate(False, False, False, False)
self.assertIsInstance(out, tuple)
self.assertIsNotNone(out[0])
self.assertIsNone(out[1])
def test_generate_contains_extract(self):
generator = self.get_generator(False, False, False, True)
out = self.c.generate(False, False, False, True)
generator.extract_data()
stack_data = generator.stack_data_template()
heat_template = generator.heat_template()
self.assertEqual(out, (heat_template, stack_data))
class StackDataTests(BaseTestCase):
def test_keypair(self):
self.mock_nova.return_value = FakeNovaManager()
generator = self.get_generator(False, False, False, True)
expected = {
'key_0': {
'type': 'OS::Nova::KeyPair',
'action': 'CREATE',
'metadata': {},
'name': 'key_0',
'resource_data': {},
'resource_id': 'key',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_keys(), expected)
def test_router(self):
self.mock_neutron.return_value = FakeNeutronManager()
generator = self.get_generator(False, False, False, True)
expected = {
'router_0': {
'type': 'OS::Neutron::Router',
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_routers(), expected)
def test_router_with_external_gateway(self):
fake = FakeNeutronManager()
fake.routers = [{'name': 'myrouter',
'id': '1234',
'admin_state_up': 'true',
'external_gateway_info': {
'network_id': '8765',
'enable_snat': 'true'}}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected = {
'router_0': {
'type': 'OS::Neutron::Router',
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE'
},
'router_0_gateway': {
'type': 'OS::Neutron::RouterGateway',
'action': 'CREATE',
'metadata': {},
'name': 'router_0_gateway',
'resource_data': {},
'resource_id': '1234:8765',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_routers(), expected)
def test_router_with_ports(self):
fake = FakeNeutronManager()
fake.ports = [{'status': 'ACTIVE',
'name': '',
'allowed_address_pairs': [],
'admin_state_up': True,
'network_id': '4444',
'extra_dhcp_opts': [],
'binding:vnic_type': 'normal',
'device_owner': 'network:router_interface',
'mac_address': 'fa:16:3e:4b:8c:98',
'fixed_ips': [{'subnet_id': '1111',
'ip_address': '10.123.2.3'}],
'id': '1234567',
'security_groups': [],
'device_id': '1234'}, ]
fake.subnets = [{'name': 'subnet_1111',
'enable_dhcp': True,
'network_id': '1234',
'dns_nameservers': [],
'allocation_pools': [{'start': '10.123.2.2',
'end': '10.123.2.30'}],
'host_routes': [],
'ip_version': 4,
'gateway_ip': '10.123.2.1',
'cidr': '10.123.2.0/27',
'id': '1111'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected = {
'router_0': {
'type': 'OS::Neutron::Router',
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE'
},
'router_0_interface_0': {
'type': 'OS::Neutron::RouterInterface',
'action': 'CREATE',
'metadata': {},
'name': 'router_0_interface_0',
'resource_data': {},
'resource_id': '1234:subnet_id=1111',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_routers(), expected)
def test_network(self):
self.mock_neutron.return_value = FakeNeutronManager()
generator = self.get_generator(False, False, False, True)
expected = {
'network_0': {
'type': 'OS::Neutron::Net',
'action': 'CREATE',
'metadata': {},
'name': 'network_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_networks(), expected)
def test_external_network(self):
fake = FakeNeutronManager()
fake.networks[0]['router:external'] = True
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
self.check_stackdata(generator._extract_networks(), {})
def test_subnet(self):
fake = FakeNeutronManager()
fake.subnets = [{'name': 'subnet_1111',
'enable_dhcp': True,
'network_id': '2222',
'dns_nameservers': [],
'allocation_pools': [{'start': '10.123.2.2',
'end': '10.123.2.30'}],
'host_routes': [],
'ip_version': 4,
'gateway_ip': '10.123.2.1',
'cidr': '10.123.2.0/27',
'id': '1111'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected = {
'subnet_0': {
'type': 'OS::Neutron::Subnet',
'action': 'CREATE',
'metadata': {},
'name': 'subnet_0',
'resource_data': {},
'resource_id': '1111',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_subnets(), expected)
def test_floatingip(self):
fake = FakeNeutronManager()
fake.floatingips = [{'router_id': '1111',
'status': 'ACTIVE',
'floating_network_id': '1234',
'fixed_ip_address': '10.0.48.251',
'floating_ip_address': '84.39.33.60',
'port_id': '4321',
'id': '2222'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(True, False, False, True)
expected = {
'floatingip_0': {
'type': 'OS::Neutron::FloatingIP',
'action': 'CREATE',
'metadata': {},
'name': 'floatingip_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_floating(), expected)
def test_security_group(self):
rules = [{'remote_group_id': None,
'direction': 'ingress',
'remote_ip_prefix': '0.0.0.0/0',
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': 22,
'port_range_min': 22,
'id': '8901',
'security_group_id': '1234'}, ]
fake = FakeNeutronManager()
fake.groups = [{'tenant_id': '7777',
'name': 'somename',
'description': 'description',
'security_group_rules': rules,
'id': '1234'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected = {
'security_group_0': {
'type': 'OS::Neutron::SecurityGroup',
'action': 'CREATE',
'metadata': {},
'name': 'security_group_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_secgroups(), expected)
def test_default_security_group(self):
rules = [{'remote_group_id': None,
'direction': 'ingress',
'remote_ip_prefix': '0.0.0.0/0',
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': 22,
'port_range_min': 22,
'id': '8901',
'security_group_id': '1234'}, ]
fake = FakeNeutronManager()
fake.groups = [{'tenant_id': '7777',
'name': 'default',
'description': 'default',
'security_group_rules': rules,
'id': '1234'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
self.check_stackdata(generator._extract_secgroups(), {})
def test_volume(self):
self.mock_cinder.return_value = FakeCinderManager()
generator = self.get_generator(False, False, False, True)
expected = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'action': 'CREATE',
'metadata': {},
'name': 'volume_0',
'resource_data': {},
'resource_id': 1234,
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_volumes(), expected)
def test_server(self):
self.mock_nova.return_value = FakeNovaManager()
generator = self.get_generator(False, False, False, True)
expected = {
'server_0': {
'type': 'OS::Nova::Server',
'action': 'CREATE',
'metadata': {},
'name': 'server_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_servers(), expected)
def test_server_with_default_security_group(self):
fake_neutron = FakeNeutronManager()
fake_nova = FakeNovaManager()
fake_neutron.groups = [{"name": "default",
"id": "1",
"security_group_rules": [],
"description": "default"}, ]
fake_nova.groups = {'server1': [FakeSecurityGroup(
name='default', description='default')]}
self.mock_neutron.return_value = fake_neutron
self.mock_nova.return_value = fake_nova
generator = self.get_generator(False, False, False, True)
expected = {
'server_0': {
'type': 'OS::Nova::Server',
'action': 'CREATE',
'metadata': {},
'name': 'server_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE'
}
}
self.check_stackdata(generator._extract_servers(), expected)
def test_servergroup(self):
self.mock_nova.return_value = FakeNovaManager()
generator = self.get_generator(False, False, False, True)
expected = {
'servergroup_0': {
'action': 'CREATE',
'metadata': {},
'name': 'servergroup_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::ServerGroup'
}
}
self.check_stackdata(generator._extract_servergroups(), expected)
class NetworkTests(BaseTestCase):
def test_keypair(self):
self.mock_nova.return_value = FakeNovaManager()
generator = self.get_generator(False, False, False, True)
expected = {
'key_0': {
'type': 'OS::Nova::KeyPair',
'properties': {
'public_key': 'ssh-rsa XXXX',
'name': 'testkey'
}
}
}
self.check_template(generator._extract_keys(), expected)
def test_router(self):
self.mock_neutron.return_value = FakeNeutronManager()
generator = self.get_generator(False, False, False, True)
expected = {
'router_0': {
'type': 'OS::Neutron::Router',
'properties': {
'name': 'myrouter',
'admin_state_up': 'true',
}
}
}
self.check_template(generator._extract_routers(), expected)
def test_router_with_external_gateway(self):
fake = FakeNeutronManager()
fake.routers = [{'name': 'myrouter',
'id': '1234',
'admin_state_up': 'true',
'external_gateway_info': {
'network_id': '8765',
'enable_snat': 'true'}}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'router_0_external_network': {
'default': '8765',
'type': 'string',
'description': 'Router external network'
}
}
expected_resources = {
'router_0': {
'type': 'OS::Neutron::Router',
'properties': {
'name': 'myrouter',
'admin_state_up': 'true',
}
},
'router_0_gateway': {
'type': 'OS::Neutron::RouterGateway',
'properties': {
'router_id': {'get_resource': 'router_0'},
'network_id': {
'get_param': 'router_0_external_network'
}
}
}
}
self.check_template(generator._extract_routers(), expected_resources,
expected_parameters)
def test_router_with_ports(self):
fake = FakeNeutronManager()
fake.ports = [{'status': 'ACTIVE',
'name': '',
'allowed_address_pairs': [],
'admin_state_up': True,
'network_id': '4444',
'extra_dhcp_opts': [],
'binding:vnic_type': 'normal',
'device_owner': 'network:router_interface',
'mac_address': 'fa:16:3e:4b:8c:98',
'fixed_ips': [{'subnet_id': '1111',
'ip_address': '10.123.2.3'}],
'id': '1234567',
'security_groups': [],
'device_id': '1234'}, ]
fake.subnets = [{'name': 'subnet_1111',
'enable_dhcp': True,
'network_id': '1234',
'dns_nameservers': [],
'allocation_pools': [{'start': '10.123.2.2',
'end': '10.123.2.30'}],
'host_routes': [],
'ip_version': 4,
'gateway_ip': '10.123.2.1',
'cidr': '10.123.2.0/27',
'id': '1111'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected = {
'router_0': {
'type': 'OS::Neutron::Router',
'properties': {
'name': 'myrouter',
'admin_state_up': 'true',
}
},
'router_0_interface_0': {
'type': 'OS::Neutron::RouterInterface',
'properties': {
'subnet_id': {'get_resource': 'subnet_0'},
'router_id': {'get_resource': 'router_0'}
}
}
}
self.check_template(generator._extract_routers(), expected)
def test_network(self):
self.mock_neutron.return_value = FakeNeutronManager()
generator = self.get_generator(False, False, False, True)
expected = {
'network_0': {
'type': 'OS::Neutron::Net',
'properties': {
'shared': False,
'name': 'mynetwork',
'admin_state_up': True
}
}
}
self.check_template(generator._extract_networks(), expected)
def test_external_network(self):
fake = FakeNeutronManager()
fake.networks[0]['router:external'] = True
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
self.check_template(generator._extract_networks(), {})
def test_subnet(self):
fake = FakeNeutronManager()
fake.subnets = [{'name': 'subnet_1111',
'enable_dhcp': True,
'network_id': '2222',
'dns_nameservers': [],
'allocation_pools': [{'start': '10.123.2.2',
'end': '10.123.2.30'}],
'host_routes': [],
'ip_version': 4,
'gateway_ip': '10.123.2.1',
'cidr': '10.123.2.0/27',
'id': '1111'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected = {
'subnet_0': {
'type': 'OS::Neutron::Subnet',
'properties': {
'network_id': {'get_resource': 'network_0'},
'allocation_pools': [{'start': '10.123.2.2',
'end': '10.123.2.30'}],
'host_routes': [],
'name': 'subnet_1111',
'enable_dhcp': True,
'ip_version': 4,
'cidr': '10.123.2.0/27',
'dns_nameservers': []
}
}
}
self.check_template(generator._extract_subnets(), expected)
def test_floatingip(self):
fake = FakeNeutronManager()
fake.floatingips = [{'router_id': '1111',
'status': 'ACTIVE',
'floating_network_id': '1234',
'fixed_ip_address': '10.0.48.251',
'floating_ip_address': '84.39.33.60',
'port_id': '4321',
'id': '2222'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(True, False, False, False)
expected_parameters = {
'external_network_for_floating_ip_0': {
'default': '1234',
'type': 'string',
'description': 'Network to allocate floating IP from'
}
}
expected_resources = {
'floatingip_0': {
'type': 'OS::Neutron::FloatingIP',
'properties': {
'floating_network_id': {
'get_param': 'external_network_for_floating_ip_0'
}
}
}
}
self.check_template(generator._extract_floating(), expected_resources,
expected_parameters)
def test_security_group(self):
rules = [
{
'remote_group_id': '1234',
'direction': 'ingress',
'remote_ip_prefix': None,
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': 65535,
'port_range_min': 1,
'id': '5678',
'security_group_id': '1234'
},
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '6789',
'security_group_id': '1234'
},
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv6',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '7890',
'security_group_id': '1234'
},
{
'remote_group_id': None,
'direction': 'ingress',
'remote_ip_prefix': '0.0.0.0/0',
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': 22,
'port_range_min': 22,
'id': '8901',
'security_group_id': '1234'
},
]
fake = FakeNeutronManager()
fake.groups = [{'tenant_id': '7777',
'name': 'toto',
'description': 'description',
'security_group_rules': rules,
'id': '1234'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, False)
expected = {
'security_group_0': {
'type': 'OS::Neutron::SecurityGroup',
'properties': {
'rules': [
{
'direction': 'ingress',
'protocol': 'tcp',
'ethertype': 'IPv4',
'port_range_max': 65535,
'port_range_min': 1,
'remote_mode': 'remote_group_id'
},
{
'ethertype': 'IPv4',
'direction': 'egress'
},
{
'ethertype': 'IPv6',
'direction': 'egress'
},
{
'direction': 'ingress',
'protocol': 'tcp',
'ethertype': 'IPv4',
'port_range_max': 22,
'port_range_min': 22,
'remote_ip_prefix': '0.0.0.0/0'
}
],
'description': 'description',
'name': 'toto'
}
}
}
self.check_template(generator._extract_secgroups(), expected)
def test_security_group_default(self):
rules = [
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv6',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '1234',
'security_group_id': '1111'
},
{
'remote_group_id': '1111',
'direction': 'ingress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv6',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '2345',
'security_group_id': '1111'
},
{
'remote_group_id': None,
'direction': 'ingress',
'remote_ip_prefix': '0.0.0.0/0',
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': 22,
'port_range_min': 22,
'id': '3456',
'security_group_id': '1111'
},
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '4567',
'security_group_id': '1111'
},
{
'remote_group_id': '1111',
'direction': 'ingress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '5678',
'security_group_id': '1111'
}
]
fake = FakeNeutronManager()
fake.groups = [{'tenant_id': '7777',
'name': 'default',
'description': 'default',
'security_group_rules': rules,
'id': '1111'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, False)
expected = {
'security_group_0': {
'type': 'OS::Neutron::SecurityGroup',
'properties': {
'rules': [
{
'ethertype': 'IPv6',
'direction': 'egress'
},
{
'ethertype': 'IPv6',
'direction': 'ingress',
'remote_mode': 'remote_group_id'
},
{
'direction': 'ingress',
'protocol': 'tcp',
'ethertype': 'IPv4',
'port_range_max': 22,
'port_range_min': 22,
'remote_ip_prefix': '0.0.0.0/0'
},
{
'ethertype': 'IPv4',
'direction': 'egress'
},
{
'ethertype': 'IPv4',
'direction': 'ingress',
'remote_mode': 'remote_group_id'
}
],
'description': 'default',
'name': '_default'
}
}
}
self.check_template(generator._extract_secgroups(), expected)
def test_security_groups(self):
rules1 = [
{
'remote_group_id': '2222',
'direction': 'ingress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '01234',
'security_group_id': '1111'
},
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv6',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '1234',
'security_group_id': '1111'
},
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '2345',
'security_group_id': '1111'
},
{
'remote_group_id': '2222',
'direction': 'ingress',
'remote_ip_prefix': None,
'protocol': 'icmp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '3456',
'security_group_id': '1111'
}
]
rules2 = [
{
'remote_group_id': '1111',
'direction': 'ingress',
'remote_ip_prefix': None,
'protocol': 'udp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': 8888,
'port_range_min': 7777,
'id': '4567',
'security_group_id': '2222'
},
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv6',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '5678',
'security_group_id': '2222'
},
{
'remote_group_id': None,
'direction': 'egress',
'remote_ip_prefix': None,
'protocol': None,
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': None,
'port_range_min': None,
'id': '6789',
'security_group_id': '2222'
},
{
'remote_group_id': '1111',
'direction': 'ingress',
'remote_ip_prefix': None,
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': '7777',
'port_range_max': 65535,
'port_range_min': 1,
'id': '7890',
'security_group_id': '2222'
}
]
fake = FakeNeutronManager()
fake.groups = [{'tenant_id': '7777',
'name': 'security_group_1',
'description': 'security_group_1',
'security_group_rules': rules1,
'id': '1111'},
{'tenant_id': '7777',
'name': 'security_group_2',
'description': 'security_group_2',
'security_group_rules': rules2,
'id': '2222'}, ]
self.mock_neutron.return_value = fake
generator = self.get_generator(False, False, False, True)
expected = {
'security_group_0': {
'type': 'OS::Neutron::SecurityGroup',
'properties': {
'rules': [
{
'remote_group_id': {
'get_resource': 'security_group_1'
},
'direction': 'ingress',
'ethertype': 'IPv4',
'remote_mode': 'remote_group_id'
},
{
'ethertype': 'IPv6',
'direction': 'egress'
},
{
'ethertype': 'IPv4',
'direction': 'egress'
},
{
'remote_group_id': {
'get_resource': 'security_group_1'
},
'direction': 'ingress',
'protocol': 'icmp',
'ethertype': 'IPv4',
'remote_mode': 'remote_group_id'
}
],
'description': 'security_group_1',
'name': 'security_group_1'
}
},
'security_group_1': {
'type': 'OS::Neutron::SecurityGroup',
'properties': {
'rules': [
{
'remote_group_id': {
'get_resource': 'security_group_0'
},
'direction': 'ingress',
'protocol': 'udp',
'ethertype': 'IPv4',
'port_range_max': 8888,
'port_range_min': 7777,
'remote_mode': 'remote_group_id'
},
{
'ethertype': 'IPv6',
'direction': 'egress'
},
{
'ethertype': 'IPv4',
'direction': 'egress'
},
{
'remote_group_id': {
'get_resource': 'security_group_0'
},
'direction': 'ingress',
'protocol': 'tcp',
'ethertype': 'IPv4',
'port_range_max': 65535,
'port_range_min': 1,
'remote_mode': 'remote_group_id'
}
],
'description': 'security_group_2',
'name': 'security_group_2'
}
}
}
self.check_template(generator._extract_secgroups(), expected)
class VolumeTests(BaseTestCase):
def setUp(self):
super(VolumeTests, self).setUp()
self.fake = FakeCinderManager()
self.mock_cinder.return_value = self.fake
def test_basic(self):
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
def test_basic_unnamed(self):
self.fake.volumes = [FakeUnnamedVolume(), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
def test_source_volid_external(self):
self.fake.volumes = [FakeVolume(source_volid=5678), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_source_volid': {
'description': 'Volume to create volume volume_0 from',
'type': 'string'
},
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'source_volid': {'get_param': 'volume_0_source_volid'},
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
def test_source_volid_included(self):
self.fake.volumes = [FakeVolume(source_volid=5678),
FakeVolume(id=5678)]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'
},
'volume_1_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_1',
'type': 'string'
}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'source_volid': {'get_resource': 'volume_1'},
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
},
'volume_1': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'volume_type': {'get_param': 'volume_1_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
def test_image(self):
metadata = {
'kernel_id': '9817',
'container_format': 'bare',
'min_ram': '0',
'ramdisk_id': '4ec7',
'disk_format': 'qcow2',
'image_name': 'cirros-0.3.1-x86_64-uec',
'image_id': '5c5c',
'checksum': 'f8a2e',
'min_disk': '0',
'size': '25'}
self.fake.volumes = [FakeVolume(bootable='true',
volume_image_metadata=metadata), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'
},
'volume_0_image': {
'default': '5c5c',
'description': 'Image to create volume volume_0 from',
'type': 'string'
}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'image': {'get_param': 'volume_0_image'},
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
def test_snapshot_id(self):
self.fake.volumes = [FakeVolume(snapshot_id=5678), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_snapshot_id': {
'default': 5678,
'description': 'Snapshot to create volume volume_0 from',
'type': 'string'
},
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'
}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'snapshot_id': {'get_param': 'volume_0_snapshot_id'},
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
def test_volume_type(self):
self.fake.volumes = [FakeVolume(volume_type='isci'), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_volume_type': {
'description': 'Volume type for volume volume_0',
'default': 'isci',
'type': 'string'
}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
def test_metadata(self):
self.fake.volumes = [FakeVolume(metadata={'key': 'value'}), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'
}
}
expected_resources = {
'volume_0': {
'type': 'OS::Cinder::Volume',
'properties': {
'name': 'vol1',
'description': 'Description',
'metadata': {'key': 'value'},
'volume_type': {'get_param': 'volume_0_volume_type'},
'size': 1
}
}
}
self.check_template(generator._extract_volumes(), expected_resources,
expected_parameters)
class ServerTests(BaseTestCase):
def setUp(self):
super(ServerTests, self).setUp()
self.fake = FakeNovaManager()
self.mock_nova.return_value = self.fake
def test_basic(self):
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'}
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_keypair(self):
self.fake.servers = [FakeServer(key_name='testkey')]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'key_name': {'get_resource': 'key_0'},
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'}
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_boot_from_volume(self):
attachments = [{'device': 'vda',
'server_id': '777',
'id': '5678',
'host_name': None,
'volume_id': '5678'}, ]
servers_args = {"id": 777,
"image": None,
"os-extended-volumes:volumes_attached": [{'id': 5678}]}
self.fake.servers = [FakeServer(**servers_args), ]
fake_cinder = FakeCinderManager()
fake_cinder.volumes = [FakeVolume(id=5678,
attachments=attachments,
bootable='true')]
self.mock_cinder.return_value = fake_cinder
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'key_name': {'get_resource': 'key_0'},
'block_device_mapping_v2': [{'volume_id': {
'get_resource': 'volume_0'}, 'device_name': 'vda'}]
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_volume_attached(self):
attachments = [{'device': '/dev/vdb',
'server_id': '777',
'id': '5678',
'host_name': None,
'volume_id': '5678'}, ]
fake_cinder = FakeCinderManager()
fake_cinder.volumes = [FakeVolume(id=5678, attachments=attachments,
bootable='false'), ]
self.mock_cinder.return_value = fake_cinder
server_args = {"id": 777,
"os-extended-volumes:volumes_attached": [{'id': 5678}]}
self.fake.servers = [FakeServer(**server_args), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'},
'block_device_mapping_v2': [{'volume_id': {
'get_resource': 'volume_0'}, 'device_name':
'/dev/vdb'}]
}
},
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_security_groups(self):
fake_neutron = FakeNeutronManager()
fake_neutron.groups = [{"name": "group1",
"id": "1",
"security_group_rules": [],
"description": "Group"}, ]
self.mock_neutron.return_value = fake_neutron
self.fake.groups = {'server1': [FakeSecurityGroup(), ]}
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'security_groups': [
{
'get_resource': 'security_group_0'
}
],
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'}
}
}
}
generator._extract_secgroups()
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_config_drive(self):
self.fake.servers = [FakeServer(config_drive="True"), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'config_drive': 'True',
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'}
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_metadata(self):
self.fake.servers = [FakeServer(metadata={"key": "value"}), ]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'metadata': {'key': 'value'},
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'}
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_networks(self):
subnet = {
"id": "4321",
"name": "private_subnet",
"network_id": "1234",
"allocation_pools": {"start": "10.0.0.2", "end": "10.0.0.254"},
"cidr": "10.0.0.0/24",
"dns_nameservers": [],
"enable_dhcp": True,
"host_routes": [],
"ip_version": 4}
fake_neutron = FakeNeutronManager()
fake_neutron.subnets = [subnet]
fake_neutron.networks = [{"id": "1234", "name": "private"}, ]
self.mock_neutron.return_value = fake_neutron
addresses = {"private": [{"addr": "10.0.0.2"}]}
self.fake.servers = [FakeServer(addresses=addresses)]
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'networks': [
{'network': {'get_resource': 'network_0'}}],
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'}
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_excluded_volume_attached(self):
attachments = [{'device': '/dev/vdb',
'server_id': '777',
'id': '5678',
'host_name': None,
'volume_id': '5678'}]
fake_cinder = FakeCinderManager()
fake_cinder.volumes = [FakeVolume(id=5678, attachments=attachments,
bootable='false'), ]
self.mock_cinder.return_value = fake_cinder
server_args = {"id": 777,
"os-extended-volumes:volumes_attached": [{'id': 5678}]}
self.fake.servers = [FakeServer(**server_args), ]
generator = self.get_generator(False, True, False, False)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
},
'volume_server1_0': {
'default': 5678,
'description': 'Volume for server server1, device '
'/dev/vdb',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'},
'block_device_mapping_v2': [{'volume_id': {
'get_param': 'volume_server1_0'}, 'device_name':
'/dev/vdb'}]
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
def test_servergroup(self):
self.fake.servers = [FakeServer()]
self.fake.servers[0].id = '12345'
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'description': 'Image to use to boot server server_0',
'default': '3333',
'type': 'string'
}
}
expected_resources = {
'server_0': {
'type': 'OS::Nova::Server',
'properties': {
'name': 'server1',
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'},
'scheduler_hints': {'group':
{'get_resource': 'servergroup_0'}}
}
}
}
self.check_template(generator._extract_servers(), expected_resources,
expected_parameters)
class GenerationTests(BaseTestCase):
def setUp(self):
super(GenerationTests, self).setUp()
self.mock_neutron.return_value = FakeNeutronManager()
self.mock_nova.return_value = FakeNovaManager()
self.mock_cinder.return_value = FakeCinderManager()
def test_generation(self):
generator = self.get_generator(False, False, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'default': '3333',
'description': 'Image to use to boot server server_0',
'type': 'string'
},
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'
}
}
expected_resources = {
'key_0': {
'properties': {
'name': 'testkey',
'public_key': 'ssh-rsa XXXX'
},
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'properties': {
'admin_state_up': True,
'name': 'mynetwork',
'shared': False
},
'type': 'OS::Neutron::Net'
},
'router_0': {
'properties': {
'admin_state_up': 'true',
'name': 'myrouter'
},
'type': 'OS::Neutron::Router'
},
'server_0': {
'properties': {
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'},
'name': 'server1'
},
'type': 'OS::Nova::Server'
},
'servergroup_0': {
'properties': {
'name': 'policy_group',
'policies': 'affinity'},
'type': 'OS::Nova::ServerGroup'
},
'volume_0': {
'properties': {
'description': 'Description',
'name': 'vol1',
'size': 1,
'volume_type': {'get_param': 'volume_0_volume_type'}
},
'type': 'OS::Cinder::Volume'
}
}
expected_data = {
'key_0': {
'action': 'CREATE',
'metadata': {},
'name': 'key_0',
'resource_data': {},
'resource_id': 'key',
'status': 'COMPLETE',
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'action': 'CREATE',
'metadata': {},
'name': 'network_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE',
'type': 'OS::Neutron::Net'
},
'router_0': {
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Neutron::Router'
},
'server_0': {
'action': 'CREATE',
'metadata': {},
'name': 'server_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::Server'
},
'servergroup_0': {
'action': 'CREATE',
'metadata': {},
'name': 'servergroup_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::ServerGroup'
},
'volume_0': {
'action': 'CREATE',
'metadata': {},
'name': 'volume_0',
'resource_data': {},
'resource_id': 1234,
'status': 'COMPLETE',
'type': 'OS::Cinder::Volume'
}
}
generator.extract_data()
self.assertEqual(generator.template['resources'], expected_resources)
self.assertEqual(generator.template['parameters'], expected_parameters)
self.assertEqual(generator.stack_data['resources'], expected_data)
def test_generation_exclude_servers(self):
generator = self.get_generator(True, False, False, True)
expected_parameters = {
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'
}
}
expected_resources = {
'key_0': {
'properties': {
'name': 'testkey',
'public_key': 'ssh-rsa XXXX'
},
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'properties': {
'admin_state_up': True,
'name': 'mynetwork',
'shared': False
},
'type': 'OS::Neutron::Net'
},
'router_0': {
'properties': {
'admin_state_up': 'true',
'name': 'myrouter'
},
'type': 'OS::Neutron::Router'
},
'servergroup_0': {
'properties': {
'name': 'policy_group',
'policies': 'affinity'
},
'type': 'OS::Nova::ServerGroup'
},
'volume_0': {
'properties': {
'description': 'Description',
'name': 'vol1',
'size': 1,
'volume_type': {'get_param': 'volume_0_volume_type'}
},
'type': 'OS::Cinder::Volume'
}
}
expected_data = {
'key_0': {
'action': 'CREATE',
'metadata': {},
'name': 'key_0',
'resource_data': {},
'resource_id': 'key',
'status': 'COMPLETE',
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'action': 'CREATE',
'metadata': {},
'name': 'network_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE',
'type': 'OS::Neutron::Net'
},
'router_0': {
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Neutron::Router'
},
'servergroup_0': {
'action': 'CREATE',
'metadata': {},
'name': 'servergroup_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::ServerGroup'
},
'volume_0': {
'action': 'CREATE',
'metadata': {},
'name': 'volume_0',
'resource_data': {},
'resource_id': 1234,
'status': 'COMPLETE',
'type': 'OS::Cinder::Volume'
}
}
generator.extract_data()
self.assertEqual(generator.template['resources'], expected_resources)
self.assertEqual(generator.template['parameters'], expected_parameters)
self.assertEqual(generator.stack_data['resources'], expected_data)
def test_generation_exclude_volumes(self):
generator = self.get_generator(False, True, False, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'default': '3333',
'description': 'Image to use to boot server server_0',
'type': 'string'
}
}
expected_resources = {
'key_0': {
'properties': {
'name': 'testkey',
'public_key': 'ssh-rsa XXXX'
},
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'properties': {
'admin_state_up': True,
'name': 'mynetwork',
'shared': False
},
'type': 'OS::Neutron::Net'
},
'router_0': {
'properties': {
'admin_state_up': 'true',
'name': 'myrouter'
},
'type': 'OS::Neutron::Router'
},
'server_0': {
'properties': {
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_resource': 'key_0'},
'name': 'server1'
},
'type': 'OS::Nova::Server'
},
'servergroup_0': {
'properties': {
'name': 'policy_group',
'policies': 'affinity'},
'type': 'OS::Nova::ServerGroup'
}
}
expected_data = {
'key_0': {
'action': 'CREATE',
'metadata': {},
'name': 'key_0',
'resource_data': {},
'resource_id': 'key',
'status': 'COMPLETE',
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'action': 'CREATE',
'metadata': {},
'name': 'network_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE',
'type': 'OS::Neutron::Net'
},
'router_0': {
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Neutron::Router'
},
'server_0': {
'action': 'CREATE',
'metadata': {},
'name': 'server_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::Server'
},
'servergroup_0': {
'action': 'CREATE',
'metadata': {},
'name': 'servergroup_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::ServerGroup'
}
}
generator.extract_data()
self.assertEqual(generator.template['resources'], expected_resources)
self.assertEqual(generator.template['parameters'], expected_parameters)
self.assertEqual(generator.stack_data['resources'], expected_data)
def test_generation_exclude_keypairs(self):
generator = self.get_generator(False, False, True, True)
expected_parameters = {
'server_0_flavor': {
'default': 'm1.small',
'description': 'Flavor to use for server server_0',
'type': 'string'
},
'server_0_image': {
'default': '3333',
'description': 'Image to use to boot server server_0',
'type': 'string'
},
'server_0_key': {
'default': 'testkey',
'description': 'Key for server server_0',
'type': 'string'
},
'volume_0_volume_type': {
'default': 'fast',
'description': 'Volume type for volume volume_0',
'type': 'string'
}
}
expected_resources = {
'network_0': {
'properties': {
'admin_state_up': True,
'name': 'mynetwork',
'shared': False
},
'type': 'OS::Neutron::Net'
},
'router_0': {
'properties': {
'admin_state_up': 'true',
'name': 'myrouter'
},
'type': 'OS::Neutron::Router'
},
'server_0': {
'properties': {
'diskConfig': 'MANUAL',
'flavor': {'get_param': 'server_0_flavor'},
'image': {'get_param': 'server_0_image'},
'key_name': {'get_param': 'server_0_key'},
'name': 'server1'
},
'type': 'OS::Nova::Server'
},
'servergroup_0': {
'properties': {
'name': 'policy_group',
'policies': 'affinity'
},
'type': 'OS::Nova::ServerGroup'
},
'volume_0': {
'properties': {
'description': 'Description',
'name': 'vol1',
'size': 1,
'volume_type': {'get_param': 'volume_0_volume_type'}
},
'type': 'OS::Cinder::Volume'
}
}
expected_data = {
'network_0': {
'action': 'CREATE',
'metadata': {},
'name': 'network_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE',
'type': 'OS::Neutron::Net'
},
'router_0': {
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Neutron::Router'
},
'server_0': {
'action': 'CREATE',
'metadata': {},
'name': 'server_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::Server'
},
'servergroup_0': {
'action': 'CREATE',
'metadata': {},
'name': 'servergroup_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::ServerGroup'
},
'volume_0': {
'action': 'CREATE',
'metadata': {},
'name': 'volume_0',
'resource_data': {},
'resource_id': 1234,
'status': 'COMPLETE',
'type': 'OS::Cinder::Volume'
}
}
generator.extract_data()
self.assertEqual(generator.template['resources'], expected_resources)
self.assertEqual(generator.template['parameters'], expected_parameters)
self.assertEqual(generator.stack_data['resources'], expected_data)
def test_generation_exclude_servers_and_volumes(self):
generator = self.get_generator(True, True, False, True)
expected_parameters = {}
expected_resources = {
'key_0': {
'properties': {
'name': 'testkey',
'public_key': 'ssh-rsa XXXX'
},
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'properties': {
'admin_state_up': True,
'name': 'mynetwork',
'shared': False
},
'type': 'OS::Neutron::Net'
},
'router_0': {
'properties': {
'admin_state_up': 'true',
'name': 'myrouter'
},
'type': 'OS::Neutron::Router'
},
'servergroup_0': {
'properties': {
'name': 'policy_group',
'policies': 'affinity'
},
'type': 'OS::Nova::ServerGroup'
}
}
expected_data = {
'key_0': {
'action': 'CREATE',
'metadata': {},
'name': 'key_0',
'resource_data': {},
'resource_id': 'key',
'status': 'COMPLETE',
'type': 'OS::Nova::KeyPair'
},
'network_0': {
'action': 'CREATE',
'metadata': {},
'name': 'network_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE',
'type': 'OS::Neutron::Net'
},
'router_0': {
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Neutron::Router'
},
'servergroup_0': {
'action': 'CREATE',
'metadata': {},
'name': 'servergroup_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::ServerGroup'
}
}
generator.extract_data()
self.assertEqual(generator.template['resources'], expected_resources)
self.assertEqual(generator.template['parameters'], expected_parameters)
self.assertEqual(generator.stack_data['resources'], expected_data)
def test_generation_exclude_servers_volumes_keypairs(self):
generator = self.get_generator(True, True, True, True)
expected_parameters = {}
expected_resources = {
'network_0': {
'properties': {
'admin_state_up': True,
'name': 'mynetwork',
'shared': False
},
'type': 'OS::Neutron::Net'
},
'router_0': {
'properties': {
'admin_state_up': 'true',
'name': 'myrouter'
},
'type': 'OS::Neutron::Router'
},
'servergroup_0': {
'properties': {
'name': 'policy_group',
'policies': 'affinity'
},
'type': 'OS::Nova::ServerGroup'
}
}
expected_data = {
'network_0': {
'action': 'CREATE',
'metadata': {},
'name': 'network_0',
'resource_data': {},
'resource_id': '2222',
'status': 'COMPLETE',
'type': 'OS::Neutron::Net'
},
'router_0': {
'action': 'CREATE',
'metadata': {},
'name': 'router_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Neutron::Router'
},
'servergroup_0': {
'action': 'CREATE',
'metadata': {},
'name': 'servergroup_0',
'resource_data': {},
'resource_id': '1234',
'status': 'COMPLETE',
'type': 'OS::Nova::ServerGroup'
}
}
generator.extract_data()
self.assertEqual(generator.template['resources'], expected_resources)
self.assertEqual(generator.template['parameters'], expected_parameters)
self.assertEqual(generator.stack_data['resources'], expected_data)