functest/opnfv_tests/vnf/epc/juju_epc.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Rebaca and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9 """Juju testcase implementation."""
  10
  11 import errno
  12 import logging
  13 import os
  14 import time
  15 import json
  16 import re
  17 import subprocess
  18 import sys
  19 import uuid
  20 from copy import deepcopy
  21 import pkg_resources
  22 import yaml
  23
  24 import six
  25 from snaps.config.flavor import FlavorConfig
  26 from snaps.config.image import ImageConfig
  27 from snaps.config.network import NetworkConfig, SubnetConfig
  28 from snaps.config.router import RouterConfig
  29 from snaps.config.security_group import (
  30     Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
  31 from snaps.config.user import UserConfig
  32 from snaps.openstack.create_flavor import OpenStackFlavor
  33 from snaps.openstack.create_image import OpenStackImage
  34 from snaps.openstack.create_network import OpenStackNetwork
  35 from snaps.openstack.create_router import OpenStackRouter
  36 from snaps.openstack.create_security_group import OpenStackSecurityGroup
  37 from snaps.openstack.create_user import OpenStackUser
  38 from snaps.openstack.utils import keystone_utils
  39 from snaps.openstack.utils import nova_utils
  40 from snaps.openstack.utils import neutron_utils
  41
  42 from functest.core import vnf
  43 from functest.opnfv_tests.openstack.snaps import snaps_utils
  44 from functest.utils import config
  45 from functest.utils import env
  46
  47 __author__ = "Amarendra Meher <amarendra@rebaca.com>"
  48 __author__ = "Soumaya K Nayek <soumaya.nayek@rebaca.com>"
  49
  50 CLOUD_TEMPLATE = """clouds:
  51   abot-epc:
  52     type: openstack
  53     auth-types: [userpass]
  54     endpoint: {url}
  55     regions:
  56       {region}:
  57         endpoint: {url}"""
  58
  59 CREDS_TEMPLATE2 = """credentials:
  60   abot-epc:
  61     default-credential: abot-epc
  62     abot-epc:
  63       auth-type: userpass
  64       password: {pass}
  65       project-domain-name: {project_domain_n}
  66       tenant-name: {tenant_n}"""
  67
  68 CREDS_TEMPLATE3 = """credentials:
  69   abot-epc:
  70     default-credential: abot-epc
  71     abot-epc:
  72       auth-type: userpass
  73       password: {pass}
  74       project-domain-name: {project_domain_n}
  75       tenant-name: {tenant_n}
  76       user-domain-name: {user_domain_n}
  77       username: {user_n}"""
  78
  79
  80 class JujuEpc(vnf.VnfOnBoarding):
  81     # pylint:disable=too-many-instance-attributes
  82     """Abot EPC deployed with JUJU Orchestrator Case"""
  83
  84     __logger = logging.getLogger(__name__)
  85
  86     juju_timeout = '3600'
  87
  88     def __init__(self, **kwargs):
  89         if "case_name" not in kwargs:
  90             kwargs["case_name"] = "juju_epc"
  91         super(JujuEpc, self).__init__(**kwargs)
  92
  93         # Retrieve the configuration
  94         self.case_dir = pkg_resources.resource_filename(
  95             'functest', 'opnfv_tests/vnf/epc')
  96         try:
  97             self.config = getattr(
  98                 config.CONF, 'vnf_{}_config'.format(self.case_name))
  99         except Exception:
 100             raise Exception("VNF config file not found")
 101         self.config_file = os.path.join(self.case_dir, self.config)
 102         self.orchestrator = dict(requirements=get_config(
 103             "orchestrator.requirements", self.config_file))
 104
 105         self.created_object = []
 106         self.details['orchestrator'] = dict(
 107             name=get_config("orchestrator.name", self.config_file),
 108             version=get_config("orchestrator.version", self.config_file),
 109             status='ERROR',
 110             result=''
 111         )
 112
 113         self.vnf = dict(
 114             descriptor=get_config("vnf.descriptor", self.config_file),
 115             requirements=get_config("vnf.requirements", self.config_file)
 116         )
 117         self.details['vnf'] = dict(
 118             descriptor_version=self.vnf['descriptor']['version'],
 119             name=get_config("vnf.name", self.config_file),
 120             version=get_config("vnf.version", self.config_file),
 121         )
 122         self.__logger.debug("VNF configuration: %s", self.vnf)
 123
 124         self.details['test_vnf'] = dict(
 125             name=get_config("vnf_test_suite.name", self.config_file),
 126             version=get_config("vnf_test_suite.version", self.config_file),
 127             tag_name=get_config("vnf_test_suite.tag_name", self.config_file)
 128         )
 129         self.public_auth_url = None
 130
 131         self.res_dir = os.path.join(
 132             getattr(config.CONF, 'dir_results'), self.case_name)
 133
 134     def _bypass_juju_netdiscovery_bug(self, name):
 135         user_creator = OpenStackUser(
 136             self.snaps_creds,
 137             UserConfig(
 138                 name=name,
 139                 password=str(uuid.uuid4()),
 140                 project_name=self.tenant_name,
 141                 domain_name=self.snaps_creds.user_domain_name,
 142                 roles={'_member_': self.tenant_name}))
 143         user_creator.create()
 144         self.created_object.append(user_creator)
 145         return user_creator
 146
 147     def _register_cloud(self):
 148         self.__logger.info("Creating Cloud for Abot-epc .....")
 149         clouds_yaml = os.path.join(self.res_dir, "clouds.yaml")
 150         cloud_data = {
 151             'url': self.public_auth_url,
 152             'region': self.snaps_creds.region_name if (
 153                 self.snaps_creds.region_name) else 'RegionOne'}
 154         with open(clouds_yaml, 'w') as yfile:
 155             yfile.write(CLOUD_TEMPLATE.format(**cloud_data))
 156         cmd = ['juju', 'add-cloud', 'abot-epc', '-f', clouds_yaml, '--replace']
 157         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 158         self.__logger.info("%s\n%s", " ".join(cmd), output)
 159
 160     def _register_credentials_v2(self):
 161         self.__logger.info("Creating Credentials for Abot-epc .....")
 162         user_creator = self._bypass_juju_netdiscovery_bug(
 163             'juju_network_discovery_bug')
 164         snaps_creds = user_creator.get_os_creds(self.snaps_creds.project_name)
 165         self.__logger.debug("snaps creds: %s", snaps_creds)
 166         credentials_yaml = os.path.join(self.res_dir, "credentials.yaml")
 167         creds_data = {
 168             'pass': snaps_creds.password,
 169             'tenant_n': snaps_creds.project_name,
 170             'user_n': snaps_creds.username}
 171         with open(credentials_yaml, 'w') as yfile:
 172             yfile.write(CREDS_TEMPLATE2.format(**creds_data))
 173         cmd = ['juju', 'add-credential', 'abot-epc', '-f', credentials_yaml,
 174                '--replace']
 175         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 176         self.__logger.info("%s\n%s", " ".join(cmd), output)
 177
 178     def _register_credentials_v3(self):
 179         self.__logger.info("Creating Credentials for Abot-epc .....")
 180         user_creator = self._bypass_juju_netdiscovery_bug(
 181             'juju_network_discovery_bug')
 182         snaps_creds = user_creator.get_os_creds(self.snaps_creds.project_name)
 183         self.__logger.debug("snaps creds: %s", snaps_creds)
 184         credentials_yaml = os.path.join(self.res_dir, "credentials.yaml")
 185         creds_data = {
 186             'pass': snaps_creds.password,
 187             'tenant_n': snaps_creds.project_name,
 188             'user_n': snaps_creds.username,
 189             'project_domain_n': snaps_creds.project_domain_name,
 190             'user_domain_n': snaps_creds.user_domain_name}
 191         with open(credentials_yaml, 'w') as yfile:
 192             yfile.write(CREDS_TEMPLATE3.format(**creds_data))
 193         cmd = ['juju', 'add-credential', 'abot-epc', '-f', credentials_yaml,
 194                '--replace']
 195         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 196         self.__logger.info("%s\n%s", " ".join(cmd), output)
 197
 198     def _add_custom_rule(self, sec_grp_name):
 199         """ To add custom rule for SCTP Traffic """
 200
 201         security_group = OpenStackSecurityGroup(
 202             self.snaps_creds,
 203             SecurityGroupConfig(
 204                 name=sec_grp_name))
 205
 206         security_group.create()
 207
 208         # Add custom security rule to the obtained Security Group
 209         self.__logger.info("Adding SCTP ingress rule to SG:%s",
 210                            security_group.sec_grp_settings.name)
 211
 212         try:
 213             security_group.add_rule(SecurityGroupRuleConfig(
 214                 sec_grp_name=sec_grp_name, direction=Direction.ingress,
 215                 protocol=Protocol.sctp))
 216         except Exception:  # pylint: disable=broad-except
 217             self.__logger.exception(
 218                 "Some issue encountered with adding SCTP security rule ...")
 219
 220     def prepare(self):
 221         """Prepare testcase (Additional pre-configuration steps)."""
 222         self.__logger.info("Additional pre-configuration steps")
 223         super(JujuEpc, self).prepare()
 224         try:
 225             os.makedirs(self.res_dir)
 226         except OSError as ex:
 227             if ex.errno != errno.EEXIST:
 228                 self.__logger.exception("Cannot create %s", self.res_dir)
 229                 raise vnf.VnfPreparationException
 230
 231         self.__logger.info("ENV:\n%s", env.string())
 232
 233         self.public_auth_url = keystone_utils.get_endpoint(
 234             self.snaps_creds, 'identity')
 235
 236         # it enforces a versioned public identity endpoint as juju simply
 237         # adds /auth/tokens wich fails vs an unversioned endpoint.
 238         if not self.public_auth_url.endswith(('v3', 'v3/', 'v2.0', 'v2.0/')):
 239             self.public_auth_url = six.moves.urllib.parse.urljoin(
 240                 self.public_auth_url, 'v3')
 241         self._register_cloud()
 242         if self.snaps_creds.identity_api_version == 3:
 243             self._register_credentials_v3()
 244         else:
 245             self._register_credentials_v2()
 246
 247     def deploy_orchestrator(self):  # pylint: disable=too-many-locals
 248         """
 249         Create network, subnet, router
 250
 251         Bootstrap juju
 252         """
 253         self.__logger.info("Deploying Juju Orchestrator")
 254         private_net_name = getattr(
 255             config.CONF, 'vnf_{}_private_net_name'.format(self.case_name))
 256         private_subnet_name = '{}-{}'.format(
 257             getattr(config.CONF,
 258                     'vnf_{}_private_subnet_name'.format(self.case_name)),
 259             self.uuid)
 260         private_subnet_cidr = getattr(
 261             config.CONF, 'vnf_{}_private_subnet_cidr'.format(self.case_name))
 262         abot_router = '{}-{}'.format(
 263             getattr(config.CONF,
 264                     'vnf_{}_external_router'.format(self.case_name)),
 265             self.uuid)
 266         self.__logger.info("Creating full network with nameserver: %s",
 267                            env.get('NAMESERVER'))
 268         subnet_settings = SubnetConfig(
 269             name=private_subnet_name,
 270             cidr=private_subnet_cidr,
 271             dns_nameservers=[env.get('NAMESERVER')])
 272         network_settings = NetworkConfig(
 273             name=private_net_name, subnet_settings=[subnet_settings])
 274         network_creator = OpenStackNetwork(self.snaps_creds, network_settings)
 275         net_id = network_creator.create().id
 276         self.created_object.append(network_creator)
 277
 278         ext_net_name = snaps_utils.get_ext_net_name(self.snaps_creds)
 279         self.__logger.info("Creating network Router ....")
 280         router_creator = OpenStackRouter(
 281             self.snaps_creds, RouterConfig(
 282                 name=abot_router,
 283                 external_gateway=ext_net_name,
 284                 internal_subnets=[subnet_settings.name]))
 285         router_creator.create()
 286         self.created_object.append(router_creator)
 287         self.__logger.info("Creating Flavor ....")
 288         flavor_settings = FlavorConfig(
 289             name=self.orchestrator['requirements']['flavor']['name'],
 290             ram=self.orchestrator['requirements']['flavor']['ram_min'],
 291             disk=10, vcpus=1)
 292         flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
 293         flavor_creator.create()
 294         self.created_object.append(flavor_creator)
 295
 296         self.__logger.info("Upload some OS images if it doesn't exist")
 297         images = get_config("tenant_images", self.config_file)
 298         self.__logger.info("Images needed for vEPC: %s", images)
 299         for image_name, image_file in six.iteritems(images):
 300             self.__logger.info("image: %s, file: %s", image_name, image_file)
 301             if image_file and image_name:
 302                 image_creator = OpenStackImage(self.snaps_creds, ImageConfig(
 303                     name=image_name, image_user='cloud', img_format='qcow2',
 304                     image_file=image_file))
 305                 image_id = image_creator.create().id
 306                 cmd = ['juju', 'metadata', 'generate-image', '-d', '/root',
 307                        '-i', image_id, '-s', image_name, '-r',
 308                        self.snaps_creds.region_name if (
 309                            self.snaps_creds.region_name) else 'RegionOne',
 310                        '-u', self.public_auth_url]
 311                 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 312                 self.__logger.info("%s\n%s", " ".join(cmd), output)
 313                 self.created_object.append(image_creator)
 314         self.__logger.info("Network ID  : %s", net_id)
 315
 316         self.__logger.info("Starting Juju Bootstrap process...")
 317         try:
 318             cmd = ['timeout', '-t', JujuEpc.juju_timeout,
 319                    'juju', 'bootstrap', 'abot-epc', 'abot-controller',
 320                    '--metadata-source', '/root',
 321                    '--constraints', 'mem=2G',
 322                    '--bootstrap-series', 'xenial',
 323                    '--config', 'network={}'.format(net_id),
 324                    '--config', 'ssl-hostname-verification=false',
 325                    '--config', 'use-floating-ip=true',
 326                    '--config', 'use-default-secgroup=true',
 327                    '--debug']
 328             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 329             self.__logger.info("%s\n%s", " ".join(cmd), output)
 330         except subprocess.CalledProcessError as cpe:
 331             self.__logger.error(
 332                 "Exception with Juju Bootstrap: %s\n%s",
 333                 cpe.cmd, cpe.output)
 334             return False
 335         except Exception:  # pylint: disable=broad-except
 336             self.__logger.exception("Some issue with Juju Bootstrap ...")
 337             return False
 338
 339         return True
 340
 341     def check_app(self, name='abot-epc-basic', status='active'):
 342         """Check application status."""
 343         cmd = ['juju', 'status', '--format', 'short', name]
 344         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 345         self.__logger.info("%s\n%s", " ".join(cmd), output)
 346         ret = re.search(r'(?=workload:({})\))'.format(status), output)
 347         if ret:
 348             self.__logger.info("%s workload is %s", name, status)
 349             return True
 350         self.__logger.error("%s workload differs from %s", name, status)
 351         return False
 352
 353     def deploy_vnf(self):
 354         """Deploy ABOT-OAI-EPC."""
 355         self.__logger.info("Upload VNFD")
 356         descriptor = self.vnf['descriptor']
 357         self.__logger.info("Get or create flavor for all Abot-EPC")
 358         flavor_settings = FlavorConfig(
 359             name=self.vnf['requirements']['flavor']['name'],
 360             ram=self.vnf['requirements']['flavor']['ram_min'],
 361             disk=10,
 362             vcpus=1)
 363         flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
 364         flavor_creator.create()
 365         self.created_object.append(flavor_creator)
 366
 367         self.__logger.info("Deploying Abot-epc bundle file ...")
 368         cmd = ['juju', 'deploy', '{}'.format(descriptor.get('file_name'))]
 369         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 370         self.__logger.info("%s\n%s", " ".join(cmd), output)
 371         self.__logger.info("Waiting for instances .....")
 372         try:
 373             cmd = ['timeout', '-t', JujuEpc.juju_timeout, 'juju-wait']
 374             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 375             self.__logger.info("%s\n%s", " ".join(cmd), output)
 376             self.__logger.info("Deployed Abot-epc on Openstack")
 377         except subprocess.CalledProcessError as cpe:
 378             self.__logger.error(
 379                 "Exception with Juju VNF Deployment: %s\n%s",
 380                 cpe.cmd, cpe.output)
 381             return False
 382         except Exception:  # pylint: disable=broad-except
 383             self.__logger.exception("Some issue with the VNF Deployment ..")
 384             return False
 385
 386         self.__logger.info("Checking status of ABot and EPC units ...")
 387         cmd = ['juju', 'status']
 388         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 389         self.__logger.debug("%s\n%s", " ".join(cmd), output)
 390         for app in ['abot-epc-basic', 'oai-epc', 'oai-hss']:
 391             if not self.check_app(app):
 392                 return False
 393
 394         nova_client = nova_utils.nova_client(self.snaps_creds)
 395         instances = get_instances(nova_client)
 396         self.__logger.info("List of Instance: %s", instances)
 397         for items in instances:
 398             metadata = get_instance_metadata(nova_client, items)
 399             if 'juju-units-deployed' in metadata:
 400                 sec_group = 'juju-{}-{}'.format(
 401                     metadata['juju-controller-uuid'],
 402                     metadata['juju-model-uuid'])
 403                 self.__logger.info("Instance: %s", sec_group)
 404                 break
 405         self.__logger.info("Adding Security group rule....")
 406         # This will add sctp rule to a common Security Group Created
 407         # by juju and shared to all deployed units.
 408         self._add_custom_rule(sec_group)
 409
 410         self.__logger.info("Transferring the feature files to Abot_node ...")
 411         cmd = ['timeout', '-t', JujuEpc.juju_timeout,
 412                'juju', 'scp', '--', '-r', '-v',
 413                '{}/featureFiles'.format(self.case_dir), 'abot-epc-basic/0:~/']
 414         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 415         self.__logger.info("%s\n%s", " ".join(cmd), output)
 416
 417         self.__logger.info("Copying the feature files within Abot_node ")
 418         cmd = ['timeout', '-t', JujuEpc.juju_timeout,
 419                'juju', 'ssh', 'abot-epc-basic/0',
 420                'sudo', 'cp', '-vfR', '~/featureFiles/*',
 421                '/etc/rebaca-test-suite/featureFiles']
 422         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 423         self.__logger.info("%s\n%s", " ".join(cmd), output)
 424         return True
 425
 426     def test_vnf(self):
 427         """Run test on ABoT."""
 428         start_time = time.time()
 429         self.__logger.info("Running VNF Test cases....")
 430         cmd = ['juju', 'run-action', 'abot-epc-basic/0', 'run',
 431                'tagnames={}'.format(self.details['test_vnf']['tag_name'])]
 432         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 433         self.__logger.info("%s\n%s", " ".join(cmd), output)
 434
 435         cmd = ['timeout', '-t', JujuEpc.juju_timeout, 'juju-wait']
 436         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 437         self.__logger.info("%s\n%s", " ".join(cmd), output)
 438
 439         duration = time.time() - start_time
 440         self.__logger.info("Getting results from Abot node....")
 441         cmd = ['timeout', '-t', JujuEpc.juju_timeout,
 442                'juju', 'scp', '--', '-v',
 443                'abot-epc-basic/0:'
 444                '/var/lib/abot-epc-basic/artifacts/TestResults.json',
 445                '{}/.'.format(self.res_dir)]
 446         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 447         self.__logger.info("%s\n%s", " ".join(cmd), output)
 448         self.__logger.info("Parsing the Test results...")
 449         res = (process_abot_test_result('{}/TestResults.json'.format(
 450             self.res_dir)))
 451         short_result = sig_test_format(res)
 452         self.__logger.info(short_result)
 453         self.details['test_vnf'].update(
 454             status='PASS', result=short_result, full_result=res,
 455             duration=duration)
 456         self.__logger.info(
 457             "Test VNF result: Passed: %d, Failed:%d, Skipped: %d",
 458             short_result['passed'],
 459             short_result['failures'], short_result['skipped'])
 460         return True
 461
 462     def _get_floating_ips(self):
 463         """Get the list of floating IPs associated with the current project"""
 464
 465         project_id = self.os_project.get_project().id
 466
 467         neutron_client = neutron_utils.neutron_client(self.snaps_creds)
 468         floating_ips = neutron_utils.get_floating_ips(neutron_client)
 469
 470         project_floating_ip_list = list()
 471         for floating_ip in floating_ips:
 472             if project_id and project_id == floating_ip.project_id:
 473                 project_floating_ip_list.append(floating_ip)
 474
 475         return project_floating_ip_list
 476
 477     def _release_floating_ips(self, fip_list):
 478         """
 479         Responsible for deleting a list of floating IPs
 480         :param fip_list: A list of SNAPS FloatingIp objects
 481         :return:
 482         """
 483         if not fip_list:
 484             return
 485
 486         neutron_client = neutron_utils.neutron_client(self.snaps_creds)
 487
 488         for floating_ip in fip_list:
 489             neutron_utils.delete_floating_ip(neutron_client, floating_ip)
 490
 491     def clean(self):
 492         """Clean created objects/functions."""
 493
 494         # Store Floating IPs of instances created by Juju
 495         fip_list = self._get_floating_ips()
 496         self.__logger.info("Floating IPs assigned to project:%s",
 497                            self.os_project.get_project().name)
 498         for floating_ip in fip_list:
 499             self.__logger.debug("%s:%s", floating_ip.ip,
 500                                 floating_ip.description)
 501
 502         try:
 503             cmd = ['juju', 'debug-log', '--replay', '--no-tail']
 504             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 505             self.__logger.debug("%s\n%s", " ".join(cmd), output)
 506             if not self.orchestrator['requirements']['preserve_setup']:
 507                 self.__logger.info("Destroying Orchestrator...")
 508                 cmd = ['timeout', '-t', JujuEpc.juju_timeout,
 509                        'juju', 'destroy-controller', '-y', 'abot-controller',
 510                        '--destroy-all-models']
 511                 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 512                 self.__logger.info("%s\n%s", " ".join(cmd), output)
 513         except subprocess.CalledProcessError as cpe:
 514             self.__logger.error(
 515                 "Exception with Juju Cleanup: %s\n%s",
 516                 cpe.cmd, cpe.output)
 517         except Exception:  # pylint: disable=broad-except
 518             self.__logger.exception("General issue during the undeployment ..")
 519
 520         if not self.orchestrator['requirements']['preserve_setup']:
 521             try:
 522                 self.__logger.info('Release floating IPs assigned by Juju...')
 523                 self._release_floating_ips(fip_list)
 524             except Exception:  # pylint: disable=broad-except
 525                 self.__logger.exception(
 526                     "Exception while releasing floating IPs ...")
 527
 528             self.__logger.info('Remove the Abot_epc OS objects ..')
 529             super(JujuEpc, self).clean()
 530
 531         return True
 532
 533
 534 # ----------------------------------------------------------
 535 #
 536 #               YAML UTILS
 537 #
 538 # -----------------------------------------------------------
 539 def get_config(parameter, file_path):
 540     """
 541     Returns the value of a given parameter in file.yaml
 542     parameter must be given in string format with dots
 543     Example: general.openstack.image_name
 544     """
 545     with open(file_path) as config_file:
 546         file_yaml = yaml.safe_load(config_file)
 547     config_file.close()
 548     value = file_yaml
 549     for element in parameter.split("."):
 550         value = value.get(element)
 551         if value is None:
 552             raise ValueError("The parameter %s is not defined in"
 553                              " reporting.yaml" % parameter)
 554     return value
 555
 556
 557 def sig_test_format(sig_test):
 558     """
 559     Process the signaling result to have a short result
 560     """
 561     nb_passed = 0
 562     nb_failures = 0
 563     nb_skipped = 0
 564     for data_test in sig_test:
 565         if data_test['result'] == "passed":
 566             nb_passed += 1
 567         elif data_test['result'] == "failed":
 568             nb_failures += 1
 569         elif data_test['result'] == "skipped":
 570             nb_skipped += 1
 571     total_sig_test_result = {}
 572     total_sig_test_result['passed'] = nb_passed
 573     total_sig_test_result['failures'] = nb_failures
 574     total_sig_test_result['skipped'] = nb_skipped
 575     return total_sig_test_result
 576
 577
 578 def process_abot_test_result(file_path):
 579     """ Process ABoT Result """
 580     with open(file_path) as test_result:
 581         data = json.load(test_result)
 582         res = []
 583         for tests in data:
 584             tests = update_data(tests)
 585             try:
 586                 flatten_steps = tests['elements'][0].pop('flatten_steps')
 587                 for steps in flatten_steps:
 588                     steps['result'] = steps['step_status']
 589                     res.append(steps)
 590             except Exception:  # pylint: disable=broad-except
 591                 logging.error("Could not post data to ElasticSearch host")
 592                 raise
 593         return res
 594
 595
 596 def update_data(obj):
 597     """ Update Result data"""
 598     try:
 599         obj['feature_file'] = os.path.splitext(os.path.basename(obj['uri']))[0]
 600
 601         for element in obj['elements']:
 602             element['final_result'] = "passed"
 603             element['flatten_steps'] = []
 604
 605             for step in element['steps']:
 606                 temp_dict = {}
 607                 step['result'][step['result']['status']] = 1
 608                 if step['result']['status'].lower() in ['fail', 'failed']:
 609                     element['final_result'] = "failed"
 610
 611                 temp_dict['feature_file'] = obj['feature_file']
 612                 temp_dict['step_name'] = step['name']
 613                 temp_dict['step_status'] = step['result']['status']
 614                 temp_dict['step_duration'] = step['result'].get('duration', 0)
 615                 temp_dict['step_' + step['result']['status']] = 1
 616                 element['flatten_steps'].append(deepcopy(temp_dict))
 617
 618             # Need to put the tag in OBJ and not ELEMENT
 619             if 'tags' in obj:
 620                 element['tags'] = deepcopy(obj['tags'])
 621                 for tag in obj['tags']:
 622                     element[tag['name']] = 1
 623             else:
 624                 for tag in element['tags']:
 625                     element[tag['name']] = 1
 626
 627     except Exception:  # pylint: disable=broad-except
 628         logging.error("Error in updating data, %s", sys.exc_info()[0])
 629         raise
 630
 631     return obj
 632
 633
 634 def get_instances(nova_client):
 635     """ To get all vm info of a project """
 636     try:
 637         instances = nova_client.servers.list()
 638         return instances
 639     except Exception as exc:  # pylint: disable=broad-except
 640         logging.error("Error [get_instances(nova_client)]: %s", exc)
 641         return None
 642
 643
 644 def get_instance_metadata(nova_client, instance):
 645     """ Get instance Metadata - Instance ID """
 646     try:
 647         instance = nova_client.servers.get(instance.id)
 648         return instance.metadata
 649     except Exception as exc:  # pylint: disable=broad-except
 650         logging.error("Error [get_instance_status(nova_client)]: %s", exc)
 651         return None