"""
import logging
-import os
import tempfile
import time
-import uuid
-import os_client_config
import paramiko
-import shade
from xtesting.core import testcase
from functest.core import tenantnetwork
from functest.utils import config
-class SingleVm1(tenantnetwork.TenantNetwork1):
+class VmReady1(tenantnetwork.TenantNetwork1):
"""Deploy a single VM reachable via ssh (scenario1)
It inherits from TenantNetwork1 which creates all network resources and
- completes it by booting a VM attached to that network.
+ prepares a future VM attached to that network.
It ensures that all testcases inheriting from SingleVm1 could work
without specific configurations (or at least read the same config data).
flavor_ram = 1024
flavor_vcpus = 1
flavor_disk = 1
- username = 'cirros'
- ssh_connect_timeout = 60
def __init__(self, **kwargs):
if "case_name" not in kwargs:
- kwargs["case_name"] = 'singlevm1'
- super(SingleVm1, self).__init__(**kwargs)
+ kwargs["case_name"] = 'vmready1'
+ super(VmReady1, self).__init__(**kwargs)
self.orig_cloud = self.cloud
self.image = None
- self.sshvm = None
self.flavor = None
- self.sec = None
- self.fip = None
- self.keypair = None
- self.ssh = paramiko.SSHClient()
- (_, self.key_filename) = tempfile.mkstemp()
def _publish_image(self):
assert self.cloud
meta=meta)
self.__logger.debug("image: %s", self.image)
+ def _create_flavor(self):
+ assert self.orig_cloud
+ self.flavor = self.orig_cloud.create_flavor(
+ '{}-flavor_{}'.format(self.case_name, self.guid),
+ getattr(config.CONF, '{}_flavor_ram'.format(self.case_name),
+ self.flavor_ram),
+ getattr(config.CONF, '{}_flavor_vcpus'.format(self.case_name),
+ self.flavor_vcpus),
+ getattr(config.CONF, '{}_flavor_disk'.format(self.case_name),
+ self.flavor_disk))
+ self.__logger.debug("flavor: %s", self.flavor)
+ self.orig_cloud.set_flavor_specs(
+ self.flavor.id, getattr(config.CONF, 'flavor_extra_specs', {}))
+
+ def run(self, **kwargs):
+ """Boot the new VM
+
+ Here are the main actions:
+ - publish the image
+ - create the flavor
+
+ Returns:
+ - TestCase.EX_OK
+ - TestCase.EX_RUN_ERROR on error
+ """
+ status = testcase.TestCase.EX_RUN_ERROR
+ try:
+ assert self.cloud
+ super(VmReady1, self).run(**kwargs)
+ self._publish_image()
+ self._create_flavor()
+ self.result = 100
+ status = testcase.TestCase.EX_OK
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception('Cannot run %s', self.case_name)
+ finally:
+ self.stop_time = time.time()
+ return status
+
+ def clean(self):
+ try:
+ assert self.orig_cloud
+ assert self.cloud
+ self.cloud.delete_image(self.image)
+ self.orig_cloud.delete_flavor(self.flavor.id)
+ self.cloud.delete_image(self.image)
+ except Exception: # pylint: disable=broad-except
+ pass
+
+
+class VmReady2(VmReady1):
+ """Deploy a single VM reachable via ssh (scenario2)
+
+ It creates new user/project before creating and configuring all tenant
+ network ressources, flavors, images, etc. required by advanced testcases.
+
+ It ensures that all testcases inheriting from SingleVm2 could work
+ without specific configurations (or at least read the same config data).
+ """
+
+ __logger = logging.getLogger(__name__)
+
+ def __init__(self, **kwargs):
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'vmready2'
+ super(VmReady2, self).__init__(**kwargs)
+ try:
+ assert self.orig_cloud
+ self.project = tenantnetwork.NewProject(
+ self.orig_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")
+ self.cloud = None
+ self.project = None
+
+ def clean(self):
+ try:
+ super(VmReady2, self).clean()
+ assert self.project
+ self.project.clean()
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("cannot clean all ressources")
+
+
+class SingleVm1(VmReady1):
+ """Deploy a single VM reachable via ssh (scenario1)
+
+ It inherits from TenantNetwork1 which creates all network resources and
+ completes it by booting a VM attached to that network.
+
+ It ensures that all testcases inheriting from SingleVm1 could work
+ without specific configurations (or at least read the same config data).
+ """
+ # pylint: disable=too-many-instance-attributes
+
+ __logger = logging.getLogger(__name__)
+ username = 'cirros'
+ ssh_connect_timeout = 60
+
+ def __init__(self, **kwargs):
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'singlevm1'
+ super(SingleVm1, self).__init__(**kwargs)
+ self.sshvm = None
+ self.sec = None
+ self.fip = None
+ self.keypair = None
+ self.ssh = paramiko.SSHClient()
+ (_, self.key_filename) = tempfile.mkstemp()
+
def create_sg_rules(self):
"""Create the security group
self.sec.id, protocol='icmp', direction='ingress')
def _boot_vm(self):
- assert self.orig_cloud
assert self.cloud
- self.flavor = self.orig_cloud.create_flavor(
- '{}-flavor_{}'.format(self.case_name, self.guid),
- getattr(config.CONF, '{}_flavor_ram'.format(self.case_name),
- self.flavor_ram),
- getattr(config.CONF, '{}_flavor_vcpus'.format(self.case_name),
- self.flavor_vcpus),
- getattr(config.CONF, '{}_flavor_disk'.format(self.case_name),
- self.flavor_disk))
- self.__logger.debug("flavor: %s", self.flavor)
- self.cloud.set_flavor_specs(
- self.flavor.id, getattr(config.CONF, 'flavor_extra_specs', {}))
-
self.keypair = self.cloud.create_keypair(
'{}-kp_{}'.format(self.case_name, self.guid))
self.__logger.debug("keypair: %s", self.keypair)
"""Boot the new VM
Here are the main actions:
- - publish the image
- add a new ssh key
- boot the VM
- create the security group
assert self.cloud
super(SingleVm1, self).run(**kwargs)
self.result = 0
- self._publish_image()
self.create_sg_rules()
self._boot_vm()
assert self._connect()
kwargs["case_name"] = 'singlevm2'
super(SingleVm2, 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"]
+ assert self.orig_cloud
+ self.project = tenantnetwork.NewProject(
+ self.orig_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(SingleVm2, self).run(**kwargs)
+ self.cloud = None
+ self.project = None
def clean(self):
try:
- assert self.cloud
- assert self.orig_cloud
super(SingleVm2, 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")
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.orig_cloud = cloud
+ self.cloud = None
+ self.case_name = case_name
+ self.guid = guid
+ self.project = None
+ self.user = None
+
+ def create(self):
+ """Create projects/users"""
+ assert self.orig_cloud
+ assert self.case_name
+ password = str(uuid.uuid4())
+ 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=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=password,
+ default_project=self.project.id,
+ domain_id=domain.id)
+ self.__logger.debug("user: %s", self.user)
+ 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"]
+
+ 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):
# pylint: disable=too-many-instance-attributes
"""Create a tenant network (scenario1)
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
- assert self.orig_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")
Code Review / functest.git / commitdiff