| | | IPv6 subnet |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_network_tests.py - CreateMultipleNetworkTests
+----------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Neutron API | Description |
++=======================================+===============+===========================================================+
+| test_network_same_name_diff_proj | 2 | Ensures that a network with the same name can be created |
+| | | against different projects |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_network_create_by_admin_to | 2 | Ensures that a network can be created by the admin user |
+| _different_project | | to another project and that a creator with the credentials|
+| | | to the other project will not create a new network with |
+| | | the same name |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_router_tests.py - CreateRouterSuccessTests
-------------------------------------------------
if not self.name:
raise RouterConfigError('Name is required')
- def dict_for_neutron(self, neutron, os_creds):
+ def dict_for_neutron(self, neutron, os_creds, project_id):
"""
Returns a dictionary object representing this object.
This is meant to be converted into JSON designed for use by the Neutron
TODO - expand automated testing to exercise all parameters
:param neutron: The neutron client to retrieve external network
information if necessary
- :param os_creds: The OpenStack credentials
+ :param os_creds: The OpenStack credentials for retrieving the keystone
+ client for looking up the project ID when the
+ self.project_name is not None
+ :param project_id: the associated project ID to use when
+ self.project_name is None
:return: the dictionary object
"""
out = dict()
keystone = keystone_utils.keystone_client(os_creds)
project = keystone_utils.get_project(
keystone=keystone, project_name=self.project_name)
- project_id = None
if project:
project_id = project.id
if project_id:
Constructor
:param name: the network's name
:param id: the network's ID
+ :param project_id: the associated project ID
:param admin_state_up: T/F - network is up when True
:param shared: T/F - network can be shared amongst other project's
:param external: T/F - network is deemed to be external
"""
self.name = kwargs.get('name')
self.id = kwargs.get('id')
+ self.project_id = kwargs.get('project_id')
self.admin_state_up = kwargs.get('admin_state_up')
self.shared = kwargs.get('shared')
self.external = kwargs.get('router:external', kwargs.get('external'))
def __eq__(self, other):
return (self.name == other.name and self.id == other.id and
+ self.project_id == other.project_id and
self.admin_state_up == other.admin_state_up and
self.shared == other.shared and
self.external == other.external and
Constructor
:param name: the network's name
:param id: the subnet's ID
+ :param project_id: the associated project ID
:param network_id: the network's ID
:param cidr: the CIDR
:param ip_version: the IP version
"""
self.name = kwargs.get('name')
self.id = kwargs.get('id')
+ self.project_id = kwargs.get('project_id')
self.network_id = kwargs.get('network_id')
self.cidr = kwargs.get('cidr')
self.ip_version = kwargs.get('ip_version')
def __eq__(self, other):
return (self.name == other.name and
self.id == other.id and
+ self.project_id == other.project_id and
self.network_id == other.network_id and
self.cidr == other.cidr and
self.ip_version == other.ip_version and
def test_construction_kwargs_1(self):
subnet = Subnet(
- **{'name': 'foo', 'id': 'bar', 'network_id': 'foo-bar'})
+ **{'name': 'foo', 'id': 'bar', 'project_id': 'proj1',
+ 'network_id': 'foo-bar'})
network = Network(
- **{'name': 'foo', 'id': 'bar', 'provider:network_type': 'flat',
- 'admin_state_up': False, 'shared': True,
- 'router:external': False, 'subnets': [subnet]})
+ **{'name': 'foo', 'id': 'bar', 'project_id': 'proj1',
+ 'provider:network_type': 'flat', 'admin_state_up': False,
+ 'shared': True, 'router:external': False, 'subnets': [subnet]})
self.assertEqual('foo', network.name)
self.assertEqual('bar', network.id)
+ self.assertEqual('proj1', network.project_id)
self.assertEqual('flat', network.type)
self.assertFalse(network.admin_state_up)
self.assertFalse(network.external)
def test_construction_kwargs_2(self):
subnet = Subnet(
- **{'name': 'foo', 'id': 'bar', 'network_id': 'foo-bar'})
+ **{'name': 'foo', 'id': 'bar', 'project_id': 'proj1',
+ 'network_id': 'foo-bar'})
network = Network(
- **{'name': 'foo', 'id': 'bar', 'type': 'flat',
- 'admin_state_up': False, 'shared': True, 'external': False,
- 'subnets': [subnet]})
+ **{'name': 'foo', 'id': 'bar', 'project_id': 'proj1',
+ 'type': 'flat', 'admin_state_up': False, 'shared': True,
+ 'external': False, 'subnets': [subnet]})
self.assertEqual('foo', network.name)
self.assertEqual('bar', network.id)
+ self.assertEqual('proj1', network.project_id)
self.assertEqual('flat', network.type)
self.assertFalse(network.admin_state_up)
self.assertFalse(network.external)
def test_construction_named(self):
subnet = Subnet(
- **{'name': 'foo', 'id': 'bar', 'network_id': 'foo-bar'})
+ **{'name': 'foo', 'id': 'bar', 'project_id': 'proj1',
+ 'network_id': 'foo-bar'})
network = Network(
- name='foo', id='bar', type='flat', admin_state_up=False,
- shared=True, external=False, subnets=[subnet])
+ name='foo', id='bar', project_id='proj1', type='flat',
+ admin_state_up=False, shared=True, external=False,
+ subnets=[subnet])
self.assertEqual('foo', network.name)
self.assertEqual('bar', network.id)
+ self.assertEqual('proj1', network.project_id)
self.assertEqual('flat', network.type)
self.assertFalse(network.admin_state_up)
self.assertFalse(network.external)
def test_construction_kwargs(self):
subnet = Subnet(
- **{'name': 'foo', 'id': 'bar', 'cidr': '10.0.0.0/24',
- 'ip_version': 4, 'gateway_ip': '10.0.0.1', 'enable_dhcp': True,
+ **{'name': 'foo', 'id': 'bar', 'project_id': 'proj1',
+ 'cidr': '10.0.0.0/24', 'ip_version': 4,
+ 'gateway_ip': '10.0.0.1', 'enable_dhcp': True,
'dns_nameservers': ['8.8.8.8'], 'host_routes': list(),
'ipv6_ra_mode': 'hello', 'ipv6_address_mode': 'world'})
self.assertEqual('foo', subnet.name)
self.assertEqual('bar', subnet.id)
+ self.assertEqual('proj1', subnet.project_id)
self.assertEqual('10.0.0.0/24', subnet.cidr)
self.assertEqual(4, subnet.ip_version)
self.assertEqual('10.0.0.1', subnet.gateway_ip)
def test_construction_named(self):
subnet = Subnet(
- name='foo', id='bar', cidr='10.0.0.0/24',
+ name='foo', id='bar', project_id='proj1', cidr='10.0.0.0/24',
ip_version=4, gateway_ip='10.0.0.1', enable_dhcp=True,
dns_nameservers=['8.8.8.8'], host_routes=list(),
ipv6_ra_mode='hello', ipv6_address_mode='world')
self.assertEqual('foo', subnet.name)
self.assertEqual('bar', subnet.id)
+ self.assertEqual('proj1', subnet.project_id)
self.assertEqual('10.0.0.0/24', subnet.cidr)
self.assertEqual(4, subnet.ip_version)
self.assertEqual('10.0.0.1', subnet.gateway_ip)
'Found existing machine with name - %s',
self.instance_settings.name)
- fips = neutron_utils.get_floating_ips(self.__neutron,
- self.__ports)
+ fips = neutron_utils.get_port_floating_ips(
+ self.__neutron, self.__ports)
for port_id, fip in fips:
settings = self.instance_settings.floating_ip_settings
for fip_setting in settings:
glance = glance_utils.glance_client(self._os_creds)
self.__vm = nova_utils.create_server(
self._nova, self.__neutron, glance, self.instance_settings,
- self.image_settings, self.keypair_settings)
+ self.image_settings, self.project_id, self.keypair_settings)
logger.info('Created instance with name - %s',
self.instance_settings.name)
if volume and self.vm_active(block=True):
vm = nova_utils.attach_volume(
self._nova, self.__neutron, self.__vm, volume,
- VOL_DETACH_TIMEOUT)
+ self.project_id, timeout=VOL_DETACH_TIMEOUT)
if vm:
self.__vm = vm
if volume:
vm = nova_utils.detach_volume(
self._nova, self.__neutron, self.__vm, volume,
- VOL_DETACH_TIMEOUT)
+ self.project_id, timeout=VOL_DETACH_TIMEOUT)
if vm:
self.__vm = vm
else:
for port_setting in port_settings:
port = neutron_utils.get_port(
- self.__neutron, port_settings=port_setting)
+ self.__neutron, port_settings=port_setting,
+ project_id=self.project_id)
if port:
ports.append((port_setting.name, port))
:return: Server object
"""
return nova_utils.get_server_object_by_id(
- self._nova, self.__neutron, self.__vm.id)
+ self._nova, self.__neutron, self.__vm.id, self.project_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.__neutron, self.__vm.id)
+ self._nova, self.__neutron, self.__vm.id, self.project_id)
return True
return False
self._nova, self.__vm, reboot_type=reboot_type)
-def generate_creator(os_creds, vm_inst, image_config, keypair_config=None):
+def generate_creator(os_creds, vm_inst, image_config, project_id,
+ keypair_config=None):
"""
Initializes an OpenStackVmInstance object
:param os_creds: the OpenStack credentials
:param vm_inst: the SNAPS-OO VmInst domain object
:param image_config: the associated ImageConfig object
+ :param project_id: the associated project ID
:param keypair_config: the associated KeypairConfig object (optional)
:return: an initialized OpenStackVmInstance object
"""
nova = nova_utils.nova_client(os_creds)
neutron = neutron_utils.neutron_client(os_creds)
derived_inst_config = settings_utils.create_vm_inst_config(
- nova, neutron, vm_inst)
+ nova, neutron, vm_inst, project_id)
derived_inst_creator = OpenStackVmInstance(
os_creds, derived_inst_config, image_config, keypair_config)
try:
self.__network = neutron_utils.get_network(
self._neutron, network_settings=self.network_settings,
- project_id=self.network_settings.get_project_id(
- self._os_creds))
+ project_id=self.project_id, os_creds=self._os_creds)
except Unauthorized as e:
logger.warn('Unable to lookup network with name %s - %s',
self.network_settings.name, e)
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
- self._neutron, port_settings=port_setting)
+ self._neutron, port_settings=port_setting,
+ project_id=self.project_id)
if port:
self.__ports.append(port)
if not self.__router:
self.__router = neutron_utils.create_router(
- self._neutron, self._os_creds, self.router_settings)
+ self._neutron, self._os_creds, self.router_settings,
+ self.project_id)
for internal_subnet_name in self.router_settings.internal_subnets:
internal_subnet = neutron_utils.get_subnet(
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
- self._neutron, port_settings=port_setting)
+ self._neutron, port_settings=port_setting,
+ project_id=self.project_id)
logger.info(
'Retrieved port %s for router - %s', port_setting.name,
self.router_settings.name)
neutron = neutron_utils.neutron_client(self._os_creds)
stack_servers = heat_utils.get_stack_servers(
- self.__heat_cli, nova, neutron, self.__stack)
+ self.__heat_cli, nova, neutron, self.__stack, self.project_id)
glance = glance_utils.glance_client(self._os_creds)
for stack_server in stack_servers:
vm_inst_settings = settings_utils.create_vm_inst_config(
- nova, neutron, stack_server)
+ nova, neutron, stack_server, self.project_id)
image_settings = settings_utils.determine_image_config(
glance, stack_server, self.image_settings)
keypair_settings = settings_utils.determine_keypair_config(
# Test default reboot which should be 'SOFT'
inst_creator.reboot()
# Lag time to allow for shutdown routine to take effect
- time.sleep(10)
+ time.sleep(15)
self.assertTrue(check_dhcp_lease(inst_creator, ip))
self.assertTrue(validate_ssh_client(inst_creator))
# Test 'SOFT' reboot
inst_creator.reboot(reboot_type=RebootType.soft)
- time.sleep(10)
+ time.sleep(15)
self.assertTrue(check_dhcp_lease(inst_creator, ip))
self.assertTrue(validate_ssh_client(inst_creator))
# Test 'HARD' reboot
inst_creator.reboot(reboot_type=RebootType.hard)
- time.sleep(10)
+ time.sleep(15)
self.assertTrue(check_dhcp_lease(inst_creator, ip))
self.assertTrue(validate_ssh_client(inst_creator))
derived_inst_creator = create_instance.generate_creator(
self.os_creds, vm_inst, self.image_creator.image_settings,
- self.keypair_creator.keypair_settings)
+ self.project_id, self.keypair_creator.keypair_settings)
derived_inst_creator.add_floating_ip(FloatingIpConfig(
name=self.floating_ip_name, port_name=self.port_1_name,
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True))
+ self.neutron, self.net_creator.network_settings.name, True,
+ project_id=self.project_id))
# Validate subnets
self.assertTrue(neutron_utils_tests.validate_subnet(
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True))
+ self.neutron, self.net_creator.network_settings.name, True,
+ self.project_id))
neutron_utils.delete_network(self.neutron,
self.net_creator.get_network())
self.assertIsNone(neutron_utils.get_network(
- self.neutron, network_settings=self.net_creator.network_settings))
+ self.neutron, network_settings=self.net_creator.network_settings,
+ os_creds=self.os_creds, project_id=self.project_id))
# This shall not throw an exception here
self.net_creator.clean()
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True))
+ self.neutron, self.net_creator.network_settings.name, True,
+ self.project_id))
# Validate subnets
self.assertTrue(neutron_utils_tests.validate_subnet(
self.net_creator.create()
retrieved_net = neutron_utils.get_network(
- self.neutron, network_settings=self.net_config.network_settings)
+ self.neutron, network_settings=self.net_config.network_settings,
+ os_creds=self.os_creds, project_id=self.project_id)
self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
self.net_creator.create()
retrieved_net = neutron_utils.get_network(
- self.neutron, network_settings=self.net_config.network_settings)
+ self.neutron, network_settings=self.net_config.network_settings,
+ os_creds=self.os_creds, project_id=self.project_id)
self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True))
+ self.neutron, self.net_creator.network_settings.name, True,
+ self.project_id))
network = self.net_creator.get_network()
self.assertEqual(1, len(network.subnets))
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
- self.neutron, net_settings.name, True))
+ self.neutron, net_settings.name, True,
+ self.net_creator.project_id))
self.assertEquals(network_type, network.type)
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
- self.neutron, net_settings.name, True))
+ self.neutron, net_settings.name, True,
+ self.net_creator.project_id))
self.assertEquals(network_type, network.type)
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
- self.neutron, net_settings.name, True))
+ self.neutron, net_settings.name, True, project_id=self.project_id))
self.assertEqual(network_type, network.type)
self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
with self.assertRaises(Exception):
self.net_creator.create()
+
+
+class CreateMultipleNetworkTests(OSIntegrationTestCase):
+ """
+ Test for the CreateNetwork class and how it interacts with networks
+ groups within other projects with the same name
+ """
+
+ def setUp(self):
+ """
+ Sets up object for test
+ """
+ super(self.__class__, self).__start__()
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.net_config = openstack_tests.get_pub_net_config(
+ net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
+
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+
+ # Initialize for cleanup
+ self.admin_net_creator = None
+ self.proj_net_creator = None
+
+ def tearDown(self):
+ """
+ Cleans the network
+ """
+ if self.admin_net_creator:
+ self.admin_net_creator.clean()
+ if self.proj_net_creator:
+ self.proj_net_creator.clean()
+
+ super(self.__class__, self).__clean__()
+
+ def test_network_same_name_diff_proj(self):
+ """
+ Tests the creation of an OpenStackNetwork with the same name
+ within a different project/tenant when not configured but implied by
+ the OSCreds.
+ """
+ # Create Network
+
+ self.admin_net_creator = OpenStackNetwork(
+ self.admin_os_creds, self.net_config.network_settings)
+ self.admin_net_creator.create()
+
+ self.proj_net_creator = OpenStackNetwork(
+ self.os_creds, self.net_config.network_settings)
+ self.proj_net_creator.create()
+
+ self.assertNotEqual(
+ self.admin_net_creator.get_network().id,
+ self.proj_net_creator.get_network().id)
+
+ admin_creator2 = OpenStackNetwork(
+ self.admin_os_creds, self.net_config.network_settings)
+ admin_creator2.create()
+ self.assertEqual(
+ self.admin_net_creator.get_network(), admin_creator2.get_network())
+
+ proj_creator2 = OpenStackNetwork(
+ self.os_creds, self.net_config.network_settings)
+ proj_creator2.create()
+ self.assertEqual(self.proj_net_creator.get_network(),
+ proj_creator2.get_network())
+
+ def test_network_create_by_admin_to_different_project(self):
+ """
+ Tests the creation of an OpenStackNetwork by the admin user and
+ initialize again with tenant credentials.
+ """
+ # Create Network
+
+ net_settings = self.net_config.network_settings
+
+ net_settings.project_name = self.os_creds.project_name
+
+ self.admin_net_creator = OpenStackNetwork(
+ self.admin_os_creds, net_settings)
+ self.admin_net_creator.create()
+
+ self.proj_net_creator = OpenStackNetwork(
+ self.os_creds, self.net_config.network_settings)
+ self.proj_net_creator.create()
+
+ self.assertEqual(
+ self.admin_net_creator.get_network().id,
+ self.proj_net_creator.get_network().id)
StackSettings, StackCreationError, StackError, OpenStackHeatStack)
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, nova_utils
+from snaps.openstack.utils import (heat_utils, neutron_utils, nova_utils,
+ keystone_utils)
__author__ = 'spisarski'
self.assertEqual(self.network_name, net_creators[0].get_network().name)
neutron = neutron_utils.neutron_client(self.os_creds)
+ admin_proj_id = keystone_utils.get_project(
+ self.keystone, self.admin_os_creds.project_name)
net_by_name = neutron_utils.get_network(
- neutron, network_name=net_creators[0].get_network().name)
+ neutron, network_name=net_creators[0].get_network().name,
+ project_id=admin_proj_id)
self.assertEqual(net_creators[0].get_network(), net_by_name)
self.assertIsNotNone(neutron_utils.get_network_by_id(
neutron, net_creators[0].get_network().id))
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, neutron, vm_inst_creators[0].get_vm_inst().id))
+ nova, neutron, vm_inst_creators[0].get_vm_inst().id,
+ vm_inst_creators[0].project_id))
class CreateStackFloatingIpTests(OSIntegrationTestCase):
router = creator.get_router()
ext_net = neutron_utils.get_network(
- self.neutron, network_name=self.ext_net_name)
+ self.neutron, network_name=self.ext_net_name,
+ project_id=self.project_id)
self.assertEqual(ext_net.id, router.external_network_id)
keystone = keystone_utils.keystone_client(self.os_creds)
self.project_id = keystone_utils.get_project(
- keystone=keystone, project_name=self.os_creds.project_name)
+ keystone=keystone, project_name=self.os_creds.project_name).id
@staticmethod
def parameterize(testcase_klass, os_creds, ext_net_name,
# add user to project
self.project_creator.assoc_user(self.user_creator.get_user())
+ self.project_id = self.project_creator.get_project().id
def __clean__(self):
"""
return out
-def get_stack_servers(heat_cli, nova, neutron, stack):
+def get_stack_servers(heat_cli, nova, neutron, stack, project_id):
"""
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
+ :param project_id: the associated project ID
:return: a list of VMInst domain objects
"""
for resource in srvr_res:
try:
server = nova_utils.get_server_object_by_id(
- nova, neutron, resource.id)
+ nova, neutron, resource.id, project_id)
if server:
out.append(server)
except NotFound:
heat_cli, res_res.id, 'OS::Nova::Server')
for res_srvr in res_res_srvrs:
server = nova_utils.get_server_object_by_id(
- nova, neutron, res_srvr.id)
+ nova, neutron, res_srvr.id, project_id)
if server:
out.append(server)
def get_network(neutron, network_settings=None, network_name=None,
- project_id=None):
+ project_id=None, os_creds=None):
"""
Returns Network SNAPS-OO domain object the first network found with
either the given attributes from the network_settings object if not None,
:param network_settings: the NetworkConfig object used to create filter
:param network_name: the name of the network to retrieve
:param project_id: the id of the network's project
+ :param os_creds: the OpenStack credentials for retrieving the project
:return: a SNAPS-OO Network domain object
"""
net_filter = dict()
elif network_name:
net_filter['name'] = network_name
- if project_id:
+ if network_settings and network_settings.project_name and os_creds:
+ net_filter['project_id'] = keystone_utils.get_project(
+ os_creds=os_creds, project_name=network_settings.project_name).id
+ elif project_id:
net_filter['project_id'] = project_id
networks = neutron.list_networks(**net_filter)
return out
-def create_router(neutron, os_creds, router_settings):
+def create_router(neutron, os_creds, router_settings, project_id):
"""
Creates a router for OpenStack
:param neutron: the client
:param router_settings: A dictionary containing the router configuration
and is responsible for creating the subnet request
JSON body
+ :param project_id: the associated project ID
:return: a SNAPS-OO Router domain object
"""
if neutron:
- json_body = router_settings.dict_for_neutron(neutron, os_creds)
+ if router_settings and router_settings.project_name:
+ keystone = keystone_utils.keystone_client(os_creds)
+ project_id = keystone_utils.get_project(
+ keystone=keystone, project_name=router_settings.project_name)
+ json_body = router_settings.dict_for_neutron(
+ neutron, os_creds, project_id)
logger.info('Creating router with name - ' + router_settings.name)
os_router = neutron.create_router(json_body)
return __map_router(neutron, os_router['router'])
logger.info('Creating port for network with name - %s',
port_settings.network_name)
os_port = neutron.create_port(body=json_body)['port']
- return Port(name=os_port['name'], id=os_port['id'],
- ips=os_port['fixed_ips'],
- mac_address=os_port['mac_address'],
- allowed_address_pairs=os_port['allowed_address_pairs'])
+ return Port(**os_port)
def delete_port(neutron, port):
neutron.delete_port(port.id)
-def get_port(neutron, port_settings=None, port_name=None):
+def get_port(neutron, port_settings=None, port_name=None, project_id=None):
"""
Returns the first port object (dictionary) found for the given query
:param neutron: the client
:param port_settings: the PortConfig object used for generating the query
:param port_name: if port_settings is None, this name is the value to place
into the query
+ :param project_id: the associated project ID
:return: a SNAPS-OO Port domain object
"""
port_filter = dict()
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)
+ network = get_network(
+ neutron, network_name=port_settings.network_name,
+ project_id=project_id)
if network:
port_filter['network_id'] = network.id
elif port_name:
port_filter['name'] = port_name
+ if project_id:
+ port_filter['project_id'] = project_id
+
ports = neutron.list_ports(**port_filter)
for port in ports['ports']:
return Port(**port)
return out
-def get_floating_ips(neutron, ports=None):
+def get_port_floating_ips(neutron, ports):
"""
- 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
- 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
+ Returns all of the floating IPs associated with the ports returned in a
+ list of tuples where the port object is in the first position and the
+ floating IP object is in the second
:param neutron: the Neutron client
:param ports: a list of tuple 2 where index 0 is the port name and index 1
is the SNAPS-OO Port object
out = list()
fips = neutron.list_floatingips()
for fip in fips['floatingips']:
- if ports:
- for port_name, port in ports:
- if port and port.id == fip['port_id']:
- out.append((port.id, FloatingIp(**fip)))
- break
- else:
- out.append(FloatingIp(**fip))
+ for port_name, port in ports:
+ if port and port.id == fip['port_id']:
+ out.append((port.id, FloatingIp(**fip)))
+ break
+ return out
+
+def get_floating_ips(neutron):
+ """
+ Returns a list of all of the floating IPs
+ :param neutron: the Neutron client
+ """
+ out = list()
+ fips = neutron.list_floatingips()
+ for fip in fips['floatingips']:
+ out.append(FloatingIp(**fip))
return out
def create_server(nova, neutron, glance, instance_config, image_config,
- keypair_config=None):
+ project_id, keypair_config=None):
"""
Creates a VM instance
:param nova: the nova client (required)
:param glance: the glance client (required)
:param instance_config: the VMInstConfig object (required)
:param image_config: the VM's ImageConfig object (required)
+ :param project_id: the associated project ID (required)
:param keypair_config: the VM's KeypairConfig object (optional)
:return: a snaps.domain.VmInst object
"""
ports = list()
for port_setting in instance_config.port_settings:
- port = neutron_utils.get_port(neutron, port_settings=port_setting)
+ port = neutron_utils.get_port(
+ neutron, port_settings=port_setting, project_id=project_id)
if port:
ports.append(port)
else:
server = nova.servers.create(**args)
- return __map_os_server_obj_to_vm_inst(neutron, server)
+ return __map_os_server_obj_to_vm_inst(neutron, server, project_id)
else:
raise NovaException(
'Cannot create instance, image cannot be located with name %s',
image_config.name)
-def get_server(nova, neutron, vm_inst_settings=None, server_name=None):
+def get_server(nova, neutron, vm_inst_settings=None, server_name=None,
+ project_id=None):
"""
Returns a VmInst object for the first server instance found.
:param nova: the Nova client
the query if not None
:param server_name: the server with this name to return if vm_inst_settings
is not None
+ :param project_id: the assocaited project ID
:return: a snaps.domain.VmInst object or None if not found
"""
search_opts = dict()
servers = nova.servers.list(search_opts=search_opts)
for server in servers:
- return __map_os_server_obj_to_vm_inst(neutron, server)
+ return __map_os_server_obj_to_vm_inst(neutron, server, project_id)
def get_server_connection(nova, vm_inst_settings=None, server_name=None):
return server.links[0]
-def __map_os_server_obj_to_vm_inst(neutron, os_server):
+def __map_os_server_obj_to_vm_inst(neutron, os_server, project_id):
"""
Returns a VmInst object for an OpenStack Server object
:param neutron: the Neutron client (when None, ports will be empty)
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)
+ network = neutron_utils.get_network(
+ neutron, network_name=net_name, project_id=project_id)
ports = neutron_utils.get_ports(neutron, network, ips)
for port in ports:
out_ports.append(port)
return None
-def get_latest_server_object(nova, neutron, server):
+def get_latest_server_object(nova, neutron, server, project_id):
"""
Returns a server with a given id
:param nova: the Nova client
:param neutron: the Neutron client
:param server: the old server object
+ :param project_id: the associated project ID
: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(neutron, server)
+ return __map_os_server_obj_to_vm_inst(neutron, server, project_id)
-def get_server_object_by_id(nova, neutron, server_id):
+def get_server_object_by_id(nova, neutron, server_id, project_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
+ :param project_id: the associated project 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(neutron, server)
+ return __map_os_server_obj_to_vm_inst(neutron, server, project_id)
def get_server_security_group_names(nova, server):
return nova.quotas.update(project_id, **update_values)
-def attach_volume(nova, neutron, server, volume, timeout=120):
+def attach_volume(nova, neutron, server, volume, project_id, timeout=120):
"""
Attaches a volume to a server. When the timeout parameter is used, a VmInst
object with the proper volume updates is returned unless it has not been
:param neutron: the neutron client
:param server: the VMInst domain object
:param volume: the Volume domain object
+ :param project_id: the associated project ID
:param timeout: denotes the amount of time to block to determine if the
has been properly attached.
:return: updated VmInst object
start_time = time.time()
while time.time() < start_time + timeout:
- vm = get_server_object_by_id(nova, neutron, server.id)
+ vm = get_server_object_by_id(nova, neutron, server.id, project_id)
for vol_dict in vm.volume_ids:
if volume.id == vol_dict['id']:
return vm
volume.id, server.id))
-def detach_volume(nova, neutron, server, volume, timeout=120):
+def detach_volume(nova, neutron, server, volume, project_id, timeout=120):
"""
Detaches a volume to a server. When the timeout parameter is used, a VmInst
object with the proper volume updates is returned unless it has not been
:param neutron: the neutron client
:param server: the VMInst domain object
:param volume: the Volume domain object
+ :param project_id: the associated project ID
:param timeout: denotes the amount of time to block to determine if the
has been properly detached.
:return: updated VmInst object
start_time = time.time()
while time.time() < start_time + timeout:
- vm = get_server_object_by_id(nova, neutron, server.id)
+ vm = get_server_object_by_id(nova, neutron, server.id, project_id)
if len(vm.volume_ids) == 0:
return vm
else:
return KeypairConfig(name=keypair.name)
-def create_vm_inst_config(nova, neutron, server):
+def create_vm_inst_config(nova, neutron, server, project_id):
"""
Returns a VmInstanceConfig object
note: if the server instance is not active, the PortSettings objects will
:param nova: the nova client
:param neutron: the neutron client
:param server: a SNAPS-OO VmInst domain object
+ :param project_id: the associated project ID
:return:
"""
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'])
+ neutron, kwargs['port_settings'], project_id)
return VmInstanceConfig(**kwargs)
return out
-def __create_floatingip_config(neutron, port_settings):
+def __create_floatingip_config(neutron, port_settings, project_id):
"""
Returns a list of FloatingIpConfig objects as they pertain to an
existing deployed server instance
fip_ports = list()
for port_setting in port_settings:
- setting_port = neutron_utils.get_port(neutron, port_setting)
+ setting_port = neutron_utils.get_port(
+ neutron, port_setting, project_id=project_id)
if setting_port:
network = neutron_utils.get_network(
neutron, network_name=port_setting.network_name)
fip_ports.append((port_setting.name, setting_port))
break
- floating_ips = neutron_utils.get_floating_ips(neutron, fip_ports)
+ floating_ips = neutron_utils.get_port_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)
nova = nova_utils.nova_client(self.os_creds)
servers = heat_utils.get_stack_servers(
- self.heat_client, nova, neutron, self.stack1)
+ self.heat_client, nova, neutron, self.stack1, self.project_id)
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, neutron, self.stack)
+ self.heat_client, nova, neutron, self.stack,
+ self.project_id)
for server in servers:
vm_settings = settings_utils.create_vm_inst_config(
- nova, neutron, server)
+ nova, neutron, server, self.project_id)
img_settings = settings_utils.determine_image_config(
glance, server,
[self.image_creator1.image_settings,
glance = glance_utils.glance_client(self.os_creds)
servers = heat_utils.get_stack_servers(
- self.heat_client, nova, neutron, self.stack)
+ self.heat_client, nova, neutron, self.stack, self.project_id)
self.assertIsNotNone(servers)
self.assertEqual(2, len(servers))
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
self.assertEqual(len(self.net_config.network_settings.subnet_settings),
len(self.network.subnets))
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
with self.assertRaises(Exception):
SubnetConfig(cidr=self.net_config.subnet_cidr)
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
Tests the neutron_utils.create_router()
"""
self.router = neutron_utils.create_router(
- self.neutron, self.os_creds, self.net_config.router_settings)
+ self.neutron, self.os_creds, self.net_config.router_settings,
+ self.project_id)
validate_router(self.neutron, self.net_config.router_settings.name,
True)
self.net_config.router_settings.name,
self.ext_net_name)
self.router = neutron_utils.create_router(
- self.neutron, self.os_creds, self.net_config.router_settings)
+ self.neutron, self.os_creds, self.net_config.router_settings,
+ self.project_id)
validate_router(self.neutron, self.net_config.router_settings.name,
True)
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
self.neutron, subnet_setting.name, subnet_setting.cidr, True))
self.router = neutron_utils.create_router(
- self.neutron, self.os_creds, self.net_config.router_settings)
+ self.neutron, self.os_creds, self.net_config.router_settings,
+ self.project_id)
validate_router(self.neutron, self.net_config.router_settings.name,
True)
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
self.router = neutron_utils.create_router(
- self.neutron, self.os_creds, self.net_config.router_settings)
+ self.neutron, self.os_creds, self.net_config.router_settings,
+ self.project_id)
validate_router(self.neutron, self.net_config.router_settings.name,
True)
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
self.router = neutron_utils.create_router(
- self.neutron, self.os_creds, self.net_config.router_settings)
+ self.neutron, self.os_creds, self.net_config.router_settings,
+ self.project_id)
validate_router(self.neutron, self.net_config.router_settings.name,
True)
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(self.neutron, subnet_setting.name,
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
self.assertEqual(self.net_config.network_settings.name,
self.network.name)
self.assertTrue(validate_network(
- self.neutron, self.net_config.network_settings.name, True))
+ self.neutron, self.net_config.network_settings.name, True,
+ self.project_id))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
"""
-def validate_network(neutron, name, exists):
+def validate_network(neutron, name, exists, project_id):
"""
Returns true if a network for a given name DOES NOT exist if the exists
parameter is false conversely true. Returns false if a network for a given
:param neutron: The neutron client
:param name: The expected network name
:param exists: Whether or not the network name should exist or not
+ :param project_id: the associated project ID
:return: True/False
"""
- network = neutron_utils.get_network(neutron, network_name=name)
+ network = neutron_utils.get_network(neutron, network_name=name,
+ project_id=project_id)
if exists and network:
return True
if not exists and not network:
self.vm_inst = nova_utils.create_server(
self.nova, self.neutron, self.glance, self.instance_settings,
- self.image_creator.image_settings)
+ self.image_creator.image_settings, self.project_id)
self.assertIsNotNone(self.vm_inst)
self.assertTrue(active)
vm_inst = nova_utils.get_latest_server_object(
- self.nova, self.neutron, self.vm_inst)
+ self.nova, self.neutron, self.vm_inst, self.project_id)
self.assertEqual(self.vm_inst.name, vm_inst.name)
self.assertEqual(self.vm_inst.id, vm_inst.id)
neutron = neutron_utils.neutron_client(self.os_creds)
self.assertIsNotNone(nova_utils.attach_volume(
self.nova, neutron, self.instance_creator.get_vm_inst(),
- self.volume_creator.get_volume()))
+ self.volume_creator.get_volume(), self.project_id))
vol_attach = None
attached = False
self.assertIsNotNone(vol_attach)
vm_attach = nova_utils.get_server_object_by_id(
- self.nova, neutron, self.instance_creator.get_vm_inst().id)
+ self.nova, neutron, self.instance_creator.get_vm_inst().id,
+ self.project_id)
# Validate Attachment
self.assertIsNotNone(vol_attach)
# Detach volume to VM
self.assertIsNotNone(nova_utils.detach_volume(
self.nova, neutron, self.instance_creator.get_vm_inst(),
- self.volume_creator.get_volume()))
+ self.volume_creator.get_volume(), self.project_id))
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, neutron, self.instance_creator.get_vm_inst().id)
+ self.nova, neutron, self.instance_creator.get_vm_inst().id,
+ self.project_id)
# Validate Detachment
self.assertIsNotNone(vol_detach)
with self.assertRaises(NovaException):
nova_utils.attach_volume(
self.nova, neutron, self.instance_creator.get_vm_inst(),
- self.volume_creator.get_volume(), 0)
+ self.volume_creator.get_volume(), self.project_id, 0)
def test_detach_volume_nowait(self):
"""
neutron = neutron_utils.neutron_client(self.os_creds)
nova_utils.attach_volume(
self.nova, neutron, self.instance_creator.get_vm_inst(),
- self.volume_creator.get_volume())
+ self.volume_creator.get_volume(), self.project_id)
# Check VmInst for attachment
latest_vm = nova_utils.get_server_object_by_id(
- self.nova, neutron, self.instance_creator.get_vm_inst().id)
+ self.nova, neutron, self.instance_creator.get_vm_inst().id,
+ self.project_id)
self.assertEqual(1, len(latest_vm.volume_ids))
# Check Volume for attachment
with self.assertRaises(NovaException):
nova_utils.detach_volume(
self.nova, neutron, self.instance_creator.get_vm_inst(),
- self.volume_creator.get_volume(), 0)
+ self.volume_creator.get_volume(), self.project_id, 0)
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.nova, self.neutron, server, self.project_id)
self.assertIsNotNone(derived_vm_settings)
self.assertIsNotNone(derived_vm_settings.port_settings)
self.assertIsNotNone(derived_vm_settings.floating_ip_settings)
Code Review / snaps.git / commitdiff