diff options
Diffstat (limited to 'functest')
| -rwxr-xr-x | functest/ci/config_functest.yaml | 1 | ||||
| -rwxr-xr-x | functest/ci/testcases.yaml | 8 | ||||
| -rw-r--r-- | functest/core/vnf_base.py | 26 | ||||
| -rwxr-xr-x | functest/opnfv_tests/openstack/refstack_client/refstack_client.py | 2 | ||||
| -rwxr-xr-x | functest/opnfv_tests/openstack/refstack_client/tempest_conf.py | 6 | ||||
| -rwxr-xr-x | functest/opnfv_tests/sdn/odl/odl.py | 69 | ||||
| -rw-r--r-- | functest/opnfv_tests/vnf/ims/clearwater_ims_base.py | 149 | ||||
| -rw-r--r-- | functest/opnfv_tests/vnf/ims/cloudify_ims.py | 142 | ||||
| -rw-r--r--[-rwxr-xr-x] | functest/opnfv_tests/vnf/ims/opera_ims.py | 457 | ||||
| -rw-r--r-- | functest/tests/unit/core/test_testcase.py (renamed from functest/tests/unit/core/test_testcase_base.py) | 7 | ||||
| -rw-r--r-- | functest/tests/unit/odl/test_odl.py | 371 | ||||
| -rw-r--r-- | functest/tests/unit/opnfv_tests/vnf/ims/test_cloudify_ims.py | 118 | ||||
| -rw-r--r-- | functest/tests/unit/opnfv_tests/vnf/ims/test_ims_base.by | 58 |
13 files changed, 681 insertions, 733 deletions
diff --git a/functest/ci/config_functest.yaml b/functest/ci/config_functest.yaml index 78f6257c..95a4408a 100755 --- a/functest/ci/config_functest.yaml +++ b/functest/ci/config_functest.yaml @@ -27,6 +27,7 @@ general: repo_parser: /home/opnfv/repos/parser repo_domino: /home/opnfv/repos/domino repo_snaps: /home/opnfv/repos/snaps + repo_opera: /home/opnfv/repos/opera repo_fds: /home/opnfv/repos/fds repo_securityscan: /home/opnfv/repos/securityscanning repo_vrouter: /home/opnfv/repos/vnfs/vrouter diff --git a/functest/ci/testcases.yaml b/functest/ci/testcases.yaml index 9a62770e..1c33b996 100755 --- a/functest/ci/testcases.yaml +++ b/functest/ci/testcases.yaml @@ -487,7 +487,7 @@ tiers: scenario: '(ocl)|(nosdn)|^(os-odl)((?!bgpvpn).)*$' run: module: 'functest.opnfv_tests.vnf.ims.cloudify_ims' - class: 'ImsVnf' + class: 'CloudifyIms' # - # name: aaa # criteria: 'ret == 0' @@ -517,17 +517,17 @@ tiers: - name: opera_ims - criteria: 'ret == 0' + criteria: 'status == "PASS"' blocking: false clean_flag: true description: >- - Evolution of vIMS + VNF deployment with OPEN-O dependencies: installer: 'unknown' scenario: 'unknown' run: module: 'functest.opnfv_tests.vnf.ims.opera_ims' - class: 'ImsVnf' + class: 'OperaIms' - name: vyos_vrouter diff --git a/functest/core/vnf_base.py b/functest/core/vnf_base.py index c374c491..3f0adcc6 100644 --- a/functest/core/vnf_base.py +++ b/functest/core/vnf_base.py @@ -7,15 +7,14 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 -import time - import inspect +import time +import functest.core.testcase as base +from functest.utils.constants import CONST import functest.utils.functest_logger as ft_logger -import functest.utils.openstack_utils as os_utils import functest.utils.functest_utils as ft_utils -import testcase as base -from functest.utils.constants import CONST +import functest.utils.openstack_utils as os_utils class VnfOnBoardingBase(base.TestCase): @@ -29,7 +28,7 @@ class VnfOnBoardingBase(base.TestCase): self.case_name = case self.cmd = cmd self.details = {} - self.data_dir = CONST.dir_functest_data + self.result_dir = CONST.dir_results self.details['orchestrator'] = {} self.details['vnf'] = {} self.details['test_vnf'] = {} @@ -39,14 +38,14 @@ class VnfOnBoardingBase(base.TestCase): 'vnf_{}_tenant_name'.format(self.case_name)) self.tenant_description = CONST.__getattribute__( 'vnf_{}_tenant_description'.format(self.case_name)) - except: + except Exception: # raise Exception("Unknown VNF case=" + self.case_name) self.logger.error("Unknown VNF case={}".format(self.case_name)) try: self.images = CONST.__getattribute__( 'vnf_{}_tenant_images'.format(self.case_name)) - except: + except Exception: self.logger.warn("No tenant image defined for this VNF") def execute(self): @@ -234,7 +233,12 @@ class VnfOnBoardingBase(base.TestCase): def step_failure(self, error_msg): part = inspect.stack()[1][3] - self.details[part]['status'] = 'FAIL' - self.details[part]['result'] = error_msg - self.logger.error("Step failure:{}".format(error_msg)) + self.logger.error("Step '%s' failed: %s", part, error_msg) + try: + part_info = self.details[part] + except KeyError: + self.details[part] = {} + part_info = self.details[part] + part_info['status'] = 'FAIL' + part_info['result'] = error_msg raise Exception(error_msg) diff --git a/functest/opnfv_tests/openstack/refstack_client/refstack_client.py b/functest/opnfv_tests/openstack/refstack_client/refstack_client.py index 101bf736..37aa9e39 100755 --- a/functest/opnfv_tests/openstack/refstack_client/refstack_client.py +++ b/functest/opnfv_tests/openstack/refstack_client/refstack_client.py @@ -184,7 +184,7 @@ class RefstackClient(testcase.TestCase): self.source_venv() self._prep_test() self.run_defcore(self.confpath, self.testlist) - res = testcase.TestcaseBase.EX_OK + res = testcase.TestCase.EX_OK except Exception as e: logger.error('Error with run: %s', e) res = testcase.TestCase.EX_RUN_ERROR diff --git a/functest/opnfv_tests/openstack/refstack_client/tempest_conf.py b/functest/opnfv_tests/openstack/refstack_client/tempest_conf.py index 66876bbf..5624ed79 100755 --- a/functest/opnfv_tests/openstack/refstack_client/tempest_conf.py +++ b/functest/opnfv_tests/openstack/refstack_client/tempest_conf.py @@ -39,15 +39,15 @@ class TempestConf(object): def main(self): try: self.generate_tempestconf() - res = testcase.TestcaseBase.EX_OK + res = testcase.TestCase.EX_OK except Exception as e: logger.error('Error with run: %s', e) - res = testcase.TestcaseBase.EX_RUN_ERROR + res = testcase.TestCase.EX_RUN_ERROR return res if __name__ == '__main__': tempestconf = TempestConf() result = tempestconf.main() - if result != testcase.TestcaseBase.EX_OK: + if result != testcase.TestCase.EX_OK: sys.exit(result) diff --git a/functest/opnfv_tests/sdn/odl/odl.py b/functest/opnfv_tests/sdn/odl/odl.py index f65118f6..ccc1101a 100755 --- a/functest/opnfv_tests/sdn/odl/odl.py +++ b/functest/opnfv_tests/sdn/odl/odl.py @@ -7,6 +7,15 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +"""Define classes required to run ODL suites. + +It has been designed for any context. But helpers are given for +running test suites in OPNFV environment. + +Example: + $ python odl.py +""" + import argparse import errno import fileinput @@ -24,8 +33,11 @@ from functest.core import testcase import functest.utils.functest_logger as ft_logger import functest.utils.openstack_utils as op_utils +__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" + class ODLResultVisitor(robot.api.ResultVisitor): + """Visitor to get result details.""" def __init__(self): self._data = [] @@ -43,10 +55,12 @@ class ODLResultVisitor(robot.api.ResultVisitor): self._data.append(output) def get_data(self): + """Get the details of the result.""" return self._data class ODLTests(testcase.TestCase): + """ODL test runner.""" repos = "/home/opnfv/repos/" odl_test_repo = os.path.join(repos, "odl_test") @@ -64,6 +78,12 @@ class ODLTests(testcase.TestCase): @classmethod def set_robotframework_vars(cls, odlusername="admin", odlpassword="admin"): + """Set credentials in csit/variables/Variables.py. + + Returns: + True if credentials are set. + False otherwise. + """ odl_variables_files = os.path.join(cls.odl_test_repo, 'csit/variables/Variables.py') try: @@ -79,6 +99,7 @@ class ODLTests(testcase.TestCase): return False def parse_results(self): + """Parse output.xml and get the details in it.""" xml_file = os.path.join(self.res_dir, 'output.xml') result = robot.api.ExecutionResult(xml_file) visitor = ODLResultVisitor() @@ -91,6 +112,34 @@ class ODLTests(testcase.TestCase): self.details['tests'] = visitor.get_data() def main(self, suites=None, **kwargs): + """Run the test suites + + It has been designed to be called in any context. + It requires the following keyword arguments: + * odlusername, + * odlpassword, + * osauthurl, + * neutronip, + * osusername, + * ostenantname, + * ospassword, + * odlip, + * odlwebport, + * odlrestconfport. + + Here are the steps: + * set all RobotFramework_variables, + * create the output directories if required, + * get the results in output.xml, + * delete temporary files. + + Args: + **kwargs: Arbitrary keyword arguments. + + Returns: + EX_OK if all suites ran well. + EX_RUN_ERROR otherwise. + """ try: if not suites: suites = self.default_suites @@ -146,6 +195,18 @@ class ODLTests(testcase.TestCase): return self.EX_RUN_ERROR def run(self, **kwargs): + """Run suites in OPNFV environment + + It basically check env vars to call main() with the keywords + required. + + Args: + **kwargs: Arbitrary keyword arguments. + + Returns: + EX_OK if all suites ran well. + EX_RUN_ERROR otherwise. + """ try: suites = self.default_suites try: @@ -191,6 +252,7 @@ class ODLTests(testcase.TestCase): class ODLParser(object): # pylint: disable=too-few-public-methods + """Parser to run ODL test suites.""" def __init__(self): self.parser = argparse.ArgumentParser() @@ -229,6 +291,13 @@ class ODLParser(object): # pylint: disable=too-few-public-methods action='store_true') def parse_args(self, argv=None): + """Parse arguments. + + It can call sys.exit if arguments are incorrect. + + Returns: + the arguments from cmdline + """ if not argv: argv = [] return vars(self.parser.parse_args(argv)) diff --git a/functest/opnfv_tests/vnf/ims/clearwater_ims_base.py b/functest/opnfv_tests/vnf/ims/clearwater_ims_base.py new file mode 100644 index 00000000..f21ce3f9 --- /dev/null +++ b/functest/opnfv_tests/vnf/ims/clearwater_ims_base.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python +# +# Copyright (c) 2017 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 os +import shutil + +import requests + +import functest.core.vnf_base as vnf_base +from functest.utils.constants import CONST +import functest.utils.functest_logger as ft_logger +import functest.utils.functest_utils as ft_utils + + +class ClearwaterOnBoardingBase(vnf_base.VnfOnBoardingBase): + + def __init__(self, project='functest', case='', repo='', cmd=''): + self.logger = ft_logger.Logger(__name__).getLogger() + super(ClearwaterOnBoardingBase, self).__init__( + project, case, repo, cmd) + self.case_dir = os.path.join(CONST.dir_functest_test, 'vnf', 'ims') + self.data_dir = CONST.dir_ims_data + self.result_dir = os.path.join(CONST.dir_results, case) + self.test_dir = CONST.dir_repo_vims_test + + if not os.path.exists(self.data_dir): + os.makedirs(self.data_dir) + 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): + output_dict = {} + self.logger.info('Configure Ellis: %s', ellis_ip) + output_dict['ellis_ip'] = ellis_ip + account_url = 'http://{0}/accounts'.format(ellis_ip) + params = {"password": "functest", + "full_name": "opnfv functest user", + "email": "functest@opnfv.org", + "signup_code": signup_code} + rq = requests.post(account_url, data=params) + output_dict['login'] = params + if rq.status_code != 201 and rq.status_code != 409: + raise Exception("Unable to create an account for number" + " provision: %s" % rq.json()['reason']) + self.logger.info('Account is created on Ellis: %s', params) + + session_url = 'http://{0}/session'.format(ellis_ip) + session_data = { + 'username': params['email'], + 'password': params['password'], + 'email': params['email'] + } + rq = requests.post(session_url, data=session_data) + if rq.status_code != 201: + raise Exception('Failed to get cookie for Ellis') + cookies = rq.cookies + self.logger.info('Cookies: %s' % cookies) + + number_url = 'http://{0}/accounts/{1}/numbers'.format( + ellis_ip, + params['email']) + self.logger.info('Create 1st calling number on Ellis') + number_res = self.create_ellis_number(number_url, cookies) + output_dict['number'] = number_res + + if two_numbers: + self.logger.info('Create 2nd calling number on Ellis') + number_res = self.create_ellis_number(number_url, cookies) + output_dict['number2'] = number_res + + return output_dict + + def create_ellis_number(self, number_url, cookies): + rq = requests.post(number_url, cookies=cookies) + + if rq.status_code != 200: + if rq and rq.json(): + reason = rq.json()['reason'] + else: + reason = rq + raise Exception("Unable to create a number: %s" % reason) + number_res = rq.json() + self.logger.info('Calling number is created: %s', number_res) + return number_res + + def run_clearwater_live_test(self, dns_ip, public_domain, + bono_ip=None, ellis_ip=None, + signup_code='secret'): + self.logger.info('Run Clearwater live test') + nameservers = ft_utils.get_resolvconf_ns() + resolvconf = ['{0}{1}{2}'.format(os.linesep, 'nameserver ', ns) + for ns in nameservers] + self.logger.debug('resolvconf: %s', resolvconf) + dns_file = '/etc/resolv.conf' + dns_file_bak = '/etc/resolv.conf.bak' + shutil.copy(dns_file, dns_file_bak) + script = ('echo -e "nameserver {0}{1}" > {2};' + 'source /etc/profile.d/rvm.sh;' + 'cd {3};' + 'rake test[{4}] SIGNUP_CODE={5}' + .format(dns_ip, + ''.join(resolvconf), + dns_file, + self.test_dir, + public_domain, + signup_code)) + if bono_ip and ellis_ip: + subscript = ' PROXY={0} ELLIS={1}'.format(bono_ip, ellis_ip) + script = '{0}{1}'.format(script, subscript) + script = ('{0}{1}'.format(script, ' --trace')) + cmd = "/bin/bash -c '{0}'".format(script) + self.logger.info('Live test cmd: %s', cmd) + output_file = os.path.join(self.result_dir, "ims_test_output.txt") + ft_utils.execute_command(cmd, + error_msg='Clearwater live test failed', + output_file=output_file) + + with open(dns_file_bak, 'r') as bak_file: + result = bak_file.read() + with open(dns_file, 'w') as f: + f.write(result) + + f = open(output_file, 'r') + result = f.read() + if result != "": + self.logger.debug(result) + + vims_test_result = "" + tempFile = os.path.join(self.test_dir, "temp.json") + try: + self.logger.debug("Trying to load test results") + with open(tempFile) as f: + vims_test_result = json.load(f) + f.close() + except Exception: + self.logger.error("Unable to retrieve test results") + + try: + os.remove(tempFile) + except Exception: + self.logger.error("Deleting file failed") + + return vims_test_result diff --git a/functest/opnfv_tests/vnf/ims/cloudify_ims.py b/functest/opnfv_tests/vnf/ims/cloudify_ims.py index f7dfd532..404f208e 100644 --- a/functest/opnfv_tests/vnf/ims/cloudify_ims.py +++ b/functest/opnfv_tests/vnf/ims/cloudify_ims.py @@ -7,42 +7,37 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 -import json import os -import requests -import subprocess import sys import time + +import requests import yaml -import functest.core.vnf_base as vnf_base +from functest.opnfv_tests.vnf.ims.clearwater import Clearwater +import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base +from functest.opnfv_tests.vnf.ims.orchestrator_cloudify import Orchestrator +from functest.utils.constants import CONST import functest.utils.functest_logger as ft_logger import functest.utils.functest_utils as ft_utils import functest.utils.openstack_utils as os_utils -from clearwater import Clearwater -from functest.utils.constants import CONST -from orchestrator_cloudify import Orchestrator - -class ImsVnf(vnf_base.VnfOnBoardingBase): +class CloudifyIms(clearwater_ims_base.ClearwaterOnBoardingBase): def __init__(self, project='functest', case='cloudify_ims', repo='', cmd=''): - super(ImsVnf, self).__init__(project, case, repo, cmd) - self.logger = ft_logger.Logger("vIMS").getLogger() - self.case_dir = os.path.join(CONST.dir_functest_test, 'vnf/ims/') - self.data_dir = CONST.dir_ims_data - self.test_dir = CONST.dir_repo_vims_test + super(CloudifyIms, self).__init__(project, case, repo, cmd) + self.logger = ft_logger.Logger(__name__).getLogger() # Retrieve the configuration try: self.config = CONST.__getattribute__( 'vnf_{}_config'.format(self.case_name)) - except: + except Exception: raise Exception("VNF config file not found") - config_file = self.case_dir + self.config + config_file = os.path.join(self.case_dir, self.config) self.orchestrator = dict( requirements=get_config("cloudify.requirements", config_file), blueprint=get_config("cloudify.blueprint", config_file), @@ -61,10 +56,6 @@ class ImsVnf(vnf_base.VnfOnBoardingBase): self.images = get_config("tenant_images", config_file) self.logger.info("Images needed for vIMS: %s" % self.images) - # vIMS Data directory creation - if not os.path.exists(self.data_dir): - os.makedirs(self.data_dir) - def deploy_orchestrator(self, **kwargs): self.logger.info("Additional pre-configuration steps") @@ -82,7 +73,7 @@ class ImsVnf(vnf_base.VnfOnBoardingBase): image_id = os_utils.get_image_id(self.glance_client, image_name) self.logger.debug("image_id: %s" % image_id) - except: + except Exception: self.logger.error("Unexpected error: %s" % sys.exc_info()[0]) if image_id == '': @@ -177,10 +168,11 @@ class ImsVnf(vnf_base.VnfOnBoardingBase): self.logger.debug("Resolvconf set") self.logger.info("Prepare virtualenv for cloudify-cli") - cmd = "chmod +x " + self.case_dir + "create_venv.sh" + venv_scrit_dir = os.path.join(self.case_dir, "create_venv.sh") + cmd = "chmod +x " + venv_scrit_dir ft_utils.execute_command(cmd) time.sleep(3) - cmd = self.case_dir + "create_venv.sh " + self.data_dir + cmd = venv_scrit_dir + " " + self.data_dir ft_utils.execute_command(cmd) cfy.download_manager_blueprint( @@ -251,99 +243,25 @@ class ImsVnf(vnf_base.VnfOnBoardingBase): self.logger.debug("Trying to get clearwater manager IP ... ") mgr_ip = os.popen(cmd).read() mgr_ip = mgr_ip.splitlines()[0] - except: + except Exception: self.step_failure("Unable to retrieve the IP of the " "cloudify manager server !") - api_url = "http://" + mgr_ip + "/api/v2" - dep_outputs = requests.get(api_url + "/deployments/" + - self.vnf['deployment_name'] + "/outputs") - dns_ip = dep_outputs.json()['outputs']['dns_ip'] - ellis_ip = dep_outputs.json()['outputs']['ellis_ip'] - - self.logger.debug("DNS ip : %s" % dns_ip) - self.logger.debug("ELLIS ip : %s" % ellis_ip) - - ellis_url = "http://" + ellis_ip + "/" - url = ellis_url + "accounts" - - params = {"password": "functest", - "full_name": "opnfv functest user", - "email": "functest@opnfv.fr", - "signup_code": "secret"} - - rq = requests.post(url, data=params) - i = 30 - while rq.status_code != 201 and i > 0: - rq = requests.post(url, data=params) - self.logger.debug("Account creation http status code: %s" - % rq.status_code) - i = i - 1 - time.sleep(10) - - if rq.status_code == 201: - url = ellis_url + "session" - rq = requests.post(url, data=params) - cookies = rq.cookies - else: - self.step_failure("Unable to create an account") - - url = ellis_url + "accounts/" + params['email'] + "/numbers" - if cookies != "": - rq = requests.post(url, cookies=cookies) - i = 24 - while rq.status_code != 200 and i > 0: - rq = requests.post(url, cookies=cookies) - self.logger.debug("Number creation http status code: %s" - % rq.status_code) - i = i - 1 - time.sleep(25) - - if rq.status_code != 200: - self.step_failure("Unable to create a number: %s" - % rq.json()['reason']) - - nameservers = ft_utils.get_resolvconf_ns() - resolvconf = "" - for ns in nameservers: - resolvconf += "\nnameserver " + ns + self.logger.info('Cloudify Manager: %s', mgr_ip) + api_url = 'http://{0}/api/v2/deployments/{1}/outputs'.format( + mgr_ip, self.vnf['deployment_name']) + dep_outputs = requests.get(api_url) + self.logger.info(api_url) + outputs = dep_outputs.json()['outputs'] + self.logger.info("Deployment outputs: %s", outputs) + dns_ip = outputs['dns_ip'] + ellis_ip = outputs['ellis_ip'] + self.config_ellis(ellis_ip) if dns_ip != "": - script = ('echo -e "nameserver ' + dns_ip + resolvconf + - '" > /etc/resolv.conf; ') - script += 'source /etc/profile.d/rvm.sh; ' - script += 'cd {0}; ' - script += ('rake test[{1}] SIGNUP_CODE="secret"') - - cmd = ("/bin/bash -c '" + - script.format(self.data_dir, self.inputs["public_domain"]) + - "'") - output_file = "output.txt" - f = open(output_file, 'w+') - subprocess.call(cmd, shell=True, stdout=f, - stderr=subprocess.STDOUT) - f.close() - - f = open(output_file, 'r') - result = f.read() - if result != "": - self.logger.debug(result) - - vims_test_result = "" - tempFile = os.path.join(self.test_dir, "temp.json") - try: - self.logger.debug("Trying to load test results") - with open(tempFile) as f: - vims_test_result = json.load(f) - f.close() - except: - self.logger.error("Unable to retrieve test results") - - try: - os.remove(tempFile) - except: - self.logger.error("Deleting file failed") - + vims_test_result = self.run_clearwater_live_test( + dns_ip=dns_ip, + public_domain=self.inputs["public_domain"]) if vims_test_result != '': return {'status': 'PASS', 'result': vims_test_result} else: @@ -352,7 +270,7 @@ class ImsVnf(vnf_base.VnfOnBoardingBase): def clean(self): self.vnf['object'].undeploy_vnf() self.orchestrator['object'].undeploy_manager() - super(ImsVnf, self).clean() + super(CloudifyIms, self).clean() def main(self, **kwargs): self.logger.info("Cloudify IMS VNF onboarding test starting") diff --git a/functest/opnfv_tests/vnf/ims/opera_ims.py b/functest/opnfv_tests/vnf/ims/opera_ims.py index 7ead401f..d022b3c7 100755..100644 --- a/functest/opnfv_tests/vnf/ims/opera_ims.py +++ b/functest/opnfv_tests/vnf/ims/opera_ims.py @@ -1,6 +1,6 @@ #!/usr/bin/env python -# Copyright (c) 2016 Orange and others. +# 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 @@ -8,381 +8,117 @@ # http://www.apache.org/licenses/LICENSE-2.0 import json -import socket -import sys +import os import time -import yaml -import functest.core.vnf_base as vnf_base -import functest.utils.functest_logger as ft_logger -import functest.utils.functest_utils as ft_utils -import functest.utils.openstack_utils as os_utils -import os +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 +import functest.utils.functest_logger as ft_logger -from org.openbaton.cli.agents.agents import MainAgent -from org.openbaton.cli.errors.errors import NfvoException +class OperaIms(clearwater_ims_base.ClearwaterOnBoardingBase): -def servertest(host, port): - args = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM) - for family, socktype, proto, canonname, sockaddr in args: - s = socket.socket(family, socktype, proto) + def __init__(self, project='functest', case='opera_ims', + repo=CONST.dir_repo_opera, cmd=''): + super(OperaIms, self).__init__(project, case, repo, cmd) + self.logger = ft_logger.Logger(__name__).getLogger() + 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: - s.connect(sockaddr) - except socket.error: - return False + 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: - s.close() - return True - -# ---------------------------------------------------------- -# -# UTILS -# -# ----------------------------------------------------------- - - -def get_config(parameter, file): - """ - Returns the value of a given parameter in file.yaml - parameter must be given in string format with dots - Example: general.openstack.image_name - """ - with open(file) as f: - file_yaml = yaml.safe_load(f) - f.close() - value = file_yaml - for element in parameter.split("."): - value = value.get(element) - if value is None: - raise ValueError("The parameter %s is not defined in" - " reporting.yaml" % parameter) - return value - - -def download_and_add_image_on_glance(glance, image_name, - image_url, data_dir): - dest_path = data_dir - if not os.path.exists(dest_path): - os.makedirs(dest_path) - file_name = image_url.rsplit('/')[-1] - if not ft_utils.download_url(image_url, dest_path): - return False - image = os_utils.create_glance_image( - glance, image_name, dest_path + file_name) - if not image: - return False - return image - - -class ImsVnf(vnf_base.VnfOnBoardingBase): - - def __init__(self, project='functest', case='orchestra_ims', - repo='', cmd=''): - super(ImsVnf, self).__init__(project, case, repo, cmd) - self.ob_password = "openbaton" - self.ob_username = "admin" - self.ob_https = False - self.ob_port = "8080" - self.ob_ip = "localhost" - self.ob_instance_id = "" - self.logger = ft_logger.Logger("orchestra_ims").getLogger() - self.case_dir = os.path.join(CONST.dir_functest_test, 'vnf/ims/') - self.data_dir = CONST.dir_ims_data - self.test_dir = CONST.dir_repo_vims_test - self.ob_projectid = "" - self.keystone_client = os_utils.get_keystone_client() - self.ob_nsr_id = "" - self.main_agent = None - # vIMS Data directory creation - if not os.path.exists(self.data_dir): - os.makedirs(self.data_dir) - # Retrieve the configuration - try: - self.config = CONST.__getattribute__( - 'vnf_{}_config'.format(self.case_name)) - except: - raise Exception("Orchestra VNF config file not found") - config_file = self.case_dir + self.config - self.imagename = get_config("openbaton.imagename", config_file) - self.market_link = get_config("openbaton.marketplace_link", - config_file) - self.images = get_config("tenant_images", config_file) - - def deploy_orchestrator(self, **kwargs): - self.logger.info("Additional pre-configuration steps") - nova_client = os_utils.get_nova_client() - neutron_client = os_utils.get_neutron_client() - glance_client = os_utils.get_glance_client() - - # needs some images - self.logger.info("Upload some OS images if it doesn't exist") - temp_dir = os.path.join(self.data_dir, "tmp/") - for image_name, image_url in self.images.iteritems(): - self.logger.info("image: %s, url: %s" % (image_name, image_url)) - try: - image_id = os_utils.get_image_id(glance_client, - image_name) - self.logger.info("image_id: %s" % image_id) - except: - self.logger.error("Unexpected error: %s" % sys.exc_info()[0]) - - if image_id == '': - self.logger.info("""%s image doesn't exist on glance repository. Try - downloading this image and upload on glance !""" % image_name) - image_id = download_and_add_image_on_glance(glance_client, - image_name, - image_url, - temp_dir) - if image_id == '': - self.step_failure( - "Failed to find or upload required OS " - "image for this deployment") - network_dic = os_utils.create_network_full(neutron_client, - "openbaton_mgmt", - "openbaton_mgmt_subnet", - "openbaton_router", - "192.168.100.0/24") - - # orchestrator VM flavor - self.logger.info("Check medium Flavor is available, if not create one") - flavor_exist, flavor_id = os_utils.get_or_create_flavor( - "m1.medium", - "4096", - '1', - '2', - public=True) - self.logger.debug("Flavor id: %s" % flavor_id) - - if not network_dic: - self.logger.error("There has been a problem when creating the " - "neutron network") - - network_id = network_dic["net_id"] - - self.logger.info("Creating floating IP for VM in advance...") - floatip_dic = os_utils.create_floating_ip(neutron_client) - floatip = floatip_dic['fip_addr'] - - if floatip is None: - self.logger.error("Cannot create floating IP.") - - userdata = "#!/bin/bash\n" - userdata += "set -x\n" - userdata += "set -e\n" - userdata += "echo \"nameserver 8.8.8.8\" >> /etc/resolv.conf\n" - userdata += "apt-get install curl\n" - userdata += ("echo \"rabbitmq_broker_ip=%s\" > ./config_file\n" - % floatip) - userdata += "echo \"mysql=no\" >> ./config_file\n" - userdata += ("echo \"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCuPXrV3" - "geeHc6QUdyUr/1Z+yQiqLcOskiEGBiXr4z76MK4abiFmDZ18OMQlc" - "fl0p3kS0WynVgyaOHwZkgy/DIoIplONVr2CKBKHtPK+Qcme2PVnCtv" - "EqItl/FcD+1h5XSQGoa+A1TSGgCod/DPo+pes0piLVXP8Ph6QS1k7S" - "ic7JDeRQ4oT1bXYpJ2eWBDMfxIWKZqcZRiGPgMIbJ1iEkxbpeaAd9O" - "4MiM9nGCPESmed+p54uYFjwEDlAJZShcAZziiZYAvMZhvAhe6USljc" - "7YAdalAnyD/jwCHuwIrUw/lxo7UdNCmaUxeobEYyyFA1YVXzpNFZya" - "XPGAAYIJwEq/ openbaton@opnfv\" >> /home/ubuntu/.ssh/aut" - "horized_keys\n") - userdata += "cat ./config_file\n" - userdata += ("curl -s http://get.openbaton.org/bootstrap " - "> ./bootstrap\n") - userdata += "export OPENBATON_COMPONENT_AUTOSTART=false\n" - bootstrap = "sh ./bootstrap release -configFile=./config_file" - userdata += bootstrap + "\n" - - userdata += ("echo \"nfvo.plugin.timeout=300000\" >> " - "/etc/openbaton/openbaton-nfvo.properties\n") - userdata += "service openbaton-nfvo restart\n" - userdata += "service openbaton-vnfm-generic restart\n" - - sg_id = os_utils.create_security_group_full(neutron_client, - "orchestra-sec-group", - "allowall") - - os_utils.create_secgroup_rule(neutron_client, sg_id, "ingress", - "icmp", 0, 255) - os_utils.create_secgroup_rule(neutron_client, sg_id, "egress", - "icmp", 0, 255) - os_utils.create_secgroup_rule(neutron_client, sg_id, "ingress", - "tcp", 1, 65535) - os_utils.create_secgroup_rule(neutron_client, sg_id, "ingress", - "udp", 1, 65535) - os_utils.create_secgroup_rule(neutron_client, sg_id, "egress", - "tcp", 1, 65535) - os_utils.create_secgroup_rule(neutron_client, sg_id, "egress", - "udp", 1, 65535) - - self.logger.info("Security group set") - - self.logger.info("Create instance....") - self.logger.info("flavor: m1.medium\n" - "image: %s\n" - "network_id: %s\n" - "userdata: %s\n" - % (self.imagename, network_id, userdata)) - - instance = os_utils.create_instance_and_wait_for_active( - "m1.medium", - os_utils.get_image_id(glance_client, self.imagename), - network_id, - "orchestra-openbaton", - config_drive=False, - userdata=userdata) + self.logger.info('OPEN-O endpoint is: %s', self.openo_msb_endpoint) - self.ob_instance_id = instance.id + def prepare(self): + pass - self.logger.info("Adding sec group to orchestra instance") - os_utils.add_secgroup_to_instance(nova_client, - self.ob_instance_id, sg_id) - - self.logger.info("Associating floating ip: '%s' to VM '%s' " - % (floatip, "orchestra-openbaton")) - if not os_utils.add_floating_ip(nova_client, instance.id, floatip): - self.logger.error("Cannot associate floating IP to VM.") - self.step_failure("Cannot associate floating IP to VM.") - - self.logger.info("Waiting for nfvo to be up and running...") - x = 0 - while x < 100: - if servertest(floatip, "8080"): - break - else: - self.logger.debug("openbaton is not started yet") - time.sleep(5) - x += 1 - - if x == 100: - self.logger.error("Openbaton is not started correctly") - self.step_failure("Openbaton is not started correctly") - - self.ob_ip = floatip - self.ob_password = "openbaton" - self.ob_username = "admin" - self.ob_https = False - self.ob_port = "8080" - - self.logger.info("Deploy orchestrator: OK") + def clean(self): + pass def deploy_vnf(self): - self.logger.info("vIMS Deployment") - - self.main_agent = MainAgent(nfvo_ip=self.ob_ip, - nfvo_port=self.ob_port, - https=self.ob_https, - version=1, - username=self.ob_username, - password=self.ob_password) - - project_agent = self.main_agent.get_agent("project", self.ob_projectid) - for p in json.loads(project_agent.find()): - if p.get("name") == "default": - self.ob_projectid = p.get("id") - break - - self.logger.debug("project id: %s" % self.ob_projectid) - if self.ob_projectid == "": - self.logger.error("Default project id was not found!") - self.step_failure("Default project id was not found!") - - vim_json = { - "name": "vim-instance", - "authUrl": os_utils.get_credentials().get("auth_url"), - "tenant": os_utils.get_credentials().get("tenant_name"), - "username": os_utils.get_credentials().get("username"), - "password": os_utils.get_credentials().get("password"), - # "keyPair": "opnfv", - # TODO change the keypair to correct value - # or upload a correct one or remove it - "securityGroups": [ - "default", - "orchestra-sec-group" - ], - "type": "openstack", - "location": { - "name": "opnfv", - "latitude": "52.525876", - "longitude": "13.314400" - } - } - - self.logger.debug("vim: %s" % vim_json) - - self.main_agent.get_agent( - "vim", - project_id=self.ob_projectid).create(entity=json.dumps(vim_json)) - - market_agent = self.main_agent.get_agent("market", - project_id=self.ob_projectid) - - nsd = {} - try: - self.logger.info("sending: %s" % self.market_link) - nsd = market_agent.create(entity=self.market_link) - self.logger.info("Onboarded nsd: " + nsd.get("name")) - except NfvoException as e: - self.step_failure(e.message) - - nsr_agent = self.main_agent.get_agent("nsr", - project_id=self.ob_projectid) - nsd_id = nsd.get('id') - if nsd_id is None: - self.step_failure("NSD not onboarded correctly") - - nsr = None try: - nsr = nsr_agent.create(nsd_id) - except NfvoException as e: - self.step_failure(e.message) - - if nsr.get('code') is not None: - self.logger.error( - "vIMS cannot be deployed: %s -> %s" % - (nsr.get('code'), nsr.get('message'))) - self.step_failure("vIMS cannot be deployed") - - i = 0 - self.logger.info("waiting NSR to go to active...") - while nsr.get("status") != 'ACTIVE' and nsr.get("status") != 'ERROR': - i += 1 - if i == 100: - self.step_failure("After %s sec the nsr did not go to active.." - % 5 * 100) - time.sleep(5) - nsr = json.loads(nsr_agent.find(nsr.get('id'))) - - if nsr.get("status") == 'ACTIVE': - deploy_vnf = {'status': "PASS", 'result': nsr} - self.logger.info("Deploy VNF: OK") + 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: - deploy_vnf = {'status': "FAIL", 'result': nsr} - self.logger.error("Deploy VNF: ERROR") - self.step_failure("Deploy vIMS failed") - self.ob_nsr_id = nsr.get("id") - return deploy_vnf + self.logger.info('vIMS deployment is kicked off') + return {'status': 'PASS', 'result': ''} - def test_vnf(self): - # Adaptations probably needed - # code used for cloudify_ims - # ruby client on jumphost calling the vIMS on the SUT - return + 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 clean(self): - self.main_agent.get_agent( - "nsr", - project_id=self.ob_projectid).delete(self.ob_nsr_id) - time.sleep(5) - os_utils.delete_instance(nova_client=os_utils.get_nova_client(), - instance_id=self.ob_instance_id) - # TODO question is the clean removing also the VM? - # I think so since is goinf to remove the tenant... - super(ImsVnf, self).clean() + 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("Orchestra IMS VNF onboarding test starting") + self.logger.info("Start to run Opera vIMS VNF onboarding test") self.execute() - self.logger.info("Orchestra IMS VNF onboarding test executed") + self.logger.info("Opera vIMS VNF onboarding test finished") if self.criteria is "PASS": return self.EX_OK else: @@ -391,10 +127,3 @@ class ImsVnf(vnf_base.VnfOnBoardingBase): def run(self): kwargs = {} return self.main(**kwargs) - - -if __name__ == '__main__': - test = ImsVnf() - test.deploy_orchestrator() - test.deploy_vnf() - test.clean() diff --git a/functest/tests/unit/core/test_testcase_base.py b/functest/tests/unit/core/test_testcase.py index 9ee1609f..32104194 100644 --- a/functest/tests/unit/core/test_testcase_base.py +++ b/functest/tests/unit/core/test_testcase.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +"""Define the classe required to fully cover testcase.""" + import logging import unittest @@ -14,9 +16,14 @@ import mock from functest.core import testcase +__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" + class TestCaseTesting(unittest.TestCase): + """The class testing TestCase.""" + # pylint: disable=missing-docstring + logging.disable(logging.CRITICAL) def setUp(self): diff --git a/functest/tests/unit/odl/test_odl.py b/functest/tests/unit/odl/test_odl.py index 6967eb1a..e08deb27 100644 --- a/functest/tests/unit/odl/test_odl.py +++ b/functest/tests/unit/odl/test_odl.py @@ -7,14 +7,16 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +"""Define the classes required to fully cover odl.""" + import errno import logging -import mock import os import StringIO import unittest from keystoneauth1.exceptions import auth_plugins +import mock from robot.errors import DataError, RobotError from robot.result import testcase as result_testcase from robot.utils.robottime import timestamp_to_secs @@ -22,9 +24,50 @@ from robot.utils.robottime import timestamp_to_secs from functest.core import testcase from functest.opnfv_tests.sdn.odl import odl +__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" + + +class ODLVisitorTesting(unittest.TestCase): + + """The class testing ODLResultVisitor.""" + # pylint: disable=missing-docstring + + logging.disable(logging.CRITICAL) + + def setUp(self): + self.visitor = odl.ODLResultVisitor() + + def test_empty(self): + self.assertFalse(self.visitor.get_data()) + + def test_ok(self): + data = {'name': 'foo', + 'parent': 'bar', + 'status': 'PASS', + 'starttime': "20161216 16:00:00.000", + 'endtime': "20161216 16:00:01.000", + 'elapsedtime': 1000, + 'text': 'Hello, World!', + 'critical': True} + test = result_testcase.TestCase(name=data['name'], + status=data['status'], + message=data['text'], + starttime=data['starttime'], + endtime=data['endtime']) + test.parent = mock.Mock() + config = {'name': data['parent'], + 'criticality.test_is_critical.return_value': data[ + 'critical']} + test.parent.configure_mock(**config) + self.visitor.visit_test(test) + self.assertEqual(self.visitor.get_data(), [data]) + class ODLTesting(unittest.TestCase): + """The super class which testing classes could inherit.""" + # pylint: disable=missing-docstring + logging.disable(logging.CRITICAL) _keystone_ip = "127.0.0.1" @@ -60,39 +103,19 @@ class ODLTesting(unittest.TestCase): 'odlrestconfport': self._odl_restconfport, 'pushtodb': False} - def test_empty_visitor(self): - visitor = odl.ODLResultVisitor() - self.assertFalse(visitor.get_data()) - def test_visitor(self): - visitor = odl.ODLResultVisitor() - data = {'name': 'foo', - 'parent': 'bar', - 'status': 'PASS', - 'starttime': "20161216 16:00:00.000", - 'endtime': "20161216 16:00:01.000", - 'elapsedtime': 1000, - 'text': 'Hello, World!', - 'critical': True} - test = result_testcase.TestCase(name=data['name'], - status=data['status'], - message=data['text'], - starttime=data['starttime'], - endtime=data['endtime']) - test.parent = mock.Mock() - config = {'name': data['parent'], - 'criticality.test_is_critical.return_value': data[ - 'critical']} - test.parent.configure_mock(**config) - visitor.visit_test(test) - self.assertEqual(visitor.get_data(), [data]) +class ODLParseResultTesting(ODLTesting): + + """The class testing ODLTests.parse_results().""" + # pylint: disable=missing-docstring @mock.patch('robot.api.ExecutionResult', side_effect=DataError) - def test_parse_results_raises_exceptions(self, *args): + def test_raises_exc(self, mock_method): with self.assertRaises(DataError): self.test.parse_results() + mock_method.assert_called_once_with() - def test_parse_results(self, *args): + def test_ok(self): config = {'name': 'dummy', 'starttime': '20161216 16:00:00.000', 'endtime': '20161216 16:00:01.000', 'status': 'PASS'} suite = mock.Mock() @@ -108,16 +131,28 @@ class ODLTesting(unittest.TestCase): self.assertEqual(self.test.details, {'description': config['name'], 'tests': []}) + +class ODLRobotTesting(ODLTesting): + + """The class testing ODLTests.set_robotframework_vars().""" + # pylint: disable=missing-docstring + @mock.patch('fileinput.input', side_effect=Exception()) - def test_set_robotframework_vars_failed(self, *args): + def test_set_vars_ko(self, mock_method): self.assertFalse(self.test.set_robotframework_vars()) + mock_method.assert_called_once_with( + os.path.join(odl.ODLTests.odl_test_repo, + 'csit/variables/Variables.py'), inplace=True) @mock.patch('fileinput.input', return_value=[]) - def test_set_robotframework_vars_empty(self, args): + def test_set_vars_empty(self, mock_method): self.assertTrue(self.test.set_robotframework_vars()) + mock_method.assert_called_once_with( + os.path.join(odl.ODLTests.odl_test_repo, + 'csit/variables/Variables.py'), inplace=True) @mock.patch('sys.stdout', new_callable=StringIO.StringIO) - def _test_set_robotframework_vars(self, msg1, msg2, *args): + def _test_set_vars(self, msg1, msg2, *args): line = mock.MagicMock() line.__iter__.return_value = [msg1] with mock.patch('fileinput.input', return_value=line) as mock_method: @@ -127,15 +162,15 @@ class ODLTesting(unittest.TestCase): 'csit/variables/Variables.py'), inplace=True) self.assertEqual(args[0].getvalue(), "{}\n".format(msg2)) - def test_set_robotframework_vars_auth_default(self): - self._test_set_robotframework_vars("AUTH = []", - "AUTH = [u'admin', u'admin']") + def test_set_vars_auth_default(self): + self._test_set_vars("AUTH = []", + "AUTH = [u'admin', u'admin']") - def test_set_robotframework_vars_auth1(self): - self._test_set_robotframework_vars("AUTH1 = []", "AUTH1 = []") + def test_set_vars_auth1(self): + self._test_set_vars("AUTH1 = []", "AUTH1 = []") @mock.patch('sys.stdout', new_callable=StringIO.StringIO) - def test_set_robotframework_vars_auth_foo(self, *args): + def test_set_vars_auth_foo(self, *args): line = mock.MagicMock() line.__iter__.return_value = ["AUTH = []"] with mock.patch('fileinput.input', return_value=line) as mock_method: @@ -146,15 +181,11 @@ class ODLTesting(unittest.TestCase): self.assertEqual(args[0].getvalue(), "AUTH = [u'{}', u'{}']\n".format('foo', 'bar')) - @classmethod - def _fake_url_for(cls, service_type='identity', **kwargs): - if service_type == 'identity': - return "http://{}:5000/v2.0".format( - ODLTesting._keystone_ip) - elif service_type == 'network': - return "http://{}:9696".format(ODLTesting._neutron_ip) - else: - return None + +class ODLMainTesting(ODLTesting): + + """The class testing ODLTests.main().""" + # pylint: disable=missing-docstring def _get_main_kwargs(self, key=None): kwargs = {'odlusername': self._odl_username, @@ -199,50 +230,50 @@ class ODLTesting(unittest.TestCase): args[2].assert_called_with( os.path.join(odl.ODLTests.res_dir, 'stdout.txt')) - def _test_main_missing_keyword(self, key): + def _test_no_keyword(self, key): kwargs = self._get_main_kwargs(key) self.assertEqual(self.test.main(**kwargs), testcase.TestCase.EX_RUN_ERROR) - def test_main_missing_odlusername(self): - self._test_main_missing_keyword('odlusername') + def test_no_odlusername(self): + self._test_no_keyword('odlusername') - def test_main_missing_odlpassword(self): - self._test_main_missing_keyword('odlpassword') + def test_no_odlpassword(self): + self._test_no_keyword('odlpassword') - def test_main_missing_neutronip(self): - self._test_main_missing_keyword('neutronip') + def test_no_neutronip(self): + self._test_no_keyword('neutronip') - def test_main_missing_osauthurl(self): - self._test_main_missing_keyword('osauthurl') + def test_no_osauthurl(self): + self._test_no_keyword('osauthurl') - def test_main_missing_osusername(self): - self._test_main_missing_keyword('osusername') + def test_no_osusername(self): + self._test_no_keyword('osusername') - def test_main_missing_ostenantname(self): - self._test_main_missing_keyword('ostenantname') + def test_no_ostenantname(self): + self._test_no_keyword('ostenantname') - def test_main_missing_ospassword(self): - self._test_main_missing_keyword('ospassword') + def test_no_ospassword(self): + self._test_no_keyword('ospassword') - def test_main_missing_odlip(self): - self._test_main_missing_keyword('odlip') + def test_no_odlip(self): + self._test_no_keyword('odlip') - def test_main_missing_odlwebport(self): - self._test_main_missing_keyword('odlwebport') + def test_no_odlwebport(self): + self._test_no_keyword('odlwebport') - def test_main_missing_odlrestconfport(self): - self._test_main_missing_keyword('odlrestconfport') + def test_no_odlrestconfport(self): + self._test_no_keyword('odlrestconfport') - def test_main_set_robotframework_vars_failed(self): + def test_set_vars_ko(self): with mock.patch.object(self.test, 'set_robotframework_vars', - return_value=False): + return_value=False) as mock_object: self._test_main(testcase.TestCase.EX_RUN_ERROR) - self.test.set_robotframework_vars.assert_called_once_with( + mock_object.assert_called_once_with( self._odl_username, self._odl_password) @mock.patch('os.makedirs', side_effect=Exception) - def test_main_makedirs_exception(self, mock_method): + def test_makedirs_exc(self, mock_method): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ self.assertRaises(Exception): @@ -250,7 +281,7 @@ class ODLTesting(unittest.TestCase): mock_method) @mock.patch('os.makedirs', side_effect=OSError) - def test_main_makedirs_oserror(self, mock_method): + def test_makedirs_oserror(self, mock_method): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True): self._test_main(testcase.TestCase.EX_RUN_ERROR, @@ -258,7 +289,7 @@ class ODLTesting(unittest.TestCase): @mock.patch('robot.run', side_effect=RobotError) @mock.patch('os.makedirs') - def test_main_robot_run_failed(self, *args): + def test_run_ko(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ mock.patch.object(odl, 'open', mock.mock_open(), @@ -268,7 +299,7 @@ class ODLTesting(unittest.TestCase): @mock.patch('robot.run') @mock.patch('os.makedirs') - def test_main_parse_results_failed(self, *args): + def test_parse_results_ko(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ mock.patch.object(odl, 'open', mock.mock_open(), @@ -280,7 +311,7 @@ class ODLTesting(unittest.TestCase): @mock.patch('os.remove', side_effect=Exception) @mock.patch('robot.run') @mock.patch('os.makedirs') - def test_main_remove_exception(self, *args): + def test_remove_exc(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ mock.patch.object(self.test, 'parse_results'), \ @@ -290,7 +321,7 @@ class ODLTesting(unittest.TestCase): @mock.patch('os.remove') @mock.patch('robot.run') @mock.patch('os.makedirs') - def test_main(self, *args): + def test_ok(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ mock.patch.object(odl, 'open', mock.mock_open(), @@ -301,7 +332,7 @@ class ODLTesting(unittest.TestCase): @mock.patch('os.remove') @mock.patch('robot.run') @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, '')) - def test_main_makedirs_oserror17(self, *args): + def test_makedirs_oserror17(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ mock.patch.object(odl, 'open', mock.mock_open(), @@ -312,7 +343,7 @@ class ODLTesting(unittest.TestCase): @mock.patch('os.remove') @mock.patch('robot.run', return_value=1) @mock.patch('os.makedirs') - def test_main_testcases_in_failure(self, *args): + def test_testcases_in_failure(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ mock.patch.object(odl, 'open', mock.mock_open(), @@ -323,7 +354,7 @@ class ODLTesting(unittest.TestCase): @mock.patch('os.remove', side_effect=OSError) @mock.patch('robot.run') @mock.patch('os.makedirs') - def test_main_remove_oserror(self, *args): + def test_remove_oserror(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ mock.patch.object(odl, 'open', mock.mock_open(), @@ -331,7 +362,23 @@ class ODLTesting(unittest.TestCase): mock.patch.object(self.test, 'parse_results'): self._test_main(testcase.TestCase.EX_OK, *args) - def _test_run_missing_env_var(self, var): + +class ODLRunTesting(ODLTesting): + + """The class testing ODLTests.run().""" + # pylint: disable=missing-docstring + + @classmethod + def _fake_url_for(cls, service_type='identity'): + if service_type == 'identity': + return "http://{}:5000/v2.0".format( + ODLTesting._keystone_ip) + elif service_type == 'network': + return "http://{}:9696".format(ODLTesting._neutron_ip) + else: + return None + + def _test_no_env_var(self, var): with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=self._fake_url_for): del os.environ[var] @@ -339,8 +386,12 @@ class ODLTesting(unittest.TestCase): testcase.TestCase.EX_RUN_ERROR) def _test_run(self, status=testcase.TestCase.EX_OK, - exception=None, odlip="127.0.0.3", odlwebport="8080", - odlrestconfport="8181"): + exception=None, **kwargs): + odlip = kwargs['odlip'] if 'odlip' in kwargs else '127.0.0.3' + odlwebport = kwargs['odlwebport'] if 'odlwebport' in kwargs else '8080' + odlrestconfport = (kwargs['odlrestconfport'] + if 'odlrestconfport' in kwargs else '8181') + with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=self._fake_url_for): if exception: @@ -358,11 +409,13 @@ class ODLTesting(unittest.TestCase): ospassword=self._os_password, ostenantname=self._os_tenantname, osusername=self._os_username) - def _test_run_defining_multiple_suites( - self, suites, - status=testcase.TestCase.EX_OK, - exception=None, odlip="127.0.0.3", odlwebport="8080", - odlrestconfport="8181"): + def _test_multiple_suites(self, suites, + status=testcase.TestCase.EX_OK, + exception=None, **kwargs): + odlip = kwargs['odlip'] if 'odlip' in kwargs else '127.0.0.3' + odlwebport = kwargs['odlwebport'] if 'odlwebport' in kwargs else '8080' + odlrestconfport = (kwargs['odlrestconfport'] + if 'odlrestconfport' in kwargs else '8181') with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=self._fake_url_for): if exception: @@ -380,31 +433,31 @@ class ODLTesting(unittest.TestCase): ospassword=self._os_password, ostenantname=self._os_tenantname, osusername=self._os_username) - def test_run_exception(self): + def test_exc(self): with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=auth_plugins.MissingAuthPlugin()): self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR) - def test_run_missing_os_auth_url(self): - self._test_run_missing_env_var("OS_AUTH_URL") + def test_no_os_auth_url(self): + self._test_no_env_var("OS_AUTH_URL") - def test_run_missing_os_username(self): - self._test_run_missing_env_var("OS_USERNAME") + def test_no_os_username(self): + self._test_no_env_var("OS_USERNAME") - def test_run_missing_os_password(self): - self._test_run_missing_env_var("OS_PASSWORD") + def test_no_os_password(self): + self._test_no_env_var("OS_PASSWORD") - def test_run_missing_os_tenant_name(self): - self._test_run_missing_env_var("OS_TENANT_NAME") + def test_no_os_tenant_name(self): + self._test_no_env_var("OS_TENANT_NAME") - def test_run_main_false(self): + def test_main_false(self): os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip self._test_run(testcase.TestCase.EX_RUN_ERROR, odlip=self._sdn_controller_ip, odlwebport=self._odl_webport) - def test_run_main_exception(self): + def test_main_exc(self): with self.assertRaises(Exception): os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip self._test_run(status=testcase.TestCase.EX_RUN_ERROR, @@ -412,147 +465,155 @@ class ODLTesting(unittest.TestCase): odlip=self._sdn_controller_ip, odlwebport=self._odl_webport) - def test_run_missing_sdn_controller_ip(self): + def test_no_sdn_controller_ip(self): with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=self._fake_url_for): self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR) - def test_run_without_installer_type(self): + def test_without_installer_type(self): os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip self._test_run(testcase.TestCase.EX_OK, odlip=self._sdn_controller_ip, odlwebport=self._odl_webport) - def test_run_redefining_suites(self): + def test_suites(self): os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip - self._test_run_defining_multiple_suites( + self._test_multiple_suites( [odl.ODLTests.basic_suite_dir], testcase.TestCase.EX_OK, odlip=self._sdn_controller_ip, odlwebport=self._odl_webport) - def test_run_fuel(self): + def test_fuel(self): os.environ["INSTALLER_TYPE"] = "fuel" self._test_run(testcase.TestCase.EX_OK, odlip=self._neutron_ip, odlwebport='8282') - def test_run_apex_missing_sdn_controller_ip(self): + def test_apex_no_controller_ip(self): with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=self._fake_url_for): os.environ["INSTALLER_TYPE"] = "apex" self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR) - def test_run_apex(self): + def test_apex(self): os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip os.environ["INSTALLER_TYPE"] = "apex" self._test_run(testcase.TestCase.EX_OK, odlip=self._sdn_controller_ip, odlwebport='8081', odlrestconfport='8081') - def test_run_netvirt_missing_sdn_controller_ip(self): + def test_netvirt_no_controller_ip(self): with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=self._fake_url_for): os.environ["INSTALLER_TYPE"] = "netvirt" self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR) - def test_run_netvirt(self): + def test_netvirt(self): os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip os.environ["INSTALLER_TYPE"] = "netvirt" self._test_run(testcase.TestCase.EX_OK, odlip=self._sdn_controller_ip, odlwebport='8081', odlrestconfport='8081') - def test_run_joid_missing_sdn_controller(self): + def test_joid_no_controller_ip(self): with mock.patch('functest.utils.openstack_utils.get_endpoint', side_effect=self._fake_url_for): os.environ["INSTALLER_TYPE"] = "joid" self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR) - def test_run_joid(self): + def test_joid(self): os.environ["SDN_CONTROLLER"] = self._sdn_controller_ip os.environ["INSTALLER_TYPE"] = "joid" self._test_run(testcase.TestCase.EX_OK, odlip=self._sdn_controller_ip, odlwebport='8080') - def test_run_compass(self, *args): + def test_compass(self): os.environ["INSTALLER_TYPE"] = "compass" self._test_run(testcase.TestCase.EX_OK, odlip=self._neutron_ip, odlwebport='8181') - def test_argparser_default(self): - parser = odl.ODLParser() - self.assertEqual(parser.parse_args(), self.defaultargs) - def test_argparser_basic(self): +class ODLArgParserTesting(ODLTesting): + + """The class testing ODLParser.""" + # pylint: disable=missing-docstring + + def setUp(self): + self.parser = odl.ODLParser() + super(ODLArgParserTesting, self).setUp() + + def test_default(self): + self.assertEqual(self.parser.parse_args(), self.defaultargs) + + def test_basic(self): self.defaultargs['neutronip'] = self._neutron_ip self.defaultargs['odlip'] = self._sdn_controller_ip - parser = odl.ODLParser() - self.assertEqual(parser.parse_args( - ["--neutronip={}".format(self._neutron_ip), - "--odlip={}".format(self._sdn_controller_ip) - ]), self.defaultargs) + self.assertEqual( + self.parser.parse_args( + ["--neutronip={}".format(self._neutron_ip), + "--odlip={}".format(self._sdn_controller_ip)]), + self.defaultargs) @mock.patch('sys.stderr', new_callable=StringIO.StringIO) - def test_argparser_fail(self, *args): + def test_fail(self, mock_method): self.defaultargs['foo'] = 'bar' - parser = odl.ODLParser() with self.assertRaises(SystemExit): - parser.parse_args(["--foo=bar"]) + self.parser.parse_args(["--foo=bar"]) + mock_method.assert_called_once_with() - def _test_argparser(self, arg, value): + def _test_arg(self, arg, value): self.defaultargs[arg] = value - parser = odl.ODLParser() - self.assertEqual(parser.parse_args(["--{}={}".format(arg, value)]), - self.defaultargs) + self.assertEqual( + self.parser.parse_args(["--{}={}".format(arg, value)]), + self.defaultargs) - def test_argparser_odlusername(self): - self._test_argparser('odlusername', 'foo') + def test_odlusername(self): + self._test_arg('odlusername', 'foo') - def test_argparser_odlpassword(self): - self._test_argparser('odlpassword', 'foo') + def test_odlpassword(self): + self._test_arg('odlpassword', 'foo') - def test_argparser_osauthurl(self): - self._test_argparser('osauthurl', 'http://127.0.0.4:5000/v2') + def test_osauthurl(self): + self._test_arg('osauthurl', 'http://127.0.0.4:5000/v2') - def test_argparser_neutronip(self): - self._test_argparser('neutronip', '127.0.0.4') + def test_neutronip(self): + self._test_arg('neutronip', '127.0.0.4') - def test_argparser_osusername(self): - self._test_argparser('osusername', 'foo') + def test_osusername(self): + self._test_arg('osusername', 'foo') - def test_argparser_ostenantname(self): - self._test_argparser('ostenantname', 'foo') + def test_ostenantname(self): + self._test_arg('ostenantname', 'foo') - def test_argparser_ospassword(self): - self._test_argparser('ospassword', 'foo') + def test_ospassword(self): + self._test_arg('ospassword', 'foo') - def test_argparser_odlip(self): - self._test_argparser('odlip', '127.0.0.4') + def test_odlip(self): + self._test_arg('odlip', '127.0.0.4') - def test_argparser_odlwebport(self): - self._test_argparser('odlwebport', '80') + def test_odlwebport(self): + self._test_arg('odlwebport', '80') - def test_argparser_odlrestconfport(self): - self._test_argparser('odlrestconfport', '80') + def test_odlrestconfport(self): + self._test_arg('odlrestconfport', '80') - def test_argparser_pushtodb(self): + def test_pushtodb(self): self.defaultargs['pushtodb'] = True - parser = odl.ODLParser() - self.assertEqual(parser.parse_args(["--{}".format('pushtodb')]), + self.assertEqual(self.parser.parse_args(["--{}".format('pushtodb')]), self.defaultargs) - def test_argparser_multiple_args(self): + def test_multiple_args(self): self.defaultargs['neutronip'] = self._neutron_ip self.defaultargs['odlip'] = self._sdn_controller_ip - parser = odl.ODLParser() - self.assertEqual(parser.parse_args( - ["--neutronip={}".format(self._neutron_ip), - "--odlip={}".format(self._sdn_controller_ip) - ]), self.defaultargs) + self.assertEqual( + self.parser.parse_args( + ["--neutronip={}".format(self._neutron_ip), + "--odlip={}".format(self._sdn_controller_ip)]), + self.defaultargs) if __name__ == "__main__": diff --git a/functest/tests/unit/opnfv_tests/vnf/ims/test_cloudify_ims.py b/functest/tests/unit/opnfv_tests/vnf/ims/test_cloudify_ims.py index e25816f0..f47ea865 100644 --- a/functest/tests/unit/opnfv_tests/vnf/ims/test_cloudify_ims.py +++ b/functest/tests/unit/opnfv_tests/vnf/ims/test_cloudify_ims.py @@ -13,7 +13,7 @@ import mock from functest.opnfv_tests.vnf.ims import cloudify_ims -class ImsVnfTesting(unittest.TestCase): +class CloudifyImsTesting(unittest.TestCase): logging.disable(logging.CRITICAL) @@ -22,7 +22,7 @@ class ImsVnfTesting(unittest.TestCase): 'os.makedirs'), \ mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' 'get_config', return_value='config_value'): - self.ims_vnf = cloudify_ims.ImsVnf() + self.ims_vnf = cloudify_ims.CloudifyIms() self.neutron_client = mock.Mock() self.glance_client = mock.Mock() self.keystone_client = mock.Mock() @@ -419,94 +419,46 @@ class ImsVnfTesting(unittest.TestCase): self.ims_vnf.test_vnf() self.assertTrue(msg in context.exception) - def test_test_vnf_create_number_failure(self): - with mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'os.popen') as m, \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'requests.get'), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'requests.post', - return_value=self.mock_post), \ - self.assertRaises(Exception) as context: - mock_obj = mock.Mock() - attrs = {'read.return_value': 'test_ip\n'} - mock_obj.configure_mock(**attrs) - m.return_value = mock_obj - - self.ims_vnf.test_vnf() - - msg = "Unable to create a number:" - self.assertTrue(msg in context.exception) - - def _get_post_status(self, url, cookies='', data=''): - ellis_url = "http://test_ellis_ip/session" - if url == ellis_url: - return self.mock_post_200 - return self.mock_post - def test_test_vnf_fail(self): with mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'os.popen') as m, \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'requests.get') as mock_get, \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'requests.post', - side_effect=self._get_post_status), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'ft_utils.get_resolvconf_ns'), \ - mock.patch('__builtin__.open', mock.mock_open()), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'subprocess.call'), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'os.remove'), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'json.load', return_value=''): - mock_obj = mock.Mock() - attrs = {'read.return_value': 'test_ip\n'} - mock_obj.configure_mock(**attrs) - m.return_value = mock_obj + 'os.popen'), \ + mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' + 'requests.get') as mock_get, \ + mock.patch.object(self.ims_vnf, 'config_ellis'), \ + mock.patch.object(self.ims_vnf, + 'run_clearwater_live_test') as clearwater_obj: + clearwater_obj.return_value = '' - mock_obj2 = mock.Mock() - attrs = {'json.return_value': {'outputs': - {'dns_ip': 'test_dns_ip', - 'ellis_ip': 'test_ellis_ip'}}} - mock_obj2.configure_mock(**attrs) - mock_get.return_value = mock_obj2 + mock_obj2 = mock.Mock() + attrs = {'json.return_value': {'outputs': + {'dns_ip': 'test_dns_ip', + 'ellis_ip': 'test_ellis_ip'}}} + mock_obj2.configure_mock(**attrs) + mock_get.return_value = mock_obj2 - self.assertEqual(self.ims_vnf.test_vnf(), - {'status': 'FAIL', 'result': ''}) + self.assertEqual(self.ims_vnf.test_vnf(), + {'status': 'FAIL', 'result': ''}) def test_test_vnf_pass(self): with mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'os.popen') as m, \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'requests.get') as mock_get, \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'requests.post', - side_effect=self._get_post_status), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'ft_utils.get_resolvconf_ns'), \ - mock.patch('__builtin__.open', mock.mock_open()), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'subprocess.call'), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'os.remove'), \ - mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' - 'json.load', return_value='vims_test_result'): - mock_obj = mock.Mock() - attrs = {'read.return_value': 'test_ip\n'} - mock_obj.configure_mock(**attrs) - m.return_value = mock_obj - - mock_obj2 = mock.Mock() - attrs = {'json.return_value': {'outputs': - {'dns_ip': 'test_dns_ip', - 'ellis_ip': 'test_ellis_ip'}}} - mock_obj2.configure_mock(**attrs) - mock_get.return_value = mock_obj2 - - self.assertEqual(self.ims_vnf.test_vnf(), - {'status': 'PASS', 'result': 'vims_test_result'}) + 'os.popen'), \ + mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' + 'requests.get') as mock_get, \ + mock.patch.object(self.ims_vnf, 'config_ellis'), \ + mock.patch.object(self.ims_vnf, + 'run_clearwater_live_test') as clearwater_obj: + clearwater_obj.return_value = 'vims_test_result' + + mock_obj2 = mock.Mock() + attrs = {'json.return_value': {'outputs': + {'dns_ip': 'test_dns_ip', + 'ellis_ip': 'test_ellis_ip'}}} + mock_obj2.configure_mock(**attrs) + mock_get.return_value = mock_obj2 + + self.assertEqual(self.ims_vnf.test_vnf(), + {'status': 'PASS', + 'result': 'vims_test_result'}) def test_download_and_add_image_on_glance_incorrect_url(self): with mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' diff --git a/functest/tests/unit/opnfv_tests/vnf/ims/test_ims_base.by b/functest/tests/unit/opnfv_tests/vnf/ims/test_ims_base.by new file mode 100644 index 00000000..9440bcdf --- /dev/null +++ b/functest/tests/unit/opnfv_tests/vnf/ims/test_ims_base.by @@ -0,0 +1,58 @@ +#!/usr/bin/env python + +# 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 logging +import unittest + +import mock + +from functest.opnfv_tests.vnf.ims import ims_base + + +class ClearwaterOnBoardingBaseTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + def setUp(self): + with mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.' + 'os.makedirs'): + self.ims_vnf = ims_base.ClearwaterOnBoardingBase() + + self.mock_post = mock.Mock() + attrs = {'status_code': 201, + 'cookies': ""} + self.mock_post.configure_mock(**attrs) + + self.mock_post_200 = mock.Mock() + attrs = {'status_code': 200, + 'cookies': ""} + self.mock_post_200.configure_mock(**attrs) + + self.mock_post_500 = mock.Mock() + attrs = {'status_code': 500, + 'cookies': ""} + self.mock_post_200.configure_mock(**attrs) + + def test_create_ellis_number_failure(self): + with mock.patch('functest.opnfv_tests.vnf.ims.ims_base.' + 'requests.post', + return_value=self.mock_post_500), \ + self.assertRaises(Exception) as context: + self.ims_vnf.create_ellis_number() + + msg = "Unable to create a number:" + self.assertTrue(msg in context.exception) + + def _get_post_status(self, url, cookies='', data=''): + ellis_url = "http://test_ellis_ip/session" + if url == ellis_url: + return self.mock_post_200 + return self.mock_post + + +if __name__ == "__main__": + unittest.main(verbosity=2) |