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