Allow getting service via any endpoint

[functest.git] / functest / opnfv_tests / openstack / vmtp / vmtp.py

diff --git a/functest/opnfv_tests/openstack/vmtp/vmtp.py b/functest/opnfv_tests/openstack/vmtp/vmtp.py
index 52d1d6d..d102f64 100644 (file)
--- a/functest/opnfv_tests/openstack/vmtp/vmtp.py
+++ b/functest/opnfv_tests/openstack/vmtp/vmtp.py
@@ -35,7 +35,7 @@ from functest.core import singlevm
 from functest.utils import env
 
 
-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,6 +49,7 @@ class Vmtp(singlevm.VmReady1):
     flavor_ram = 2048
     flavor_vcpus = 1
     flavor_disk = 0
+    create_server_timeout = 300
 
     def __init__(self, **kwargs):
         if "case_name" not in kwargs:
@@ -58,6 +59,26 @@ class Vmtp(singlevm.VmReady1):
         (_, self.privkey_filename) = tempfile.mkstemp()
         (_, self.pubkey_filename) = tempfile.mkstemp()
 
+    def check_requirements(self):
+        if len(self.orig_cloud.list_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='{}-router_{}'.format(self.case_name, self.guid),
+            ext_gateway_net_id=self.ext_net.id)
+        self.__logger.debug("router: %s", self.router)
+
     def generate_keys(self):
         """Generate Keys
 
@@ -83,14 +104,14 @@ 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)
+            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),
@@ -100,21 +121,40 @@ class Vmtp(singlevm.VmReady1):
             vmtp_conf["security_group_name"] = "pns-security{}".format(
                 self.guid)
             vmtp_conf["dns_nameservers"] = [env.get('NAMESERVER')]
+            vmtp_conf["generic_retry_count"] = self.create_server_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
+        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"] = "{}/v3".format(new_env["OS_AUTH_URL"])
+        try:
+            del new_env['OS_TENANT_NAME']
+            del new_env['OS_TENANT_ID']
+        except Exception:  # pylint: disable=broad-except
+            pass
         cmd = ['vmtp', '-d', '--json', '{}/vmtp.json'.format(self.res_dir),
                '-c', self.config]
-        output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
+        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)
+        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:
             self.details = json.load(res_file)
@@ -124,8 +164,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(Vmtp, self).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 +183,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,9 +195,9 @@ class Vmtp(singlevm.VmReady1):
     def clean(self):
         try:
             assert self.cloud
+            super(Vmtp, self).clean()
             os.remove(self.privkey_filename)
             os.remove(self.pubkey_filename)
-            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))
         except Exception:  # pylint: disable=broad-except