ports = list()
for port_setting in port_settings:
- port = neutron_utils.get_port_by_name(self.__neutron,
- port_setting.name)
+ port = neutron_utils.get_port(
+ self.__neutron, port_settings=port_setting)
if port:
ports.append((port_setting.name, port))
elif not cleanup:
import logging
from keystoneclient.exceptions import NotFound
-from snaps.openstack.utils import keystone_utils
+from snaps.openstack.utils import keystone_utils, neutron_utils
__author__ = 'spisarski'
:return: void
"""
if self.__project:
+ # Delete security group 'default' if exists
+ neutron = neutron_utils.neutron_client(self.__os_creds)
+ default_sec_grp = neutron_utils.get_security_group(
+ neutron, 'default',
+ tenant_id=self.__project.id)
+ if default_sec_grp:
+ try:
+ neutron_utils.delete_security_group(
+ neutron, default_sec_grp)
+ except:
+ pass
+
+ # Delete Project
try:
keystone_utils.delete_project(self.__keystone, self.__project)
except NotFound:
'Subnet not found with name ' + internal_subnet_name)
for port_setting in self.router_settings.port_settings:
- port = neutron_utils.get_port_by_name(self.__neutron,
- port_setting.name)
+ port = neutron_utils.get_port(
+ self.__neutron, port_settings=port_setting)
logger.info(
'Retrieved port %s for router - %s', port_setting.name,
self.router_settings.name)
self.assertTrue(verify_router_attributes(router, self.router_creator))
+ # Instantiate second identical creator to ensure a second router
+ # has not been created
+ router_creator2 = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
+ router2 = router_creator2.create()
+ self.assertIsNotNone(self.router_creator.get_router(), router2)
+
def test_create_router_external_network(self):
"""
Test creation of a router connected to an external network and a
self.role = None
if self.use_keystone:
- self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.keystone = keystone_utils.keystone_client(self.admin_os_creds)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())[:-19]
project_name = guid + '-proj'
self.project_creator = deploy_utils.create_project(
- self.admin_os_creds, ProjectSettings(name=project_name))
+ self.admin_os_creds, ProjectSettings(
+ name=project_name,
+ domain=self.admin_os_creds.project_domain_name))
self.user_creator = deploy_utils.create_user(
self.admin_os_creds, UserSettings(
name=guid + '-user', password=guid,
project_name=project_name, roles={
- 'admin': self.project_creator.project_settings.name}))
+ 'admin': self.project_creator.project_settings.name},
+ domain_name=self.admin_os_creds.user_domain_name))
self.os_creds = self.user_creator.get_os_creds(
self.project_creator.project_settings.name)
neutron.delete_port(port.id)
-def get_port_by_name(neutron, port_name):
+def get_port(neutron, port_settings=None, port_name=None):
"""
- Returns the first port object (dictionary) found with a given name
+ Returns the first port object (dictionary) found for the given query
:param neutron: the client
- :param port_name: the name of the port to retrieve
+ :param port_settings: the PortSettings object used for generating the query
+ :param port_name: if port_settings is None, this name is the value to place
+ into the query
:return: a SNAPS-OO Port domain object
"""
- ports = neutron.list_ports(**{'name': port_name})
+ port_filter = dict()
+
+ if port_settings:
+ port_filter['name'] = port_settings.name
+ if port_settings.admin_state_up:
+ port_filter['admin_state_up'] = port_settings.admin_state_up
+ if port_settings.device_id:
+ port_filter['device_id'] = port_settings.device_id
+ if port_settings.mac_address:
+ port_filter['mac_address'] = port_settings.mac_address
+ elif port_name:
+ port_filter['name'] = port_name
+
+ ports = neutron.list_ports(**port_filter)
for port in ports['ports']:
- if port['name'] == port_name:
- return Port(name=port['name'], id=port['id'],
- ips=port['fixed_ips'], mac_address=port['mac_address'])
+ return Port(name=port['name'], id=port['id'],
+ ips=port['fixed_ips'], mac_address=port['mac_address'])
return None
neutron.delete_security_group(sec_grp.id)
-def get_security_group(neutron, name):
+def get_security_group(neutron, name, tenant_id=None):
"""
Returns the first security group object of the given name else None
:param neutron: the client
"""
logger.info('Retrieving security group with name - ' + name)
- groups = neutron.list_security_groups(**{'name': name})
+ filter = {'name': name}
+ if tenant_id:
+ filter['tenant_id'] = tenant_id
+ groups = neutron.list_security_groups(**filter)
for group in groups['security_groups']:
if group['name'] == name:
return SecurityGroup(**group)
ports = list()
for port_setting in instance_settings.port_settings:
- ports.append(neutron_utils.get_port_by_name(
- neutron, port_setting.name))
+ ports.append(neutron_utils.get_port(
+ neutron, port_settings=port_setting))
nics = []
for port in ports:
kv = dict()
from snaps.openstack.create_project import ProjectSettings
from snaps.openstack.create_user import UserSettings
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
-from snaps.openstack.utils import keystone_utils
+from snaps.openstack.utils import keystone_utils, neutron_utils
__author__ = 'spisarski'
Cleans the remote OpenStack objects
"""
if self.project:
- keystone_utils.delete_project(self.keystone, self.project)
+ neutron = neutron_utils.neutron_client(self.os_creds)
+ default_sec_grp = neutron_utils.get_security_group(
+ neutron, 'default',
+ tenant_id=self.project.id)
+ if default_sec_grp:
+ try:
+ neutron_utils.delete_security_group(
+ neutron, default_sec_grp)
+ except:
+ pass
+
+ keystone_utils.delete_project(self.keystone, self.project)
if self.user:
keystone_utils.delete_user(self.keystone, self.user)
Code Review / snaps.git / commitdiff