functest/opnfv_tests/vnf/ims/clearwater.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright (c) 2017 All rights reserved
   4 # This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 #
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 """Ease testing any Clearwater deployment"""
  11
  12 import logging
  13 import os
  14 import re
  15 import time
  16
  17 import pkg_resources
  18 import requests
  19
  20 from functest.utils import config
  21 import functest.utils.functest_utils as ft_utils
  22
  23 __author__ = ("Valentin Boucher <valentin.boucher@orange.com>, "
  24               "Helen Yao <helanyao@gmail.com>")
  25
  26
  27 class ClearwaterTesting():
  28     """vIMS clearwater base usable by several orchestrators"""
  29
  30     def __init__(self, case_name, bono_ip, ellis_ip):
  31         self.logger = logging.getLogger(__name__)
  32         self.case_dir = pkg_resources.resource_filename(
  33             'functest', 'opnfv_tests/vnf/ims')
  34         self.data_dir = getattr(config.CONF, 'dir_ims_data')
  35         self.result_dir = os.path.join(
  36             getattr(config.CONF, 'dir_results'), case_name)
  37         self.test_dir = getattr(config.CONF, 'dir_repo_vims_test')
  38
  39         if not os.path.exists(self.data_dir):
  40             os.makedirs(self.data_dir)
  41         if not os.path.exists(self.result_dir):
  42             os.makedirs(self.result_dir)
  43
  44         self.ellis_ip = ellis_ip
  45         self.bono_ip = bono_ip
  46
  47     def availability_check(self, signup_code='secret', two_numbers=False):
  48         """Create one or two numbers"""
  49         assert self.ellis_ip
  50         output_dict = {}
  51         self.logger.debug('Ellis IP: %s', self.ellis_ip)
  52         output_dict['ellis_ip'] = self.ellis_ip
  53         account_url = f'http://{self.ellis_ip}/accounts'
  54         params = {"password": "functest",
  55                   "full_name": "opnfv functest user",
  56                   "email": "functest@opnfv.org",
  57                   "signup_code": signup_code}
  58         output_dict['login'] = params
  59
  60         number_res = self._create_ellis_account(account_url, params)
  61         output_dict['number'] = number_res
  62
  63         session_url = f'http://{self.ellis_ip}/session'
  64         session_data = {
  65             'username': params['email'],
  66             'password': params['password'],
  67             'email': params['email']
  68         }
  69         cookies = self._get_ellis_session_cookies(session_url, session_data)
  70
  71         number_url = (
  72             f"http://{self.ellis_ip}/accounts/{params['email']}/numbers")
  73         self.logger.debug('Create 1st calling number on Ellis')
  74         number_res = self._create_ellis_number(number_url, cookies)
  75
  76         if two_numbers:
  77             self.logger.debug('Create 2nd calling number on Ellis')
  78             number_res = self._create_ellis_number(number_url, cookies)
  79             output_dict['number2'] = number_res
  80
  81         return output_dict
  82
  83     def _create_ellis_account(self, account_url, params):
  84         i = 80
  85         for iloop in range(i):
  86             try:
  87                 req = requests.post(account_url, data=params)
  88                 if req.status_code == 201:
  89                     account_res = req.json()
  90                     self.logger.info(
  91                         'Account %s is created on Ellis\n%s',
  92                         params.get('full_name'), account_res)
  93                     return account_res
  94                 raise Exception("Cannot create ellis account")
  95             except Exception:  # pylint: disable=broad-except
  96                 self.logger.info(
  97                     "try %s: cannot create ellis account", iloop + 1)
  98                 time.sleep(30)
  99         raise Exception(
 100             f"Unable to create an account {params.get('full_name')}")
 101
 102     def _get_ellis_session_cookies(self, session_url, params):
 103         i = 15
 104         for iloop in range(i):
 105             try:
 106                 req = requests.post(session_url, data=params)
 107                 if req.status_code == 201:
 108                     cookies = req.cookies
 109                     self.logger.debug('cookies: %s', cookies)
 110                     return cookies
 111                 raise Exception('Failed to get cookies for Ellis')
 112             except Exception:  # pylint: disable=broad-except
 113                 self.logger.info(
 114                     "try %s: cannot get cookies for Ellis", iloop + 1)
 115                 time.sleep(10)
 116         raise Exception('Failed to get cookies for Ellis')
 117
 118     def _create_ellis_number(self, number_url, cookies):
 119         i = 30
 120         for iloop in range(i):
 121             try:
 122                 req = requests.post(number_url, cookies=cookies)
 123                 if req.status_code == 200:
 124                     number_res = req.json()
 125                     self.logger.info(
 126                         'Calling number is created: %s', number_res)
 127                     return number_res
 128                 if req and req.json():
 129                     reason = req.json()['reason']
 130                 else:
 131                     reason = req
 132                 self.logger.info("cannot create a number: %s", reason)
 133                 raise Exception('Failed to create a number')
 134             except Exception:  # pylint: disable=broad-except
 135                 self.logger.info(
 136                     "try %s: cannot create a number", iloop + 1)
 137                 time.sleep(25)
 138         raise Exception('Failed to create a number')
 139
 140     def run_clearwater_live_test(self, public_domain, signup_code='secret'):
 141         """Run the Clearwater live tests
 142
 143         It first runs dnsmasq to reach clearwater services by FQDN and then the
 144         Clearwater live tests. All results are saved in ims_test_output.txt.
 145
 146         Returns:
 147             - a dict containing the overall results
 148             - None on error
 149         """
 150         # pylint: disable=too-many-locals,too-many-arguments
 151         self.logger.info('Run Clearwater live test')
 152         script = (f'cd {self.test_dir};'
 153                   f'rake test[{public_domain}] SIGNUP_CODE={signup_code}')
 154         if self.bono_ip and self.ellis_ip:
 155             subscript = f' PROXY={self.bono_ip} ELLIS={self.ellis_ip}'
 156             script = f'{script}{subscript}'
 157         script = f'{script} --trace'
 158         cmd = f"/bin/sh -c '{script}'"
 159         self.logger.debug('Live test cmd: %s', cmd)
 160         output_file = os.path.join(self.result_dir, "ims_test_output.txt")
 161         ft_utils.execute_command(cmd,
 162                                  error_msg='Clearwater live test failed',
 163                                  output_file=output_file)
 164
 165         with open(output_file, 'r', encoding='utf-8') as ofile:
 166             result = ofile.read()
 167
 168         if result != "":
 169             self.logger.debug(result)
 170
 171         vims_test_result = {}
 172         try:
 173             grp = re.search(
 174                 r'^(\d+) failures out of (\d+) tests run.*\n'
 175                 r'(\d+) tests skipped$', result, re.MULTILINE | re.DOTALL)
 176             assert grp
 177             vims_test_result["failures"] = int(grp.group(1))
 178             vims_test_result["total"] = int(grp.group(2))
 179             vims_test_result["skipped"] = int(grp.group(3))
 180             vims_test_result['passed'] = (
 181                 int(grp.group(2)) - int(grp.group(3)) - int(grp.group(1)))
 182             if vims_test_result['total'] - vims_test_result['skipped'] > 0:
 183                 vnf_test_rate = vims_test_result['passed'] / (
 184                     vims_test_result['total'] - vims_test_result['skipped'])
 185             else:
 186                 vnf_test_rate = 0
 187         except Exception:  # pylint: disable=broad-except
 188             self.logger.exception("Cannot parse live tests results")
 189             return None, 0
 190         return vims_test_result, vnf_test_rate