diff options
Diffstat (limited to 'functest/opnfv_tests/openstack/vmtp/vmtp.py')
| -rw-r--r-- | functest/opnfv_tests/openstack/vmtp/vmtp.py | 117 |
1 files changed, 88 insertions, 29 deletions
diff --git a/functest/opnfv_tests/openstack/vmtp/vmtp.py b/functest/opnfv_tests/openstack/vmtp/vmtp.py index 326a4f383..9833cc72a 100644 --- a/functest/opnfv_tests/openstack/vmtp/vmtp.py +++ b/functest/opnfv_tests/openstack/vmtp/vmtp.py @@ -33,9 +33,10 @@ from xtesting.core import testcase from functest.core import singlevm from functest.utils import env +from functest.utils import functest_utils -class Vmtp(singlevm.VmReady1): +class Vmtp(singlevm.VmReady2): """Class to run Vmtp_ as an OPNFV Functest testcase .. _Vmtp: http://vmtp.readthedocs.io/en/latest/ @@ -49,28 +50,50 @@ class Vmtp(singlevm.VmReady1): flavor_ram = 2048 flavor_vcpus = 1 flavor_disk = 0 + create_server_timeout = 300 + ssh_retry_timeout = 240 def __init__(self, **kwargs): if "case_name" not in kwargs: kwargs["case_name"] = 'vmtp' - super(Vmtp, self).__init__(**kwargs) - self.config = "{}/vmtp.conf".format(self.res_dir) + super().__init__(**kwargs) + self.config = f"{self.res_dir}/vmtp.conf" (_, self.privkey_filename) = tempfile.mkstemp() (_, self.pubkey_filename) = tempfile.mkstemp() + def check_requirements(self): + if self.count_hypervisors() < 2: + self.__logger.warning("Vmtp requires at least 2 hypervisors") + self.is_skipped = True + self.project.clean() + + def create_network_resources(self): + """Create router + + It creates a router which gateway is the external network detected. + + Raises: expection on error + """ + assert self.cloud + assert self.ext_net + self.router = self.cloud.create_router( + name=f'{self.case_name}-router_{self.guid}', + ext_gateway_net_id=self.ext_net.id) + self.__logger.debug("router: %s", self.router) + def generate_keys(self): """Generate Keys Raises: Exception on error """ assert self.cloud - name = "vmtp_{}".format(self.guid) + name = f"vmtp_{self.guid}" self.__logger.info("Creating keypair with name: '%s'", name) keypair = self.cloud.create_keypair(name) self.__logger.debug("keypair: %s", keypair) - with open(self.privkey_filename, 'w') as key_file: + with open(self.privkey_filename, 'w', encoding='utf-8') as key_file: key_file.write(keypair.private_key) - with open(self.pubkey_filename, 'w') as key_file: + with open(self.pubkey_filename, 'w', encoding='utf-8') as key_file: key_file.write(keypair.public_key) self.cloud.delete_keypair(keypair.id) @@ -83,40 +106,66 @@ class Vmtp(singlevm.VmReady1): if not os.path.exists(self.res_dir): os.makedirs(self.res_dir) cmd = ['vmtp', '-sc'] - output = subprocess.check_output(cmd) + output = subprocess.check_output(cmd).decode("utf-8") self.__logger.info("%s\n%s", " ".join(cmd), output) - with open(self.config, "w+") as conf: - vmtp_conf = yaml.load(output) + with open(self.config, "w+", encoding='utf-8') as conf: + vmtp_conf = yaml.full_load(output) vmtp_conf["private_key_file"] = self.privkey_filename vmtp_conf["public_key_file"] = self.pubkey_filename vmtp_conf["image_name"] = str(self.image.name) - vmtp_conf["router_name"] = "pns_router_{}".format(self.guid) + vmtp_conf["router_name"] = str(self.router.name) vmtp_conf["flavor_type"] = str(self.flavor.name) vmtp_conf["internal_network_name"] = [ - "pns-internal-net_{}".format(self.guid), - "pns-internal-net2_{}".format(self.guid)] - vmtp_conf["vm_name_client"] = "TestClient_{}".format(self.guid) - vmtp_conf["vm_name_server"] = "TestServer_{}".format(self.guid) - vmtp_conf["security_group_name"] = "pns-security{}".format( - self.guid) + f"pns-internal-net_{self.guid}", + f"pns-internal-net2_{self.guid}"] + vmtp_conf["vm_name_client"] = f"TestClient_{self.guid}" + vmtp_conf["vm_name_server"] = f"TestServer_{self.guid}" + vmtp_conf["security_group_name"] = f"pns-security{self.guid}" vmtp_conf["dns_nameservers"] = [env.get('NAMESERVER')] + vmtp_conf["generic_retry_count"] = self.create_server_timeout // 2 + vmtp_conf["ssh_retry_count"] = self.ssh_retry_timeout // 2 conf.write(yaml.dump(vmtp_conf)) def run_vmtp(self): + # pylint: disable=unexpected-keyword-arg """Run Vmtp and generate charts Raises: Exception on error """ assert self.cloud - cmd = ['vmtp', '-d', '--json', '{}/vmtp.json'.format(self.res_dir), + new_env = dict( + os.environ, + OS_USERNAME=self.project.user.name, + OS_PROJECT_NAME=self.project.project.name, + OS_PROJECT_ID=self.project.project.id, + OS_PROJECT_DOMAIN_NAME=self.project.domain.name, + OS_USER_DOMAIN_NAME=self.project.domain.name, + OS_PASSWORD=self.project.password) + if not new_env["OS_AUTH_URL"].endswith(('v3', 'v3/')): + new_env["OS_AUTH_URL"] = f'{new_env["OS_AUTH_URL"]}/v3' + try: + del new_env['OS_TENANT_NAME'] + del new_env['OS_TENANT_ID'] + except Exception: # pylint: disable=broad-except + pass + cmd = ['vmtp', '-d', '--json', f'{self.res_dir}/vmtp.json', '-c', self.config] - output = subprocess.check_output(cmd, stderr=subprocess.STDOUT) + if env.get("VMTP_HYPERVISORS"): + hypervisors = functest_utils.convert_ini_to_list( + env.get("VMTP_HYPERVISORS")) + for hypervisor in hypervisors: + cmd.extend(["--hypervisor", hypervisor]) + self.__logger.debug("cmd: %s", cmd) + output = subprocess.check_output( + cmd, stderr=subprocess.STDOUT, env=new_env).decode("utf-8") self.__logger.info("%s\n%s", " ".join(cmd), output) - cmd = ['vmtp_genchart', '-c', '{}/vmtp.html'.format(self.res_dir), - '{}/vmtp.json'.format(self.res_dir)] - output = subprocess.check_output(cmd, stderr=subprocess.STDOUT) + cmd = ['vmtp_genchart', '-c', f'{self.res_dir}/vmtp.html', + f'{self.res_dir}/vmtp.json'] + output = subprocess.check_output( + cmd, stderr=subprocess.STDOUT).decode("utf-8") self.__logger.info("%s\n%s", " ".join(cmd), output) - with open('{}/vmtp.json'.format(self.res_dir), 'r') as res_file: + with open(f'{self.res_dir}/vmtp.json', 'r', + encoding='utf-8') as res_file: self.details = json.load(res_file) def run(self, **kwargs): @@ -124,8 +173,18 @@ class Vmtp(singlevm.VmReady1): status = testcase.TestCase.EX_RUN_ERROR try: assert self.cloud - self.image = self.publish_image() - self.flavor = self.create_flavor() + assert super().run(**kwargs) == self.EX_OK + status = testcase.TestCase.EX_RUN_ERROR + if self.orig_cloud.get_role("admin"): + role_name = "admin" + elif self.orig_cloud.get_role("Admin"): + role_name = "Admin" + else: + raise Exception("Cannot detect neither admin nor Admin") + self.orig_cloud.grant_role( + role_name, user=self.project.user.id, + project=self.project.project.id, + domain=self.project.domain.id) self.generate_keys() self.write_config() self.run_vmtp() @@ -133,7 +192,8 @@ class Vmtp(singlevm.VmReady1): status = testcase.TestCase.EX_OK except subprocess.CalledProcessError as cpe: self.__logger.error( - "Exception when calling %s\n%s", cpe.cmd, cpe.output) + "Exception when calling %s\n%s", cpe.cmd, + cpe.output.decode("utf-8")) self.result = 0 except Exception: # pylint: disable=broad-except self.__logger.exception("Cannot run vmtp") @@ -144,11 +204,10 @@ class Vmtp(singlevm.VmReady1): def clean(self): try: assert self.cloud + super().clean() os.remove(self.privkey_filename) os.remove(self.pubkey_filename) - if self.image: - self.cloud.delete_image(self.image) - self.cloud.delete_network("pns-internal-net_{}".format(self.guid)) - self.cloud.delete_network("pns-internal-net2_{}".format(self.guid)) + self.cloud.delete_network(f"pns-internal-net_{self.guid}") + self.cloud.delete_network(f"pns-internal-net2_{self.guid}") except Exception: # pylint: disable=broad-except pass |