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