"""
def test_construction_positional(self):
- vm_inst = VmInst('name', 'id', '456', '123', dict(), 'kp-name',
+ vm_inst = VmInst('name', 'id', '456', '123', list(), 'kp-name',
['foo', 'bar'], ['123', '456'])
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(list(), vm_inst.ports)
self.assertEqual('kp-name', vm_inst.keypair_name)
self.assertEqual(['foo', 'bar'], vm_inst.sec_grp_names)
self.assertEqual(['123', '456'], vm_inst.volume_ids)
def test_construction_named(self):
vm_inst = VmInst(
volume_ids=['123', '456'], sec_grp_names=['foo', 'bar'],
- networks=dict(), inst_id='id', name='name', flavor_id='123',
+ ports=list(), 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(list(), vm_inst.ports)
self.assertEqual('kp-name', vm_inst.keypair_name)
self.assertEqual(['foo', 'bar'], vm_inst.sec_grp_names)
self.assertEqual(['123', '456'], vm_inst.volume_ids)
SNAPS domain object for Images. Should contain attributes that
are shared amongst cloud providers
"""
- def __init__(self, name, inst_id, image_id, flavor_id, networks,
+ def __init__(self, name, inst_id, image_id, flavor_id, ports,
keypair_name, sec_grp_names, volume_ids):
"""
Constructor
:param inst_id: the instance's id
: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 ports: list of SNAPS-OO Port domain objects associated with this
+ server instance
:param keypair_name: the name of the associated keypair
:param sec_grp_names: list of security group names
:param volume_ids: list of attached volume IDs
self.id = inst_id
self.image_id = image_id
self.flavor_id = flavor_id
- self.networks = networks
+ self.ports = ports
self.keypair_name = keypair_name
self.sec_grp_names = sec_grp_names
self.volume_ids = volume_ids
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.ports == other.ports and
self.keypair_name == other.keypair_name and
self.sec_grp_names == other.sec_grp_names and
self.volume_ids == other.volume_ids)
within the project
"""
server = nova_utils.get_server(
- self._nova, vm_inst_settings=self.instance_settings)
+ self._nova, self.__neutron,
+ vm_inst_settings=self.instance_settings)
if server:
if server.name == self.instance_settings.name:
self.__vm = server
if volume and self.vm_active(block=True):
timeout = 30
vm = nova_utils.attach_volume(
- self._nova, self.__vm, volume, timeout)
+ self._nova, self.__neutron, self.__vm, volume, timeout)
if vm:
self.__vm = vm
if volume:
try:
vm = nova_utils.detach_volume(
- self._nova, self.__vm, volume, 30)
+ self._nova, self.__neutron, self.__vm, volume, 30)
if vm:
self.__vm = vm
else:
Returns the latest version of this server object from OpenStack
:return: Server object
"""
- return nova_utils.get_server_object_by_id(self._nova, self.__vm.id)
+ return nova_utils.get_server_object_by_id(
+ self._nova, self.__neutron, self.__vm.id)
def get_console_output(self):
"""
STATUS_ACTIVE, block, self.instance_settings.vm_boot_timeout,
poll_interval):
self.__vm = nova_utils.get_server_object_by_id(
- self._nova, self.__vm.id)
+ self._nova, self.__neutron, self.__vm.id)
return True
return False
out = list()
nova = nova_utils.nova_client(self._os_creds)
+ neutron = neutron_utils.neutron_client(self._os_creds)
stack_servers = heat_utils.get_stack_servers(
- self.__heat_cli, nova, self.__stack)
+ self.__heat_cli, nova, neutron, self.__stack)
- neutron = neutron_utils.neutron_client(self._os_creds)
glance = glance_utils.glance_client(self._os_creds)
for stack_server in stack_servers:
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
self.nova = nova_utils.nova_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
vm_inst = self.inst_creator.create()
self.assertIsNotNone(nova_utils.get_server(
- self.nova, vm_inst_settings=instance_settings))
+ self.nova, self.neutron, vm_inst_settings=instance_settings))
# Delete instance
nova_utils.delete_vm_instance(self.nova, vm_inst)
self.assertTrue(self.inst_creator.vm_deleted(block=True))
self.assertIsNone(nova_utils.get_server(
- self.nova, vm_inst_settings=instance_settings))
+ self.nova, self.neutron, vm_inst_settings=instance_settings))
# Exception should not be thrown
self.inst_creator.clean()
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
self.nova = nova_utils.nova_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
os_image_settings = openstack_tests.cirros_image_settings(
name=guid + '-image', image_metadata=self.image_metadata)
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(nova_utils.get_server(
- self.nova, vm_inst_settings=instance_settings))
+ self.nova, self.neutron, vm_inst_settings=instance_settings))
self.assertIsNotNone(vm_inst)
self.assertEqual(1, len(vm_inst.volume_ids))
vm_inst = self.inst_creator.create(block=True)
self.assertIsNotNone(nova_utils.get_server(
- self.nova, vm_inst_settings=instance_settings))
+ self.nova, self.neutron, vm_inst_settings=instance_settings))
self.assertIsNotNone(vm_inst)
self.assertEqual(2, len(vm_inst.volume_ids))
vm_inst_creators[0].get_vm_inst().name)
nova = nova_utils.nova_client(self.admin_os_creds)
+ neutron = neutron_utils.neutron_client(self.admin_os_creds)
vm_inst_by_name = nova_utils.get_server(
- nova, server_name=vm_inst_creators[0].get_vm_inst().name)
+ nova, neutron, 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))
+ nova, neutron, vm_inst_creators[0].get_vm_inst().id))
class CreateStackFloatingIpTests(OSIntegrationTestCase):
return out
-def get_stack_servers(heat_cli, nova, stack):
+def get_stack_servers(heat_cli, nova, neutron, stack):
"""
Returns a list of VMInst domain objects associated with a Stack
:param heat_cli: the OpenStack heat client object
:param nova: the OpenStack nova client object
+ :param neutron: the OpenStack neutron client object
:param stack: the SNAPS-OO Stack domain object
:return: a list of VMInst domain objects
"""
resources = get_resources(heat_cli, stack, 'OS::Nova::Server')
for resource in resources:
try:
- server = nova_utils.get_server_object_by_id(nova, resource.id)
+ server = nova_utils.get_server_object_by_id(
+ nova, neutron, resource.id)
if server:
out.append(server)
except NotFound:
server = nova.servers.create(**args)
- return __map_os_server_obj_to_vm_inst(server)
+ return __map_os_server_obj_to_vm_inst(neutron, server)
else:
raise NovaException(
'Cannot create instance, image cannot be located with name %s',
image_config.name)
-def get_server(nova, vm_inst_settings=None, server_name=None):
+def get_server(nova, neutron, vm_inst_settings=None, server_name=None):
"""
Returns a VmInst object for the first server instance found.
:param nova: the Nova client
+ :param neutron: the Neutron client
:param vm_inst_settings: the VmInstanceConfig object from which to build
the query if not None
:param server_name: the server with this name to return if vm_inst_settings
servers = nova.servers.list(search_opts=search_opts)
for server in servers:
- return __map_os_server_obj_to_vm_inst(server)
+ return __map_os_server_obj_to_vm_inst(neutron, server)
def get_server_connection(nova, vm_inst_settings=None, server_name=None):
return server.links[0]
-def __map_os_server_obj_to_vm_inst(os_server):
+def __map_os_server_obj_to_vm_inst(neutron, os_server):
"""
Returns a VmInst object for an OpenStack Server object
+ :param neutron: the Neutron client (when None, ports will be empty)
:param os_server: the OpenStack server object
:return: an equivalent SNAPS-OO VmInst domain object
"""
if sec_group.get('name'):
sec_grp_names.append(sec_group.get('name'))
+ out_ports = list()
+ if len(os_server.networks) > 0:
+ for net_name, ips in os_server.networks.items():
+ network = neutron_utils.get_network(neutron, network_name=net_name)
+ ports = neutron_utils.get_ports(neutron, network, ips)
+ for port in ports:
+ out_ports.append(port)
+
volumes = None
if hasattr(os_server, 'os-extended-volumes:volumes_attached'):
volumes = getattr(os_server, 'os-extended-volumes:volumes_attached')
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,
+ ports=out_ports, keypair_name=os_server.key_name,
sec_grp_names=sec_grp_names, volume_ids=volumes)
return None
-def get_latest_server_object(nova, server):
+def get_latest_server_object(nova, neutron, server):
"""
Returns a server with a given id
:param nova: the Nova client
+ :param neutron: the Neutron client
:param server: the old server object
:return: the list of servers or None if not found
"""
server = __get_latest_server_os_object(nova, server)
- return __map_os_server_obj_to_vm_inst(server)
+ return __map_os_server_obj_to_vm_inst(neutron, server)
-def get_server_object_by_id(nova, server_id):
+def get_server_object_by_id(nova, neutron, server_id):
"""
Returns a server with a given id
:param nova: the Nova client
+ :param neutron: the Neutron 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)
+ return __map_os_server_obj_to_vm_inst(neutron, server)
def get_server_security_group_names(nova, server):
return vm._info
return None
+
def reboot_server(nova, server, reboot_type=None):
"""
Returns a dictionary of a VMs info as returned by OpenStack
return nova.quotas.update(project_id, **update_values)
-def attach_volume(nova, server, volume, timeout=None):
+def attach_volume(nova, neutron, server, volume, timeout=None):
"""
Attaches a volume to a server
:param nova: the nova client
+ :param neutron: the neutron client
:param server: the VMInst domain object
:param volume: the Volume domain object
:param timeout: denotes the amount of time to block to determine if the
if timeout:
start_time = time.time()
while time.time() < start_time + timeout:
- vm = get_server_object_by_id(nova, server.id)
+ vm = get_server_object_by_id(nova, neutron, server.id)
for vol_dict in vm.volume_ids:
if volume.id == vol_dict['id']:
return vm
return None
else:
- return get_server_object_by_id(nova, server.id)
+ return get_server_object_by_id(nova, neutron, server.id)
-def detach_volume(nova, server, volume, timeout=None):
+def detach_volume(nova, neutron, server, volume, timeout=None):
"""
Attaches a volume to a server
:param nova: the nova client
+ :param neutron: the neutron client
:param server: the VMInst domain object
:param volume: the Volume domain object
:param timeout: denotes the amount of time to block to determine if the
if timeout:
start_time = time.time()
while time.time() < start_time + timeout:
- vm = get_server_object_by_id(nova, server.id)
+ vm = get_server_object_by_id(nova, neutron, server.id)
found = False
for vol_dict in vm.volume_ids:
if volume.id == vol_dict['id']:
return None
else:
- return get_server_object_by_id(nova, server.id)
+ return get_server_object_by_id(nova, neutron, server.id)
class RebootType(enum.Enum):
if network:
ext_net_name = network.name
- ports_tuple_list = list()
+ out_ports = list()
if router.port_subnets:
for port, subnets in router.port_subnets:
network = neutron_utils.get_network_by_id(
ip_addrs.append(ext_fixed_ips['ip_address'])
else:
for ip in port.ips:
- ip_addrs.append(ip)
+ ip_addrs.append(ip['ip_address'])
- ip_list = list()
- if len(ip_addrs) > 0:
- for ip_addr in ip_addrs:
- if isinstance(ip_addr, dict):
- ip_list.append(ip_addr['ip_address'])
- else:
- ip_list.append(ip_addr)
+ ports = neutron_utils.get_ports(neutron, network, ip_addrs)
+ for out_port in ports:
+ out_ports.append(out_port)
- ports_tuple_list.append((network, ip_list))
-
- port_settings = __create_port_config(neutron, ports_tuple_list)
+ port_settings = __create_port_configs(neutron, out_ports)
filtered_settings = list()
for port_setting in port_settings:
kwargs['name'] = server.name
kwargs['flavor'] = flavor_name
- net_tuples = list()
- for net_name, ips in server.networks.items():
- network = neutron_utils.get_network(neutron, network_name=net_name)
- if network:
- net_tuples.append((network, ips))
-
- kwargs['port_settings'] = __create_port_config(
- neutron, net_tuples)
+ kwargs['port_settings'] = __create_port_configs(neutron, server.ports)
kwargs['security_group_names'] = server.sec_grp_names
kwargs['floating_ip_settings'] = __create_floatingip_config(
neutron, kwargs['port_settings'])
return VmInstanceConfig(**kwargs)
-def __create_port_config(neutron, networks):
+def __create_port_configs(neutron, ports):
"""
Returns a list of PortConfig objects based on the networks parameter
:param neutron: the neutron client
- :param networks: a list of tuples where #1 is the SNAPS Network domain
- object and #2 is a list of IP addresses
+ :param ports: a list of SNAPS-OO Port domain objects
:return:
"""
out = list()
- for network, ips in networks:
- ports = neutron_utils.get_ports(neutron, network, ips)
- for port in ports:
- if port.device_owner != 'network:dhcp':
- ip_addrs = list()
- for ip_dict in port.ips:
- subnet = neutron_utils.get_subnet_by_id(
- neutron, ip_dict['subnet_id'])
- ip_addrs.append({'subnet_name': subnet.name,
- 'ip': ip_dict['ip_address']})
-
- 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
- kwargs['ip_addrs'] = ip_addrs
- out.append(PortConfig(**kwargs))
+ for port in ports:
+ if port.device_owner != 'network:dhcp':
+ ip_addrs = list()
+ for ip_dict in port.ips:
+ subnet = neutron_utils.get_subnet_by_id(
+ neutron, ip_dict['subnet_id'])
+ ip_addrs.append({'subnet_name': subnet.name,
+ 'ip': ip_dict['ip_address']})
+
+ network = neutron_utils.get_network_by_id(neutron, port.network_id)
+ 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
+ kwargs['ip_addrs'] = ip_addrs
+ out.append(PortConfig(**kwargs))
return out
nova = nova_utils.nova_client(self.os_creds)
servers = heat_utils.get_stack_servers(
- self.heat_client, nova, self.stack1)
+ self.heat_client, nova, neutron, self.stack1)
self.assertIsNotNone(servers)
self.assertEqual(1, len(servers))
self.assertEqual(self.vm_inst_name, servers[0].name)
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)
+ self.heat_client, nova, neutron, self.stack)
for server in servers:
vm_settings = settings_utils.create_vm_inst_config(
nova, neutron, server)
glance = glance_utils.glance_client(self.os_creds)
servers = heat_utils.get_stack_servers(
- self.heat_client, nova, self.stack)
+ self.heat_client, nova, neutron, self.stack)
self.assertIsNotNone(servers)
self.assertEqual(2, len(servers))
iters += 1
self.assertTrue(active)
- vm_inst = nova_utils.get_latest_server_object(self.nova, self.vm_inst)
+ vm_inst = nova_utils.get_latest_server_object(
+ self.nova, self.neutron, self.vm_inst)
self.assertEqual(self.vm_inst.name, vm_inst.name)
self.assertEqual(self.vm_inst.id, vm_inst.id)
self.assertEqual(0, len(self.volume_creator.get_volume().attachments))
# Attach volume to VM
+ neutron = neutron_utils.neutron_client(self.os_creds)
nova_utils.attach_volume(
- self.nova, self.instance_creator.get_vm_inst(),
+ self.nova, neutron, self.instance_creator.get_vm_inst(),
self.volume_creator.get_volume())
time.sleep(10)
vol_attach = cinder_utils.get_volume_by_id(
self.cinder, self.volume_creator.get_volume().id)
vm_attach = nova_utils.get_server_object_by_id(
- self.nova, self.instance_creator.get_vm_inst().id)
+ self.nova, neutron, self.instance_creator.get_vm_inst().id)
# Detach volume to VM
nova_utils.detach_volume(
- self.nova, self.instance_creator.get_vm_inst(),
+ self.nova, neutron, self.instance_creator.get_vm_inst(),
self.volume_creator.get_volume())
time.sleep(10)
vol_detach = cinder_utils.get_volume_by_id(
self.cinder, self.volume_creator.get_volume().id)
vm_detach = nova_utils.get_server_object_by_id(
- self.nova, self.instance_creator.get_vm_inst().id)
+ self.nova, neutron, self.instance_creator.get_vm_inst().id)
# Validate Attachment
self.assertIsNotNone(vol_attach)
self.inst_creator.create(block=True)
server = nova_utils.get_server(
- self.nova, vm_inst_settings=self.inst_creator.instance_settings)
+ self.nova, self.neutron,
+ vm_inst_settings=self.inst_creator.instance_settings)
derived_vm_settings = settings_utils.create_vm_inst_config(
self.nova, self.neutron, server)
self.assertIsNotNone(derived_vm_settings)
self.inst_creator.create(block=True)
server = nova_utils.get_server(
- self.nova, vm_inst_settings=self.inst_creator.instance_settings)
+ self.nova, self.neutron,
+ vm_inst_settings=self.inst_creator.instance_settings)
derived_image_settings = settings_utils.determine_image_config(
self.glance, server, [self.image_creator.image_settings])
self.assertIsNotNone(derived_image_settings)