functest/opnfv_tests/vnf/ims/clearwater_ims_base.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 import json
  10 import logging
  11 import os
  12 import pkg_resources
  13 import shlex
  14 import shutil
  15 import subprocess
  16 import time
  17
  18 import requests
  19
  20 import functest.core.vnf as vnf
  21 from functest.utils.constants import CONST
  22 import functest.utils.functest_utils as ft_utils
  23
  24 __author__ = ("Valentin Boucher <valentin.boucher@orange.com>, "
  25               "Helen Yao <helanyao@gmail.com>")
  26
  27
  28 class ClearwaterOnBoardingBase(vnf.VnfOnBoarding):
  29     """ vIMS clearwater base usable by several orchestrators"""
  30
  31     def __init__(self, **kwargs):
  32         self.logger = logging.getLogger(__name__)
  33         super(ClearwaterOnBoardingBase, self).__init__(**kwargs)
  34         self.case_dir = pkg_resources.resource_filename(
  35             'functest', 'opnfv_tests/vnf/ims')
  36         self.data_dir = CONST.__getattribute__('dir_ims_data')
  37         self.result_dir = os.path.join(CONST.__getattribute__('dir_results'),
  38                                        self.case_name)
  39         self.test_dir = CONST.__getattribute__('dir_repo_vims_test')
  40
  41         if not os.path.exists(self.data_dir):
  42             os.makedirs(self.data_dir)
  43         if not os.path.exists(self.result_dir):
  44             os.makedirs(self.result_dir)
  45
  46     def config_ellis(self, ellis_ip, signup_code='secret', two_numbers=False):
  47         output_dict = {}
  48         self.logger.debug('Configure Ellis: %s', ellis_ip)
  49         output_dict['ellis_ip'] = ellis_ip
  50         account_url = 'http://{0}/accounts'.format(ellis_ip)
  51         params = {"password": "functest",
  52                   "full_name": "opnfv functest user",
  53                   "email": "functest@opnfv.org",
  54                   "signup_code": signup_code}
  55         rq = requests.post(account_url, data=params)
  56         output_dict['login'] = params
  57         if rq.status_code != 201 and rq.status_code != 409:
  58             raise Exception("Unable to create an account for number provision")
  59         self.logger.debug('Account is created on Ellis: %s', params)
  60
  61         session_url = 'http://{0}/session'.format(ellis_ip)
  62         session_data = {
  63             'username': params['email'],
  64             'password': params['password'],
  65             'email': params['email']
  66         }
  67         rq = requests.post(session_url, data=session_data)
  68         if rq.status_code != 201:
  69             raise Exception('Failed to get cookie for Ellis')
  70         cookies = rq.cookies
  71         self.logger.debug('Cookies: %s', cookies)
  72
  73         number_url = 'http://{0}/accounts/{1}/numbers'.format(
  74                      ellis_ip,
  75                      params['email'])
  76         self.logger.debug('Create 1st calling number on Ellis')
  77         i = 30
  78         while rq.status_code != 200 and i > 0:
  79             try:
  80                 number_res = self.create_ellis_number(number_url, cookies)
  81                 break
  82             except:
  83                 if i == 1:
  84                     raise Exception("Unable to create a number")
  85                 self.logger.warn("Unable to create a number. Retry ..")
  86                 time.sleep(25)
  87             i = i - 1
  88         output_dict['number'] = number_res
  89
  90         if two_numbers:
  91             self.logger.debug('Create 2nd calling number on Ellis')
  92             number_res = self.create_ellis_number(number_url, cookies)
  93             output_dict['number2'] = number_res
  94
  95         return output_dict
  96
  97     def create_ellis_number(self, number_url, cookies):
  98         rq = requests.post(number_url, cookies=cookies)
  99
 100         if rq.status_code != 200:
 101             if rq and rq.json():
 102                 reason = rq.json()['reason']
 103             else:
 104                 reason = rq
 105             raise Exception("Unable to create a number: %s" % reason)
 106         number_res = rq.json()
 107         self.logger.info('Calling number is created: %s', number_res)
 108         return number_res
 109
 110     def run_clearwater_live_test(self, dns_ip, public_domain,
 111                                  bono_ip=None, ellis_ip=None,
 112                                  signup_code='secret'):
 113         self.logger.info('Run Clearwater live test')
 114         dns_file = '/etc/resolv.conf'
 115         dns_file_bak = '/etc/resolv.conf.bak'
 116         self.logger.debug('Backup %s -> %s', dns_file, dns_file_bak)
 117         shutil.copy(dns_file, dns_file_bak)
 118         cmd = ("dnsmasq -d -u root --server=/clearwater.opnfv/{0} "
 119                "-r /etc/resolv.conf.bak".format(dns_ip))
 120         dnsmasq_process = subprocess.Popen(shlex.split(cmd))
 121         script = ('echo -e "nameserver {0}" > {1};'
 122                   'cd {2};'
 123                   'rake test[{3}] SIGNUP_CODE={4}'
 124                   .format('127.0.0.1',
 125                           dns_file,
 126                           self.test_dir,
 127                           public_domain,
 128                           signup_code))
 129         if bono_ip and ellis_ip:
 130             subscript = ' PROXY={0} ELLIS={1}'.format(bono_ip, ellis_ip)
 131             script = '{0}{1}'.format(script, subscript)
 132         script = ('{0}{1}'.format(script, ' --trace'))
 133         cmd = "/bin/bash -c '{0}'".format(script)
 134         self.logger.debug('Live test cmd: %s', cmd)
 135         output_file = os.path.join(self.result_dir, "ims_test_output.txt")
 136         ft_utils.execute_command(cmd,
 137                                  error_msg='Clearwater live test failed',
 138                                  output_file=output_file)
 139         dnsmasq_process.kill()
 140         with open(dns_file_bak, 'r') as bak_file:
 141             result = bak_file.read()
 142             with open(dns_file, 'w') as f:
 143                 f.write(result)
 144
 145         f = open(output_file, 'r')
 146         result = f.read()
 147         if result != "":
 148             self.logger.debug(result)
 149
 150         vims_test_result = ""
 151         tempFile = os.path.join(self.test_dir, "temp.json")
 152         try:
 153             self.logger.debug("Trying to load test results")
 154             with open(tempFile) as f:
 155                 vims_test_result = json.load(f)
 156             f.close()
 157         except Exception:
 158             self.logger.error("Unable to retrieve test results")
 159
 160         try:
 161             os.remove(tempFile)
 162         except Exception:
 163             self.logger.error("Deleting file failed")
 164
 165         return vims_test_result