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 = 'http://{0}/accounts'.format(self.ellis_ip)
  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 = 'http://{0}/session'.format(self.ellis_ip)
  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 = 'http://{0}/accounts/{1}/numbers'.format(
  72             self.ellis_ip, params['email'])
  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             "Unable to create an account {}".format(
 101                 params.get('full_name')))
 102
 103     def _get_ellis_session_cookies(self, session_url, params):
 104         i = 15
 105         for iloop in range(i):
 106             try:
 107                 req = requests.post(session_url, data=params)
 108                 if req.status_code == 201:
 109                     cookies = req.cookies
 110                     self.logger.debug('cookies: %s', cookies)
 111                     return cookies
 112                 raise Exception('Failed to get cookies for Ellis')
 113             except Exception:  # pylint: disable=broad-except
 114                 self.logger.info(
 115                     "try %s: cannot get cookies for Ellis", iloop + 1)
 116                 time.sleep(10)
 117         raise Exception('Failed to get cookies for Ellis')
 118
 119     def _create_ellis_number(self, number_url, cookies):
 120         i = 30
 121         for iloop in range(i):
 122             try:
 123                 req = requests.post(number_url, cookies=cookies)
 124                 if req.status_code == 200:
 125                     number_res = req.json()
 126                     self.logger.info(
 127                         'Calling number is created: %s', number_res)
 128                     return number_res
 129                 if req and req.json():
 130                     reason = req.json()['reason']
 131                 else:
 132                     reason = req
 133                 self.logger.info("cannot create a number: %s", reason)
 134                 raise Exception('Failed to create a number')
 135             except Exception:  # pylint: disable=broad-except
 136                 self.logger.info(
 137                     "try %s: cannot create a number", iloop + 1)
 138                 time.sleep(25)
 139         raise Exception('Failed to create a number')
 140
 141     def run_clearwater_live_test(self, public_domain, signup_code='secret'):
 142         """Run the Clearwater live tests
 143
 144         It first runs dnsmasq to reach clearwater services by FQDN and then the
 145         Clearwater live tests. All results are saved in ims_test_output.txt.
 146
 147         Returns:
 148             - a dict containing the overall results
 149             - None on error
 150         """
 151         # pylint: disable=too-many-locals,too-many-arguments
 152         self.logger.info('Run Clearwater live test')
 153         script = ('cd {0};'
 154                   'rake test[{1}] SIGNUP_CODE={2}'
 155                   .format(self.test_dir,
 156                           public_domain,
 157                           signup_code))
 158         if self.bono_ip and self.ellis_ip:
 159             subscript = ' PROXY={0} ELLIS={1}'.format(
 160                 self.bono_ip, self.ellis_ip)
 161             script = '{0}{1}'.format(script, subscript)
 162         script = ('{0}{1}'.format(script, ' --trace'))
 163         cmd = "/bin/bash -c '{0}'".format(script)
 164         self.logger.debug('Live test cmd: %s', cmd)
 165         output_file = os.path.join(self.result_dir, "ims_test_output.txt")
 166         ft_utils.execute_command(cmd,
 167                                  error_msg='Clearwater live test failed',
 168                                  output_file=output_file)
 169
 170         with open(output_file, 'r') as ofile:
 171             result = ofile.read()
 172
 173         if result != "":
 174             self.logger.debug(result)
 175
 176         vims_test_result = {}
 177         try:
 178             grp = re.search(
 179                 r'^(\d+) failures out of (\d+) tests run.*\n'
 180                 r'(\d+) tests skipped$', result, re.MULTILINE | re.DOTALL)
 181             assert grp
 182             vims_test_result["failures"] = int(grp.group(1))
 183             vims_test_result["total"] = int(grp.group(2))
 184             vims_test_result["skipped"] = int(grp.group(3))
 185             vims_test_result['passed'] = (
 186                 int(grp.group(2)) - int(grp.group(3)) - int(grp.group(1)))
 187             if vims_test_result['total'] - vims_test_result['skipped'] > 0:
 188                 vnf_test_rate = vims_test_result['passed'] / (
 189                     vims_test_result['total'] - vims_test_result['skipped'])
 190             else:
 191                 vnf_test_rate = 0
 192         except Exception:  # pylint: disable=broad-except
 193             self.logger.exception("Cannot parse live tests results")
 194             return None, 0
 195         return vims_test_result, vnf_test_rate