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