-# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs")
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
# and others. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
def test_name_flavor_port_only(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
settings = VmInstanceSettings(name='foo', flavor='bar', port_settings=[port_settings])
- self.assertEquals('foo', settings.name)
- self.assertEquals('bar', settings.flavor)
- self.assertEquals(1, len(settings.port_settings))
- self.assertEquals('foo-port', settings.port_settings[0].name)
- self.assertEquals('bar-net', settings.port_settings[0].network_name)
- self.assertEquals(0, len(settings.security_group_names))
- self.assertEquals(0, len(settings.floating_ip_settings))
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('bar', settings.flavor)
+ self.assertEqual(1, len(settings.port_settings))
+ self.assertEqual('foo-port', settings.port_settings[0].name)
+ self.assertEqual('bar-net', settings.port_settings[0].network_name)
+ self.assertEqual(0, len(settings.security_group_names))
+ self.assertEqual(0, len(settings.floating_ip_settings))
self.assertIsNone(settings.sudo_user)
- self.assertEquals(900, settings.vm_boot_timeout)
- self.assertEquals(300, settings.vm_delete_timeout)
- self.assertEquals(180, settings.ssh_connect_timeout)
+ self.assertEqual(900, settings.vm_boot_timeout)
+ self.assertEqual(300, settings.vm_delete_timeout)
+ self.assertEqual(180, settings.ssh_connect_timeout)
self.assertIsNone(settings.availability_zone)
def test_config_with_name_flavor_port_only(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
settings = VmInstanceSettings(config={'name': 'foo', 'flavor': 'bar', 'ports': [port_settings]})
- self.assertEquals('foo', settings.name)
- self.assertEquals('bar', settings.flavor)
- self.assertEquals(1, len(settings.port_settings))
- self.assertEquals('foo-port', settings.port_settings[0].name)
- self.assertEquals('bar-net', settings.port_settings[0].network_name)
- self.assertEquals(0, len(settings.security_group_names))
- self.assertEquals(0, len(settings.floating_ip_settings))
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('bar', settings.flavor)
+ self.assertEqual(1, len(settings.port_settings))
+ self.assertEqual('foo-port', settings.port_settings[0].name)
+ self.assertEqual('bar-net', settings.port_settings[0].network_name)
+ self.assertEqual(0, len(settings.security_group_names))
+ self.assertEqual(0, len(settings.floating_ip_settings))
self.assertIsNone(settings.sudo_user)
- self.assertEquals(900, settings.vm_boot_timeout)
- self.assertEquals(300, settings.vm_delete_timeout)
- self.assertEquals(180, settings.ssh_connect_timeout)
+ self.assertEqual(900, settings.vm_boot_timeout)
+ self.assertEqual(300, settings.vm_delete_timeout)
+ self.assertEqual(180, settings.ssh_connect_timeout)
self.assertIsNone(settings.availability_zone)
def test_all(self):
security_group_names=['sec_grp_1'], floating_ip_settings=[fip_settings],
sudo_user='joe', vm_boot_timeout=999, vm_delete_timeout=333,
ssh_connect_timeout=111, availability_zone='server name')
- self.assertEquals('foo', settings.name)
- self.assertEquals('bar', settings.flavor)
- self.assertEquals(1, len(settings.port_settings))
- self.assertEquals('foo-port', settings.port_settings[0].name)
- self.assertEquals('bar-net', settings.port_settings[0].network_name)
- self.assertEquals(1, len(settings.security_group_names))
- self.assertEquals('sec_grp_1', settings.security_group_names[0])
- self.assertEquals(1, len(settings.floating_ip_settings))
- self.assertEquals('foo-fip', settings.floating_ip_settings[0].name)
- self.assertEquals('bar-port', settings.floating_ip_settings[0].port_name)
- self.assertEquals('foo-bar-router', settings.floating_ip_settings[0].router_name)
- self.assertEquals('joe', settings.sudo_user)
- self.assertEquals(999, settings.vm_boot_timeout)
- self.assertEquals(333, settings.vm_delete_timeout)
- self.assertEquals(111, settings.ssh_connect_timeout)
- self.assertEquals('server name', settings.availability_zone)
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('bar', settings.flavor)
+ self.assertEqual(1, len(settings.port_settings))
+ self.assertEqual('foo-port', settings.port_settings[0].name)
+ self.assertEqual('bar-net', settings.port_settings[0].network_name)
+ self.assertEqual(1, len(settings.security_group_names))
+ self.assertEqual('sec_grp_1', settings.security_group_names[0])
+ self.assertEqual(1, len(settings.floating_ip_settings))
+ self.assertEqual('foo-fip', settings.floating_ip_settings[0].name)
+ self.assertEqual('bar-port', settings.floating_ip_settings[0].port_name)
+ self.assertEqual('foo-bar-router', settings.floating_ip_settings[0].router_name)
+ self.assertEqual('joe', settings.sudo_user)
+ self.assertEqual(999, settings.vm_boot_timeout)
+ self.assertEqual(333, settings.vm_delete_timeout)
+ self.assertEqual(111, settings.ssh_connect_timeout)
+ self.assertEqual('server name', settings.availability_zone)
def test_config_all(self):
port_settings = PortSettings(name='foo-port', network_name='bar-net')
'floating_ips': [fip_settings], 'sudo_user': 'joe',
'vm_boot_timeout': 999, 'vm_delete_timeout': 333,
'ssh_connect_timeout': 111, 'availability_zone': 'server name'})
- self.assertEquals('foo', settings.name)
- self.assertEquals('bar', settings.flavor)
- self.assertEquals(1, len(settings.port_settings))
- self.assertEquals('foo-port', settings.port_settings[0].name)
- self.assertEquals('bar-net', settings.port_settings[0].network_name)
- self.assertEquals(1, len(settings.security_group_names))
- self.assertEquals(1, len(settings.floating_ip_settings))
- self.assertEquals('foo-fip', settings.floating_ip_settings[0].name)
- self.assertEquals('bar-port', settings.floating_ip_settings[0].port_name)
- self.assertEquals('foo-bar-router', settings.floating_ip_settings[0].router_name)
- self.assertEquals('joe', settings.sudo_user)
- self.assertEquals(999, settings.vm_boot_timeout)
- self.assertEquals(333, settings.vm_delete_timeout)
- self.assertEquals(111, settings.ssh_connect_timeout)
- self.assertEquals('server name', settings.availability_zone)
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('bar', settings.flavor)
+ self.assertEqual(1, len(settings.port_settings))
+ self.assertEqual('foo-port', settings.port_settings[0].name)
+ self.assertEqual('bar-net', settings.port_settings[0].network_name)
+ self.assertEqual(1, len(settings.security_group_names))
+ self.assertEqual(1, len(settings.floating_ip_settings))
+ self.assertEqual('foo-fip', settings.floating_ip_settings[0].name)
+ self.assertEqual('bar-port', settings.floating_ip_settings[0].port_name)
+ self.assertEqual('foo-bar-router', settings.floating_ip_settings[0].router_name)
+ self.assertEqual('joe', settings.sudo_user)
+ self.assertEqual(999, settings.vm_boot_timeout)
+ self.assertEqual(333, settings.vm_delete_timeout)
+ self.assertEqual(111, settings.ssh_connect_timeout)
+ self.assertEqual('server name', settings.availability_zone)
class FloatingIpSettingsUnitTests(unittest.TestCase):
def test_name_port_router_only(self):
settings = FloatingIpSettings(name='foo', port_name='foo-port', router_name='bar-router')
- self.assertEquals('foo', settings.name)
- self.assertEquals('foo-port', settings.port_name)
- self.assertEquals('bar-router', settings.router_name)
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('foo-port', settings.port_name)
+ self.assertEqual('bar-router', settings.router_name)
self.assertIsNone(settings.subnet_name)
self.assertTrue(settings.provisioning)
def test_config_with_name_port_router_only(self):
settings = FloatingIpSettings(config={'name': 'foo', 'port_name': 'foo-port', 'router_name': 'bar-router'})
- self.assertEquals('foo', settings.name)
- self.assertEquals('foo-port', settings.port_name)
- self.assertEquals('bar-router', settings.router_name)
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('foo-port', settings.port_name)
+ self.assertEqual('bar-router', settings.router_name)
self.assertIsNone(settings.subnet_name)
self.assertTrue(settings.provisioning)
def test_all(self):
settings = FloatingIpSettings(name='foo', port_name='foo-port', router_name='bar-router',
subnet_name='bar-subnet', provisioning=False)
- self.assertEquals('foo', settings.name)
- self.assertEquals('foo-port', settings.port_name)
- self.assertEquals('bar-router', settings.router_name)
- self.assertEquals('bar-subnet', settings.subnet_name)
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('foo-port', settings.port_name)
+ self.assertEqual('bar-router', settings.router_name)
+ self.assertEqual('bar-subnet', settings.subnet_name)
self.assertFalse(settings.provisioning)
def test_config_all(self):
settings = FloatingIpSettings(config={'name': 'foo', 'port_name': 'foo-port', 'router_name': 'bar-router',
'subnet_name': 'bar-subnet', 'provisioning': False})
- self.assertEquals('foo', settings.name)
- self.assertEquals('foo-port', settings.port_name)
- self.assertEquals('bar-router', settings.router_name)
- self.assertEquals('bar-subnet', settings.subnet_name)
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('foo-port', settings.port_name)
+ self.assertEqual('bar-router', settings.router_name)
+ self.assertEqual('bar-subnet', settings.subnet_name)
self.assertFalse(settings.provisioning)
self.floating_ip_name = guid + 'fip1'
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.network_creator = None
self.flavor_creator = None
self.inst_creator = None
# Create Image
# Set the default image settings, then set any custom parameters sent from the app
- self.os_image_settings = openstack_tests.cirros_url_image(name=guid + '-image')
-
- if self.image_metadata:
- if self.image_metadata['disk_url']:
- self.os_image_settings.url = self.image_metadata['disk_url']
- if self.image_metadata['extra_properties']:
- self.os_image_settings.extra_properties = self.image_metadata['extra_properties']
+ self.os_image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
try:
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(OpenStackImage(self.os_creds, self.os_image_settings))
- self.image_creators[-1].create()
+ self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator.create()
# Create Network
self.network_creator = OpenStackNetwork(self.os_creds, self.priv_net_config.network_settings)
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
if os.path.isfile(self.keypair_pub_filepath):
os.remove(self.keypair_pub_filepath)
if os.path.isfile(self.keypair_priv_filepath):
os.remove(self.keypair_priv_filepath)
+ if self.network_creator:
+ try:
+ self.network_creator.clean()
+ except Exception as e:
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
+
if self.flavor_creator:
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
vm = self.inst_creator.create()
ip = self.inst_creator.get_port_ip(self.port_settings.name)
found = False
timeout = 160
start_time = time.time()
- match_value = 'Lease of.*obtained'
- logger.info("Looking for expression %s in the console log" % match_value)
+ logger.info("Looking for IP %s in the console log" % ip)
full_log = ''
while timeout > time.time() - start_time:
output = vm.get_console_output()
full_log = full_log + output
- if re.search(match_value, output):
+ if re.search(ip, output):
logger.info('DHCP lease obtained logged in console')
- if ip in output:
- logger.info('With correct IP address')
- found = True
- else:
- logger.error('With incorrect IP address')
+ found = True
break
if not found:
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
self.nova = nova_utils.nova_client(self.os_creds)
- self.os_image_settings = openstack_tests.cirros_url_image(name=guid + '-image')
+ self.os_image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
net_config = openstack_tests.get_priv_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name)
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.flavor_creator = None
- self.net_creator = None
+ self.network_creator = None
self.inst_creator = None
try:
# Create Image
- # Set any custom parameters sent from the app
- if self.image_metadata:
- if self.image_metadata['disk_url']:
- self.os_image_settings.url = self.image_metadata['disk_url']
- if self.image_metadata['extra_properties']:
- self.os_image_settings.extra_properties = self.image_metadata['extra_properties']
-
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(OpenStackImage(self.os_creds, self.os_image_settings))
- self.image_creators[-1].create()
+ self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
if self.flavor_creator:
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
- if self.net_creator:
+ if self.network_creator:
try:
- self.net_creator.clean()
+ self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings)
+ self.os_creds, instance_settings, self.image_creator.image_settings)
vm_inst = self.inst_creator.create()
- self.assertEquals(1, len(nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
+ self.assertEqual(1, len(nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
# Delete instance
nova_utils.delete_vm_instance(self.nova, vm_inst)
self.assertTrue(self.inst_creator.vm_deleted(block=True))
- self.assertEquals(0, len(nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
+ self.assertEqual(0, len(nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
# Exception should not be thrown
self.inst_creator.clean()
self.floating_ip_name = guid + 'fip1'
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.network_creator = None
self.router_creator = None
self.flavor_creator = None
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.os_image_settings = openstack_tests.cirros_url_image(name=guid + '-image')
-
+ self.os_image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
try:
# Create Image
- # Set any custom parameters sent from the app
- if self.image_metadata:
- if self.image_metadata['disk_url']:
- self.os_image_settings.url = self.image_metadata['disk_url']
- if self.image_metadata['extra_properties']:
- self.os_image_settings.extra_properties = self.image_metadata['extra_properties']
-
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(OpenStackImage(self.os_creds, self.os_image_settings))
- self.image_creators[-1].create()
+ self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator.create()
# Create Network
self.network_creator = OpenStackNetwork(self.os_creds, self.pub_net_config.network_settings)
try:
inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
if self.keypair_creator:
try:
self.keypair_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning keypair with message - ' + e.message)
+ logger.error('Unexpected exception cleaning keypair with message - ' + str(e))
if os.path.isfile(self.keypair_pub_filepath):
os.remove(self.keypair_pub_filepath)
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
if self.router_creator:
try:
self.router_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning router with message - ' + e.message)
+ logger.error('Unexpected exception cleaning router with message - ' + str(e))
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
router_name=self.pub_net_config.router_settings.name)])
inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings,
+ self.os_creds, instance_settings, self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
self.inst_creators.append(inst_creator)
vm_inst = inst_creator.create()
- self.assertEquals(ip_1, inst_creator.get_port_ip(self.port_1_name))
+ self.assertEqual(ip_1, inst_creator.get_port_ip(self.port_1_name))
self.assertTrue(inst_creator.vm_active(block=True))
- self.assertEquals(vm_inst, inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst, inst_creator.get_vm_inst())
def test_ssh_client_fip_before_active(self):
"""
router_name=self.pub_net_config.router_settings.name)])
inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings,
+ self.os_creds, instance_settings, self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
self.inst_creators.append(inst_creator)
vm_inst = inst_creator.create()
self.assertIsNotNone(vm_inst)
self.assertTrue(inst_creator.vm_active(block=True))
- self.assertEquals(vm_inst, inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst, inst_creator.get_vm_inst())
self.assertTrue(validate_ssh_client(inst_creator))
router_name=self.pub_net_config.router_settings.name)])
inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings,
+ self.os_creds, instance_settings, self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
self.inst_creators.append(inst_creator)
self.assertIsNotNone(vm_inst)
self.assertTrue(inst_creator.vm_active(block=True))
- self.assertEquals(vm_inst, inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst, inst_creator.get_vm_inst())
self.assertTrue(validate_ssh_client(inst_creator))
- # TODO - Determine how allowed_address_pairs is supposed to operate before continuing this test
- # see http://docs.openstack.org/developer/dragonflow/specs/allowed_address_pairs.html for a functional description
- # def test_allowed_address_port_access(self):
- # """
- # Tests to ensure that setting allowed_address_pairs on a port functions as designed
- # """
- # port_settings_1 = PortSettings(
- # name=self.port_1_name + '-1', network_name=self.pub_net_config.network_settings.name)
- #
- # instance_settings_1 = VmInstanceSettings(
- # name=self.vm_inst_name + '-1', flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings_1],
- # floating_ip_settings=[FloatingIpSettings(
- # name=self.floating_ip_name + '-1', port_name=port_settings_1.name,
- # router_name=self.pub_net_config.router_settings.name)])
- #
- # inst_creator_1 = OpenStackVmInstance(
- # self.os_creds, instance_settings_1, self.image_creators[-1].image_settings,
- # keypair_settings=self.keypair_creator.keypair_settings)
- # self.inst_creators.append(inst_creator_1)
- #
- # # block=True will force the create() method to block until the
- # vm_inst_1 = inst_creator_1.create(block=True)
- # self.assertIsNotNone(vm_inst_1)
- #
- # port_settings_1 = PortSettings(
- # name=self.port_1_name + '-1', network_name=self.pub_net_config.network_settings.name)
- #
- # instance_settings_1 = VmInstanceSettings(
- # name=self.vm_inst_name + '-1', flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings_1],
- # floating_ip_settings=[FloatingIpSettings(
- # name=self.floating_ip_name + '-1', port_name=port_settings_1.name,
- # router_name=self.pub_net_config.router_settings.name)])
- #
- # inst_creator_1 = OpenStackVmInstance(
- # self.os_creds, instance_settings_1, self.image_creators[-1].image_settings,
- # keypair_settings=self.keypair_creator.keypair_settings)
- # self.inst_creators.append(inst_creator_1)
- # inst_creator_1.create(block=True)
- #
- # ip = inst_creator_1.get_port_ip(port_settings_1.name,
- # subnet_name=self.pub_net_config.network_settings.subnet_settings[0].name)
- # self.assertIsNotNone(ip)
- # mac_addr = inst_creator_1.get_port_mac(port_settings_1.name)
- # self.assertIsNotNone(mac_addr)
- #
- # allowed_address_pairs = [{'ip_address': ip, 'mac_address': mac_addr}]
- #
- # # Create VM that can be accessed by vm_inst_1
- # port_settings_2 = PortSettings(
- # name=self.port_1_name + '-2', network_name=self.pub_net_config.network_settings.name,
- # allowed_address_pairs=allowed_address_pairs)
- #
- # instance_settings_2 = VmInstanceSettings(
- # name=self.vm_inst_name + '-2', flavor=self.flavor_creator.flavor_settings.name,
- # port_settings=[port_settings_2])
- #
- # inst_creator_2 = OpenStackVmInstance(
- # self.os_creds, instance_settings_2, self.image_creators[-1].image_settings)
- # self.inst_creators.append(inst_creator_2)
- # inst_creator_2.create(block=True)
- #
- # # Create VM that cannot be accessed by vm_inst_1
- # ip = '10.55.0.101'
- # mac_addr = '0a:1b:2c:3d:4e:5f'
- # invalid_address_pairs = [{'ip_address': ip, 'mac_address': mac_addr}]
- #
- # port_settings_3 = PortSettings(
- # name=self.port_1_name + '-3', network_name=self.pub_net_config.network_settings.name,
- # allowed_address_pairs=invalid_address_pairs)
- #
- # instance_settings_3 = VmInstanceSettings(
- # name=self.vm_inst_name + '-3', flavor=self.flavor_creator.flavor_settings.name,
- # port_settings=[port_settings_3])
- #
- # inst_creator_3 = OpenStackVmInstance(
- # self.os_creds, instance_settings_3, self.image_creators[-1].image_settings)
- # self.inst_creators.append(inst_creator_3)
- # inst_creator_3.create(block=True)
- #
- # print 'foo'
- # I expected that this feature would block/allow traffic from specific endpoints (VMs). In this case, I would
- # expect inst_1 to be able to access inst_2 but not inst_3; however, they all can access each other.
- # TODO - Add validation
-
class CreateInstancePortManipulationTests(OSIntegrationTestCase):
"""
self.floating_ip_name = guid + 'fip1'
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.network_creator = None
self.flavor_creator = None
self.inst_creator = None
self.net_config = openstack_tests.get_priv_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name)
- self.os_image_settings = openstack_tests.cirros_url_image(name=guid + '-image')
+ self.os_image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
try:
# Create Image
- # Set any custom parameters sent from the app
- if self.image_metadata:
- if self.image_metadata['disk_url']:
- self.os_image_settings.url = self.image_metadata['disk_url']
- if self.image_metadata['extra_properties']:
- self.os_image_settings.extra_properties = self.image_metadata['extra_properties']
-
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(OpenStackImage(self.os_creds, self.os_image_settings))
- self.image_creators[-1].create()
+ self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator.create()
# Create Network
self.network_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings)
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
if self.flavor_creator:
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
- self.inst_creator.create()
+ self.image_creator.image_settings)
+ self.inst_creator.create(block=True)
- self.assertEquals(ip, self.inst_creator.get_port_ip(
+ self.assertEqual(ip, self.inst_creator.get_port_ip(
self.port_1_name, subnet_name=self.net_config.network_settings.subnet_settings[0].name))
def test_set_custom_invalid_ip_one_subnet(self):
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
- self.inst_creator.create()
+ self.image_creator.image_settings)
+ self.inst_creator.create(block=True)
- self.assertEquals(mac_addr, self.inst_creator.get_port_mac(self.port_1_name))
+ self.assertEqual(mac_addr, self.inst_creator.get_port_mac(self.port_1_name))
def test_set_custom_invalid_mac(self):
"""
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings)
+ self.os_creds, instance_settings, self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
- self.inst_creator.create()
+ self.image_creator.image_settings)
+ self.inst_creator.create(block=True)
- self.assertEquals(ip, self.inst_creator.get_port_ip(
+ self.assertEqual(ip, self.inst_creator.get_port_ip(
self.port_1_name, subnet_name=self.net_config.network_settings.subnet_settings[0].name))
- self.assertEquals(mac_addr, self.inst_creator.get_port_mac(self.port_1_name))
+ self.assertEqual(mac_addr, self.inst_creator.get_port_mac(self.port_1_name))
def test_set_allowed_address_pairs(self):
"""
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
- self.inst_creator.create()
+ self.image_creator.image_settings)
+ self.inst_creator.create(block=True)
port = self.inst_creator.get_port_by_name(port_settings.name)
self.assertIsNotNone(port)
self.assertIsNotNone(port['port'].get('allowed_address_pairs'))
- self.assertEquals(1, len(port['port']['allowed_address_pairs']))
+ self.assertEqual(1, len(port['port']['allowed_address_pairs']))
validation_utils.objects_equivalent(pair, port['port']['allowed_address_pairs'][0])
def test_set_allowed_address_pairs_bad_mac(self):
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
with self.assertRaises(Exception):
self.inst_creator.create()
self.port_base_name = guid + 'port'
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.flavor_creator = None
self.network_creator = None
self.inst_creators = list()
self.priv_net_config = openstack_tests.get_priv_net_config(
net_name=guid + '-priv-net', subnet_name=guid + '-priv-subnet')
- self.os_image_settings = openstack_tests.cirros_url_image(name=guid + '-image')
+ self.os_image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
try:
# Create Network
self.flavor_creator.create()
# Create Image
- # Set any custom parameters sent from the app
- if self.image_metadata:
- if self.image_metadata['disk_url']:
- self.os_image_settings.url = self.image_metadata['disk_url']
- if self.image_metadata['extra_properties']:
- self.os_image_settings.extra_properties = self.image_metadata['extra_properties']
-
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(OpenStackImage(self.os_creds, self.os_image_settings))
- self.image_creators[-1].create()
+ self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator.create()
except Exception as e:
self.tearDown()
try:
inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
if self.flavor_creator:
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
name=inst_name, flavor=self.flavor_creator.flavor_settings.name, availability_zone=zone,
port_settings=[port_settings])
inst_creator = OpenStackVmInstance(
- self.admin_os_creds, instance_settings, self.image_creators[-1].image_settings)
+ self.admin_os_creds, instance_settings, self.image_creator.image_settings)
self.inst_creators.append(inst_creator)
inst_creator.create()
vm = creator.get_vm_inst()
deployed_zone = vm._info['OS-EXT-AZ:availability_zone']
deployed_host = vm._info['OS-EXT-SRV-ATTR:host']
- self.assertEquals(zone, deployed_zone + ':' + deployed_host)
+ self.assertEqual(zone, deployed_zone + ':' + deployed_host)
index += 1
super(self.__class__, self).__start__()
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.network_creators = list()
self.router_creators = list()
self.flavor_creator = None
self.pub_net_config = openstack_tests.get_pub_net_config(
net_name=self.guid + '-pub-net', subnet_name=self.guid + '-pub-subnet',
router_name=self.guid + '-pub-router', external_net=self.ext_net_name)
+
image_name = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.os_image_settings = openstack_tests.centos_url_image(name=image_name)
+ self.os_image_settings = openstack_tests.centos_image_settings(name=image_name,
+ image_metadata=self.image_metadata)
try:
# Create Image
- # Set any custom parameters sent from the app
- if self.image_metadata:
- if self.image_metadata['disk_url']:
- self.os_image_settings.url = self.image_metadata['disk_url']
- if self.image_metadata['extra_properties']:
- self.os_image_settings.extra_properties = self.image_metadata['extra_properties']
-
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(OpenStackImage(self.os_creds, self.os_image_settings))
- self.image_creators[-1].create()
+ self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator.create()
# First network is public
self.network_creators.append(OpenStackNetwork(self.os_creds, self.pub_net_config.network_settings))
self.keypair_creator.create()
except Exception as e:
self.tearDown()
- raise Exception(e.message)
+ raise Exception(str(e))
def tearDown(self):
"""
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
if self.keypair_creator:
try:
self.keypair_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning keypair with message - ' + e.message)
+ logger.error('Unexpected exception cleaning keypair with message - ' + str(e))
if os.path.isfile(self.keypair_pub_filepath):
os.remove(self.keypair_pub_filepath)
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
for router_creator in self.router_creators:
try:
router_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning router with message - ' + e.message)
+ logger.error('Unexpected exception cleaning router with message - ' + str(e))
for network_creator in self.network_creators:
try:
network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
router_name=self.pub_net_config.router_settings.name)])
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings,
+ self.os_creds, instance_settings, self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
vm_inst = self.inst_creator.create(block=True)
- self.assertEquals(vm_inst, self.inst_creator.get_vm_inst())
+ self.assertEqual(vm_inst, self.inst_creator.get_vm_inst())
# Effectively blocks until VM has been properly activated
self.assertTrue(self.inst_creator.vm_active(block=True))
# Effectively blocks until VM's ssh port has been opened
self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
+ # TODO - Refactor config_nics() to return a status that can be validated here.
self.inst_creator.config_nics()
# TODO - *** ADD VALIDATION HERE ***
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = self.guid + '-inst'
self.nova = nova_utils.nova_client(self.os_creds)
- self.os_image_settings = openstack_tests.cirros_url_image(name=self.guid + '-image')
+ self.os_image_settings = openstack_tests.cirros_image_settings(
+ name=self.guid + '-image', image_metadata=self.image_metadata)
self.vm_inst_name = self.guid + '-inst'
self.port_1_name = self.guid + 'port-1'
router_name=self.guid + '-pub-router', external_net=self.ext_net_name)
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.flavor_creator = None
self.network_creator = None
self.router_creator = None
try:
# Create Image
- # Set any custom parameters sent from the app
- if self.image_metadata:
- if self.image_metadata['disk_url']:
- self.os_image_settings.url = self.image_metadata['disk_url']
- if self.image_metadata['extra_properties']:
- self.os_image_settings.extra_properties = self.image_metadata['extra_properties']
-
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(OpenStackImage(self.os_creds, self.os_image_settings))
- self.image_creators[-1].create()
+ self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator.create()
# Create Network
self.network_creator = OpenStackNetwork(self.os_creds, net_config.network_settings)
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
for sec_grp_creator in self.sec_grp_creators:
try:
sec_grp_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning security group with message - ' + e.message)
+ logger.error('Unexpected exception cleaning security group with message - ' + str(e))
if self.flavor_creator:
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
instance_settings = VmInstanceSettings(
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
instance_settings = VmInstanceSettings(
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name,
security_group_names=[sec_grp_settings.name], port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
instance_settings = VmInstanceSettings(
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name,
security_group_names=[sec_grp_settings.name], port_settings=[self.port_settings])
self.inst_creator = OpenStackVmInstance(self.os_creds, instance_settings,
- self.image_creators[-1].image_settings)
+ self.image_creator.image_settings)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(vm_inst)
router_name=guid + '-pub-router', external_net=self.ext_net_name)
# Initialize for tearDown()
- self.image_creators = list()
+ self.image_creator = None
self.network_creator = None
self.flavor_creator = None
self.inst_creator = None
try:
- # Create Images
- # Set properties
- properties = {}
- if self.image_metadata and self.image_metadata['extra_properties']:
- properties = self.image_metadata['extra_properties']
-
- # Create the kernel image
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=self.image_name+'_kernel',
- url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel')
-
- if self.image_metadata and self.image_metadata['kernel_url']:
- kernel_image_settings.url = self.image_metadata['kernel_url']
-
- self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
-
- # Create the ramdisk image
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=self.image_name+'_ramdisk',
- url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs')
- if self.image_metadata and self.image_metadata['ramdisk_url']:
- ramdisk_image_settings.url = self.image_metadata['ramdisk_url']
-
- self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- self.assertIsNotNone(ramdisk_image)
-
- # Create the main image
- os_image_settings = openstack_tests.cirros_url_image(
+ image_settings = openstack_tests.cirros_image_settings(
name=self.image_name,
- url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img')
- if self.image_metadata and self.image_metadata['disk_url']:
- os_image_settings.url = self.image_metadata['disk_url']
+ image_metadata={'disk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img',
+ 'kernel_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel',
+ 'ramdisk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs'})
- properties['kernel_id'] = kernel_image.id
- properties['ramdisk_id'] = ramdisk_image.id
- os_image_settings.extra_properties = properties
- self.image_creators.append(OpenStackImage(self.os_creds, os_image_settings))
- created_image = self.image_creators[-1].create()
- self.assertIsNotNone(created_image)
+ self.image_creator = OpenStackImage(self.os_creds, image_settings)
+ self.image_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
try:
self.inst_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning VM instance with message - ' + e.message)
+ logger.error('Unexpected exception cleaning VM instance with message - ' + str(e))
if self.flavor_creator:
try:
self.flavor_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning flavor with message - ' + e.message)
+ logger.error('Unexpected exception cleaning flavor with message - ' + str(e))
if self.network_creator:
try:
self.network_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning network with message - ' + e.message)
+ logger.error('Unexpected exception cleaning network with message - ' + str(e))
- if self.image_creators:
+ if self.image_creator:
try:
- while self.image_creators:
- self.image_creators[0].clean()
- self.image_creators.pop(0)
+ self.image_creator.clean()
except Exception as e:
- logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ logger.error('Unexpected exception cleaning image with message - ' + str(e))
super(self.__class__, self).__clean__()
- def test_create_delete_instance_from_three_part_image(self):
+ def test_create_instance_from_three_part_image(self):
"""
Tests the creation of an OpenStack instance from a 3-part image.
"""
# The last created image is the main image from which we create the instance
self.inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings)
+ self.os_creds, instance_settings, self.image_creator.image_settings)
vm_inst = self.inst_creator.create()
- self.assertEquals(1, len(nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
-
+ self.assertIsNotNone(vm_inst)
self.assertTrue(self.inst_creator.vm_active(block=True))
-
- # Delete instance
- nova_utils.delete_vm_instance(self.nova, vm_inst)
-
- self.assertTrue(self.inst_creator.vm_deleted(block=True))
- self.assertEquals(0, len(nova_utils.get_servers_by_name(self.nova, instance_settings.name)))
-
- # Exception should not be thrown
- self.inst_creator.clean()