import uuid
import os
+from neutronclient.common.exceptions import InvalidIpForSubnetClient
from snaps import file_utils
from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
from snaps.openstack.create_image import OpenStackImage, ImageSettings
from snaps.openstack.create_instance import (
- VmInstanceSettings, OpenStackVmInstance, FloatingIpSettings)
+ VmInstanceSettings, OpenStackVmInstance, FloatingIpSettings,
+ VmInstanceSettingsError, FloatingIpSettingsError)
from snaps.openstack.create_keypairs import OpenStackKeypair, KeypairSettings
from snaps.openstack.create_network import OpenStackNetwork, PortSettings
from snaps.openstack.create_router import OpenStackRouter
"""
def test_no_params(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(VmInstanceSettingsError):
VmInstanceSettings()
def test_empty_config(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(VmInstanceSettingsError):
VmInstanceSettings(config=dict())
def test_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(VmInstanceSettingsError):
VmInstanceSettings(name='foo')
def test_config_with_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(VmInstanceSettingsError):
VmInstanceSettings(config={'name': 'foo'})
def test_name_flavor_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(VmInstanceSettingsError):
VmInstanceSettings(name='foo', flavor='bar')
def test_config_with_name_flavor_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(VmInstanceSettingsError):
VmInstanceSettings(config={'name': 'foo', 'flavor': 'bar'})
def test_name_flavor_port_only(self):
"""
def test_no_params(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings()
def test_empty_config(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(**dict())
def test_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(name='foo')
def test_config_with_name_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(**{'name': 'foo'})
def test_name_port_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(name='foo', port_name='bar')
def test_config_with_name_port_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(**{'name': 'foo', 'port_name': 'bar'})
def test_name_router_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(name='foo', router_name='bar')
def test_config_with_name_router_only(self):
- with self.assertRaises(Exception):
+ with self.assertRaises(FloatingIpSettingsError):
FloatingIpSettings(**{'name': 'foo', 'router_name': 'bar'})
def test_name_port_router_only(self):
self.inst_creator = OpenStackVmInstance(
self.os_creds, instance_settings,
self.image_creator.image_settings)
- vm = self.inst_creator.create()
+ self.inst_creator.create()
ip = self.inst_creator.get_port_ip(self.port_settings.name)
self.assertIsNotNone(ip)
self.assertTrue(self.inst_creator.vm_active(block=True))
- self.assertTrue(check_dhcp_lease(self.nova, vm, ip))
+ self.assertTrue(check_dhcp_lease(self.inst_creator, ip))
class CreateInstanceSimpleTests(OSIntegrationTestCase):
self.assertTrue(inst_creator.vm_active(block=True))
ip = inst_creator.get_port_ip(port_settings.name)
- self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(inst_creator, ip))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
self.assertTrue(inst_creator.vm_active(block=True))
ip = inst_creator.get_port_ip(port_settings.name)
- self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(inst_creator, ip))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
self.assertTrue(validate_ssh_client(inst_creator))
+ def test_ssh_client_fip_second_creator(self):
+ """
+ Tests the ability to access a VM via SSH and a floating IP via a
+ creator that is identical to the original creator.
+ """
+ port_settings = PortSettings(
+ name=self.port_1_name,
+ network_name=self.pub_net_config.network_settings.name)
+
+ instance_settings = VmInstanceSettings(
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings],
+ floating_ip_settings=[FloatingIpSettings(
+ name=self.floating_ip_name, port_name=self.port_1_name,
+ router_name=self.pub_net_config.router_settings.name)])
+
+ inst_creator = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
+ keypair_settings=self.keypair_creator.keypair_settings)
+ self.inst_creators.append(inst_creator)
+
+ # block=True will force the create() method to block until the
+ vm_inst = inst_creator.create(block=True)
+ self.assertIsNotNone(vm_inst)
+
+ self.assertTrue(inst_creator.vm_active(block=True))
+
+ ip = inst_creator.get_port_ip(port_settings.name)
+ self.assertTrue(check_dhcp_lease(inst_creator, ip))
+
+ inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
+ self.assertEqual(vm_inst, inst_creator.get_vm_inst())
+
+ self.assertTrue(validate_ssh_client(inst_creator))
+
+ inst_creator2 = OpenStackVmInstance(
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
+ keypair_settings=self.keypair_creator.keypair_settings)
+ inst_creator2.create()
+ self.assertTrue(validate_ssh_client(inst_creator2))
+
class CreateInstancePortManipulationTests(OSIntegrationTestCase):
"""
self.os_creds, instance_settings,
self.image_creator.image_settings)
- with self.assertRaises(Exception):
+ with self.assertRaises(InvalidIpForSubnetClient):
self.inst_creator.create()
def test_set_custom_valid_mac(self):
"""
from snaps.openstack.utils import nova_utils
nova = nova_utils.nova_client(self.admin_os_creds)
- zones = nova_utils.get_nova_availability_zones(nova)
+ zone_hosts = nova_utils.get_availability_zone_hosts(nova)
# Create Instance on each server/zone
ctr = 0
- for zone in zones:
+ for zone in zone_hosts:
inst_name = self.vm_inst_name + '-' + zone
ctr += 1
port_settings = PortSettings(
# Validate instances to ensure they've been deployed to the correct
# server
index = 0
- for zone in zones:
+ for zone in zone_hosts:
creator = self.inst_creators[index]
self.assertTrue(creator.vm_active(block=True))
- vm = creator.get_os_vm_server_obj()
- deployed_zone = vm._info['OS-EXT-AZ:availability_zone']
- deployed_host = vm._info['OS-EXT-SRV-ATTR:host']
+ info = creator.get_vm_info()
+ deployed_zone = info['OS-EXT-AZ:availability_zone']
+ deployed_host = info['OS-EXT-SRV-ATTR:host']
self.assertEqual(zone, deployed_zone + ':' + deployed_host)
index += 1
self.assertTrue(self.inst_creator.vm_active(block=True))
ip = self.inst_creator.get_port_ip(ports_settings[0].name)
- self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(self.inst_creator, ip))
# Add security group to VM
self.inst_creator.add_security_group(
:param nova: the nova client
:param vm_inst: the VmInst domain object
:param sec_grp_name: the name of the security group to validate
- :return:
+ :return: T/F
"""
- vm = nova_utils.get_latest_server_os_object(nova, vm_inst)
- if not hasattr(vm, 'security_groups'):
- return False
-
- found = False
- for sec_grp_dict in vm.security_groups:
- if sec_grp_name in sec_grp_dict['name']:
- found = True
- break
- return found
+ sec_grp_names = nova_utils.get_server_security_group_names(nova, vm_inst)
+ for name in sec_grp_names:
+ if sec_grp_name == name:
+ return True
+ return False
def validate_ssh_client(instance_creator):
if ssh_active:
ssh_client = instance_creator.ssh_client()
if ssh_client:
- out = ssh_client.exec_command('pwd')[1]
+ try:
+ out = ssh_client.exec_command('pwd')[1]
+ finally:
+ ssh_client.close()
else:
return False
self.assertTrue(self.inst_creator.vm_active(block=True))
-def check_dhcp_lease(nova_client, vm_domain, ip, timeout=160):
+def check_dhcp_lease(inst_creator, ip, timeout=160):
"""
Returns true if the expected DHCP lease has been acquired
- :param nova_client: the nova client
- :param vm_domain: the SNAPS VM instance domain object
+ :param inst_creator: the SNAPS OpenStackVmInstance object
:param ip: the IP address to look for
:param timeout: how long to query for IP address
:return:
logger.info("Looking for IP %s in the console log" % ip)
full_log = ''
while timeout > time.time() - start_time:
- vm = nova_utils.get_latest_server_os_object(nova_client, vm_domain)
- output = vm.get_console_output()
+ output = inst_creator.get_console_output()
full_log = full_log + output
if re.search(ip, output):
logger.info('DHCP lease obtained logged in console')
Code Review / snaps.git / blobdiff