| | | a flavor properly with all supported settings |
+---------------------------------------+---------------+-----------------------------------------------------------+
-heat_utils_tests.py - HeatUtilsCreateStackTests
------------------------------------------------
+heat_utils_tests.py - HeatUtilsCreateSimpleStackTests
+-----------------------------------------------------
+---------------------------------------+---------------+-----------------------------------------------------------+
-| Test Name | Glance API | Description |
+| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
| test_create_stack | 1 | Tests the heat_utils.create_stack() with a test template |
+---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_stack_x2 | 1 | Tests the heat_utils.create_stack() with a test template |
+| | | and attempts to deploy a second time w/o actually |
+| | | deploying any objects |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
+heat_utils_tests.py - HeatUtilsCreateComplexStackTests
+------------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_get_settings_from_stack | 1 | Tests the heat_utils functions that are responsible for |
+| | | reverse engineering settings objects of the types deployed|
+| | | by Heat |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
+settings_utils_tests.py - SettingsUtilsNetworkingTests
+------------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | API | Description |
++=======================================+===============+===========================================================+
+| test_derive_net_settings_no_subnet | Neutron 2 | Tests to ensure that derived NetworkSettings from an |
+| | | OpenStack network are correct without a subnet |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_derive_net_settings_two_subnets | Neutron 2 | Tests to ensure that derived NetworkSettings from an |
+| | | OpenStack network are correct with two subnets |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
+settings_utils_tests.py - SettingsUtilsVmInstTests
+--------------------------------------------------
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | API | Description |
++=======================================+===============+===========================================================+
+| test_derive_vm_inst_settings | Neutron 2 | Tests to ensure that derived VmInstanceSettings from an |
+| | | OpenStack VM instance is correct |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_derive_image_settings | Neutron 2 | Tests to ensure that derived ImageSettings from an |
+| | | OpenStack VM instance is correct |
++---------------------------------------+---------------+-----------------------------------------------------------+
| test_create_same_stack | 2 | Ensures that a Heat stack with the same name cannot be |
| | | created 2x |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_same_stack | 2 | Ensures that a Heat stack with the same name cannot be |
-| | | created 2x |
+| test_retrieve_network_creators | 2 | Ensures that an OpenStackHeatStack instance can return an |
+| | | OpenStackNetwork instance configured as deployed |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_retrieve_vm_inst_creators | 2 | Ensures that an OpenStackHeatStack instance can return an |
+| | | OpenStackVmInstance instance configured as deployed |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
+create_stack_tests.py - CreateComplexStackTests
+-----------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Neutron API | Description |
++=======================================+===============+===========================================================+
+| test_connect_via_ssh_heat_vm | 2 | Ensures that two OpenStackHeatStack instances can return |
+| | | OpenStackVmInstance instances one configured with a |
+| | | floating IP and keypair and can be access via SSH |
+---------------------------------------+---------------+-----------------------------------------------------------+
create_stack_tests.py - CreateStackNegativeTests
----------------------
Ensures that all required members are included when constructing a
-Stack domain object
+Stack domain object (for Heat)
ResourceDomainObjectTests
-------------------------
Ensures that all required members are included when constructing a
-Resource domain object
+Resource domain object (for Heat)
+
+OutputDomainObjectTests
+-----------------------
+
+Ensures that all required members are included when constructing a
+Output domain object (for Heat)
FloatingIpSettingsUnitTests
---------------------------
SNAPS domain object for Keypairs. Should contain attributes that
are shared amongst cloud providers
"""
- def __init__(self, name, id, public_key):
+ def __init__(self, name, kp_id, public_key, fingerprint=None):
"""
Constructor
:param name: the keypair's name
- :param id: the keypair's id
+ :param kp_id: the keypair's id
+ :param public_key: the keypair's public key
+ :param fingerprint: the keypair's host fingerprint
"""
self.name = name
- self.id = id
+ self.id = kp_id
self.public_key = public_key
+ self.fingerprint = fingerprint
def __eq__(self, other):
return (self.name == other.name and self.id == other.id and
Constructor
:param name: the security group's name
:param id: the security group's id
- :param ips: a list of IP addresses
+ :param description: description
+ :param ips|fixed_ips: a list of IP addresses
+ :param mac_address: the port's MAC addresses
+ :param allowed_address_pairs: the port's allowed_address_pairs value
+ :param admin_state_up: T|F whether or not the port is up
+ :param device_id: device's ID
+ :param device_owner: device's owner
+ :param network_id: associated network ID
+ :param port_security_enabled: T|F whether or not the port security is
+ enabled
+ :param security_groups: the security group IDs associated with port
+ :param project_id: the associated project/tenant ID
"""
self.name = kwargs.get('name')
self.id = kwargs.get('id')
- self.ips = kwargs.get('ips')
+ self.description = kwargs.get('description')
+ self.ips = kwargs.get('ips', kwargs.get('fixed_ips'))
self.mac_address = kwargs.get('mac_address')
self.allowed_address_pairs = kwargs.get('allowed_address_pairs')
+ self.admin_state_up = kwargs.get('admin_state_up')
+ self.device_id = kwargs.get('device_id')
+ self.device_owner = kwargs.get('device_owner')
+ self.network_id = kwargs.get('network_id')
+ self.port_security_enabled = kwargs.get('port_security_enabled')
+ self.security_groups = kwargs.get('security_groups')
+ self.project_id = kwargs.get('tenant_id', kwargs.get('project_id'))
def __eq__(self, other):
return (self.name == other.name and self.id == other.id and
class Resource:
"""
- SNAPS domain object for resources created by a heat template
+ SNAPS domain object for a resource created by a heat template
"""
def __init__(self, resource_type, resource_id):
"""
"""
self.type = resource_type
self.id = resource_id
+
+
+class Output:
+ """
+ SNAPS domain object for an output defined by a heat template
+ """
+ def __init__(self, **kwargs):
+ """
+ Constructor
+ :param description: the output description
+ :param output_key: the output's key
+ """
+ self.description = kwargs.get('description')
+ self.key = kwargs.get('output_key')
+ self.value = kwargs.get('output_value')
"""
def test_construction_positional(self):
- keypair = Keypair('foo', '123-456', 'foo-bar')
+ keypair = Keypair('foo', '123-456', 'foo-bar', '01:02:03')
self.assertEqual('foo', keypair.name)
self.assertEqual('123-456', keypair.id)
self.assertEqual('foo-bar', keypair.public_key)
+ self.assertEqual('01:02:03', keypair.fingerprint)
def test_construction_named(self):
- keypair = Keypair(public_key='foo-bar', id='123-456', name='foo')
+ keypair = Keypair(fingerprint='01:02:03', public_key='foo-bar',
+ kp_id='123-456', name='foo')
self.assertEqual('foo', keypair.name)
self.assertEqual('123-456', keypair.id)
self.assertEqual('foo-bar', keypair.public_key)
+ self.assertEqual('01:02:03', keypair.fingerprint)
Tests the construction of the snaps.domain.network.Port class
"""
- def test_construction_kwargs(self):
+ def test_construction_ips_kwargs(self):
ips = ['10', '11']
port = Port(
- **{'name': 'name', 'id': 'id', 'ips': ips})
- self.assertEqual('name', port.name)
- self.assertEqual('id', port.id)
+ **{'name': 'foo', 'id': 'bar', 'description': 'test desc',
+ 'ips': ips, 'mac_address': 'abc123',
+ 'allowed_address_pairs': list(), 'admin_state_up': False,
+ 'device_id': 'def456', 'device_owner': 'joe',
+ 'network_id': 'ghi789', 'port_security_enabled': False,
+ 'security_groups': list(), 'tenant_id': 'jkl098'})
+ self.assertEqual('foo', port.name)
+ self.assertEqual('bar', port.id)
+ self.assertEqual('test desc', port.description)
self.assertEqual(ips, port.ips)
+ self.assertEqual('abc123', port.mac_address)
+ self.assertEqual(list(), port.allowed_address_pairs)
+ self.assertFalse(port.admin_state_up)
+ self.assertEqual('def456', port.device_id)
+ self.assertEqual('joe', port.device_owner)
+ self.assertEqual('ghi789', port.network_id)
+ self.assertFalse(port.port_security_enabled)
+ self.assertEqual(list(), port.security_groups)
+ self.assertEqual(list(), port.security_groups)
- def test_construction_named(self):
+ def test_construction_fixed_ips_kwargs(self):
+ ips = ['10', '11']
+ port = Port(
+ **{'name': 'foo', 'id': 'bar', 'description': 'test desc',
+ 'fixed_ips': ips, 'mac_address': 'abc123',
+ 'allowed_address_pairs': list(), 'admin_state_up': False,
+ 'device_id': 'def456', 'device_owner': 'joe',
+ 'network_id': 'ghi789', 'port_security_enabled': False,
+ 'security_groups': list(), 'tenant_id': 'jkl098'})
+ self.assertEqual('foo', port.name)
+ self.assertEqual('bar', port.id)
+ self.assertEqual('test desc', port.description)
+ self.assertEqual(ips, port.ips)
+ self.assertEqual('abc123', port.mac_address)
+ self.assertEqual(list(), port.allowed_address_pairs)
+ self.assertFalse(port.admin_state_up)
+ self.assertEqual('def456', port.device_id)
+ self.assertEqual('joe', port.device_owner)
+ self.assertEqual('ghi789', port.network_id)
+ self.assertFalse(port.port_security_enabled)
+ self.assertEqual(list(), port.security_groups)
+ self.assertEqual(list(), port.security_groups)
+
+ def test_construction_named_ips(self):
ips = ['10', '11']
- port = Port(ips=ips, id='id', name='name')
- self.assertEqual('name', port.name)
- self.assertEqual('id', port.id)
+ port = Port(
+ mac_address='abc123', description='test desc', ips=ips, id='bar',
+ name='foo', allowed_address_pairs=list(), admin_state_up=False,
+ device_id='def456', device_owner='joe', network_id='ghi789',
+ port_security_enabled=False, security_groups=list(),
+ tenant_id='jkl098')
+ self.assertEqual('foo', port.name)
+ self.assertEqual('bar', port.id)
+ self.assertEqual('test desc', port.description)
+ self.assertEqual(ips, port.ips)
+ self.assertEqual('abc123', port.mac_address)
+ self.assertEqual(list(), port.allowed_address_pairs)
+ self.assertFalse(port.admin_state_up)
+ self.assertEqual('def456', port.device_id)
+ self.assertEqual('joe', port.device_owner)
+ self.assertEqual('ghi789', port.network_id)
+ self.assertFalse(port.port_security_enabled)
+ self.assertEqual(list(), port.security_groups)
+ self.assertEqual(list(), port.security_groups)
+
+ def test_construction_named_fixed_ips(self):
+ ips = ['10', '11']
+ port = Port(
+ mac_address='abc123', description='test desc', fixed_ips=ips,
+ id='bar', name='foo', allowed_address_pairs=list(),
+ admin_state_up=False, device_id='def456', device_owner='joe',
+ network_id='ghi789', port_security_enabled=False,
+ security_groups=list(), tenant_id='jkl098')
+ self.assertEqual('foo', port.name)
+ self.assertEqual('bar', port.id)
+ self.assertEqual('test desc', port.description)
self.assertEqual(ips, port.ips)
+ self.assertEqual('abc123', port.mac_address)
+ self.assertEqual(list(), port.allowed_address_pairs)
+ self.assertFalse(port.admin_state_up)
+ self.assertEqual('def456', port.device_id)
+ self.assertEqual('joe', port.device_owner)
+ self.assertEqual('ghi789', port.network_id)
+ self.assertFalse(port.port_security_enabled)
+ self.assertEqual(list(), port.security_groups)
+ self.assertEqual(list(), port.security_groups)
class RouterDomainObjectTests(unittest.TestCase):
# limitations under the License.
import unittest
-from snaps.domain.stack import Stack, Resource
+from snaps.domain.stack import Stack, Resource, Output
class StackDomainObjectTests(unittest.TestCase):
resource = Resource(resource_id='bar', resource_type='foo')
self.assertEqual('foo', resource.type)
self.assertEqual('bar', resource.id)
+
+
+class OutputDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.Resource class
+ """
+
+ def test_construction_kwargs(self):
+ kwargs = {'description': 'foo', 'output_key': 'test_key',
+ 'output_value': 'bar'}
+ resource = Output(**kwargs)
+ self.assertEqual('foo', resource.description)
+ self.assertEqual('test_key', resource.key)
+ self.assertEqual('bar', resource.value)
+
+ def test_construction_named(self):
+ resource = Output(description='foo', output_key='test_key',
+ output_value='bar')
+ self.assertEqual('foo', resource.description)
+ self.assertEqual('test_key', resource.key)
+ self.assertEqual('bar', resource.value)
"""
def test_construction_positional(self):
- vm_inst = VmInst('name', 'id', dict())
+ vm_inst = VmInst('name', 'id', '456', '123', dict(), 'kp-name', list())
self.assertEqual('name', vm_inst.name)
self.assertEqual('id', vm_inst.id)
+ self.assertEqual('456', vm_inst.image_id)
+ self.assertEqual('123', vm_inst.flavor_id)
self.assertEqual(dict(), vm_inst.networks)
+ self.assertEqual('kp-name', vm_inst.keypair_name)
+ self.assertEqual(list(), vm_inst.sec_grp_names)
def test_construction_named(self):
- vm_inst = VmInst(networks=dict(), inst_id='id', name='name')
+ vm_inst = VmInst(sec_grp_names=list(), networks=dict(), inst_id='id',
+ name='name', flavor_id='123', image_id='456',
+ keypair_name='kp-name')
self.assertEqual('name', vm_inst.name)
self.assertEqual('id', vm_inst.id)
+ self.assertEqual('456', vm_inst.image_id)
+ self.assertEqual('123', vm_inst.flavor_id)
self.assertEqual(dict(), vm_inst.networks)
+ self.assertEqual('kp-name', vm_inst.keypair_name)
+ self.assertEqual(list(), vm_inst.sec_grp_names)
class FloatingIpDomainObjectTests(unittest.TestCase):
Tests the construction of the snaps.domain.test.Image class
"""
- def test_construction_positional(self):
- vm_inst = FloatingIp('id-123', '10.0.0.1')
- self.assertEqual('id-123', vm_inst.id)
- self.assertEqual('10.0.0.1', vm_inst.ip)
+ def test_construction_kwargs_ip_proj(self):
+ kwargs = {'id': 'foo', 'description': 'bar', 'ip': '192.168.122.3',
+ 'fixed_ip_address': '10.0.0.3',
+ 'floating_network_id': 'id_of_net', 'port_id': 'id_of_port',
+ 'router_id': 'id_of_router', 'project_id': 'id_of_proj'}
+ vm_inst = FloatingIp(**kwargs)
+ self.assertEqual('foo', vm_inst.id)
+ self.assertEqual('bar', vm_inst.description)
+ self.assertEqual('192.168.122.3', vm_inst.ip)
+ self.assertEqual('10.0.0.3', vm_inst.fixed_ip_address)
+ self.assertEqual('id_of_net', vm_inst.floating_network_id)
+ self.assertEqual('id_of_port', vm_inst.port_id)
+ self.assertEqual('id_of_router', vm_inst.router_id)
+ self.assertEqual('id_of_proj', vm_inst.project_id)
- def test_construction_named(self):
- vm_inst = FloatingIp(ip='10.0.0.1', inst_id='id-123')
- self.assertEqual('id-123', vm_inst.id)
- self.assertEqual('10.0.0.1', vm_inst.ip)
+ def test_construction_kwargs_fixed_ip_tenant(self):
+ kwargs = {'id': 'foo', 'description': 'bar',
+ 'floating_ip_address': '192.168.122.3',
+ 'fixed_ip_address': '10.0.0.3',
+ 'floating_network_id': 'id_of_net', 'port_id': 'id_of_port',
+ 'router_id': 'id_of_router', 'tenant_id': 'id_of_proj'}
+ vm_inst = FloatingIp(**kwargs)
+ self.assertEqual('foo', vm_inst.id)
+ self.assertEqual('bar', vm_inst.description)
+ self.assertEqual('192.168.122.3', vm_inst.ip)
+ self.assertEqual('10.0.0.3', vm_inst.fixed_ip_address)
+ self.assertEqual('id_of_net', vm_inst.floating_network_id)
+ self.assertEqual('id_of_port', vm_inst.port_id)
+ self.assertEqual('id_of_router', vm_inst.router_id)
+ self.assertEqual('id_of_proj', vm_inst.project_id)
+
+ def test_construction_named_ip_proj(self):
+ vm_inst = FloatingIp(
+ id='foo', description='bar', ip='192.168.122.3',
+ fixed_ip_address='10.0.0.3', floating_network_id='id_of_net',
+ port_id='id_of_port', router_id='id_of_router',
+ project_id='id_of_proj')
+ self.assertEqual('foo', vm_inst.id)
+ self.assertEqual('bar', vm_inst.description)
+ self.assertEqual('192.168.122.3', vm_inst.ip)
+ self.assertEqual('10.0.0.3', vm_inst.fixed_ip_address)
+ self.assertEqual('id_of_net', vm_inst.floating_network_id)
+ self.assertEqual('id_of_port', vm_inst.port_id)
+ self.assertEqual('id_of_router', vm_inst.router_id)
+ self.assertEqual('id_of_proj', vm_inst.project_id)
+
+ def test_construction_kwargs_named_fixed_ip_tenant(self):
+ vm_inst = FloatingIp(
+ id='foo', description='bar', floating_ip_address='192.168.122.3',
+ fixed_ip_address='10.0.0.3', floating_network_id='id_of_net',
+ port_id='id_of_port', router_id='id_of_router',
+ tenant_id='id_of_proj')
+ self.assertEqual('foo', vm_inst.id)
+ self.assertEqual('bar', vm_inst.description)
+ self.assertEqual('192.168.122.3', vm_inst.ip)
+ self.assertEqual('10.0.0.3', vm_inst.fixed_ip_address)
+ self.assertEqual('id_of_net', vm_inst.floating_network_id)
+ self.assertEqual('id_of_port', vm_inst.port_id)
+ self.assertEqual('id_of_router', vm_inst.router_id)
+ self.assertEqual('id_of_proj', vm_inst.project_id)
SNAPS domain object for Images. Should contain attributes that
are shared amongst cloud providers
"""
- def __init__(self, name, inst_id, networks):
+ def __init__(self, name, inst_id, image_id, flavor_id, networks,
+ keypair_name, sec_grp_names):
"""
Constructor
:param name: the image's name
:param inst_id: the instance's id
- :param networks: dict of networks where the key is the subnet name and
+ :param image_id: the instance's image id
+ :param flavor_id: the ID used to spawn this instance
+ :param networks: dict of networks where the key is the network name and
value is a list of associated IPs
+ :param keypair_name: the name of the associated keypair
+ :param sec_grp_names: list of security group names
"""
self.name = name
self.id = inst_id
+ self.image_id = image_id
+ self.flavor_id = flavor_id
self.networks = networks
+ self.keypair_name = keypair_name
+ self.sec_grp_names = sec_grp_names
+
+ def __eq__(self, other):
+ return (self.name == other.name and
+ self.id == other.id and
+ self.image_id == other.image_id and
+ self.flavor_id == other.flavor_id and
+ self.networks == other.networks and
+ self.keypair_name == other.keypair_name and
+ self.sec_grp_names == other.sec_grp_names)
class FloatingIp:
SNAPS domain object for Images. Should contain attributes that
are shared amongst cloud providers
"""
- def __init__(self, inst_id, ip):
+ def __init__(self, **kwargs):
"""
Constructor
- :param inst_id: the floating ip's id
- :param ip: the IP address
+ :param id: the floating ip's id
+ :param description: the description
+ :param ip|floating_ip_address: the Floating IP address mapped to the
+ 'ip' attribute
+ :param fixed_ip_address: the IP address of the tenant network
+ :param floating_network_id: the ID of the external network
+ :param port_id: the ID of the associated port
+ :param router_id: the ID of the associated router
+ :param project_id|tenant_id: the ID of the associated project mapped to
+ the attribute 'project_id'
+ :param
"""
- self.id = inst_id
- self.ip = ip
+ self.id = kwargs.get('id')
+ self.description = kwargs.get('description')
+ self.ip = kwargs.get('ip', kwargs.get('floating_ip_address'))
+ self.fixed_ip_address = kwargs.get('fixed_ip_address')
+ self.floating_network_id = kwargs.get('floating_network_id')
+ self.port_id = kwargs.get('port_id')
+ self.router_id = kwargs.get('router_id')
+ self.project_id = kwargs.get('project_id', kwargs.get('tenant_id'))
# limitations under the License.
import os
import logging
+
+from cryptography.hazmat.primitives import serialization
+
try:
import urllib.request as urllib
except ImportError:
raise
try:
with open(dest, 'wb') as download_file:
- logger.debug('Saving file to - ' + os.path.abspath(download_file.name))
+ logger.debug('Saving file to - %s',
+ os.path.abspath(download_file.name))
response = __get_url_response(url)
download_file.write(response.read())
return download_file
download_file.close()
+def save_keys_to_files(keys=None, pub_file_path=None, priv_file_path=None):
+ """
+ Saves the generated RSA generated keys to the filesystem
+ :param keys: the keys to save generated by cryptography
+ :param pub_file_path: the path to the public keys
+ :param priv_file_path: the path to the private keys
+ """
+ if keys:
+ if pub_file_path:
+ # To support '~'
+ pub_expand_file = os.path.expanduser(pub_file_path)
+ pub_dir = os.path.dirname(pub_expand_file)
+
+ if not os.path.isdir(pub_dir):
+ os.mkdir(pub_dir)
+
+ public_handle = None
+ try:
+ public_handle = open(pub_expand_file, 'wb')
+ public_bytes = keys.public_key().public_bytes(
+ serialization.Encoding.OpenSSH,
+ serialization.PublicFormat.OpenSSH)
+ public_handle.write(public_bytes)
+ finally:
+ if public_handle:
+ public_handle.close()
+
+ os.chmod(pub_expand_file, 0o400)
+ logger.info("Saved public key to - " + pub_expand_file)
+ if priv_file_path:
+ # To support '~'
+ priv_expand_file = os.path.expanduser(priv_file_path)
+ priv_dir = os.path.dirname(priv_expand_file)
+ if not os.path.isdir(priv_dir):
+ os.mkdir(priv_dir)
+
+ private_handle = None
+ try:
+ private_handle = open(priv_expand_file, 'wb')
+ private_handle.write(
+ keys.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.TraditionalOpenSSL,
+ encryption_algorithm=serialization.NoEncryption()))
+ finally:
+ if private_handle:
+ private_handle.close()
+
+ os.chmod(priv_expand_file, 0o400)
+ logger.info("Saved private key to - " + priv_expand_file)
+
+
+def save_string_to_file(string, file_path, mode=None):
+ """
+ Stores
+ :param string: the string contents to store
+ :param file_path: the file path to create
+ :param mode: the file's mode
+ :return: the file object
+ """
+ save_file = open(file_path, 'w')
+ try:
+ save_file.write(string)
+ if mode:
+ os.chmod(file_path, mode)
+ return save_file
+ finally:
+ save_file.close()
+
+
def get_content_length(url):
"""
Returns the number of bytes to be downloaded from the given URL
def create(self, cleanup=False, block=False):
"""
Creates a VM instance
- :param cleanup: When true, only perform lookups for OpenStack objects.
+ :param cleanup: When true, this object is initialized only via queries,
+ else objects will be created when the queries return
+ None. The name of this parameter should be changed to
+ something like 'readonly' as the same goes with all of
+ the other creator classes.
:param block: Thread will block until instance has either become
active, error, or timeout waiting.
Additionally, when True, floating IPs will not be applied
fips = neutron_utils.get_floating_ips(self.__neutron,
self.__ports)
- for port_name, fip in fips:
+ for port_id, fip in fips:
settings = self.instance_settings.floating_ip_settings
for fip_setting in settings:
- if port_name == fip_setting.port_name:
+ if port_id == fip_setting.port_id:
self.__floating_ip_dict[fip_setting.name] = fip
+ else:
+ port = neutron_utils.get_port_by_id(
+ self.__neutron, port_id)
+ if port and port.name == fip_setting.port_name:
+ self.__floating_ip_dict[fip_setting.name] = fip
def __create_vm(self, block=False):
"""
# Cleanup ports
for name, port in self.__ports:
- logger.info('Deleting Port - ' + name)
+ logger.info('Deleting Port with ID - %S ' + port.id)
try:
neutron_utils.delete_port(self.__neutron, port)
except PortNotFoundClient as e:
for port_setting in port_settings:
port = neutron_utils.get_port(
self.__neutron, port_settings=port_setting)
+ if not port:
+ network = neutron_utils.get_network(
+ self.__neutron, network_name=port_setting.network_name)
+ net_ports = neutron_utils.get_ports(self.__neutron, network)
+ for net_port in net_ports:
+ if port_setting.mac_address == net_port.mac_address:
+ port = net_port
+ break
if port:
ports.append((port_setting.name, port))
elif not cleanup:
Returns the latest version of this server object from OpenStack
:return: Server object
"""
- return self.__vm
+ return nova_utils.get_server_object_by_id(self.__nova, self.__vm.id)
def get_console_output(self):
"""
def vm_active(self, block=False, poll_interval=POLL_INTERVAL):
"""
- Returns true when the VM status returns the value of
- expected_status_code
+ Returns true when the VM status returns the value of the constant
+ STATUS_ACTIVE
:param block: When true, thread will block until active or timeout
value in seconds has been exceeded (False)
:param poll_interval: The polling interval in seconds
:return: T/F
"""
if not self.__vm:
- return False
+ if expected_status_code == STATUS_DELETED:
+ return True
+ else:
+ return False
status = nova_utils.get_server_status(self.__nova, self.__vm)
if not status:
"""
Constructor
:param name: the name of the VM
- :param flavor: the VM's flavor
+ :param flavor: the VM's flavor name
:param port_settings: the port configuration settings (required)
:param security_group_names: a set of names of the security groups to
add to the VM
"""
self.name = kwargs.get('name')
self.port_name = kwargs.get('port_name')
+ self.port_id = kwargs.get('port_id')
self.router_name = kwargs.get('router_name')
self.subnet_name = kwargs.get('subnet_name')
if kwargs.get('provisioning') is not None:
else:
self.provisioning = True
- if not self.name or not self.port_name or not self.router_name:
+ # if not self.name or not self.port_name or not self.router_name:
+ if not self.name or not self.router_name:
+ raise FloatingIpSettingsError(
+ 'The attributes name, port_name and router_name are required')
+
+ if not self.port_name and not self.port_id:
raise FloatingIpSettingsError(
- 'The attributes name, port_name and router_name are required '
- 'for FloatingIPSettings')
+ 'The attributes port_name or port_id are required')
class VmInstanceSettingsError(Exception):
self.__keypair = nova_utils.upload_keypair(
self.__nova, self.keypair_settings.name,
nova_utils.public_key_openssh(keys))
- nova_utils.save_keys_to_files(
+ file_utils.save_keys_to_files(
keys, self.keypair_settings.public_filepath,
self.keypair_settings.private_filepath)
def __init__(self, **kwargs):
"""
- Constructor - all parameters are optional
- :param name: A symbolic name for the port.
+ Constructor
+ :param name: A symbolic name for the port (optional).
:param network_name: The name of the network on which to create the
- port.
+ port (required).
:param admin_state_up: A boolean value denoting the administrative
status of the port. True = up / False = down
:param project_name: The name of the project who owns the network.
self.device_owner = kwargs.get('device_owner')
self.device_id = kwargs.get('device_id')
- if not self.name or not self.network_name:
+ if not self.network_name:
raise PortSettingsError(
- 'The attributes neutron, name, and network_name are required '
- 'for PortSettings')
+ 'The attribute network_name is required')
def __set_fixed_ips(self, neutron):
"""
from heatclient.exc import HTTPNotFound
-from snaps.openstack.create_network import (
- OpenStackNetwork, NetworkSettings, SubnetSettings)
+from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.utils import nova_utils, settings_utils, glance_utils
+
+from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.utils import heat_utils, neutron_utils
__author__ = 'spisarski'
STATUS_CREATE_FAILED = 'CREATE_FAILED'
STATUS_CREATE_COMPLETE = 'CREATE_COMPLETE'
STATUS_DELETE_COMPLETE = 'DELETE_COMPLETE'
+STATUS_DELETE_FAILED = 'DELETE_FAILED'
class OpenStackHeatStack:
Class responsible for creating an heat stack in OpenStack
"""
- def __init__(self, os_creds, stack_settings):
+ def __init__(self, os_creds, stack_settings, image_settings=None,
+ keypair_settings=None):
"""
Constructor
:param os_creds: The OpenStack connection credentials
:param stack_settings: The stack settings
+ :param image_settings: A list of ImageSettings objects that were used
+ for spawning this stack
+ :param image_settings: A list of ImageSettings objects that were used
+ for spawning this stack
+ :param keypair_settings: A list of KeypairSettings objects that were
+ used for spawning this stack
:return:
"""
self.__os_creds = os_creds
self.stack_settings = stack_settings
+
+ if image_settings:
+ self.image_settings = image_settings
+ else:
+ self.image_settings = None
+
+ if image_settings:
+ self.keypair_settings = keypair_settings
+ else:
+ self.keypair_settings = None
+
self.__stack = None
self.__heat_cli = None
"""
if self.__stack:
try:
+ logger.info('Deleting stack - %s' + self.__stack.name)
+ heat_utils.delete_stack(self.__heat_cli, self.__stack)
+
+ try:
+ self.stack_deleted(block=True)
+ except StackError as e:
+ # Stack deletion seems to fail quite a bit
+ logger.warn('Stack did not delete properly - %s', e)
+
+ # Delete VMs first
+ for vm_inst_creator in self.get_vm_inst_creators():
+ try:
+ vm_inst_creator.clean()
+ if not vm_inst_creator.vm_deleted(block=True):
+ logger.warn('Unable to deleted VM - %s',
+ vm_inst_creator.get_vm_inst().name)
+ except:
+ logger.warn('Unexpected error deleting VM - %s ',
+ vm_inst_creator.get_vm_inst().name)
+
+ logger.info('Attempting to delete again stack - %s',
+ self.__stack.name)
+
+ # Delete Stack again
heat_utils.delete_stack(self.__heat_cli, self.__stack)
+ deleted = self.stack_deleted(block=True)
+ if not deleted:
+ raise StackError(
+ 'Stack could not be deleted ' + self.__stack.name)
except HTTPNotFound:
pass
- self.__stack = None
+ self.__stack = None
def get_stack(self):
"""
object
:return:
"""
- return heat_utils.get_stack_outputs(self.__heat_cli, self.__stack.id)
+ return heat_utils.get_outputs(self.__heat_cli, self.__stack)
def get_status(self):
"""
if not timeout:
timeout = self.stack_settings.stack_create_timeout
return self._stack_status_check(STATUS_CREATE_COMPLETE, block, timeout,
- poll_interval)
+ poll_interval, STATUS_CREATE_FAILED)
+
+ def stack_deleted(self, block=False, timeout=None,
+ poll_interval=POLL_INTERVAL):
+ """
+ Returns true when the stack status returns the value of
+ expected_status_code
+ :param block: When true, thread will block until active or timeout
+ value in seconds has been exceeded (False)
+ :param timeout: The timeout value
+ :param poll_interval: The polling interval in seconds
+ :return: T/F
+ """
+ if not timeout:
+ timeout = self.stack_settings.stack_create_timeout
+ return self._stack_status_check(STATUS_DELETE_COMPLETE, block, timeout,
+ poll_interval, STATUS_DELETE_FAILED)
def get_network_creators(self):
"""
self.__heat_cli, neutron, self.__stack)
for stack_network in stack_networks:
- net_settings = self.__create_network_settings(
+ net_settings = settings_utils.create_network_settings(
neutron, stack_network)
net_creator = OpenStackNetwork(self.__os_creds, net_settings)
out.append(net_creator)
return out
- def __create_network_settings(self, neutron, network):
+ def get_vm_inst_creators(self, heat_keypair_option=None):
"""
- Returns a NetworkSettings object
- :param neutron: the neutron client
- :param network: a SNAPS-OO Network domain object
- :return:
+ Returns a list of VM Instance creator objects as configured by the heat
+ template
+ :return: list() of OpenStackVmInstance objects
"""
- return NetworkSettings(
- name=network.name, network_type=network.type,
- subnet_settings=self.__create_subnet_settings(neutron, network))
- def __create_subnet_settings(self, neutron, network):
- """
- Returns a list of SubnetSettings objects for a given network
- :param neutron: the OpenStack neutron client
- :param network: the SNAPS-OO Network domain object
- :return: a list
- """
out = list()
+ nova = nova_utils.nova_client(self.__os_creds)
+
+ stack_servers = heat_utils.get_stack_servers(
+ self.__heat_cli, nova, self.__stack)
+
+ neutron = neutron_utils.neutron_client(self.__os_creds)
+ glance = glance_utils.glance_client(self.__os_creds)
+
+ for stack_server in stack_servers:
+ vm_inst_settings = settings_utils.create_vm_inst_settings(
+ nova, neutron, stack_server)
+ image_settings = settings_utils.determine_image_settings(
+ glance, stack_server, self.image_settings)
+ keypair_settings = settings_utils.determine_keypair_settings(
+ self.__heat_cli, self.__stack, stack_server,
+ keypair_settings=self.keypair_settings,
+ priv_key_key=heat_keypair_option)
+ vm_inst_creator = OpenStackVmInstance(
+ self.__os_creds, vm_inst_settings, image_settings,
+ keypair_settings)
+ out.append(vm_inst_creator)
+ vm_inst_creator.create(cleanup=True)
- subnets = neutron_utils.get_subnets_by_network(neutron, network)
- for subnet in subnets:
- kwargs = dict()
- kwargs['cidr'] = subnet.cidr
- kwargs['ip_version'] = subnet.ip_version
- kwargs['name'] = subnet.name
- kwargs['start'] = subnet.start
- kwargs['end'] = subnet.end
- kwargs['gateway_ip'] = subnet.gateway_ip
- kwargs['enable_dhcp'] = subnet.enable_dhcp
- kwargs['dns_nameservers'] = subnet.dns_nameservers
- kwargs['host_routes'] = subnet.host_routes
- kwargs['ipv6_ra_mode'] = subnet.ipv6_ra_mode
- kwargs['ipv6_address_mode'] = subnet.ipv6_address_mode
- out.append(SubnetSettings(**kwargs))
return out
def _stack_status_check(self, expected_status_code, block, timeout,
- poll_interval):
+ poll_interval, fail_status):
"""
Returns true when the stack status returns the value of
expected_status_code
value in seconds has been exceeded (False)
:param timeout: The timeout value
:param poll_interval: The polling interval in seconds
+ :param fail_status: Returns false if the fail_status code is found
:return: T/F
"""
# sleep and wait for stack status change
start = time.time() - timeout
while timeout > time.time() - start:
- status = self._status(expected_status_code)
+ status = self._status(expected_status_code, fail_status)
if status:
logger.debug(
'Stack is active with name - ' + self.stack_settings.name)
'Timeout checking for stack status for ' + expected_status_code)
return False
- def _status(self, expected_status_code):
+ def _status(self, expected_status_code, fail_status=STATUS_CREATE_FAILED):
"""
Returns True when active else False
:param expected_status_code: stack status evaluated with this string
'Cannot stack status for stack with ID - ' + self.__stack.id)
return False
- if status == STATUS_CREATE_FAILED:
- raise StackCreationError('Stack had an error during deployment')
+ if fail_status and status == fail_status:
+ raise StackError('Stack had an error')
logger.debug('Stack status is - ' + status)
return status == expected_status_code
"""
Exception to be thrown when an stack cannot be created
"""
+
+
+class StackError(Exception):
+ """
+ General exception
+ """
with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(**{'name': 'foo', 'router_name': 'bar'})
- def test_name_port_router_only(self):
+ def test_name_port_router_name_only(self):
settings = FloatingIpSettings(name='foo', port_name='foo-port',
router_name='bar-router')
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
+ self.assertIsNone(settings.port_id)
+ self.assertEqual('bar-router', settings.router_name)
+ self.assertIsNone(settings.subnet_name)
+ self.assertTrue(settings.provisioning)
+
+ def test_name_port_router_id_only(self):
+ settings = FloatingIpSettings(name='foo', port_id='foo-port',
+ router_name='bar-router')
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('foo-port', settings.port_id)
+ self.assertIsNone(settings.port_name)
self.assertEqual('bar-router', settings.router_name)
self.assertIsNone(settings.subnet_name)
self.assertTrue(settings.provisioning)
'router_name': 'bar-router'})
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
+ self.assertIsNone(settings.port_id)
self.assertEqual('bar-router', settings.router_name)
self.assertIsNone(settings.subnet_name)
self.assertTrue(settings.provisioning)
provisioning=False)
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
+ self.assertIsNone(settings.port_id)
self.assertEqual('bar-router', settings.router_name)
self.assertEqual('bar-subnet', settings.subnet_name)
self.assertFalse(settings.provisioning)
'provisioning': False})
self.assertEqual('foo', settings.name)
self.assertEqual('foo-port', settings.port_name)
+ self.assertIsNone(settings.port_id)
self.assertEqual('bar-router', settings.router_name)
self.assertEqual('bar-subnet', settings.subnet_name)
self.assertFalse(settings.provisioning)
self.assertEqual(ip_1, inst_creator.get_port_ip(self.port_1_name))
self.assertTrue(inst_creator.vm_active(block=True))
- self.assertEqual(vm_inst, inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst.id, inst_creator.get_vm_inst().id)
def test_ssh_client_fip_before_active(self):
"""
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
- self.assertEqual(vm_inst, inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst.id, inst_creator.get_vm_inst().id)
self.assertTrue(validate_ssh_client(inst_creator))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
- self.assertEqual(vm_inst, inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst.id, inst_creator.get_vm_inst().id)
self.assertTrue(validate_ssh_client(inst_creator))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
- self.assertEqual(vm_inst, inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst.id, inst_creator.get_vm_inst().id)
self.assertTrue(validate_ssh_client(inst_creator))
vm_inst = self.inst_creator.create(block=True)
- self.assertEqual(vm_inst, self.inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst.id, self.inst_creator.get_vm_inst().id)
# Effectively blocks until VM has been properly activated
self.assertTrue(self.inst_creator.vm_active(block=True))
:return:
"""
keys = nova_utils.create_keys()
- nova_utils.save_keys_to_files(keys=keys,
+ file_utils.save_keys_to_files(keys=keys,
pub_file_path=self.pub_file_path)
self.keypair_creator = OpenStackKeypair(
self.os_creds, KeypairSettings(name=self.keypair_name,
:return:
"""
keys = nova_utils.create_keys()
- nova_utils.save_keys_to_files(
+ file_utils.save_keys_to_files(
keys=keys, pub_file_path=self.pub_file_path,
priv_file_path=self.priv_file_path)
self.keypair_creator = OpenStackKeypair(
:return:
"""
keys = nova_utils.create_keys()
- nova_utils.save_keys_to_files(
+ file_utils.save_keys_to_files(
keys=keys, pub_file_path=self.pub_file_path,
priv_file_path=self.priv_file_path)
self.keypair_creator = OpenStackKeypair(
from snaps.openstack import create_stack
from snaps.openstack.create_stack import StackSettings, StackSettingsError
-from snaps.openstack.tests import openstack_tests
+from snaps.openstack.tests import openstack_tests, create_instance_tests
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
-from snaps.openstack.utils import heat_utils, neutron_utils
+from snaps.openstack.utils import heat_utils, neutron_utils, nova_utils
__author__ = 'spisarski'
class CreateStackSuccessTests(OSIntegrationTestCase):
"""
- Test for the CreateStack class defined in create_stack.py
+ Tests for the CreateStack class defined in create_stack.py
"""
def setUp(self):
self.network_name = self.guid + '-net'
self.subnet_name = self.guid + '-subnet'
+ self.vm_inst_name = self.guid + '-inst'
+
self.env_values = {
'image_name': self.image_creator.image_settings.name,
'flavor_name': self.flavor_creator.flavor_settings.name,
'net_name': self.network_name,
- 'subnet_name': self.subnet_name}
+ 'subnet_name': self.subnet_name,
+ 'inst_name': self.vm_inst_name}
self.heat_tmplt_path = pkg_resources.resource_filename(
'snaps.openstack.tests.heat', 'test_heat_template.yaml')
self.assertIsNotNone(retrieved_stack)
self.assertEqual(created_stack.name, retrieved_stack.name)
self.assertEqual(created_stack.id, retrieved_stack.id)
- self.assertIsNotNone(self.stack_creator.get_outputs())
- self.assertEquals(0, len(self.stack_creator.get_outputs()))
-
- resources = heat_utils.get_resources(
- self.heat_cli, self.stack_creator.get_stack())
- self.assertIsNotNone(resources)
- self.assertEqual(4, len(resources))
+ self.assertEqual(0, len(self.stack_creator.get_outputs()))
def test_create_stack_template_dict(self):
"""
self.assertIsNotNone(retrieved_stack)
self.assertEqual(created_stack.name, retrieved_stack.name)
self.assertEqual(created_stack.id, retrieved_stack.id)
- self.assertIsNotNone(self.stack_creator.get_outputs())
- self.assertEquals(0, len(self.stack_creator.get_outputs()))
+ self.assertEqual(0, len(self.stack_creator.get_outputs()))
def test_create_delete_stack(self):
"""
self.assertIsNotNone(retrieved_stack)
self.assertEqual(created_stack.name, retrieved_stack.name)
self.assertEqual(created_stack.id, retrieved_stack.id)
- self.assertIsNotNone(self.stack_creator.get_outputs())
- self.assertEquals(0, len(self.stack_creator.get_outputs()))
+ self.assertEqual(0, len(self.stack_creator.get_outputs()))
self.assertEqual(create_stack.STATUS_CREATE_COMPLETE,
self.stack_creator.get_status())
self.assertIsNotNone(retrieved_stack)
self.assertEqual(created_stack1.name, retrieved_stack.name)
self.assertEqual(created_stack1.id, retrieved_stack.id)
- self.assertIsNotNone(self.stack_creator.get_outputs())
self.assertEqual(0, len(self.stack_creator.get_outputs()))
# Should be retrieving the instance data
self.assertIsNotNone(subnet_by_id)
self.assertEqual(subnet_by_name, subnet_by_id)
+ def test_retrieve_vm_inst_creators(self):
+ """
+ Tests the creation of an OpenStack stack from Heat template file and
+ the retrieval of the network creator.
+ """
+ stack_settings = StackSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid) + '-stack',
+ template_path=self.heat_tmplt_path,
+ env_values=self.env_values)
+ self.stack_creator = create_stack.OpenStackHeatStack(self.heat_creds,
+ stack_settings)
+ created_stack = self.stack_creator.create()
+ self.assertIsNotNone(created_stack)
+
+ vm_inst_creators = self.stack_creator.get_vm_inst_creators()
+ self.assertIsNotNone(vm_inst_creators)
+ self.assertEqual(1, len(vm_inst_creators))
+ self.assertEqual(self.vm_inst_name,
+ vm_inst_creators[0].get_vm_inst().name)
+
+ nova = nova_utils.nova_client(self.admin_os_creds)
+ vm_inst_by_name = nova_utils.get_server(
+ nova, server_name=vm_inst_creators[0].get_vm_inst().name)
+ self.assertEqual(vm_inst_creators[0].get_vm_inst(), vm_inst_by_name)
+ self.assertIsNotNone(nova_utils.get_server_object_by_id(
+ nova, vm_inst_creators[0].get_vm_inst().id))
+
+
+class CreateComplexStackTests(OSIntegrationTestCase):
+ """
+ Tests for the CreateStack class defined in create_stack.py
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateStack object that is responsible for downloading
+ and creating an OS stack file within OpenStack
+ """
+ super(self.__class__, self).__start__()
+
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+
+ self.heat_creds = self.admin_os_creds
+ self.heat_creds.project_name = self.admin_os_creds.project_name
+
+ self.heat_cli = heat_utils.heat_client(self.heat_creds)
+ self.stack_creator = None
+
+ self.image_creator = OpenStackImage(
+ self.heat_creds, openstack_tests.cirros_image_settings(
+ name=self.guid + '-image',
+ image_metadata=self.image_metadata))
+ self.image_creator.create()
+
+ self.network_name = self.guid + '-net'
+ self.subnet_name = self.guid + '-subnet'
+ self.flavor1_name = self.guid + '-flavor1'
+ self.flavor2_name = self.guid + '-flavor2'
+ self.vm_inst1_name = self.guid + '-inst1'
+ self.vm_inst2_name = self.guid + '-inst2'
+ self.keypair_name = self.guid + '-kp'
+
+ self.env_values = {
+ 'image1_name': self.image_creator.image_settings.name,
+ 'image2_name': self.image_creator.image_settings.name,
+ 'flavor1_name': self.flavor1_name,
+ 'flavor2_name': self.flavor2_name,
+ 'net_name': self.network_name,
+ 'subnet_name': self.subnet_name,
+ 'inst1_name': self.vm_inst1_name,
+ 'inst2_name': self.vm_inst2_name,
+ 'keypair_name': self.keypair_name}
+
+ self.heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'floating_ip_heat_template.yaml')
+
+ def tearDown(self):
+ """
+ Cleans the stack and downloaded stack file
+ """
+ if self.stack_creator:
+ try:
+ self.stack_creator.clean()
+ except:
+ pass
+
+ if self.image_creator:
+ try:
+ self.image_creator.clean()
+ except:
+ pass
+
+ super(self.__class__, self).__clean__()
+
+ def test_connect_via_ssh_heat_vm(self):
+ """
+ Tests the creation of an OpenStack stack from Heat template file and
+ the retrieval of two VM instance creators and attempt to connect via
+ SSH to the first one with a floating IP.
+ """
+ stack_settings = StackSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid) + '-stack',
+ template_path=self.heat_tmplt_path,
+ env_values=self.env_values)
+ self.stack_creator = create_stack.OpenStackHeatStack(
+ self.heat_creds, stack_settings,
+ [self.image_creator.image_settings])
+ created_stack = self.stack_creator.create()
+ self.assertIsNotNone(created_stack)
+
+ vm_inst_creators = self.stack_creator.get_vm_inst_creators(
+ heat_keypair_option='private_key')
+ self.assertIsNotNone(vm_inst_creators)
+ self.assertEqual(2, len(vm_inst_creators))
+
+ for vm_inst_creator in vm_inst_creators:
+ if vm_inst_creator.get_vm_inst().name == self.vm_inst1_name:
+ self.assertTrue(
+ create_instance_tests.validate_ssh_client(vm_inst_creator))
+ else:
+ vm_settings = vm_inst_creator.instance_settings
+ self.assertEqual(0, len(vm_settings.floating_ip_settings))
+
class CreateStackNegativeTests(OSIntegrationTestCase):
"""
--- /dev/null
+##############################################################################
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+##############################################################################
+heat_template_version: 2015-04-30
+
+description: >
+ Sample template with two VMs instantiated against different images and
+ flavors on the same network and the first one has a floating IP
+
+parameters:
+ image1_name:
+ type: string
+ label: Image ID for first VM
+ description: Image name to be used for first instance
+ default: image_1
+ image2_name:
+ type: string
+ label: Image ID for second VM
+ description: Image name to be used for second instance
+ default: image_2
+ flavor1_name:
+ type: string
+ label: Instance Flavor for first VM
+ description: Flavor name for the first instance
+ default: m1.small
+ flavor2_name:
+ type: string
+ label: Instance Flavor for second VM
+ description: Flavor name for the second instance
+ default: m1.med
+ net_name:
+ type: string
+ label: Test network name
+ description: The name of the stack's network
+ default: test_net
+ subnet_name:
+ type: string
+ label: Test subnet name
+ description: The name of the stack's subnet
+ default: test_subnet
+ router_name:
+ type: string
+ label: Test router name
+ description: The name of the stack's router
+ default: mgmt_router
+ keypair_name:
+ type: string
+ label: Keypair name
+ description: The name of the stack's keypair
+ default: keypair_name
+ inst1_name:
+ type: string
+ label: First VM name
+ description: The name of the first VM to be spawned
+ default: test_vm1
+ inst2_name:
+ type: string
+ label: Second VM name
+ description: The name of the second VM to be spawned
+ default: test_vm2
+ external_net_name:
+ type: string
+ description: Name of the external network which management network will connect to
+ default: external
+
+resources:
+ flavor1:
+ type: OS::Nova::Flavor
+ properties:
+ ram: 4096
+ vcpus: 4
+ disk: 4
+ flavor2:
+ type: OS::Nova::Flavor
+ properties:
+ ram: 4096
+ vcpus: 4
+ disk: 4
+
+ network:
+ type: OS::Neutron::Net
+ properties:
+ name: { get_param: net_name }
+
+ subnet:
+ type: OS::Neutron::Subnet
+ properties:
+ name: { get_param: subnet_name }
+ ip_version: 4
+ cidr: 10.1.2.0/24
+ network: { get_resource: network }
+
+ management_router:
+ type: OS::Neutron::Router
+ properties:
+ name: { get_param: router_name }
+ external_gateway_info:
+ network: { get_param: external_net_name }
+
+ management_router_interface:
+ type: OS::Neutron::RouterInterface
+ properties:
+ router: { get_resource: management_router }
+ subnet: { get_resource: subnet }
+
+ floating_ip:
+ type: OS::Neutron::FloatingIP
+ properties:
+ floating_network: { get_param: external_net_name }
+
+ floating_ip_association:
+ type: OS::Nova::FloatingIPAssociation
+ properties:
+ floating_ip: { get_resource: floating_ip }
+ server_id: {get_resource: vm1}
+
+ keypair:
+ type: OS::Nova::KeyPair
+ properties:
+ name: { get_param: keypair_name }
+ save_private_key: True
+
+ vm1:
+ type: OS::Nova::Server
+ depends_on: [subnet, keypair, flavor1]
+ properties:
+ name: { get_param: inst1_name }
+ image: { get_param: image1_name }
+ flavor: { get_resource: flavor1 }
+ key_name: {get_resource: keypair}
+ networks:
+ - network: { get_resource: network }
+
+ vm2:
+ type: OS::Nova::Server
+ depends_on: [subnet, flavor2]
+ properties:
+ name: { get_param: inst2_name }
+ image: { get_param: image2_name }
+ flavor: { get_resource: flavor2 }
+ key_name: {get_resource: keypair}
+ networks:
+ - network: { get_resource: network }
+
+outputs:
+ private_key:
+ description: "SSH Private Key"
+ value: { get_attr: [ keypair, private_key ]}
+##############################################################################
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+##############################################################################
heat_template_version: 2015-04-30
description: Simple template to deploy a single compute instance
label: Test subnet name
description: The name of the stack's subnet
default: test_subnet
+ inst_name:
+ type: string
+ label: Test VM name
+ description: The name of the spawned vm
+ default: test_vm
resources:
private_net:
my_instance:
type: OS::Nova::Server
properties:
+ name: { get_param: inst_name }
image: { get_param: image_name }
flavor: { get_param: flavor_name }
networks:
import yaml
from heatclient.client import Client
from heatclient.common.template_format import yaml_loader
+from novaclient.exceptions import NotFound
from oslo_serialization import jsonutils
from snaps import file_utils
-from snaps.domain.stack import Stack, Resource
+from snaps.domain.stack import Stack, Resource, Output
-from snaps.openstack.utils import keystone_utils, neutron_utils
+from snaps.openstack.utils import keystone_utils, neutron_utils, nova_utils
__author__ = 'spisarski'
return heat_cli.stacks.get(stack_id).stack_status
-def get_stack_outputs(heat_cli, stack_id):
- """
- Returns a domain Stack object for a given ID
- :param heat_cli: the OpenStack heat client
- :param stack_id: the ID of the heat stack to retrieve
- :return: the Stack domain object else None
- """
- stack = heat_cli.stacks.get(stack_id)
- return stack.outputs
-
-
def create_stack(heat_cli, stack_settings):
"""
Executes an Ansible playbook to the given host
return out
+def get_outputs(heat_cli, stack):
+ """
+ Returns all of the SNAPS-OO Output domain objects for the defined outputs
+ for given stack
+ :param heat_cli: the OpenStack heat client
+ :param stack: the SNAPS-OO Stack domain object
+ :return: a list
+ """
+ out = list()
+
+ os_stack = heat_cli.stacks.get(stack.id)
+
+ outputs = None
+ if os_stack:
+ outputs = os_stack.outputs
+
+ if outputs:
+ for output in outputs:
+ out.append(Output(**output))
+
+ return out
+
+
def get_stack_networks(heat_cli, neutron, stack):
"""
Returns an instance of NetworkSettings for each network owned by this stack
return out
+def get_stack_servers(heat_cli, nova, stack):
+ """
+ Returns an instance of NetworkSettings for each network owned by this stack
+ :param heat_cli: the OpenStack heat client object
+ :param nova: the OpenStack nova client object
+ :param stack: the SNAPS-OO Stack domain object
+ :return: a list of NetworkSettings
+ """
+
+ out = list()
+ resources = get_resources(heat_cli, stack)
+ for resource in resources:
+ if resource.type == 'OS::Nova::Server':
+ try:
+ server = nova_utils.get_server_object_by_id(
+ nova, resource.id)
+ if server:
+ out.append(server)
+ except NotFound:
+ logger.warn(
+ 'VmInst cannot be located with ID %s', resource.id)
+
+ return out
+
+
def parse_heat_template_str(tmpl_str):
"""
Takes a heat template string, performs some simple validation and returns a
neutron.delete_router(router=router.id)
+def get_router_by_id(neutron, router_id):
+ """
+ Returns a router with a given ID, else None if not found
+ :param neutron: the client
+ :param router_id: the Router ID
+ :return: a SNAPS-OO Router domain object
+ """
+ router = neutron.show_router(router_id)
+ if router:
+ return Router(**router['router'])
+
+
def get_router(neutron, router_settings=None, router_name=None):
"""
Returns the first router object (dictionary) found the given the settings
port_filter = dict()
if port_settings:
- port_filter['name'] = port_settings.name
+ if port_settings.name and len(port_settings.name) > 0:
+ port_filter['name'] = port_settings.name
if port_settings.admin_state_up:
port_filter['admin_state_up'] = port_settings.admin_state_up
if port_settings.device_id:
port_filter['device_id'] = port_settings.device_id
if port_settings.mac_address:
port_filter['mac_address'] = port_settings.mac_address
+ if port_settings.network_name:
+ network = get_network(neutron,
+ network_name=port_settings.network_name)
+ port_filter['network_id'] = network.id
elif port_name:
port_filter['name'] = port_name
ports = neutron.list_ports(**port_filter)
for port in ports['ports']:
- return Port(name=port['name'], id=port['id'],
- ips=port['fixed_ips'], mac_address=port['mac_address'])
+ return Port(**port)
+ return None
+
+
+def get_port_by_id(neutron, port_id):
+ """
+ Returns a SNAPS-OO Port domain object for the given ID or none if not found
+ :param neutron: the client
+ :param port_id: the to query
+ :return: a SNAPS-OO Port domain object or None
+ """
+ port = neutron.show_port(port_id)
+ if port:
+ return Port(**port['port'])
return None
+def get_ports(neutron, network, ips=None):
+ """
+ Returns a list of SNAPS-OO Port objects for all OpenStack Port objects that
+ are associated with the 'network' parameter
+ :param neutron: the client
+ :param network: SNAPS-OO Network domain object
+ :param ips: the IPs to lookup if not None
+ :return: a SNAPS-OO Port domain object or None if not found
+ """
+ out = list()
+ ports = neutron.list_ports(**{'network_id': network.id})
+ for port in ports['ports']:
+ if ips:
+ for fixed_ips in port['fixed_ips']:
+ if ('ip_address' in fixed_ips and
+ fixed_ips['ip_address'] in ips) or ips is None:
+ out.append(Port(**port))
+ break
+ else:
+ out.append(Port(**port))
+
+ return out
+
+
def create_security_group(neutron, keystone, sec_grp_settings):
"""
Creates a security group object in OpenStack
Returns all of the floating IPs
When ports is not None, FIPs returned must be associated with one of the
ports in the list and a tuple 2 where the first element being the port's
- name and the second being the FloatingIp SNAPS-OO domain object.
+ ID and the second being the FloatingIp SNAPS-OO domain object.
When ports is None, all known FloatingIp SNAPS-OO domain objects will be
returned in a list
:param neutron: the Neutron client
- :param ports: a list of SNAPS-OO Port objects to join
- :return: a list of tuple 2 (port_name, SNAPS FloatingIp) objects when ports
+ :param ports: a list of tuple 2 where index 0 is the port name and index 1
+ is the SNAPS-OO Port object
+ :return: a list of tuple 2 (port_id, SNAPS FloatingIp) objects when ports
is not None else a list of Port objects
"""
out = list()
for fip in fips['floatingips']:
if ports:
for port_name, port in ports:
- if fip['port_id'] == port.id:
- out.append((port.name, FloatingIp(
- inst_id=fip['id'], ip=fip['floating_ip_address'])))
+ if port and port.id == fip['port_id']:
+ out.append((port.id, FloatingIp(**fip)))
break
else:
- out.append(FloatingIp(inst_id=fip['id'],
- ip=fip['floating_ip_address']))
+ out.append(FloatingIp(**fip))
return out
body={'floatingip':
{'floating_network_id': ext_net.id}})
- return FloatingIp(inst_id=fip['floatingip']['id'],
+ return FloatingIp(id=fip['floatingip']['id'],
ip=fip['floatingip']['floating_ip_address'])
else:
raise NeutronException(
floating_ip.ip)
os_fip = __get_os_floating_ip(neutron, floating_ip)
if os_fip:
- return FloatingIp(
- inst_id=os_fip['id'], ip=os_fip['floating_ip_address'])
+ return FloatingIp(id=os_fip['id'], ip=os_fip['floating_ip_address'])
def __get_os_floating_ip(neutron, floating_ip):
def get_network_quotas(neutron, project_id):
"""
Returns a list of all available keypairs
- :param nova: the Nova client
+ :param neutron: the neutron client
:param project_id: the project's ID of the quotas to lookup
:return: an object of type NetworkQuotas or None if not found
"""
args['availability_zone'] = instance_settings.availability_zone
server = nova.servers.create(**args)
- return VmInst(name=server.name, inst_id=server.id,
- networks=server.networks)
+
+ return __map_os_server_obj_to_vm_inst(server)
else:
raise NovaException(
'Cannot create instance, image cannot be located with name %s',
servers = nova.servers.list(search_opts=search_opts)
for server in servers:
- return VmInst(name=server.name, inst_id=server.id,
- networks=server.networks)
+ return __map_os_server_obj_to_vm_inst(server)
+
+
+def __map_os_server_obj_to_vm_inst(os_server):
+ """
+ Returns a VmInst object for an OpenStack Server object
+ :param os_server: the OpenStack server object
+ :return: an equivalent SNAPS-OO VmInst domain object
+ """
+ sec_grp_names = list()
+ # VM must be active for 'security_groups' attr to be initialized
+ if hasattr(os_server, 'security_groups'):
+ for sec_group in os_server.security_groups:
+ if sec_group.get('name'):
+ sec_grp_names.append(sec_group.get('name'))
+
+ return VmInst(
+ name=os_server.name, inst_id=os_server.id,
+ image_id=os_server.image['id'], flavor_id=os_server.flavor['id'],
+ networks=os_server.networks, keypair_name=os_server.key_name,
+ sec_grp_names=sec_grp_names)
def __get_latest_server_os_object(nova, server):
:param server: the domain VmInst object
:return: the list of servers or None if not found
"""
- return nova.servers.get(server.id)
+ return __get_latest_server_os_object_by_id(nova, server.id)
+
+
+def __get_latest_server_os_object_by_id(nova, server_id):
+ """
+ Returns a server with a given id
+ :param nova: the Nova client
+ :param server_id: the server's ID
+ :return: the list of servers or None if not found
+ """
+ return nova.servers.get(server_id)
def get_server_status(nova, server):
:return: the list of servers or None if not found
"""
server = __get_latest_server_os_object(nova, server)
- return VmInst(name=server.name, inst_id=server.id,
- networks=server.networks)
+ return __map_os_server_obj_to_vm_inst(server)
+
+
+def get_server_object_by_id(nova, server_id):
+ """
+ Returns a server with a given id
+ :param nova: the Nova client
+ :param server_id: the server's id
+ :return: an SNAPS-OO VmInst object or None if not found
+ """
+ server = __get_latest_server_os_object_by_id(nova, server_id)
+ return __map_os_server_obj_to_vm_inst(server)
def get_server_security_group_names(nova, server):
serialization.PublicFormat.OpenSSH)
-def save_keys_to_files(keys=None, pub_file_path=None, priv_file_path=None):
- """
- Saves the generated RSA generated keys to the filesystem
- :param keys: the keys to save generated by cryptography
- :param pub_file_path: the path to the public keys
- :param priv_file_path: the path to the private keys
- """
- if keys:
- if pub_file_path:
- # To support '~'
- pub_expand_file = os.path.expanduser(pub_file_path)
- pub_dir = os.path.dirname(pub_expand_file)
-
- if not os.path.isdir(pub_dir):
- os.mkdir(pub_dir)
-
- public_handle = None
- try:
- public_handle = open(pub_expand_file, 'wb')
- public_bytes = keys.public_key().public_bytes(
- serialization.Encoding.OpenSSH,
- serialization.PublicFormat.OpenSSH)
- public_handle.write(public_bytes)
- finally:
- if public_handle:
- public_handle.close()
-
- os.chmod(pub_expand_file, 0o400)
- logger.info("Saved public key to - " + pub_expand_file)
- if priv_file_path:
- # To support '~'
- priv_expand_file = os.path.expanduser(priv_file_path)
- priv_dir = os.path.dirname(priv_expand_file)
- if not os.path.isdir(priv_dir):
- os.mkdir(priv_dir)
-
- private_handle = None
- try:
- private_handle = open(priv_expand_file, 'wb')
- private_handle.write(
- keys.private_bytes(
- encoding=serialization.Encoding.PEM,
- format=serialization.PrivateFormat.TraditionalOpenSSL,
- encryption_algorithm=serialization.NoEncryption()))
- finally:
- if private_handle:
- private_handle.close()
-
- os.chmod(priv_expand_file, 0o400)
- logger.info("Saved private key to - " + priv_expand_file)
-
-
def upload_keypair_file(nova, name, file_path):
"""
Uploads a public key from a file
"""
logger.info('Creating keypair with name - ' + name)
os_kp = nova.keypairs.create(name=name, public_key=key.decode('utf-8'))
- return Keypair(name=os_kp.name, id=os_kp.id, public_key=os_kp.public_key)
+ return Keypair(name=os_kp.name, kp_id=os_kp.id,
+ public_key=os_kp.public_key, fingerprint=os_kp.fingerprint)
def keypair_exists(nova, keypair_obj):
"""
try:
os_kp = nova.keypairs.get(keypair_obj)
- return Keypair(name=os_kp.name, id=os_kp.id,
+ return Keypair(name=os_kp.name, kp_id=os_kp.id,
public_key=os_kp.public_key)
except:
return None
for keypair in keypairs:
if keypair.name == name:
- return Keypair(name=keypair.name, id=keypair.id,
+ return Keypair(name=keypair.name, kp_id=keypair.id,
public_key=keypair.public_key)
return None
nova.servers.delete(vm_inst.id)
-def __get_os_flavor(nova, flavor):
+def __get_os_flavor(nova, flavor_id):
"""
Returns to OpenStack flavor object by name
:param nova: the Nova client
- :param flavor: the SNAPS flavor domain object
+ :param flavor_id: the flavor's ID value
:return: the OpenStack Flavor object
"""
try:
- return nova.flavors.get(flavor.id)
+ return nova.flavors.get(flavor_id)
except NotFound:
return None
:param flavor: the SNAPS flavor domain object
:return: the SNAPS Flavor domain object
"""
- os_flavor = __get_os_flavor(nova, flavor)
+ os_flavor = __get_os_flavor(nova, flavor.id)
if os_flavor:
return Flavor(
name=os_flavor.name, id=os_flavor.id, ram=os_flavor.ram,
return None
+def get_flavor_by_id(nova, flavor_id):
+ """
+ Returns to OpenStack flavor object by name
+ :param nova: the Nova client
+ :param flavor_id: the flavor ID value
+ :return: the SNAPS Flavor domain object
+ """
+ os_flavor = __get_os_flavor(nova, flavor_id)
+ if os_flavor:
+ return Flavor(
+ name=os_flavor.name, id=os_flavor.id, ram=os_flavor.ram,
+ disk=os_flavor.disk, vcpus=os_flavor.vcpus,
+ ephemeral=os_flavor.ephemeral, swap=os_flavor.swap,
+ rxtx_factor=os_flavor.rxtx_factor, is_public=os_flavor.is_public)
+
+
def __get_os_flavor_by_name(nova, name):
"""
Returns to OpenStack flavor object by name
:param flavor: the SNAPS flavor domain object
:param metadata: the metadata to set
"""
- os_flavor = __get_os_flavor(nova, flavor)
+ os_flavor = __get_os_flavor(nova, flavor.id)
if os_flavor:
os_flavor.set_keys(metadata)
:param nova: the Nova client
:param flavor: the SNAPS flavor domain object
"""
- os_flavor = __get_os_flavor(nova, flavor)
+ os_flavor = __get_os_flavor(nova, flavor.id)
if os_flavor:
return os_flavor.get_keys()
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import uuid
+
+from snaps import file_utils
+from snaps.openstack.create_instance import (
+ VmInstanceSettings, FloatingIpSettings)
+from snaps.openstack.create_keypairs import KeypairSettings
+from snaps.openstack.create_network import (
+ PortSettings, SubnetSettings, NetworkSettings)
+from snaps.openstack.utils import (
+ neutron_utils, nova_utils, heat_utils, glance_utils)
+
+
+def create_network_settings(neutron, network):
+ """
+ Returns a NetworkSettings object
+ :param neutron: the neutron client
+ :param network: a SNAPS-OO Network domain object
+ :return:
+ """
+ return NetworkSettings(
+ name=network.name, network_type=network.type,
+ subnet_settings=create_subnet_settings(neutron, network))
+
+
+def create_subnet_settings(neutron, network):
+ """
+ Returns a list of SubnetSettings objects for a given network
+ :param neutron: the OpenStack neutron client
+ :param network: the SNAPS-OO Network domain object
+ :return: a list
+ """
+ out = list()
+
+ subnets = neutron_utils.get_subnets_by_network(neutron, network)
+ for subnet in subnets:
+ kwargs = dict()
+ kwargs['cidr'] = subnet.cidr
+ kwargs['ip_version'] = subnet.ip_version
+ kwargs['name'] = subnet.name
+ kwargs['start'] = subnet.start
+ kwargs['end'] = subnet.end
+ kwargs['gateway_ip'] = subnet.gateway_ip
+ kwargs['enable_dhcp'] = subnet.enable_dhcp
+ kwargs['dns_nameservers'] = subnet.dns_nameservers
+ kwargs['host_routes'] = subnet.host_routes
+ kwargs['ipv6_ra_mode'] = subnet.ipv6_ra_mode
+ kwargs['ipv6_address_mode'] = subnet.ipv6_address_mode
+ out.append(SubnetSettings(**kwargs))
+ return out
+
+
+def create_vm_inst_settings(nova, neutron, server):
+ """
+ Returns a NetworkSettings object
+ :param nova: the nova client
+ :param neutron: the neutron client
+ :param server: a SNAPS-OO VmInst domain object
+ :return:
+ """
+
+ flavor_name = nova_utils.get_flavor_by_id(nova, server.flavor_id)
+
+ kwargs = dict()
+ kwargs['name'] = server.name
+ kwargs['flavor'] = flavor_name
+ kwargs['port_settings'] = __create_port_settings(
+ neutron, server.networks)
+ kwargs['security_group_names'] = server.sec_grp_names
+ kwargs['floating_ip_settings'] = __create_floatingip_settings(
+ neutron, kwargs['port_settings'])
+
+ return VmInstanceSettings(**kwargs)
+
+
+def __create_port_settings(neutron, networks):
+ """
+ Returns a list of port settings based on the networks parameter
+ :param neutron: the neutron client
+ :param networks: a dict where the key is the network name and the value
+ is a list of IP addresses
+ :return:
+ """
+ out = list()
+
+ for net_name, ips in networks.items():
+ network = neutron_utils.get_network(neutron, network_name=net_name)
+ ports = neutron_utils.get_ports(neutron, network, ips)
+ for port in ports:
+ kwargs = dict()
+ if port.name:
+ kwargs['name'] = port.name
+ kwargs['network_name'] = network.name
+ kwargs['mac_address'] = port.mac_address
+ kwargs['allowed_address_pairs'] = port.allowed_address_pairs
+ kwargs['admin_state_up'] = port.admin_state_up
+ out.append(PortSettings(**kwargs))
+
+ return out
+
+
+def __create_floatingip_settings(neutron, port_settings):
+ """
+ Returns a list of FloatingIPSettings objects as they pertain to an
+ existing deployed server instance
+ :param neutron: the neutron client
+ :param port_settings: list of SNAPS-OO PortSettings objects
+ :return: a list of FloatingIPSettings objects or an empty list if no
+ floating IPs have been created
+ """
+ base_fip_name = 'fip-'
+ fip_ctr = 1
+ out = list()
+
+ fip_ports = list()
+ for port_setting in port_settings:
+ setting_port = neutron_utils.get_port(neutron, port_setting)
+ if setting_port:
+ network = neutron_utils.get_network(
+ neutron, network_name=port_setting.network_name)
+ network_ports = neutron_utils.get_ports(neutron, network)
+ if network_ports:
+ for setting_port in network_ports:
+ if port_setting.mac_address == setting_port.mac_address:
+ fip_ports.append((port_setting.name, setting_port))
+ break
+
+ floating_ips = neutron_utils.get_floating_ips(neutron, fip_ports)
+
+ for port_id, floating_ip in floating_ips:
+ router = neutron_utils.get_router_by_id(neutron, floating_ip.router_id)
+ setting_port = neutron_utils.get_port_by_id(
+ neutron, floating_ip.port_id)
+ kwargs = dict()
+ kwargs['name'] = base_fip_name + str(fip_ctr)
+ kwargs['port_name'] = setting_port.name
+ kwargs['port_id'] = setting_port.id
+ kwargs['router_name'] = router.name
+
+ if setting_port:
+ for ip_dict in setting_port.ips:
+ if ('ip_address' in ip_dict and
+ 'subnet_id' in ip_dict and
+ ip_dict['ip_address'] == floating_ip.fixed_ip_address):
+ subnet = neutron_utils.get_subnet_by_id(
+ neutron, ip_dict['subnet_id'])
+ if subnet:
+ kwargs['subnet_name'] = subnet.name
+
+ out.append(FloatingIpSettings(**kwargs))
+
+ fip_ctr += 1
+
+ return out
+
+
+def determine_image_settings(glance, server, image_settings):
+ """
+ Returns a ImageSettings object from the list that matches the name in one
+ of the image_settings parameter
+ :param glance: the glance client
+ :param server: a SNAPS-OO VmInst domain object
+ :param image_settings: list of ImageSettings objects
+ :return: ImageSettings or None
+ """
+ if image_settings:
+ for image_setting in image_settings:
+ image = glance_utils.get_image_by_id(glance, server.image_id)
+ if image and image.name == image_setting.name:
+ return image_setting
+
+
+def determine_keypair_settings(heat_cli, stack, server, keypair_settings=None,
+ priv_key_key=None):
+ """
+ Returns a KeypairSettings object from the list that matches the
+ server.keypair_name value in the keypair_settings parameter if not None,
+ else if the output_key is not None, the output's value when contains the
+ string 'BEGIN RSA PRIVATE KEY', this value will be stored into a file and
+ encoded into the KeypairSettings object returned
+ :param heat_cli: the OpenStack heat client
+ :param stack: a SNAPS-OO Stack domain object
+ :param server: a SNAPS-OO VmInst domain object
+ :param keypair_settings: list of KeypairSettings objects
+ :param priv_key_key: the stack options that holds the private key value
+ :return: KeypairSettings or None
+ """
+ # Existing keypair being used by Heat Template
+ if keypair_settings:
+ for keypair_setting in keypair_settings:
+ if server.keypair_name == keypair_setting.name:
+ return keypair_setting
+
+ # Keypair created by Heat template
+ if priv_key_key:
+ outputs = heat_utils.get_outputs(heat_cli, stack)
+ for output in outputs:
+ if output.key == priv_key_key:
+ # Save to file
+ guid = uuid.uuid4()
+ key_file = file_utils.save_string_to_file(
+ output.value, str(guid), 0o400)
+
+ # Use outputs, file and resources for the KeypairSettings
+ return KeypairSettings(
+ name=server.keypair_name, private_filepath=key_file.name)
from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_instance import OpenStackVmInstance
from snaps.openstack.create_stack import StackSettings
from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
-from snaps.openstack.utils import heat_utils, neutron_utils
+from snaps.openstack.utils import (
+ heat_utils, neutron_utils, nova_utils, settings_utils, glance_utils)
__author__ = 'spisarski'
Tests to ensure that the heat client can communicate with the cloud
"""
- def test_nova_connect_success(self):
+ def test_heat_connect_success(self):
"""
Tests to ensure that the proper credentials can connect.
"""
for stack in stacks:
print stack
- def test_nova_connect_fail(self):
+ def test_heat_connect_fail(self):
"""
Tests to ensure that the improper credentials cannot connect.
"""
print stack
-class HeatUtilsCreateStackTests(OSComponentTestCase):
+class HeatUtilsCreateSimpleStackTests(OSComponentTestCase):
"""
Test basic Heat functionality
"""
stack_name2 = guid + '-stack2'
self.network_name = guid + '-net'
self.subnet_name = guid + '-subnet'
+ self.vm_inst_name = guid + '-inst'
self.image_creator = OpenStackImage(
self.os_creds, openstack_tests.cirros_image_settings(
env_values = {'image_name': self.image_creator.image_settings.name,
'flavor_name': self.flavor_creator.flavor_settings.name,
'net_name': self.network_name,
- 'subnet_name': self.subnet_name}
+ 'subnet_name': self.subnet_name,
+ 'inst_name': self.vm_inst_name}
heat_tmplt_path = pkg_resources.resource_filename(
'snaps.openstack.tests.heat', 'test_heat_template.yaml')
self.stack_settings1 = StackSettings(
self.stack1.id)
self.assertEqual(self.stack1, stack_query_3)
- outputs = heat_utils.get_stack_outputs(
- self.heat_client, self.stack1.id)
+ resources = heat_utils.get_resources(self.heat_client, self.stack1)
+ self.assertIsNotNone(resources)
+ self.assertEqual(4, len(resources))
+
+ outputs = heat_utils.get_outputs(self.heat_client, self.stack1)
self.assertIsNotNone(outputs)
self.assertEqual(0, len(outputs))
+ # Wait until stack deployment has completed
end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
-
is_active = False
while time.time() < end_time:
status = heat_utils.get_stack_status(self.heat_client,
self.assertTrue(is_active)
- resources = heat_utils.get_resources(self.heat_client, self.stack1)
- self.assertIsNotNone(resources)
- self.assertEqual(4, len(resources))
-
neutron = neutron_utils.neutron_client(self.os_creds)
networks = heat_utils.get_stack_networks(
self.heat_client, neutron, self.stack1)
self.assertEqual(1, len(subnets))
self.assertEqual(self.subnet_name, subnets[0].name)
+ nova = nova_utils.nova_client(self.os_creds)
+ servers = heat_utils.get_stack_servers(
+ self.heat_client, nova, self.stack1)
+ self.assertIsNotNone(servers)
+ self.assertEqual(1, len(servers))
+ self.assertEqual(self.vm_inst_name, servers[0].name)
+
def test_create_stack_x2(self):
"""
Tests the creation of an OpenStack keypair that does not exist.
self.stack1.id)
self.assertEqual(self.stack1, stack1_query_3)
- outputs = heat_utils.get_stack_outputs(self.heat_client,
- self.stack1.id)
- self.assertIsNotNone(outputs)
- self.assertEqual(0, len(outputs))
-
end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
-
is_active = False
while time.time() < end_time:
status = heat_utils.get_stack_status(self.heat_client,
self.stack2.id)
self.assertEqual(self.stack2, stack2_query_3)
- outputs = heat_utils.get_stack_outputs(self.heat_client,
- self.stack2.id)
- self.assertIsNotNone(outputs)
- self.assertEqual(0, len(outputs))
-
end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
is_active = False
time.sleep(3)
self.assertTrue(is_active)
+
+
+class HeatUtilsCreateComplexStackTests(OSComponentTestCase):
+ """
+ Test basic Heat functionality
+ """
+
+ def setUp(self):
+ """
+ Instantiates OpenStack instances that cannot be spawned by Heat
+ """
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ stack_name = guid + '-stack'
+ self.network_name = guid + '-net'
+ self.subnet_name = guid + '-subnet'
+ self.vm_inst1_name = guid + '-inst1'
+ self.vm_inst2_name = guid + '-inst2'
+ self.flavor1_name = guid + '-flavor1'
+ self.flavor2_name = guid + '-flavor2'
+ self.keypair_name = guid + '-keypair'
+
+ self.image_creator1 = OpenStackImage(
+ self.os_creds, openstack_tests.cirros_image_settings(
+ name=guid + '-image1', image_metadata=self.image_metadata))
+ self.image_creator1.create()
+
+ self.image_creator2 = OpenStackImage(
+ self.os_creds, openstack_tests.cirros_image_settings(
+ name=guid + '-image2', image_metadata=self.image_metadata))
+ self.image_creator2.create()
+
+ env_values = {'image1_name': self.image_creator1.image_settings.name,
+ 'image2_name': self.image_creator2.image_settings.name,
+ 'flavor1_name': self.flavor1_name,
+ 'flavor2_name': self.flavor2_name,
+ 'net_name': self.network_name,
+ 'subnet_name': self.subnet_name,
+ 'keypair_name': self.keypair_name,
+ 'inst1_name': self.vm_inst1_name,
+ 'inst2_name': self.vm_inst2_name,
+ 'external_net_name': self.ext_net_name}
+ heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'floating_ip_heat_template.yaml')
+ stack_settings = StackSettings(
+ name=stack_name, template_path=heat_tmplt_path,
+ env_values=env_values)
+ self.heat_client = heat_utils.heat_client(self.os_creds)
+ self.stack = heat_utils.create_stack(self.heat_client, stack_settings)
+
+ # Wait until stack deployment has completed
+ end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
+ is_active = False
+ while time.time() < end_time:
+ status = heat_utils.get_stack_status(self.heat_client,
+ self.stack.id)
+ if status == create_stack.STATUS_CREATE_COMPLETE:
+ is_active = True
+ break
+ elif status == create_stack.STATUS_CREATE_FAILED:
+ is_active = False
+ break
+
+ time.sleep(3)
+ self.assertTrue(is_active)
+
+ def tearDown(self):
+ """
+ Cleans the image and downloaded image file
+ """
+ if self.stack:
+ try:
+ heat_utils.delete_stack(self.heat_client, self.stack)
+ # Wait until stack deployment has completed
+ end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
+ is_deleted = False
+ while time.time() < end_time:
+ status = heat_utils.get_stack_status(self.heat_client,
+ self.stack.id)
+ if status == create_stack.STATUS_DELETE_COMPLETE:
+ is_deleted = True
+ break
+ elif status == create_stack.STATUS_DELETE_FAILED:
+ is_deleted = False
+ break
+
+ time.sleep(3)
+
+ if not is_deleted:
+ nova = nova_utils.nova_client(self.os_creds)
+ neutron = neutron_utils.neutron_client(self.os_creds)
+ glance = glance_utils.glance_client(self.os_creds)
+ servers = heat_utils.get_stack_servers(
+ self.heat_client, nova, self.stack)
+ for server in servers:
+ vm_settings = settings_utils.create_vm_inst_settings(
+ nova, neutron, server)
+ img_settings = settings_utils.determine_image_settings(
+ glance, server,
+ [self.image_creator1.image_settings,
+ self.image_creator2.image_settings])
+ vm_creator = OpenStackVmInstance(
+ self.os_creds, vm_settings, img_settings)
+ vm_creator.create(cleanup=False)
+ vm_creator.clean()
+ vm_creator.vm_deleted(block=True)
+
+ heat_utils.delete_stack(self.heat_client, self.stack)
+ time.sleep(20)
+ except:
+ raise
+
+ if self.image_creator1:
+ try:
+ self.image_creator1.clean()
+ except:
+ pass
+
+ if self.image_creator2:
+ try:
+ self.image_creator2.clean()
+ except:
+ pass
+
+ def test_get_settings_from_stack(self):
+ """
+ Tests that a heat template with floating IPs and can have the proper
+ settings derived from settings_utils.py.
+ """
+ resources = heat_utils.get_resources(self.heat_client, self.stack)
+ self.assertIsNotNone(resources)
+ self.assertEqual(11, len(resources))
+
+ options = heat_utils.get_outputs(self.heat_client, self.stack)
+ self.assertIsNotNone(options)
+ self.assertEqual(1, len(options))
+
+ neutron = neutron_utils.neutron_client(self.os_creds)
+ networks = heat_utils.get_stack_networks(
+ self.heat_client, neutron, self.stack)
+ self.assertIsNotNone(networks)
+ self.assertEqual(1, len(networks))
+ self.assertEqual(self.network_name, networks[0].name)
+
+ network_settings = settings_utils.create_network_settings(
+ neutron, networks[0])
+ self.assertIsNotNone(network_settings)
+ self.assertEqual(self.network_name, network_settings.name)
+
+ nova = nova_utils.nova_client(self.os_creds)
+ glance = glance_utils.glance_client(self.os_creds)
+
+ servers = heat_utils.get_stack_servers(
+ self.heat_client, nova, self.stack)
+ self.assertIsNotNone(servers)
+ self.assertEqual(2, len(servers))
+
+ image_settings = settings_utils.determine_image_settings(
+ glance, servers[0],
+ [self.image_creator1.image_settings,
+ self.image_creator2.image_settings])
+
+ self.assertIsNotNone(image_settings)
+ if image_settings.name.endswith('1'):
+ self.assertEqual(
+ self.image_creator1.image_settings.name, image_settings.name)
+ else:
+ self.assertEqual(
+ self.image_creator2.image_settings.name, image_settings.name)
+
+ image_settings = settings_utils.determine_image_settings(
+ glance, servers[1],
+ [self.image_creator1.image_settings,
+ self.image_creator2.image_settings])
+ if image_settings.name.endswith('1'):
+ self.assertEqual(
+ self.image_creator1.image_settings.name, image_settings.name)
+ else:
+ self.assertEqual(
+ self.image_creator2.image_settings.name, image_settings.name)
+
+ keypair1_settings = settings_utils.determine_keypair_settings(
+ self.heat_client, self.stack, servers[0],
+ priv_key_key='private_key')
+ self.assertIsNotNone(keypair1_settings)
+ self.assertEqual(self.keypair_name, keypair1_settings.name)
+
+ keypair2_settings = settings_utils.determine_keypair_settings(
+ self.heat_client, self.stack, servers[1],
+ priv_key_key='private_key')
+ self.assertIsNotNone(keypair2_settings)
+ self.assertEqual(self.keypair_name, keypair2_settings.name)
def test_create_port_null_name(self):
"""
- Tests the neutron_utils.create_port() function for an Exception when
- the port name value is None
+ Tests the neutron_utils.create_port() when the port name value is None
"""
self.network = neutron_utils.create_network(
self.neutron, self.os_creds, self.net_config.network_settings)
self.assertTrue(validate_subnet(
self.neutron, subnet_setting.name, subnet_setting.cidr, True))
- with self.assertRaises(Exception):
- self.port = neutron_utils.create_port(
- self.neutron, self.os_creds,
- PortSettings(
- network_name=self.net_config.network_settings.name,
- ip_addrs=[{
- 'subnet_name': subnet_setting.name,
- 'ip': ip_1}]))
+ self.port = neutron_utils.create_port(
+ self.neutron, self.os_creds,
+ PortSettings(
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{
+ 'subnet_name': subnet_setting.name,
+ 'ip': ip_1}]))
+
+ port = neutron_utils.get_port_by_id(self.neutron, self.port.id)
+ self.assertEqual(self.port, port)
def test_create_port_null_network_object(self):
"""
import uuid
import os
+import time
+
+from snaps import file_utils
+from snaps.openstack import create_instance
from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
from snaps.openstack.create_image import OpenStackImage
from snaps.openstack.create_instance import VmInstanceSettings
Tests that the generated RSA keys are properly saved to files
:return:
"""
- nova_utils.save_keys_to_files(self.keys, self.pub_key_file_path,
+ file_utils.save_keys_to_files(self.keys, self.pub_key_file_path,
self.priv_key_file_path)
self.keypair = nova_utils.upload_keypair_file(self.nova,
self.keypair_name,
self.assertIsNotNone(self.vm_inst)
+ # Wait until instance is ACTIVE
+ iters = 0
+ active = False
+ while iters < 60:
+ if create_instance.STATUS_ACTIVE == nova_utils.get_server_status(
+ self.nova, self.vm_inst):
+ active = True
+ break
+
+ time.sleep(3)
+ iters += 1
+
+ self.assertTrue(active)
vm_inst = nova_utils.get_latest_server_object(self.nova, self.vm_inst)
self.assertEqual(self.vm_inst.name, vm_inst.name)
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+import os
+import uuid
+
+from snaps.openstack import (
+ create_image, create_network, create_router, create_flavor,
+ create_keypairs, create_instance)
+from snaps.openstack.create_network import (
+ NetworkSettings, OpenStackNetwork, SubnetSettings)
+from snaps.openstack.create_security_group import (
+ SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup,
+ SecurityGroupSettings)
+from snaps.openstack.tests import openstack_tests
+from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
+from snaps.openstack.utils import (
+ neutron_utils, settings_utils, nova_utils, glance_utils)
+
+__author__ = 'spisarski'
+
+logger = logging.getLogger('nova_utils_tests')
+
+
+class SettingsUtilsNetworkingTests(OSComponentTestCase):
+ """
+ Tests the ability to reverse engineer NetworkSettings objects from existing
+ networks deployed to OpenStack
+ """
+
+ def setUp(self):
+ """
+ Instantiates OpenStack instances that cannot be spawned by Heat
+ """
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.network_name = guid + '-net'
+ self.subnet_name = guid + '-subnet'
+ self.net_creator = None
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+
+ def tearDown(self):
+ """
+ Cleans the image and downloaded image file
+ """
+ if self.net_creator:
+ try:
+ self.net_creator.clean()
+ except:
+ pass
+
+ def test_derive_net_settings_no_subnet(self):
+ """
+ Validates the utility function settings_utils#create_network_settings
+ returns an acceptable NetworkSettings object and ensures that the
+ new settings object will not cause the new OpenStackNetwork instance
+ to create another network
+ """
+ net_settings = NetworkSettings(name=self.network_name)
+ self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
+ network = self.net_creator.create()
+
+ derived_settings = settings_utils.create_network_settings(
+ self.neutron, network)
+
+ self.assertIsNotNone(derived_settings)
+ self.assertEqual(net_settings.name, derived_settings.name)
+ self.assertEqual(net_settings.admin_state_up,
+ derived_settings.admin_state_up)
+ self.assertEqual(net_settings.external, derived_settings.external)
+ self.assertEqual(len(net_settings.subnet_settings),
+ len(derived_settings.subnet_settings))
+
+ net_creator = OpenStackNetwork(self.os_creds, derived_settings)
+ derived_network = net_creator.create()
+
+ self.assertEqual(network, derived_network)
+
+ def test_derive_net_settings_two_subnets(self):
+ """
+ Validates the utility function settings_utils#create_network_settings
+ returns an acceptable NetworkSettings object
+ """
+ subnet_settings = list()
+ subnet_settings.append(SubnetSettings(name='sub1', cidr='10.0.0.0/24'))
+ subnet_settings.append(SubnetSettings(name='sub2', cidr='10.0.1.0/24'))
+ net_settings = NetworkSettings(name=self.network_name,
+ subnet_settings=subnet_settings)
+ self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
+ network = self.net_creator.create()
+
+ derived_settings = settings_utils.create_network_settings(
+ self.neutron, network)
+
+ self.assertIsNotNone(derived_settings)
+ self.assertEqual(net_settings.name, derived_settings.name)
+ self.assertEqual(net_settings.admin_state_up,
+ derived_settings.admin_state_up)
+ self.assertEqual(net_settings.external, derived_settings.external)
+ self.assertEqual(len(net_settings.subnet_settings),
+ len(derived_settings.subnet_settings))
+
+ # Validate the first subnet
+ orig_sub1 = net_settings.subnet_settings[0]
+ found = False
+ for derived_sub in derived_settings.subnet_settings:
+ if orig_sub1.name == derived_sub.name:
+ self.assertEqual(orig_sub1.cidr, derived_sub.cidr)
+ found = True
+
+ self.assertTrue(found)
+
+ # Validate the second subnet
+ orig_sub2 = net_settings.subnet_settings[1]
+ found = False
+ for derived_sub in derived_settings.subnet_settings:
+ if orig_sub2.name == derived_sub.name:
+ self.assertEqual(orig_sub2.cidr, derived_sub.cidr)
+ self.assertEqual(orig_sub2.ip_version, derived_sub.ip_version)
+ found = True
+
+ self.assertTrue(found)
+
+
+class SettingsUtilsVmInstTests(OSComponentTestCase):
+ """
+ Tests the ability to reverse engineer VmInstanceSettings objects from
+ existing VMs/servers deployed to OpenStack
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
+ """
+ # super(self.__class__, self).__start__()
+
+ self.nova = nova_utils.nova_client(self.os_creds)
+ self.glance = glance_utils.glance_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.keypair_priv_filepath = 'tmp/' + guid
+ self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
+ self.keypair_name = guid + '-kp'
+ self.vm_inst_name = guid + '-inst'
+ self.test_file_local_path = 'tmp/' + guid + '-hello.txt'
+ self.port_1_name = guid + '-port-1'
+ self.port_2_name = guid + '-port-2'
+ self.floating_ip_name = guid + 'fip1'
+
+ # Setup members to cleanup just in case they don't get created
+ self.inst_creator = None
+ self.keypair_creator = None
+ self.sec_grp_creator = None
+ self.flavor_creator = None
+ self.router_creator = None
+ self.network_creator = None
+ self.image_creator = None
+
+ try:
+ # Create Image
+ os_image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-' + '-image',
+ image_metadata=self.image_metadata)
+ self.image_creator = create_image.OpenStackImage(self.os_creds,
+ os_image_settings)
+ self.image_creator.create()
+
+ # First network is public
+ self.pub_net_config = openstack_tests.get_pub_net_config(
+ net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
+ router_name=guid + '-pub-router',
+ external_net=self.ext_net_name)
+
+ self.network_creator = create_network.OpenStackNetwork(
+ self.os_creds, self.pub_net_config.network_settings)
+ self.network_creator.create()
+
+ # Create routers
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, self.pub_net_config.router_settings)
+ self.router_creator.create()
+
+ # Create Flavor
+ self.flavor_creator = create_flavor.OpenStackFlavor(
+ self.os_creds,
+ create_flavor.FlavorSettings(name=guid + '-flavor-name',
+ ram=256, disk=1, vcpus=1))
+ self.flavor_creator.create()
+
+ # Create Key/Pair
+ self.keypair_creator = create_keypairs.OpenStackKeypair(
+ self.os_creds, create_keypairs.KeypairSettings(
+ name=self.keypair_name,
+ public_filepath=self.keypair_pub_filepath,
+ private_filepath=self.keypair_priv_filepath))
+ self.keypair_creator.create()
+
+ # Create Security Group
+ sec_grp_name = guid + '-sec-grp'
+ rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
+ protocol=Protocol.icmp)
+ rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
+ protocol=Protocol.tcp,
+ port_range_min=22,
+ port_range_max=22)
+ self.sec_grp_creator = OpenStackSecurityGroup(
+ self.os_creds,
+ SecurityGroupSettings(name=sec_grp_name,
+ rule_settings=[rule1, rule2]))
+ self.sec_grp_creator.create()
+
+ # Create instance
+ ports_settings = list()
+ ports_settings.append(
+ create_network.PortSettings(
+ name=self.port_1_name,
+ network_name=self.pub_net_config.network_settings.name))
+
+ instance_settings = create_instance.VmInstanceSettings(
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=ports_settings,
+ floating_ip_settings=[create_instance.FloatingIpSettings(
+ name=self.floating_ip_name, port_name=self.port_1_name,
+ router_name=self.pub_net_config.router_settings.name)])
+
+ self.inst_creator = create_instance.OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
+ keypair_settings=self.keypair_creator.keypair_settings)
+ except:
+ self.tearDown()
+ raise
+
+ def tearDown(self):
+ """
+ Cleans the created objects
+ """
+ if self.inst_creator:
+ try:
+ self.inst_creator.clean()
+ except:
+ pass
+
+ if self.sec_grp_creator:
+ try:
+ self.sec_grp_creator.clean()
+ except:
+ pass
+
+ if self.keypair_creator:
+ try:
+ self.keypair_creator.clean()
+ except:
+ pass
+
+ if self.flavor_creator:
+ try:
+ self.flavor_creator.clean()
+ except:
+ pass
+
+ if os.path.isfile(self.keypair_pub_filepath):
+ try:
+ os.remove(self.keypair_pub_filepath)
+ except:
+ pass
+
+ if os.path.isfile(self.keypair_priv_filepath):
+ try:
+ os.remove(self.keypair_priv_filepath)
+ except:
+ pass
+
+ if self.router_creator:
+ try:
+ self.router_creator.clean()
+ except:
+ pass
+
+ if self.network_creator:
+ try:
+ self.network_creator.clean()
+ except:
+ pass
+
+ if self.image_creator and not self.image_creator.image_settings.exists:
+ try:
+ self.image_creator.clean()
+ except:
+ pass
+
+ if os.path.isfile(self.test_file_local_path):
+ os.remove(self.test_file_local_path)
+
+ # super(self.__class__, self).__clean__()
+
+ def test_derive_vm_inst_settings(self):
+ """
+ Validates the utility function settings_utils#create_vm_inst_settings
+ returns an acceptable VmInstanceSettings object
+ """
+ self.inst_creator.create(block=True)
+
+ server = nova_utils.get_server(
+ self.nova, vm_inst_settings=self.inst_creator.instance_settings)
+ derived_vm_settings = settings_utils.create_vm_inst_settings(
+ self.nova, self.neutron, server)
+ self.assertIsNotNone(derived_vm_settings)
+ self.assertIsNotNone(derived_vm_settings.port_settings)
+ self.assertIsNotNone(derived_vm_settings.floating_ip_settings)
+
+ def test_derive_image_settings(self):
+ """
+ Validates the utility function settings_utils#create_image_settings
+ returns an acceptable ImageSettings object
+ """
+ self.inst_creator.create(block=True)
+
+ server = nova_utils.get_server(
+ self.nova, vm_inst_settings=self.inst_creator.instance_settings)
+ derived_image_settings = settings_utils.determine_image_settings(
+ self.glance, server, [self.image_creator.image_settings])
+ self.assertIsNotNone(derived_image_settings)
+ self.assertEqual(self.image_creator.image_settings.name,
+ derived_image_settings.name)
CreateSecurityGroupTests, SecurityGroupRuleSettingsUnitTests,
SecurityGroupSettingsUnitTests)
from snaps.openstack.tests.create_stack_tests import (
- StackSettingsUnitTests, CreateStackSuccessTests, CreateStackNegativeTests)
+ StackSettingsUnitTests, CreateStackSuccessTests, CreateStackNegativeTests,
+ CreateComplexStackTests)
from snaps.openstack.tests.create_user_tests import (
UserSettingsUnitTests, CreateUserSuccessTests)
from snaps.openstack.tests.os_source_file_test import (
from snaps.openstack.utils.tests.glance_utils_tests import (
GlanceSmokeTests, GlanceUtilsTests)
from snaps.openstack.utils.tests.heat_utils_tests import (
- HeatUtilsCreateStackTests, HeatSmokeTests)
+ HeatSmokeTests, HeatUtilsCreateSimpleStackTests,
+ HeatUtilsCreateComplexStackTests)
from snaps.openstack.utils.tests.keystone_utils_tests import (
KeystoneSmokeTests, KeystoneUtilsTests)
from snaps.openstack.utils.tests.neutron_utils_tests import (
CreateFlavorTests, os_creds=os_creds, ext_net_name=ext_net_name,
log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- HeatUtilsCreateStackTests, os_creds=os_creds,
+ HeatUtilsCreateSimpleStackTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level,
+ image_metadata=image_metadata))
+ suite.addTest(OSComponentTestCase.parameterize(
+ HeatUtilsCreateComplexStackTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level,
image_metadata=image_metadata))
ext_net_name=ext_net_name, use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateComplexStackTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
AnsibleProvisioningTests, os_creds=os_creds,
ext_net_name=ext_net_name, use_keystone=use_keystone,
self.tmpFile = self.test_dir + '/bar.txt'
self.tmp_file_opened = None
- if not os.path.exists(self.tmpFile):
- self.tmp_file_opened = open(self.tmpFile, 'wb')
def tearDown(self):
if self.tmp_file_opened:
Ensure the file_utils.fileExists() method returns false with a
directory
"""
+ if not os.path.exists(self.tmpFile):
+ self.tmp_file_opened = open(self.tmpFile, 'wb')
+
if not os.path.exists(self.tmpFile):
os.makedirs(self.tmpFile)
self.assertEqual('http://foo:5000/v2.0/', os_env_dict['OS_AUTH_URL'])
self.assertEqual('admin', os_env_dict['OS_USERNAME'])
self.assertEqual('admin', os_env_dict['OS_TENANT_NAME'])
+
+ def test_write_str_to_file(self):
+ """
+ Ensure the file_utils.fileExists() method returns false with a
+ directory
+ """
+ test_val = 'test string'
+
+ test_file = file_utils.save_string_to_file(
+ test_val, self.tmpFile)
+ result1 = file_utils.file_exists(self.tmpFile)
+ self.assertTrue(result1)
+ result2 = file_utils.file_exists(test_file.name)
+ self.assertTrue(result2)
+
+ file_contents = file_utils.read_file(self.tmpFile)
+ self.assertEqual(test_val, file_contents)
Code Review / snaps.git / commitdiff