"Helen Yao <helanyao@gmail.com>")
-class ClearwaterOnBoardingBase(object):
+class ClearwaterTesting(object):
"""vIMS clearwater base usable by several orchestrators"""
- def __init__(self, case_name):
+ def __init__(self, case_name, ellis_ip):
self.logger = logging.getLogger(__name__)
self.case_dir = pkg_resources.resource_filename(
'functest', 'opnfv_tests/vnf/ims')
if not os.path.exists(self.result_dir):
os.makedirs(self.result_dir)
- def config_ellis(self, ellis_ip, signup_code='secret', two_numbers=False):
+ self.ellis_ip = ellis_ip
+
+ def availability_check_by_creating_numbers(self,
+ signup_code='secret',
+ two_numbers=False):
"""Create one or two numbers"""
+ assert self.ellis_ip
output_dict = {}
- self.logger.debug('Configure Ellis: %s', ellis_ip)
- output_dict['ellis_ip'] = ellis_ip
- account_url = 'http://{0}/accounts'.format(ellis_ip)
+ self.logger.debug('Ellis IP: %s', self.ellis_ip)
+ output_dict['ellis_ip'] = self.ellis_ip
+ account_url = 'http://{0}/accounts'.format(self.ellis_ip)
params = {"password": "functest",
"full_name": "opnfv functest user",
"email": "functest@opnfv.org",
"signup_code": signup_code}
- req = requests.post(account_url, data=params)
output_dict['login'] = params
- if req.status_code != 201 and req.status_code != 409:
- raise Exception(
- "Unable to create an account {}\n{}".format(
- params, req.text))
- self.logger.debug(
- 'Account %s is created on Ellis\n%s', params, req.json())
-
- session_url = 'http://{0}/session'.format(ellis_ip)
+
+ number_res = self._create_ellis_account(account_url, params)
+ output_dict['number'] = number_res
+
+ session_url = 'http://{0}/session'.format(self.ellis_ip)
session_data = {
'username': params['email'],
'password': params['password'],
'email': params['email']
}
- req = requests.post(session_url, data=session_data)
- if req.status_code != 201:
- raise Exception('Failed to get cookie for Ellis\n{}'.format(
- req.text))
- cookies = req.cookies
- self.logger.debug('Cookies: %s', cookies)
+ cookies = self._get_ellis_session_cookies(session_url, session_data)
number_url = 'http://{0}/accounts/{1}/numbers'.format(
- ellis_ip, params['email'])
+ self.ellis_ip, params['email'])
self.logger.debug('Create 1st calling number on Ellis')
- i = 30
- while req.status_code != 200 and i > 0:
- try:
- number_res = self._create_ellis_number(number_url, cookies)
- break
- except Exception: # pylint: disable=broad-except
- if i == 1:
- self.logger.exception("Unable to create a number")
- raise Exception("Unable to create a number")
- self.logger.info("Unable to create a number. Retry ..")
- time.sleep(25)
- i = i - 1
- output_dict['number'] = number_res
+ number_res = self._create_ellis_number(number_url, cookies)
if two_numbers:
self.logger.debug('Create 2nd calling number on Ellis')
return output_dict
+ def _create_ellis_account(self, account_url, params):
+ i = 50
+ for iloop in range(i):
+ try:
+ req = requests.post(account_url, data=params)
+ if req.status_code == 201:
+ account_res = req.json()
+ self.logger.info(
+ 'Account %s is created on Ellis\n%s',
+ params.get('full_name'), account_res)
+ return account_res
+ else:
+ raise Exception("Cannot create ellis account")
+ except Exception: # pylint: disable=broad-except
+ self.logger.info(
+ "try %s: cannot create ellis account", iloop + 1)
+ time.sleep(25)
+ raise Exception(
+ "Unable to create an account {}".format(
+ params.get('full_name')))
+
+ def _get_ellis_session_cookies(self, session_url, params):
+ i = 15
+ for iloop in range(i):
+ try:
+ req = requests.post(session_url, data=params)
+ if req.status_code == 201:
+ cookies = req.cookies
+ self.logger.debug('cookies: %s', cookies)
+ return cookies
+ else:
+ raise Exception('Failed to get cookies for Ellis')
+ except Exception: # pylint: disable=broad-except
+ self.logger.info(
+ "try %s: cannot get cookies for Ellis", iloop + 1)
+ time.sleep(10)
+ raise Exception('Failed to get cookies for Ellis')
+
def _create_ellis_number(self, number_url, cookies):
- req = requests.post(number_url, cookies=cookies)
-
- if req.status_code != 200:
- if req and req.json():
- reason = req.json()['reason']
- else:
- reason = req
- raise Exception("Unable to create a number: %s" % reason)
- number_res = req.json()
- self.logger.info('Calling number is created: %s', number_res)
- return number_res
+ i = 30
+ for iloop in range(i):
+ try:
+ req = requests.post(number_url, cookies=cookies)
+ if req.status_code == 200:
+ number_res = req.json()
+ self.logger.info(
+ 'Calling number is created: %s', number_res)
+ return number_res
+ else:
+ if req and req.json():
+ reason = req.json()['reason']
+ else:
+ reason = req
+ self.logger.info("cannot create a number: %s", reason)
+ raise Exception('Failed to create a number')
+ except Exception: # pylint: disable=broad-except
+ self.logger.info(
+ "try %s: cannot create a number", iloop + 1)
+ time.sleep(25)
+ raise Exception('Failed to create a number')
def run_clearwater_live_test(self, dns_ip, public_domain,
bono_ip=None, ellis_ip=None,
import six
from functest.core import cloudify
-from functest.opnfv_tests.vnf.ims import clearwater_ims_base
+from functest.opnfv_tests.vnf.ims import clearwater
from functest.utils import config
from functest.utils import env
self.case_dir = pkg_resources.resource_filename(
'functest', 'opnfv_tests/vnf/ims')
config_file = os.path.join(self.case_dir, self.config)
- self.orchestrator = dict(
- requirements=get_config("orchestrator.requirements", config_file),
- )
+
self.details['orchestrator'] = dict(
name=get_config("orchestrator.name", config_file),
version=get_config("orchestrator.version", config_file),
status='ERROR',
result=''
)
- self.__logger.debug("Orchestrator configuration %s", self.orchestrator)
+
self.vnf = dict(
descriptor=get_config("vnf.descriptor", config_file),
- inputs=get_config("vnf.inputs", config_file),
- requirements=get_config("vnf.requirements", config_file)
+ inputs=get_config("vnf.inputs", config_file)
)
self.details['vnf'] = dict(
descriptor_version=self.vnf['descriptor']['version'],
self.image_alt = None
self.flavor_alt = None
+ self.clearwater = None
def check_requirements(self):
if env.get('NEW_USER_ROLE').lower() == "admin":
execution = wait_for_execution(
self.cfy_client, execution, self.__logger, timeout=3600)
- duration = time.time() - start_time
-
self.__logger.info(execution)
- if execution.status == 'terminated':
- self.details['vnf'].update(status='PASS', duration=duration)
- self.result += 1/3 * 100
- result = True
- else:
- self.details['vnf'].update(status='FAIL', duration=duration)
- result = False
- return result
+ if execution.status != 'terminated':
+ self.details['vnf'].update(status='FAIL',
+ duration=time.time() - start_time)
+ return False
+
+ ellis_ip = self.cfy_client.deployments.outputs.get(
+ self.vnf['descriptor'].get('name'))['outputs']['ellis_ip']
+ self.clearwater = clearwater.ClearwaterTesting(self.case_name,
+ ellis_ip)
+ self.clearwater.availability_check_by_creating_numbers()
+
+ self.details['vnf'].update(status='PASS',
+ duration=time.time() - start_time)
+ self.result += 1/3 * 100
+ return True
def test_vnf(self):
"""Run test on clearwater ims instance."""
start_time = time.time()
- testing = clearwater_ims_base.ClearwaterOnBoardingBase(self.case_name)
- outputs = self.cfy_client.deployments.outputs.get(
- self.vnf['descriptor'].get('name'))['outputs']
- dns_ip = outputs['dns_ip']
- ellis_ip = outputs['ellis_ip']
- testing.config_ellis(ellis_ip)
+ dns_ip = self.cfy_client.deployments.outputs.get(
+ self.vnf['descriptor'].get('name'))['outputs']['dns_ip']
if not dns_ip:
return False
- short_result = testing.run_clearwater_live_test(
+ short_result = self.clearwater.run_clearwater_live_test(
dns_ip=dns_ip,
public_domain=self.vnf['inputs']["public_domain"])
duration = time.time() - start_time
Code Review / functest.git / commitdiff