from functest.utils import config
from functest.utils import env
-from functest.utils import functest_utils
+
+
+class NewProject(object):
+ """Ease creating new projects/users"""
+ # pylint: disable=too-many-instance-attributes
+
+ __logger = logging.getLogger(__name__)
+
+ def __init__(self, cloud, case_name, guid):
+ self.cloud = None
+ self.orig_cloud = cloud
+ self.case_name = case_name
+ self.guid = guid
+ self.project = None
+ self.user = None
+ self.password = None
+ self.domain = None
+
+ def create(self):
+ """Create projects/users"""
+ assert self.orig_cloud
+ assert self.case_name
+ self.password = str(uuid.uuid4())
+ self.domain = self.orig_cloud.get_domain(
+ 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),
+ description="Created by OPNFV Functest: {}".format(
+ self.case_name),
+ domain_id=self.domain.id)
+ self.__logger.debug("project: %s", self.project)
+ self.user = self.orig_cloud.create_user(
+ name='{}-user_{}'.format(self.case_name, self.guid),
+ password=self.password,
+ domain_id=self.domain.id)
+ self.__logger.debug("user: %s", self.user)
+ self.orig_cloud.grant_role(
+ "Member", 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']['username'] = self.user.name
+ osconfig.cloud_config['clouds']['envvars']['password'] = self.password
+ self.__logger.debug("cloud_config %s", osconfig.cloud_config)
+ self.cloud = shade.OpenStackCloud(
+ cloud_config=osconfig.get_one_cloud())
+ self.__logger.debug("new cloud %s", self.cloud.auth)
+
+ def clean(self):
+ """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)
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Cannot clean all ressources")
class TenantNetwork1(testcase.TestCase):
__logger = logging.getLogger(__name__)
cidr = '192.168.0.0/24'
+ shared_network = False
def __init__(self, **kwargs):
if "case_name" not in kwargs:
self.ext_net = None
self.__logger.exception("Cannot connect to Cloud")
try:
- self.ext_net = functest_utils.get_external_network(self.cloud)
+ self.ext_net = self.get_external_network(self.cloud)
except Exception: # pylint: disable=broad-except
self.__logger.exception("Cannot get the external network")
self.guid = str(uuid.uuid4())
self.subnet = None
self.router = None
+ @staticmethod
+ def get_external_network(cloud):
+ """
+ Returns the configured external network name or
+ the first retrieved external network name
+ """
+ assert cloud
+ if env.get("EXTERNAL_NETWORK"):
+ network = cloud.get_network(
+ env.get("EXTERNAL_NETWORK"), {"router:external": True})
+ if network:
+ return network
+ networks = cloud.list_networks({"router:external": True})
+ if networks:
+ return networks[0]
+ return None
+
def _create_network_ressources(self):
assert self.cloud
assert self.ext_net
config.CONF, '{}_segmentation_id'.format(self.case_name))
self.network = self.cloud.create_network(
'{}-net_{}'.format(self.case_name, self.guid),
- provider=provider)
+ provider=provider,
+ shared=self.shared_network)
self.__logger.debug("network: %s", self.network)
self.subnet = self.cloud.create_subnet(
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)
except Exception: # pylint: disable=broad-except
self.__logger.exception("cannot clean all ressources")
super(TenantNetwork2, self).__init__(**kwargs)
try:
assert self.cloud
- self.domain = self.cloud.get_domain(
- name_or_id=self.cloud.auth.get(
- "project_domain_name", "Default"))
- except Exception: # pylint: disable=broad-except
- self.domain = None
- self.__logger.exception("Cannot connect to Cloud")
- self.project = None
- self.user = None
- self.orig_cloud = None
- self.password = str(uuid.uuid4())
-
- def run(self, **kwargs):
- assert self.cloud
- assert self.domain
- try:
- self.project = self.cloud.create_project(
- name='{}-project_{}'.format(self.case_name, self.guid),
- description="Created by OPNFV Functest: {}".format(
- self.case_name),
- domain_id=self.domain.id)
- self.__logger.debug("project: %s", self.project)
- self.user = self.cloud.create_user(
- name='{}-user_{}'.format(self.case_name, self.guid),
- password=self.password,
- default_project=self.project.id,
- domain_id=self.domain.id)
- self.__logger.debug("user: %s", self.user)
- self.orig_cloud = self.cloud
- os.environ["OS_USERNAME"] = self.user.name
- os.environ["OS_PROJECT_NAME"] = self.user.default_project_id
- cloud_config = os_client_config.get_config()
- self.cloud = shade.OpenStackCloud(cloud_config=cloud_config)
- os.environ["OS_USERNAME"] = self.orig_cloud.auth["username"]
- os.environ["OS_PROJECT_NAME"] = self.orig_cloud.auth[
- "project_name"]
+ self.project = NewProject(
+ self.cloud, self.case_name, self.guid)
+ self.project.create()
+ self.cloud = self.project.cloud
except Exception: # pylint: disable=broad-except
self.__logger.exception("Cannot create user or project")
- return testcase.TestCase.EX_RUN_ERROR
- return super(TenantNetwork2, self).run(**kwargs)
+ self.cloud = None
+ self.project = None
def clean(self):
try:
- assert self.cloud
super(TenantNetwork2, self).clean()
- assert self.user.id
- assert self.project.id
- self.orig_cloud.delete_user(self.user.id)
- self.orig_cloud.delete_project(self.project.id)
+ 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 ressources")
Code Review / functest.git / blobdiff