"""Ease deploying tenant networks
-It offers a simple way to create all tenant network ressources required by a
+It offers a simple way to create all tenant network resources required by a
testcase (including all Functest ones):
+
- TenantNetwork1 selects the user and the project set as env vars
- - TenantNetwork2 creates a user and project to isolate the same ressources
+ - TenantNetwork2 creates a user and project to isolate the same resources
This classes could be reused by more complexed scenarios (Single VM)
"""
# pylint: disable=too-many-instance-attributes
__logger = logging.getLogger(__name__)
- default_member = "Member"
def __init__(self, cloud, case_name, guid):
self.cloud = None
self.password = None
self.domain = None
self.role = None
+ self.role_name = None
+ self.default_member = env.get('NEW_USER_ROLE')
def create(self):
"""Create projects/users"""
name_or_id=self.orig_cloud.auth.get(
"project_domain_name", "Default"))
self.project = self.orig_cloud.create_project(
- name='{}-project_{}'.format(self.case_name, self.guid),
+ name='{}-project_{}'.format(self.case_name[:18], self.guid),
description="Created by OPNFV Functest: {}".format(
self.case_name),
domain_id=self.domain.id)
domain_id=self.domain.id)
self.__logger.debug("user: %s", self.user)
try:
- if not self.orig_cloud.get_role(
- self.default_member) and not self.orig_cloud.get_role(
- self.default_member.lower()):
+ if self.orig_cloud.get_role(self.default_member):
+ self.role_name = self.default_member
+ elif self.orig_cloud.get_role(self.default_member.lower()):
+ self.role_name = self.default_member.lower()
+ else:
raise Exception("Cannot detect {}".format(self.default_member))
except Exception: # pylint: disable=broad-except
self.__logger.info("Creating default role %s", self.default_member)
self.role = self.orig_cloud.create_role(self.default_member)
+ self.role_name = self.role.name
self.__logger.debug("role: %s", self.role)
self.orig_cloud.grant_role(
- self.default_member, user=self.user.id, project=self.project.id,
+ self.role_name, user=self.user.id, project=self.project.id,
domain=self.domain.id)
osconfig = os_client_config.config.OpenStackConfig()
osconfig.cloud_config[
'clouds']['envvars']['project_name'] = self.project.name
+ osconfig.cloud_config[
+ 'clouds']['envvars']['project_id'] = self.project.id
osconfig.cloud_config['clouds']['envvars']['username'] = self.user.name
osconfig.cloud_config['clouds']['envvars']['password'] = self.password
self.__logger.debug("cloud_config %s", osconfig.cloud_config)
"""Remove projects/users"""
try:
assert self.orig_cloud
- assert self.user.id
- assert self.project.id
- self.orig_cloud.delete_user(self.user.id)
- self.orig_cloud.delete_project(self.project.id)
+ if self.user:
+ self.orig_cloud.delete_user(self.user.id)
+ if self.project:
+ self.orig_cloud.delete_project(self.project.id)
if self.role:
self.orig_cloud.delete_role(self.role.id)
+ secgroups = self.orig_cloud.list_security_groups(
+ filters={'name': 'default',
+ 'project_id': self.project.id})
+ if secgroups:
+ sec_id = secgroups[0].id
+ self.orig_cloud.delete_security_group(sec_id)
except Exception: # pylint: disable=broad-except
- self.__logger.exception("Cannot clean all ressources")
+ self.__logger.exception("Cannot clean all resources")
class TenantNetwork1(testcase.TestCase):
# pylint: disable=too-many-instance-attributes
"""Create a tenant network (scenario1)
- It creates and configures all tenant network ressources required by
+ It creates and configures all tenant network resources required by
advanced testcases (subnet, network and router).
It ensures that all testcases inheriting from TenantNetwork1 could work
"""
__logger = logging.getLogger(__name__)
- cidr = '192.168.0.0/24'
+ cidr = '192.168.120.0/24'
shared_network = False
def __init__(self, **kwargs):
getattr(config.CONF, 'dir_results'), self.case_name)
try:
cloud_config = os_client_config.get_config()
- self.cloud = shade.OpenStackCloud(cloud_config=cloud_config)
+ self.cloud = self.orig_cloud = shade.OpenStackCloud(
+ cloud_config=cloud_config)
except Exception: # pylint: disable=broad-except
- self.cloud = None
+ self.cloud = self.orig_cloud = None
self.ext_net = None
self.__logger.exception("Cannot connect to Cloud")
try:
@staticmethod
def get_external_network(cloud):
"""
- Returns the configured external network name or
+ Return the configured external network name or
the first retrieved external network name
"""
assert cloud
return networks[0]
return None
- def _create_network_ressources(self):
+ @staticmethod
+ def get_default_role(cloud, member="Member"):
+ """Get the default role
+
+ It also tests the role in lowercase to avoid possible conflicts.
+ """
+ role = cloud.get_role(member)
+ if not role:
+ role = cloud.get_role(member.lower())
+ return role
+
+ @staticmethod
+ def get_public_auth_url(cloud):
+ """Get Keystone public endpoint"""
+ keystone_id = cloud.search_services('keystone')[0].id
+ endpoint = cloud.search_endpoints(
+ filters={'interface': 'public',
+ 'service_id': keystone_id})[0].url
+ return endpoint
+
+ def create_network_resources(self):
+ """Create all tenant network resources
+
+ It creates a router which gateway is the external network detected.
+ The new subnet is attached to that router.
+
+ Raises: expection on error
+ """
assert self.cloud
assert self.ext_net
provider = {}
if hasattr(config.CONF, '{}_segmentation_id'.format(self.case_name)):
provider["segmentation_id"] = getattr(
config.CONF, '{}_segmentation_id'.format(self.case_name))
- self.network = self.cloud.create_network(
+ domain = self.orig_cloud.get_domain(
+ name_or_id=self.orig_cloud.auth.get(
+ "project_domain_name", "Default"))
+ project = self.orig_cloud.get_project(
+ self.cloud.auth['project_name'],
+ domain_id=domain.id)
+ self.network = self.orig_cloud.create_network(
'{}-net_{}'.format(self.case_name, self.guid),
- provider=provider,
+ provider=provider, project_id=project.id,
shared=self.shared_network)
self.__logger.debug("network: %s", self.network)
try:
assert self.cloud
self.start_time = time.time()
- self._create_network_ressources()
+ self.create_network_resources()
self.result = 100
status = testcase.TestCase.EX_OK
except Exception: # pylint: disable=broad-except
def clean(self):
try:
assert self.cloud
- self.cloud.remove_router_interface(self.router, self.subnet.id)
- self.cloud.delete_router(self.router.id)
- self.cloud.delete_subnet(self.subnet.id)
- self.cloud.delete_network(self.network.id)
+ if self.router:
+ if self.subnet:
+ self.cloud.remove_router_interface(
+ self.router, self.subnet.id)
+ self.cloud.delete_router(self.router.id)
+ if self.subnet:
+ self.cloud.delete_subnet(self.subnet.id)
+ if self.network:
+ self.cloud.delete_network(self.network.id)
except Exception: # pylint: disable=broad-except
- self.__logger.exception("cannot clean all ressources")
+ self.__logger.exception("cannot clean all resources")
class TenantNetwork2(TenantNetwork1):
"""Create a tenant network (scenario2)
It creates new user/project before creating and configuring all tenant
- network ressources required by a testcase (subnet, network and router).
+ network resources required by a testcase (subnet, network and router).
It ensures that all testcases inheriting from TenantNetwork2 could work
without network specific configurations (or at least read the same config
assert self.project
self.project.clean()
except Exception: # pylint: disable=broad-except
- self.__logger.exception("Cannot clean all ressources")
+ self.__logger.exception("Cannot clean all resources")
Code Review / functest.git / blobdiff