functest/opnfv_tests/vnf/ims/opera_ims.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 HUAWEI TECHNOLOGIES CO.,LTD and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 import json
  11 import logging
  12 import os
  13 import time
  14
  15 from opera import openo_connect
  16 import requests
  17
  18 import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base
  19 from functest.utils.constants import CONST
  20
  21
  22 class OperaIms(clearwater_ims_base.ClearwaterOnBoardingBase):
  23
  24     def __init__(self, **kwargs):
  25         if "case_name" not in kwargs:
  26             kwargs["case_name"] = "opera_ims"
  27         super(OperaIms, self).__init__(**kwargs)
  28         self.logger = logging.getLogger(__name__)
  29         self.ellis_file = os.path.join(
  30             CONST.__getattribute__('dir_results'), 'ellis.info')
  31         self.live_test_file = os.path.join(
  32             CONST.__getattribute__('dir_results'), 'live_test_report.json')
  33         try:
  34             self.openo_msb_endpoint = os.environ['OPENO_MSB_ENDPOINT']
  35         except KeyError:
  36             raise Exception('OPENO_MSB_ENDPOINT is not specified,'
  37                             ' put it as <OPEN-O ip>:<port>')
  38         else:
  39             self.logger.info('OPEN-O endpoint is: %s', self.openo_msb_endpoint)
  40
  41     def prepare(self):
  42         pass
  43
  44     def clean(self):
  45         pass
  46
  47     def deploy_vnf(self):
  48         try:
  49             openo_connect.create_service(self.openo_msb_endpoint,
  50                                          'functest_opera',
  51                                          'VNF for functest testing')
  52         except Exception as e:
  53             self.logger.error(e)
  54             return {'status': 'FAIL', 'result': e}
  55         else:
  56             self.logger.info('vIMS deployment is kicked off')
  57             return {'status': 'PASS', 'result': ''}
  58
  59     def dump_info(self, info_file, result):
  60         with open(info_file, 'w') as f:
  61             self.logger.debug('Save information to file: %s', info_file)
  62             json.dump(result, f)
  63
  64     def test_vnf(self):
  65         vnfm_ip = openo_connect.get_vnfm_ip(self.openo_msb_endpoint)
  66         self.logger.info('VNFM IP: %s', vnfm_ip)
  67         vnf_status_url = 'http://{0}:5000/api/v1/model/status'.format(vnfm_ip)
  68         vnf_alive = False
  69         retry = 40
  70
  71         self.logger.info('Check the VNF status')
  72         while retry > 0:
  73             rq = requests.get(vnf_status_url, timeout=90)
  74             response = rq.json()
  75             vnf_alive = response['vnf_alive']
  76             msg = response['msg']
  77             self.logger.info(msg)
  78             if vnf_alive:
  79                 break
  80             self.logger.info('check again in one and half a minute...')
  81             retry = retry - 1
  82             time.sleep(90)
  83
  84         if not vnf_alive:
  85             raise Exception('VNF failed to start: {0}'.format(msg))
  86
  87         ellis_config_url = ('http://{0}:5000/api/v1/model/ellis/configure'
  88                             .format(vnfm_ip))
  89         rq = requests.get(ellis_config_url, timeout=90)
  90         if rq.json() and not rq.json()['ellis_ok']:
  91             self.logger.error(rq.json()['data'])
  92             raise Exception('Failed to configure Ellis')
  93
  94         self.logger.info('Get Clearwater deployment detail')
  95         vnf_info_url = ('http://{0}:5000/api/v1/model/output'
  96                         .format(vnfm_ip))
  97         rq = requests.get(vnf_info_url, timeout=90)
  98         data = rq.json()['data']
  99         self.logger.info(data)
 100         bono_ip = data['bono_ip']
 101         ellis_ip = data['ellis_ip']
 102         dns_ip = data['dns_ip']
 103         result = self.config_ellis(ellis_ip, 'signup', True)
 104         self.logger.debug('Ellis Result: %s', result)
 105         self.dump_info(self.ellis_file, result)
 106
 107         if dns_ip:
 108             vims_test_result = self.run_clearwater_live_test(
 109                 dns_ip,
 110                 'clearwater.local',
 111                 bono_ip,
 112                 ellis_ip,
 113                 'signup')
 114             if vims_test_result != '':
 115                 self.dump_info(self.live_test_file, vims_test_result)
 116                 return {'status': 'PASS', 'result': vims_test_result}
 117             else:
 118                 return {'status': 'FAIL', 'result': ''}
 119
 120     def main(self, **kwargs):
 121         self.logger.info("Start to run Opera vIMS VNF onboarding test")
 122         self.execute()
 123         self.logger.info("Opera vIMS VNF onboarding test finished")
 124         if self.result is "PASS":
 125             return self.EX_OK
 126         else:
 127             return self.EX_RUN_ERROR
 128
 129     def run(self):
 130         kwargs = {}
 131         return self.main(**kwargs)