RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/snaps ${REPOS_DIR}/snaps
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/securityscanning ${REPOS_DIR}/securityscanning
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/releng ${REPOS_DIR}/releng
+RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/opera ${REPOS_DIR}/opera
# OpenStack repositories
RUN git clone --depth 1 -b $OPENSTACK_TAG https://github.com/openstack/networking-bgpvpn ${REPOS_DIR}/bgpvpn
RUN cd ${REPOS_DIR}/barometer \
&& pip install .
+RUN pip install ${REPOS_DIR}/opera
+
RUN find ${FUNCTEST_REPO_DIR} -name "*.py" \
-not -path "*tests/unit*" \
-not -path "*functest_venv*" \
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/snaps ${REPOS_DIR}/snaps
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/securityscanning ${REPOS_DIR}/securityscanning
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/releng ${REPOS_DIR}/releng
+RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/opera ${REPOS_DIR}/opera
# OpenStack repositories
RUN git clone --depth 1 -b $OPENSTACK_TAG https://github.com/openstack/networking-bgpvpn ${REPOS_DIR}/bgpvpn
# SNAPS integration
RUN pip install -e ${REPOS_DIR}/snaps/
+RUN pip install ${REPOS_DIR}/opera
+
# SFC integration
RUN /bin/bash -c ". ${REPOS_DIR}/sfc/sfc/tests/functest/setup_scripts/tacker_client_install.sh"
RUN cd ${REPOS_DIR}/sfc && pip install -e .
+-----------------------------------------+
| |
- | Testcase_base |
+ | TestCase |
| |
| - init() |
| - run() |
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
scenario: '(ocl)|(nosdn)|^(os-odl)((?!bgpvpn).)*$'
run:
module: 'functest.opnfv_tests.vnf.ims.cloudify_ims'
- class: 'ImsVnf'
+ class: 'CloudifyIms'
# -
# name: aaa
# criteria: 'ret == 0'
-
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
# 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):
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'] = {}
'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):
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)
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
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)
# 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
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 = []
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")
@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:
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()
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
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:
class ODLParser(object): # pylint: disable=too-few-public-methods
+ """Parser to run ODL test suites."""
def __init__(self):
self.parser = argparse.ArgumentParser()
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))
--- /dev/null
+#!/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
# 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),
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")
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 == '':
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(
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:
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")
#!/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
# 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:
def run(self):
kwargs = {}
return self.main(**kwargs)
-
-
-if __name__ == '__main__':
- test = ImsVnf()
- test.deploy_orchestrator()
- test.deploy_vnf()
- test.clean()
# 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
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):
# 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
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"
'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()
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:
'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:
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,
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):
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,
@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(),
@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(),
@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'), \
@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(),
@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(),
@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(),
@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(),
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]
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:
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:
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,
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__":
from functest.opnfv_tests.vnf.ims import cloudify_ims
-class ImsVnfTesting(unittest.TestCase):
+class CloudifyImsTesting(unittest.TestCase):
logging.disable(logging.CRITICAL)
'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()
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.'
--- /dev/null
+#!/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)
Code Review / functest.git / commitdiff