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