Valentin Boucher <valentin.boucher@orange.com>
Viktor Tikkanen <viktor.tikkanen@nokia.com>
Mei Mei <meimei@huawei.com>
+Linda Wang <wangwulin@huawei.com>
Additional contributors:
-Linda Wang <wangwulin@huawei.com>
Georgios Paraskevopoulos <georgepar.91@gmail.com>
Romanos Skiadas <rom.skiad@gmail.com>
Michael Polenchuk <mpolenchuk@mirantis.com>
RUN pip install --upgrade pip && easy_install -U setuptools==30.0.0
-RUN mkdir -p ${REPOS_DIR} \
+RUN mkdir -p ${FUNCTEST_REPO_DIR} \
&& mkdir -p ${REPOS_VNFS_DIR} \
&& mkdir -p ${FUNCTEST_BASE_DIR}/results \
&& mkdir -p ${FUNCTEST_CONF_DIR} \
RUN git config --global http.sslVerify false
+COPY thirdparty-requirements.txt thirdparty-requirements.txt
+RUN pip install \
+ git+https://gerrit.opnfv.org/gerrit/functest@$BRANCH#egg=functest \
+ git+https://gerrit.opnfv.org/gerrit/releng@$BRANCH#egg=opnfv\&subdirectory=modules \
+ git+https://gerrit.opnfv.org/gerrit/snaps@$BRANCH#egg=snaps \
+ -r thirdparty-requirements.txt \
+ && rm thirdparty-requirements.txt
+
+# Hook required by hardcoded paths in Functest
+RUN ln -s /usr/local/lib/python2.7/dist-packages/functest ${FUNCTEST_REPO_DIR}/functest
+
# OPNFV repositories
-RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/functest ${REPOS_DIR}/functest
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/copper ${REPOS_DIR}/copper
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/sdnvpn ${REPOS_DIR}/sdnvpn
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/domino ${REPOS_DIR}/domino
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/securityscanning ${REPOS_DIR}/securityscanning
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 git clone --depth 1 -b $KINGBIRD_TAG https://github.com/openstack/kingbird.git ${REPOS_DIR}/kingbird
-RUN git clone --depth 1 -b $RALLY_TAG https://github.com/openstack/rally.git ${REPOS_DIR}/rally
-RUN git clone --depth 1 -b $TEMPEST_TAG https://github.com/openstack/tempest.git ${REPOS_DIR}/tempest
-RUN git clone https://github.com/openstack/refstack-client ${REPOS_DIR}/refstack-client
-
# other repositories
+RUN git clone --depth 1 -b $TEMPEST_TAG https://github.com/openstack/tempest.git ${REPOS_DIR}/tempest
RUN git clone --depth 1 -b $ODL_TAG https://git.opendaylight.org/gerrit/p/integration/test.git ${REPOS_DIR}/odl_test
RUN git clone --depth 1 -b $VIMS_TAG https://github.com/boucherv-orange/clearwater-live-test ${REPOS_VNFS_DIR}/vims-test
RUN git clone --depth 1 -b $VROUTER_TAG https://github.com/oolorg/opnfv-functest-vrouter.git ${REPOS_VNFS_DIR}/vrouter
RUN git clone --depth 1 https://github.com/wuwenbin2/OnosSystemTest.git ${REPOS_DIR}/onos
-RUN pip install -r ${REPOS_DIR}/rally/requirements.txt
-RUN pip install -r ${REPOS_DIR}/tempest/requirements.txt
-
-RUN cd ${FUNCTEST_REPO_DIR} \
- && pip install -r requirements.txt \
- && pip install -e .
-
RUN pip install ${REPOS_DIR}/opera
-RUN find ${FUNCTEST_REPO_DIR} -name "*.py" \
+RUN find -L ${FUNCTEST_REPO_DIR} -name "*.py" \
-not -path "*tests/unit*" \
-not -path "*functest_venv*" \
|xargs grep -L __main__ |cut -d\: -f 1 |xargs chmod -c 644 \
- && find ${FUNCTEST_REPO_DIR} -name "*.sh" \
+ && find -L ${FUNCTEST_REPO_DIR} -name "*.sh" \
-not -path "*functest_venv*" \
|xargs grep -L \#\! |cut -d\: -f 1 |xargs chmod -c 644
-RUN find ${FUNCTEST_REPO_DIR} -name "*.py" \
+RUN find -L ${FUNCTEST_REPO_DIR} -name "*.py" \
-not -path "*tests/unit*" \
-not -path "*functest_venv*" \
|xargs grep __main__ |cut -d\: -f 1 |xargs chmod -c 755 \
- && find ${FUNCTEST_REPO_DIR} -name "*.sh" \
+ && find -L ${FUNCTEST_REPO_DIR} -name "*.sh" \
-not -path "*functest_venv*" \
|xargs grep \#\! |cut -d\: -f 1 |xargs chmod -c 755
RUN /bin/bash ${REPOS_DIR}/parser/tests/parser_install.sh ${REPOS_DIR}
-RUN ${REPOS_DIR}/rally/install_rally.sh --yes
+RUN wget -q https://raw.githubusercontent.com/openstack/rally/${RALLY_TAG}/install_rally.sh \
+ && bash install_rally.sh --branch ${RALLY_TAG} --yes && rm install_rally.sh
-RUN /bin/bash ${REPOS_DIR}/functest/docker/add_images.sh
+RUN add_images.sh
RUN gpg --keyserver hkp://p80.pool.sks-keyservers.net:80 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3
RUN curl -L https://get.rvm.io | bash -s stable
# SDNVPN integration
RUN cd ${REPOS_DIR}/sdnvpn && pip install -e .
-RUN cd ${REPOS_DIR}/bgpvpn && pip install -e .
-
-# Kingbird integration
-RUN cd ${REPOS_DIR}/kingbird && pip install -e .
-RUN cd ${FUNCTEST_REPO_DIR} && pip install -r kingbird_requirements.txt
# refstack-client integration
-RUN cd ${REPOS_DIR}/refstack-client && ./setup_env -t ${REFSTACK_TAG}
+RUN wget -q https://raw.githubusercontent.com/openstack/refstack-client/master/setup_env \
+ && bash setup_env -t ${REFSTACK_TAG} && rm setup_env
RUN /bin/bash -c ". /etc/profile.d/rvm.sh \
&& cd ${REPOS_VNFS_DIR}/vims-test \
RUN pip install --upgrade pip && easy_install -U setuptools==30.0.0
-RUN mkdir -p ${REPOS_DIR} \
+RUN mkdir -p ${FUNCTEST_REPO_DIR} \
&& mkdir -p ${REPOS_VNFS_DIR} \
&& mkdir -p ${FUNCTEST_BASE_DIR}/results \
&& mkdir -p ${FUNCTEST_BASE_DIR}/conf \
RUN git config --global http.sslVerify false
+COPY thirdparty-requirements.txt thirdparty-requirements.txt
+RUN pip install \
+ git+https://gerrit.opnfv.org/gerrit/functest@$BRANCH#egg=functest \
+ git+https://gerrit.opnfv.org/gerrit/releng@$BRANCH#egg=opnfv\&subdirectory=modules \
+ git+https://gerrit.opnfv.org/gerrit/snaps@$BRANCH#egg=snaps \
+ -r thirdparty-requirements.txt && rm thirdparty-requirements.txt
+
# OPNFV repositories
-RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/functest ${REPOS_DIR}/functest
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/copper ${REPOS_DIR}/copper
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/sdnvpn ${REPOS_DIR}/sdnvpn
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/domino ${REPOS_DIR}/domino
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/securityscanning ${REPOS_DIR}/securityscanning
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 git clone --depth 1 -b $KINGBIRD_TAG https://github.com/openstack/kingbird.git ${REPOS_DIR}/kingbird
-RUN git clone --depth 1 -b $RALLY_TAG https://github.com/openstack/rally.git ${REPOS_DIR}/rally
-RUN git clone --depth 1 -b $TEMPEST_TAG https://github.com/openstack/tempest.git ${REPOS_DIR}/tempest
-RUN git clone https://github.com/openstack/refstack-client ${REPOS_DIR}/refstack-client
-
# other repositories
+RUN git clone --depth 1 -b $TEMPEST_TAG https://github.com/openstack/tempest.git ${REPOS_DIR}/tempest
RUN git clone --depth 1 -b $ODL_TAG https://git.opendaylight.org/gerrit/p/integration/test.git ${REPOS_DIR}/odl_test
RUN git clone --depth 1 -b $VIMS_TAG https://github.com/boucherv-orange/clearwater-live-test ${REPOS_VNFS_DIR}/vims-test
RUN git clone --depth 1 https://github.com/wuwenbin2/OnosSystemTest.git ${REPOS_DIR}/onos
-RUN pip install -r ${REPOS_DIR}/rally/requirements.txt
-RUN pip install -r ${REPOS_DIR}/tempest/requirements.txt
-
-RUN cd ${FUNCTEST_REPO_DIR} \
- && pip install -r requirements.txt \
- && pip install -e .
-
-RUN find ${FUNCTEST_REPO_DIR} -name "*.py" \
+RUN find -L ${FUNCTEST_REPO_DIR} -name "*.py" \
-not -path "*tests/unit*" \
-not -path "*functest_venv*" \
|xargs grep -L __main__ |cut -d\: -f 1 |xargs chmod -c 644 \
- && find ${FUNCTEST_REPO_DIR} -name "*.sh" \
+ && find -L ${FUNCTEST_REPO_DIR} -name "*.sh" \
-not -path "*functest_venv*" \
|xargs grep -L \#\! |cut -d\: -f 1 |xargs chmod -c 644
-RUN find ${FUNCTEST_REPO_DIR} -name "*.py" \
+RUN find -L ${FUNCTEST_REPO_DIR} -name "*.py" \
-not -path "*tests/unit*" \
-not -path "*functest_venv*" \
|xargs grep __main__ |cut -d\: -f 1 |xargs chmod -c 755 \
- && find ${FUNCTEST_REPO_DIR} -name "*.sh" \
+ && find -L ${FUNCTEST_REPO_DIR} -name "*.sh" \
-not -path "*functest_venv*" \
|xargs grep \#\! |cut -d\: -f 1 |xargs chmod -c 755
RUN /bin/bash ${REPOS_DIR}/parser/tests/parser_install.sh ${REPOS_DIR}
-RUN ${REPOS_DIR}/rally/install_rally.sh --yes
+RUN wget -q https://raw.githubusercontent.com/openstack/rally/${RALLY_TAG}/install_rally.sh \
+ && bash install_rally.sh --branch ${RALLY_TAG} --yes && rm install_rally.sh
-RUN /bin/bash ${REPOS_DIR}/functest/docker/add_images.sh
+RUN add_images.sh
RUN gpg --keyserver hkp://p80.pool.sks-keyservers.net:80 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3
RUN curl -L https://get.rvm.io | bash -s stable
# SDNVPN integration
RUN cd ${REPOS_DIR}/sdnvpn && pip install -e .
-RUN cd ${REPOS_DIR}/bgpvpn && pip install -e .
-
-# Kingbird integration
-RUN cd ${REPOS_DIR}/kingbird && pip install -e .
# refstack-client integration
-RUN cd ${REPOS_DIR}/refstack-client && ./setup_env -t ${REFSTACK_TAG}
+RUN wget -q https://raw.githubusercontent.com/openstack/refstack-client/master/setup_env \
+ && bash setup_env -t ${REFSTACK_TAG} && rm setup_env
RUN /bin/bash -c ". /etc/profile.d/rvm.sh \
&& cd ${REPOS_VNFS_DIR}/vims-test \
--- /dev/null
+git+https://gerrit.opnfv.org/gerrit/barometer#egg=baro_tests
+networking-bgpvpn
+kingbird
+rally
+tempest>=15.0.0 # Apache-2.0
+git+https://github.com/openstack/refstack-client#egg=refstack-client
general:
dir:
- # Relative to the path where the repo is cloned:
- vping: functest/opnfv_tests/openstack/vping
- dir_odl: functest/opnfv_tests/sdn/odl
- rally: functest/opnfv_tests/openstack/rally
- tempest_cases: functest/opnfv_tests/openstack/tempest/custom_tests
- onos: functest/opnfv_tests/sdn/onos/teston
- onos_sfc: functest/opnfv_tests/sdn/onos/sfc
-
- # Absolute path
home: /home/opnfv
repos: /home/opnfv/repos
- repo_functest: /home/opnfv/repos/functest
dir_repo_rally: /home/opnfv/repos/rally
repo_tempest: /home/opnfv/repos/tempest
dir_repo_releng: /home/opnfv/repos/releng
repo_securityscan: /home/opnfv/repos/securityscanning
repo_vrouter: /home/opnfv/repos/vnfs/vrouter
functest: /home/opnfv/functest
- functest_test: /home/opnfv/repos/functest/functest/opnfv_tests
results: /home/opnfv/functest/results
- functest_logging_cfg: /home/opnfv/repos/functest/functest/ci/logging.ini
functest_conf: /home/opnfv/functest/conf
functest_data: /home/opnfv/functest/data
ims_data: /home/opnfv/functest/data/ims/
neutron_private_subnet_gateway: 192.168.120.254
neutron_router_name: functest-router
- functest:
- testcases_yaml: /home/opnfv/repos/functest/functest/ci/testcases.yaml
-
snaps:
use_keystone: True
use_floating_ips: True
subnet_cidr: 192.168.140.0/24
router_name: rally-router
-refstack:
- tempest_conf_path: openstack/refstack_client/refstack_tempest.conf
- defcore_list: openstack/refstack_client/defcore.txt
-
vnf:
aaa:
tenant_name: aaa
import logging
import logging.config
import os
+import pkg_resources
import re
import subprocess
import sys
pod_arch = None
arch_filter = ['aarch64']
-CONFIG_FUNCTEST_PATH = CONST.__getattribute__('CONFIG_FUNCTEST_YAML')
-CONFIG_PATCH_PATH = os.path.join(os.path.dirname(
- CONFIG_FUNCTEST_PATH), "config_patch.yaml")
-CONFIG_AARCH64_PATCH_PATH = os.path.join(os.path.dirname(
- CONFIG_FUNCTEST_PATH), "config_aarch64_patch.yaml")
-RALLY_CONF_PATH = os.path.join("/etc/rally/rally.conf")
-RALLY_AARCH64_PATCH_PATH = os.path.join(os.path.dirname(
- CONFIG_FUNCTEST_PATH), "rally_aarch64_patch.conf")
+CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
+ 'functest', 'ci/config_functest.yaml')
+CONFIG_PATCH_PATH = pkg_resources.resource_filename(
+ 'functest', 'ci/config_patch.yaml')
+CONFIG_AARCH64_PATCH_PATH = pkg_resources.resource_filename(
+ 'functest', 'ci/config_aarch64_patch.yaml')
+RALLY_CONF_PATH = "/etc/rally/rally.conf"
+RALLY_AARCH64_PATCH_PATH = pkg_resources.resource_filename(
+ 'functest', 'ci/rally_aarch64_patch.conf')
class PrepareEnvParser(object):
global handler
global pod_arch
- installer_params_yaml = os.path.join(
- CONST.__getattribute__('dir_repo_functest'),
- 'functest/ci/installer_params.yaml')
+ installer_params_yaml = pkg_resources.resource_filename(
+ 'functest', 'ci/installer_params.yaml')
if (CONST.__getattribute__('INSTALLER_IP') and
CONST.__getattribute__('INSTALLER_TYPE') and
CONST.__getattribute__('INSTALLER_TYPE') in
def verify_deployment():
print_separator()
logger.info("Verifying OpenStack services...")
- cmd = ("%s/functest/ci/check_os.sh" %
- CONST.__getattribute__('dir_repo_functest'))
+ cmd = ("%s" % pkg_resources.resource_filename(
+ 'functest', 'ci/check_os.sh'))
logger.debug("Executing command: %s" % cmd)
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
if __name__ == '__main__':
- logging.config.fileConfig(
- CONST.__getattribute__('dir_functest_logging_cfg'))
+ logging.config.fileConfig(pkg_resources.resource_filename(
+ 'functest', 'ci/logging.ini'))
parser = PrepareEnvParser()
args = parser.parse_args(sys.argv[1:])
sys.exit(main(**args))
import logging
import logging.config
import os
+import pkg_resources
import re
import sys
_tiers = tb.TierBuilder(
CONST.__getattribute__('INSTALLER_TYPE'),
CONST.__getattribute__('DEPLOY_SCENARIO'),
- CONST.__getattribute__("functest_testcases_yaml"))
+ pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
if kwargs['noclean']:
self.clean_flag = False
if __name__ == '__main__':
- logging.config.fileConfig(
- CONST.__getattribute__('dir_functest_logging_cfg'))
+ logging.config.fileConfig(pkg_resources.resource_filename(
+ 'functest', 'ci/logging.ini'))
parser = RunTestsParser()
args = parser.parse_args(sys.argv[1:])
runner = Runner()
installer: 'fuel'
scenario: 'kvm_ovs_dpdk_bar'
run:
- module: 'functest.opnfv_tests.features.barometer'
+ module: 'baro_tests.barometer'
class: 'BarometerCollectd'
-
import click
import logging.config
+import pkg_resources
from functest.cli.commands.cli_env import CliEnv
from functest.cli.commands.cli_os import CliOpenStack
from functest.cli.commands.cli_testcase import CliTestcase
from functest.cli.commands.cli_tier import CliTier
-from functest.utils.constants import CONST
CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
@click.group(context_settings=CONTEXT_SETTINGS)
@click.version_option(version='opnfv colorado.0.1 ')
def cli():
- logging.config.fileConfig(
- CONST.__getattribute__('dir_functest_logging_cfg'))
+ logging.config.fileConfig(pkg_resources.resource_filename(
+ 'functest', 'ci/logging.ini'))
_env = CliEnv()
#
import os
+import pkg_resources
import click
-import git
import prettytable
from functest.utils.constants import CONST
else:
answer = raw_input("Invalid answer. Please type [y|n]\n")
- cmd = ("python %s/functest/ci/prepare_env.py start" %
- CONST.__getattribute__('dir_repo_functest'))
+ cmd = ("python %s start" % pkg_resources.resource_filename(
+ 'functest', 'ci/prepare_env.py'))
ft_utils.execute_command(cmd)
def show(self):
installer_info = ("%s, %s" % (install_type, installer_ip))
scenario = _get_value(CONST.__getattribute__('DEPLOY_SCENARIO'))
node = _get_value(CONST.__getattribute__('NODE_NAME'))
- repo_h = git.Repo(CONST.__getattribute__('dir_repo_functest')).head
- if repo_h.is_detached:
- git_branch = 'detached from FETCH_HEAD'
- git_hash = repo_h.commit.hexsha
- else:
- branch = repo_h.reference
- git_branch = branch.name
- git_hash = branch.commit.hexsha
is_debug = _get_value(CONST.__getattribute__('CI_DEBUG'), 'false')
build_tag = CONST.__getattribute__('BUILD_TAG')
if build_tag is not None:
msg.add_row(['INSTALLER', installer_info])
msg.add_row(['SCENARIO', scenario])
msg.add_row(['POD', node])
- msg.add_row(['GIT BRANCH', git_branch])
- msg.add_row(['GIT HASH', git_hash])
if build_tag:
msg.add_row(['BUILD TAG', build_tag])
msg.add_row(['DEBUG FLAG', is_debug])
import os
+import pkg_resources
import click
def check(self):
self.ping_endpoint()
- cmd = os.path.join(CONST.__getattribute__('dir_repo_functest'),
- "functest/ci/check_os.sh")
- ft_utils.execute_command(cmd, verbose=False)
+ ft_utils.execute_command("sh %s" % pkg_resources.resource_filename(
+ 'functest', 'ci/check_os.sh'), verbose=False)
def snapshot_create(self):
self.ping_endpoint()
""" global variables """
import os
+import pkg_resources
import click
self.tiers = tb.TierBuilder(
CONST.__getattribute__('INSTALLER_TYPE'),
CONST.__getattribute__('DEPLOY_SCENARIO'),
- CONST.__getattribute__('functest_testcases_yaml'))
+ pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
def list(self):
summary = ""
else:
tests = testname.split(",")
for test in tests:
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
flags, test))
ft_utils.execute_command(cmd)
""" global variables """
import os
+import pkg_resources
import click
self.tiers = tb.TierBuilder(
CONST.__getattribute__('INSTALLER_TYPE'),
CONST.__getattribute__('DEPLOY_SCENARIO'),
- CONST.__getattribute__('functest_testcases_yaml'))
+ pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
def list(self):
summary = ""
click.echo("Functest environment is not ready. "
"Run first 'functest env prepare'")
else:
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
- flags, tiername))
+ cmd = ("python %s %s -t %s" % (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'), flags, tiername))
ft_utils.execute_command(cmd)
+++ /dev/null
-from setuptools import setup
-
-setup(
- name='functest',
- version='colorado.0.1',
- py_modules=['cli_base'],
- include_package_data=True,
- install_requires=[
- 'click',
- ],
- entry_points='''
- [console_scripts]
- functest=cli_base:cli
- ''',
-)
stream=stream, verbosity=2).run(self.suite)
self.__logger.debug("\n\n%s", stream.getvalue())
self.stop_time = time.time()
- self.details = {"failures": result.failures,
- "errors": result.errors}
+ self.details = {
+ "testsRun": result.testsRun,
+ "failures": len(result.failures),
+ "errors": len(result.errors),
+ "stream": stream.getvalue()}
self.result = 100 * (
(result.testsRun - (len(result.failures) +
len(result.errors))) /
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
-import inspect
+"""Define the parent class of all VNF TestCases."""
+
import logging
import time
import functest.core.testcase as base
from functest.utils.constants import CONST
-import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_utils as os_utils
+__author__ = ("Morgan Richomme <morgan.richomme@orange.com>, "
+ "Valentin Boucher <valentin.boucher@orange.com>")
+
+
+class VnfPreparationException(Exception):
+ """Raise when VNF preparation cannot be executed."""
+
+
+class OrchestratorDeploymentException(Exception):
+ """Raise when orchestrator cannot be deployed."""
+
+
+class VnfDeploymentException(Exception):
+ """Raise when VNF cannot be deployed."""
+
+
+class VnfTestException(Exception):
+ """Raise when VNF cannot be tested."""
+
class VnfOnBoarding(base.TestCase):
+ """Base model for VNF test cases."""
__logger = logging.getLogger(__name__)
def __init__(self, **kwargs):
super(VnfOnBoarding, self).__init__(**kwargs)
- self.repo = kwargs.get('repo', '')
- self.cmd = kwargs.get('cmd', '')
- self.details = {}
- self.result_dir = CONST.__getattribute__('dir_results')
- self.details_step_mapping = dict(
- deploy_orchestrator='orchestrator',
- deploy_vnf='vnf',
- test_vnf='test_vnf',
- prepare='prepare_env')
- self.details['prepare_env'] = {}
- self.details['orchestrator'] = {}
- self.details['vnf'] = {}
- self.details['test_vnf'] = {}
- self.images = {}
- try:
- self.tenant_name = CONST.__getattribute__(
- 'vnf_{}_tenant_name'.format(self.case_name))
- self.tenant_description = CONST.__getattribute__(
- 'vnf_{}_tenant_description'.format(self.case_name))
- 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 Exception:
- self.__logger.warn("No tenant image defined for this VNF")
-
- def execute(self):
+ self.tenant_created = False
+ self.user_created = False
+ self.tenant_name = CONST.__getattribute__(
+ 'vnf_{}_tenant_name'.format(self.case_name))
+ self.tenant_description = CONST.__getattribute__(
+ 'vnf_{}_tenant_description'.format(self.case_name))
+
+ def run(self, **kwargs):
+ """
+ Run of the VNF test case:
+
+ * Deploy an orchestrator if needed (e.g. heat, cloudify, ONAP),
+ * Deploy the VNF,
+ * Perform tests on the VNF
+
+ A VNF test case is successfull when the 3 steps are PASS
+ If one of the step is FAIL, the test case is FAIL
+
+ Returns:
+ TestCase.EX_OK if result is 'PASS'.
+ TestCase.EX_TESTCASE_FAILED otherwise.
+ """
self.start_time = time.time()
- # Prepare the test (Create Tenant, User, ...)
+
try:
- self.__logger.info("Create VNF Onboarding environment")
self.prepare()
- except Exception:
- self.__logger.error("Error during VNF Onboarding environment"
- "creation", exc_info=True)
+ if (self.deploy_orchestrator() and
+ self.deploy_vnf() and
+ self.test_vnf()):
+ self.stop_time = time.time()
+ # Calculation with different weight depending on the steps TODO
+ self.result = 100
+ return base.TestCase.EX_OK
+ else:
+ self.result = 0
+ return base.TestCase.EX_TESTCASE_FAILED
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Exception on VNF testing")
return base.TestCase.EX_TESTCASE_FAILED
- # Deploy orchestrator
- try:
- self.__logger.info("Deploy orchestrator (if necessary)")
- orchestrator_ready_time = time.time()
- res_orchestrator = self.deploy_orchestrator()
- # orchestrator is not mandatory
- if res_orchestrator is not None:
- self.details['orchestrator']['status'] = (
- res_orchestrator['status'])
- self.details['orchestrator']['result'] = (
- res_orchestrator['result'])
- self.details['orchestrator']['duration'] = round(
- orchestrator_ready_time - self.start_time, 1)
- except Exception:
- self.__logger.warn("Problem with the Orchestrator", exc_info=True)
-
- # Deploy VNF
- try:
- self.__logger.info("Deploy VNF " + self.case_name)
- res_deploy_vnf = self.deploy_vnf()
- vnf_ready_time = time.time()
- self.details['vnf']['status'] = res_deploy_vnf['status']
- self.details['vnf']['result'] = res_deploy_vnf['result']
- self.details['vnf']['duration'] = round(
- vnf_ready_time - orchestrator_ready_time, 1)
- except Exception:
- self.__logger.error("Error during VNF deployment", exc_info=True)
- return base.TestCase.EX_TESTCASE_FAILED
+ def prepare(self):
+ """
+ Prepare the environment for VNF testing:
- # Test VNF
- try:
- self.__logger.info("Test VNF")
- res_test_vnf = self.test_vnf()
- test_vnf_done_time = time.time()
- self.details['test_vnf']['status'] = res_test_vnf['status']
- self.details['test_vnf']['result'] = res_test_vnf['result']
- self.details['test_vnf']['duration'] = round(
- test_vnf_done_time - vnf_ready_time, 1)
- except Exception:
- self.__logger.error("Error when running VNF tests", exc_info=True)
- return base.TestCase.EX_TESTCASE_FAILED
+ * Creation of a user,
+ * Creation of a tenant,
+ * Allocation admin role to the user on this tenant
- # Clean the system
- self.clean()
- self.stop_time = time.time()
+ Returns base.TestCase.EX_OK if preparation is successfull
- exit_code = self.parse_results()
- self.log_results()
- return exit_code
+ Raise VnfPreparationException in case of problem
+ """
+ try:
+ self.__logger.info("Prepare VNF: %s, description: %s",
+ self.tenant_name, self.tenant_description)
+ admin_creds = os_utils.get_credentials()
+ keystone_client = os_utils.get_keystone_client()
+ self.tenant_created = os_utils.get_or_create_tenant_for_vnf(
+ keystone_client, self.tenant_name, self.tenant_description)
+ self.user_created = os_utils.get_or_create_user_for_vnf(
+ keystone_client, self.tenant_name)
+ creds = admin_creds.copy()
+ creds.update({
+ "tenant": self.tenant_name,
+ "username": self.tenant_name,
+ "password": self.tenant_name
+ })
+ return base.TestCase.EX_OK
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Exception raised during VNF preparation")
+ raise VnfPreparationException
+
+ def deploy_orchestrator(self):
+ """
+ Deploy an orchestrator (optional).
+
+ If function overwritten
+ raise orchestratorDeploymentException if error during orchestrator
+ deployment
+ """
+ self.__logger.info("Deploy orchestrator (if necessary)")
+ return True
- # prepare state could consist in the creation of the resources
- # a dedicated user
- # a dedicated tenant
- # dedicated images
- def prepare(self):
- self.creds = os_utils.get_credentials()
- self.keystone_client = os_utils.get_keystone_client()
-
- self.__logger.info(
- "Prepare OpenStack plateform(create tenant and user)")
- admin_user_id = os_utils.get_user_id(self.keystone_client,
- self.creds['username'])
- if not admin_user_id:
- self.step_failure("Failed to get id of {0}".format(
- self.creds['username']))
-
- tenant_id = os_utils.get_tenant_id(self.keystone_client,
- self.tenant_name)
- if not tenant_id:
- tenant_id = os_utils.create_tenant(self.keystone_client,
- self.tenant_name,
- self.tenant_description)
- if not tenant_id:
- self.step_failure("Failed to get or create {0} tenant".format(
- self.tenant_name))
- roles_name = ["admin", "Admin"]
- role_id = ''
- for role_name in roles_name:
- if not role_id:
- role_id = os_utils.get_role_id(self.keystone_client,
- role_name)
-
- if not role_id:
- self.step_failure("Failed to get id for {0} role".format(
- role_name))
-
- if not os_utils.add_role_user(self.keystone_client, admin_user_id,
- role_id, tenant_id):
- self.step_failure("Failed to add {0} on tenant".format(
- self.creds['username']))
-
- user_id = os_utils.get_or_create_user(self.keystone_client,
- self.tenant_name,
- self.tenant_name,
- tenant_id)
- if not user_id:
- self.step_failure("Failed to get or create {0} user".format(
- self.tenant_name))
-
- os_utils.add_role_user(self.keystone_client, user_id,
- role_id, tenant_id)
-
- self.__logger.info("Update OpenStack creds informations")
- self.admin_creds = self.creds.copy()
- self.admin_creds.update({
- "tenant": self.tenant_name
- })
- self.neutron_client = os_utils.get_neutron_client(self.admin_creds)
- self.nova_client = os_utils.get_nova_client(self.admin_creds)
- self.creds.update({
- "tenant": self.tenant_name,
- "username": self.tenant_name,
- "password": self.tenant_name,
- })
-
- # orchestrator is not mandatory to deploy and test VNF
- def deploy_orchestrator(self, **kwargs):
- pass
-
- # TODO see how to use built-in exception from releng module
def deploy_vnf(self):
+ """
+ Deploy the VNF
+
+ This function MUST be implemented by vnf test cases.
+ The details section MAY be updated in the vnf test cases.
+
+ The deployment can be executed via a specific orchestrator
+ or using nuild-in orchestrators such as:
+
+ * heat, openbaton, cloudify (available on all scenario),
+ * open-o (on open-o scenarios)
+
+ Returns:
+ True if the VNF is properly deployed
+ False if the VNF is not deployed
+
+ Raise VnfDeploymentException if error during VNF deployment
+ """
self.__logger.error("VNF must be deployed")
- raise Exception("VNF not deployed")
+ raise VnfDeploymentException
def test_vnf(self):
+ """
+ Test the VNF
+
+ This function MUST be implemented by vnf test cases.
+ The details section MAY be updated in the vnf test cases.
+
+ Once a VNF is deployed, it is assumed that specific test suite can be
+ run to validate the VNF.
+ Please note that the same test suite can be used on several test case
+ (e.g. clearwater test suite can be used whatever the orchestrator used
+ for the deployment)
+
+ Returns:
+ True if VNF tests are PASS
+ False if test suite is FAIL
+
+ Raise VnfTestException if error during VNF test
+ """
self.__logger.error("VNF must be tested")
- raise Exception("VNF not tested")
+ raise VnfTestException
- # clean before openstack clean run
def clean(self):
- self.__logger.info("test cleaning")
+ """
+ Clean VNF test case.
- def parse_results(self):
- exit_code = self.EX_OK
- self.result = "PASS"
- self.__logger.info(self.details)
- # The 2 VNF steps must be OK to get a PASS result
- if (self.details['vnf']['status'] is not "PASS" or
- self.details['test_vnf']['status'] is not "PASS"):
- exit_code = self.EX_RUN_ERROR
- self.result = "FAIL"
- return exit_code
-
- def log_results(self):
- ft_utils.logger_test_results(self.project_name,
- self.case_name,
- self.result,
- self.details)
-
- def step_failure(self, error_msg):
- part = inspect.stack()[1][3]
- self.__logger.error("Step {0} failed: {1}".format(part, error_msg))
- try:
- step_name = self.details_step_mapping[part]
- part_info = self.details[step_name]
- except KeyError:
- self.details[part] = {}
- part_info = self.details[part]
- part_info['status'] = 'FAIL'
- part_info['result'] = error_msg
- raise Exception(error_msg)
+ It is up to the test providers to delete resources used for the tests.
+ By default we clean:
+
+ * the user,
+ * the tenant
+ """
+ self.__logger.info("test cleaning")
+ keystone_client = os_utils.get_keystone_client()
+ if self.tenant_created:
+ os_utils.delete_tenant(keystone_client, self.tenant_name)
+ if self.user_created:
+ os_utils.delete_user(keystone_client, self.tenant_name)
+++ /dev/null
-#!/usr/bin/env python
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-
-import logging
-
-from baro_tests import collectd
-
-import functest.core.feature as base
-
-
-class BarometerCollectd(base.Feature):
- '''
- Class for executing barometercollectd testcase.
- '''
-
- __logger = logging.getLogger(__name__)
-
- def execute(self):
- return collectd.main(self.__logger)
import json
import logging
import os
+import pkg_resources
import re
import subprocess
import time
GLANCE_IMAGE_FORMAT = CONST.__getattribute__('openstack_image_disk_format')
FLAVOR_NAME = "m1.tiny"
- RALLY_DIR = os.path.join(
- CONST.__getattribute__('dir_repo_functest'),
- CONST.__getattribute__('dir_rally'))
- RALLY_SCENARIO_DIR = os.path.join(RALLY_DIR, "scenario")
- TEMPLATE_DIR = os.path.join(RALLY_SCENARIO_DIR, "templates")
- SUPPORT_DIR = os.path.join(RALLY_SCENARIO_DIR, "support")
+ RALLY_DIR = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/rally')
+ RALLY_SCENARIO_DIR = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/rally/scenario')
+ TEMPLATE_DIR = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/rally/scenario/templates')
+ SUPPORT_DIR = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/rally/scenario/support')
USERS_AMOUNT = 2
TENANTS_AMOUNT = 3
ITERATIONS_AMOUNT = 10
import argparse
import logging
import os
+import pkg_resources
import re
import sys
import subprocess
if "case_name" not in kwargs:
kwargs["case_name"] = "refstack_defcore"
super(RefstackClient, self).__init__(**kwargs)
- self.FUNCTEST_TEST = CONST.__getattribute__('dir_functest_test')
- self.CONF_PATH = CONST.__getattribute__('refstack_tempest_conf_path')
- self.DEFCORE_LIST = CONST.__getattribute__('refstack_defcore_list')
+ self.CONF_PATH = pkg_resources.resource_filename(
+ 'functest',
+ 'opnfv_tests/openstack/refstack_client/refstack_tempest.conf')
+ self.FUNCTEST_TEST = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests')
+ self.DEFCORE_LIST = 'openstack/refstack_client/defcore.txt'
self.confpath = os.path.join(self.FUNCTEST_TEST,
self.CONF_PATH)
- self.defcorelist = os.path.join(self.FUNCTEST_TEST,
- self.DEFCORE_LIST)
+ self.defcorelist = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/refstack_client/defcore.txt')
def source_venv(self):
class RefstackClientParser(object):
def __init__(self):
- self.FUNCTEST_TEST = CONST.__getattribute__('dir_functest_test')
- self.CONF_PATH = CONST.__getattribute__('refstack_tempest_conf_path')
- self.DEFCORE_LIST = CONST.__getattribute__('refstack_defcore_list')
+ self.FUNCTEST_TEST = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests')
+ self.CONF_PATH = pkg_resources.resource_filename(
+ 'functest',
+ 'opnfv_tests/openstack/refstack_client/refstack_tempest.conf')
+ self.DEFCORE_LIST = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/refstack_client/defcore.txt')
self.confpath = os.path.join(self.FUNCTEST_TEST,
self.CONF_PATH)
self.defcorelist = os.path.join(self.FUNCTEST_TEST,
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
import logging
-import os
+import pkg_resources
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils import openstack_utils
self.DEPLOYMENT_ID = conf_utils.get_verifier_deployment_id()
self.DEPLOYMENT_DIR = conf_utils.get_verifier_deployment_dir(
self.VERIFIER_ID, self.DEPLOYMENT_ID)
- self.confpath = os.path.join(
- CONST.__getattribute__('dir_functest_test'),
- CONST.__getattribute__('refstack_tempest_conf_path'))
+ self.confpath = pkg_resources.resource_filename(
+ 'functest',
+ 'opnfv_tests/openstack/refstack_client/refstack_tempest.conf')
def generate_tempestconf(self):
try:
import ConfigParser
import logging
import os
+import pkg_resources
import re
import shutil
import subprocess
IMAGE_ID_ALT = None
FLAVOR_ID_ALT = None
-REPO_PATH = CONST.__getattribute__('dir_repo_functest')
GLANCE_IMAGE_PATH = os.path.join(
CONST.__getattribute__('dir_functest_images'),
CONST.__getattribute__('openstack_image_file_name'))
-TEMPEST_TEST_LIST_DIR = CONST.__getattribute__('dir_tempest_cases')
TEMPEST_RESULTS_DIR = os.path.join(CONST.__getattribute__('dir_results'),
'tempest')
-TEMPEST_CUSTOM = os.path.join(REPO_PATH, TEMPEST_TEST_LIST_DIR,
- 'test_list.txt')
-TEMPEST_BLACKLIST = os.path.join(REPO_PATH, TEMPEST_TEST_LIST_DIR,
- 'blacklist.txt')
-TEMPEST_DEFCORE = os.path.join(REPO_PATH, TEMPEST_TEST_LIST_DIR,
- 'defcore_req.txt')
+TEMPEST_CUSTOM = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/tempest/custom_tests/test_list.txt')
+TEMPEST_BLACKLIST = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/tempest/custom_tests/blacklist.txt')
+TEMPEST_DEFCORE = pkg_resources.resource_filename(
+ 'functest',
+ 'opnfv_tests/openstack/tempest/custom_tests/defcore_req.txt')
TEMPEST_RAW_LIST = os.path.join(TEMPEST_RESULTS_DIR, 'test_raw_list.txt')
TEMPEST_LIST = os.path.join(TEMPEST_RESULTS_DIR, 'test_list.txt')
REFSTACK_RESULTS_DIR = os.path.join(CONST.__getattribute__('dir_results'),
with open(conf_file, 'wb') as config_file:
config.write(config_file)
- confpath = os.path.join(
- CONST.__getattribute__('dir_functest_test'),
- CONST.__getattribute__('refstack_tempest_conf_path'))
+ confpath = pkg_resources.resource_filename(
+ 'functest',
+ 'opnfv_tests/openstack/refstack_client/refstack_tempest.conf')
shutil.copyfile(conf_file, confpath)
from datetime import datetime
import logging
import os
+import pkg_resources
import time
import uuid
self.logger = logging.getLogger(self.__class__.__name__)
- self.functest_repo = CONST.__getattribute__('dir_repo_functest')
+ self.functest_repo = pkg_resources.resource_filename(
+ 'functest', '..')
self.guid = ''
if CONST.__getattribute__('vping_unique_names'):
self.guid = '-' + str(uuid.uuid4())
self.os_creds = openstack_tests.get_credentials(
os_env_file=CONST.__getattribute__('openstack_creds'))
- self.repo = CONST.__getattribute__('dir_vping')
-
self.creators = list()
self.image_creator = None
self.network_creator = None
# http://www.apache.org/licenses/LICENSE-2.0
import argparse
-import os
+import pkg_resources
from scp import SCPClient
import sys
import time
"""
self.logger.info("Trying to transfer ping.sh")
scp = SCPClient(ssh.get_transport())
- local_path = self.functest_repo + "/" + self.repo
- ping_script = os.path.join(local_path, "ping.sh")
+ ping_script = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/vping/ping.sh')
try:
scp.put(ping_script, "~/")
except:
import logging
import os
+import pkg_resources
import re
import subprocess
import shutil
onos_sfc_image_path = os.path.join(
CONST.__getattribute__('dir_functest_images'),
CONST.__getattribute__('onos_sfc_image_file_name'))
- onos_sfc_path = os.path.join(CONST.__getattribute__('dir_repo_functest'),
- CONST.__getattribute__('dir_onos_sfc'))
+ onos_sfc_path = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/sdn/onos/sfc')
installer_type = CONST.__getattribute__('INSTALLER_TYPE')
logger = logging.getLogger(__name__)
# http://www.apache.org/licenses/LICENSE-2.0
import logging
-import sys
-import argparse
-
-import functest.core.testcase as testcase
import functest.core.vnf as vnf
class AaaVnf(vnf.VnfOnBoarding):
+ """AAA VNF sample"""
logger = logging.getLogger(__name__)
def deploy_orchestrator(self):
self.logger.info("No VNFM needed to deploy a free radius here")
- return None
+ return True
-# TODO see how to use build in exception form releng module
def deploy_vnf(self):
self.logger.info("Freeradius VNF deployment")
- # TODO apt-get update + config tuning
- deploy_vnf = {}
- deploy_vnf['status'] = "PASS"
- deploy_vnf['result'] = {}
- return deploy_vnf
+ # find a way to deploy freeradius and tester (heat,manual, ..)
+ deploy_vnf = {'status': 'PASS', 'version': 'xxxx'}
+ self.details['deploy_vnf'] = deploy_vnf
+ return True
def test_vnf(self):
self.logger.info("Run test towards freeradius")
- # TODO: once the freeradius is deployed..make some tests
- test_vnf = {}
- test_vnf['status'] = "PASS"
- test_vnf['result'] = {}
- return test_vnf
-
- def main(self, **kwargs):
- self.logger.info("AAA VNF onboarding")
- self.execute()
- if self.result is "PASS":
- return self.EX_OK
- else:
- return self.EX_RUN_ERROR
-
- def run(self):
- kwargs = {}
- return self.main(**kwargs)
-
-
-if __name__ == '__main__':
- logging.basicConfig()
- parser = argparse.ArgumentParser()
- args = vars(parser.parse_args())
- aaa_vnf = AaaVnf()
- try:
- result = aaa_vnf.main(**args)
- if result != testcase.TestCase.EX_OK:
- sys.exit(result)
- if args['pushtodb']:
- sys.exit(aaa_vnf.push_to_db())
- except Exception:
- sys.exit(testcase.TestCase.EX_RUN_ERROR)
+ # once the freeradius is deployed..make some tests
+ test_vnf = {'status': 'PASS', 'version': 'xxxx'}
+ self.details['test_vnf'] = test_vnf
+ return True
import json
import logging
import os
+import pkg_resources
import shutil
import requests
from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
+__author__ = ("Valentin Boucher <valentin.boucher@orange.com>, "
+ "Helen Yao <helanyao@gmail.com>")
+
class ClearwaterOnBoardingBase(vnf.VnfOnBoarding):
+ """ vIMS clearwater base usable by several orchestrators"""
def __init__(self, **kwargs):
self.logger = logging.getLogger(__name__)
super(ClearwaterOnBoardingBase, self).__init__(**kwargs)
- 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, self.case_name)
- self.test_dir = CONST.dir_repo_vims_test
+ self.case_dir = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/vnf/ims')
+ self.data_dir = CONST.__getattribute__('dir_ims_data')
+ self.result_dir = os.path.join(CONST.__getattribute__('dir_results'),
+ self.case_name)
+ self.test_dir = CONST.__getattribute__('dir_repo_vims_test')
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir)
if rq.status_code != 201:
raise Exception('Failed to get cookie for Ellis')
cookies = rq.cookies
- self.logger.info('Cookies: %s' % cookies)
+ self.logger.info('Cookies: %s', cookies)
number_url = 'http://{0}/accounts/{1}/numbers'.format(
ellis_ip,
import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_utils as os_utils
+__author__ = "Valentin Boucher <valentin.boucher@orange.com>"
+
class CloudifyIms(clearwater_ims_base.ClearwaterOnBoardingBase):
+ """Clearwater vIMS deployed with Cloudify Orchestrator Case"""
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs["case_name"] = "cloudify_ims"
super(CloudifyIms, self).__init__(**kwargs)
self.logger = logging.getLogger(__name__)
+ self.neutron_client = ''
+ self.glance_client = ''
+ self.keystone_client = ''
+ self.nova_client = ''
# Retrieve the configuration
try:
blueprint=get_config("cloudify.blueprint", config_file),
inputs=get_config("cloudify.inputs", config_file)
)
- self.logger.debug("Orchestrator configuration: %s" % self.orchestrator)
+ self.logger.debug("Orchestrator configuration: %s", self.orchestrator)
self.vnf = dict(
blueprint=get_config("clearwater.blueprint", config_file),
deployment_name=get_config("clearwater.deployment_name",
inputs=get_config("clearwater.inputs", config_file),
requirements=get_config("clearwater.requirements", config_file)
)
- self.logger.debug("VNF configuration: %s" % self.vnf)
+ self.logger.debug("VNF configuration: %s", self.vnf)
self.images = get_config("tenant_images", config_file)
- self.logger.info("Images needed for vIMS: %s" % self.images)
+ self.logger.info("Images needed for vIMS: %s", self.images)
- def deploy_orchestrator(self, **kwargs):
+ def deploy_orchestrator(self):
self.logger.info("Additional pre-configuration steps")
self.neutron_client = os_utils.get_neutron_client(self.admin_creds)
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))
+ self.logger.info("image: %s, url: %s", image_name, image_url)
try:
image_id = os_utils.get_image_id(self.glance_client,
image_name)
- self.logger.debug("image_id: %s" % image_id)
+ self.logger.debug("image_id: %s", image_id)
except Exception:
- self.logger.error("Unexpected error: %s" % sys.exc_info()[0])
+ 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(self.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")
+ self.logger.info("""%s image does not exist on glance repo.
+ Try downloading this image
+ and upload on glance !""",
+ image_name)
+ image_id = os_utils.download_and_add_image_on_glance(
+ self.glance_client,
+ image_name,
+ image_url,
+ temp_dir)
# Need to extend quota
self.logger.info("Update security group quota for this tenant")
tenant_id = os_utils.get_tenant_id(self.keystone_client,
self.tenant_name)
- self.logger.debug("Tenant id found %s" % tenant_id)
+ self.logger.debug("Tenant id found %s", tenant_id)
if not os_utils.update_sg_quota(self.neutron_client,
tenant_id, 50, 100):
- self.step_failure("Failed to update security group quota" +
+ self.logger.error("Failed to update security group quota"
" for tenant " + self.tenant_name)
+
self.logger.debug("group quota extended")
# start the deployment of cloudify
public_auth_url = os_utils.get_endpoint('identity')
- self.logger.debug("CFY inputs: %s" % self.orchestrator['inputs'])
+ self.logger.debug("CFY inputs: %s", self.orchestrator['inputs'])
cfy = Orchestrator(self.data_dir, self.orchestrator['inputs'])
self.orchestrator['object'] = cfy
self.logger.debug("Orchestrator object created")
- self.logger.debug("Tenant name: %s" % self.tenant_name)
+ self.logger.debug("Tenant name: %s", self.tenant_name)
cfy.set_credentials(username=self.tenant_name,
password=self.tenant_name,
# orchestrator VM flavor
self.logger.info("Check Flavor is available, if not create one")
- self.logger.debug("Flavor details %s " %
+ self.logger.debug("Flavor details %s ",
self.orchestrator['requirements']['ram_min'])
flavor_exist, flavor_id = os_utils.get_or_create_flavor(
"m1.large",
'50',
'2',
public=True)
- self.logger.debug("Flavor id: %s" % flavor_id)
+ self.logger.debug("Flavor id: %s", flavor_id)
if not flavor_id:
self.logger.info("Available flavors are: ")
self.logger.info(self.nova_client.flavor.list())
- self.step_failure("Failed to find required flavor"
- "for this deployment")
+
cfy.set_flavor_id(flavor_id)
self.logger.debug("Flavor OK")
image_id = os_utils.get_image_id(
self.glance_client,
self.orchestrator['requirements']['os_image'])
- self.logger.debug("Orchestrator image id: %s" % image_id)
+ self.logger.debug("Orchestrator image id: %s", image_id)
if image_id == '':
self.logger.error("CFY image not found")
- self.step_failure("Failed to find required OS image"
- " for cloudify manager")
- else:
- self.step_failure("Failed to find required OS image"
- " for cloudify manager")
cfy.set_image_id(image_id)
self.logger.debug("Orchestrator image set")
self.logger.debug("Get External network")
ext_net = os_utils.get_external_net(self.neutron_client)
- self.logger.debug("External network: %s" % ext_net)
- if not ext_net:
- self.step_failure("Failed to get external network")
+ self.logger.debug("External network: %s", ext_net)
cfy.set_external_network_name(ext_net)
self.logger.debug("CFY External network set")
'30',
'1',
public=True)
- self.logger.debug("Flavor id: %s" % flavor_id)
+ self.logger.debug("Flavor id: %s", flavor_id)
if not flavor_id:
self.logger.info("Available flavors are: ")
self.logger.info(self.nova_client.flavor.list())
- self.step_failure("Failed to find required flavor"
- " for this deployment")
cw.set_flavor_id(flavor_id)
if 'os_image' in self.vnf['requirements'].keys():
image_id = os_utils.get_image_id(
self.glance_client, self.vnf['requirements']['os_image'])
- if image_id == '':
- self.step_failure("Failed to find required OS image"
- " for clearwater VMs")
- else:
- self.step_failure("Failed to find required OS image"
- " for clearwater VMs")
cw.set_image_id(image_id)
-
ext_net = os_utils.get_external_net(self.neutron_client)
- if not ext_net:
- self.step_failure("Failed to get external network")
-
cw.set_external_network_name(ext_net)
error = cw.deploy_vnf(self.vnf['blueprint'])
mgr_ip = os.popen(cmd).read()
mgr_ip = mgr_ip.splitlines()[0]
except Exception:
- self.step_failure("Unable to retrieve the IP of the "
- "cloudify manager server !")
+ self.logger.exception("Unable to retrieve the IP of the "
+ "cloudify manager server !")
self.logger.info('Cloudify Manager: %s', mgr_ip)
api_url = 'http://{0}/api/v2/deployments/{1}/outputs'.format(
if dns_ip != "":
vims_test_result = self.run_clearwater_live_test(
dns_ip=dns_ip,
- public_domain=self.inputs["public_domain"])
+ public_domain="") # self.inputs["public_domain"]
if vims_test_result != '':
return {'status': 'PASS', 'result': vims_test_result}
else:
self.orchestrator['object'].undeploy_manager()
super(CloudifyIms, self).clean()
- def main(self, **kwargs):
- self.logger.info("Cloudify IMS VNF onboarding test starting")
- self.execute()
- self.logger.info("Cloudify IMS VNF onboarding test executed")
- if self.result is "PASS":
- return self.EX_OK
- else:
- return self.EX_RUN_ERROR
-
- def run(self):
- kwargs = {}
- return self.main(**kwargs)
-
# ----------------------------------------------------------
#
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
import requests
import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base
+from functest.utils.constants import CONST
class OperaIms(clearwater_ims_base.ClearwaterOnBoardingBase):
kwargs["case_name"] = "opera_ims"
super(OperaIms, self).__init__(**kwargs)
self.logger = logging.getLogger(__name__)
- self.ellis_file = os.path.join(self.result_dir, 'ellis.info')
- self.live_test_file = os.path.join(self.result_dir,
- 'live_test_report.json')
+ self.ellis_file = os.path.join(
+ CONST.__getattribute__('dir_results'), 'ellis.info')
+ self.live_test_file = os.path.join(
+ CONST.__getattribute__('dir_results'), 'live_test_report.json')
try:
self.openo_msb_endpoint = os.environ['OPENO_MSB_ENDPOINT']
except KeyError:
import json
import logging
import os
+import pkg_resources
import socket
import sys
import time
import yaml
import functest.core.vnf as vnf
-import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_utils as os_utils
from functest.utils.constants import CONST
from org.openbaton.cli.agents.agents import MainAgent
from org.openbaton.cli.errors.errors import NfvoException
+
+__author__ = "Pauls, Michael <michael.pauls@fokus.fraunhofer.de>"
# ----------------------------------------------------------
#
# UTILS
# -----------------------------------------------------------
-def get_config(parameter, file):
+def get_config(parameter, my_file):
"""
Returns the value of a given parameter in file.yaml
parameter must be given in string format with dots
value = value.get(element)
if value is None:
raise ValueError("The parameter %s is not defined in"
- " %s" % (parameter, file))
+ " %s" % (parameter, my_file))
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
-
-
def servertest(host, port):
args = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM)
for family, socktype, proto, canonname, sockaddr in args:
class ImsVnf(vnf.VnfOnBoarding):
+ """OpenIMS VNF deployed with openBaton orchestrator"""
def __init__(self, project='functest', case_name='orchestra_ims',
repo='', cmd=''):
super(ImsVnf, self).__init__(project, case_name, repo, cmd)
+ self.logger = logging.getLogger(__name__)
+ self.logger.info("Orchestra IMS VNF onboarding test starting")
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 = logging.getLogger(__name__)
- 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.case_dir = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/vnf/ims/')
+ self.data_dir = CONST.__getattribute__('dir_ims_data')
+ self.test_dir = CONST.__getattribute__('dir_repo_vims_test')
self.ob_projectid = ""
self.keystone_client = os_utils.get_keystone_client()
self.ob_nsr_id = ""
self.userdata_file = get_config("openbaton.userdata.file",
config_file)
- def deploy_orchestrator(self, **kwargs):
+ def deploy_orchestrator(self):
self.logger.info("Additional pre-configuration steps")
nova_client = os_utils.get_nova_client()
neutron_client = os_utils.get_neutron_client()
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))
+ 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)
+ self.logger.info("image_id: %s", image_id)
except BaseException:
- self.logger.error("Unexpected error: %s" % sys.exc_info()[0])
+ 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)
+ self.logger.info("""%s image doesn't exist on glance
+ repository. Try downloading this image
+ and upload on glance !""" % image_name)
+ image_id = os_utils.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")
+ self.logger.error("Failed to find or upload required OS "
+ "image for this deployment")
+ return False
+
network_dic = os_utils.create_network_full(neutron_client,
"openbaton_mgmt",
"openbaton_mgmt_subnet",
if floatip is None:
self.logger.error("Cannot create floating IP.")
+ return False
userdata = "#!/bin/bash\n"
userdata += "echo \"Executing userdata...\"\n"
self.logger.info("flavor: m1.medium\n"
"image: %s\n"
"network_id: %s\n"
- "userdata: %s\n"
- % (self.imagename, network_id, userdata))
+ "userdata: %s\n",
+ self.imagename,
+ network_id,
+ userdata)
instance = os_utils.create_instance_and_wait_for_active(
"orchestra",
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"))
+ 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.step_failure("Cannot associate floating IP to VM.")
+ self.logger.error("Cannot associate floating IP to VM.")
+ return False
self.logger.info("Waiting for Open Baton NFVO to be up and running...")
x = 0
x += 1
if x == 100:
- self.step_failure("Open Baton is not started correctly")
+ self.logger.error("Open Baton is not started correctly")
self.ob_ip = floatip
self.ob_password = "openbaton"
self.details["orchestrator"] = {
'status': "PASS", 'result': "Deploy Open Baton NFVO: OK"}
self.logger.info("Deploy Open Baton NFVO: OK")
+ return True
def deploy_vnf(self):
self.logger.info("Starting vIMS Deployment...")
'20',
'1',
public=True)
- self.logger.debug("Flavor id: %s" % flavor_id)
+ self.logger.debug("Flavor id: %s", flavor_id)
self.logger.info("Getting project 'default'...")
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")
- self.logger.info("Found project 'default': %s" % p)
+ self.logger.info("Found project 'default': %s", p)
break
- self.logger.debug("project id: %s" % self.ob_projectid)
+ self.logger.debug("project id: %s", self.ob_projectid)
if self.ob_projectid == "":
- self.step_failure("Default project id was not found!")
+ self.logger.error("Default project id was not found!")
creds = os_utils.get_credentials()
- self.logger.info("PoP creds: %s" % creds)
+ self.logger.info("PoP creds: %s", creds)
if os_utils.is_keystone_v3():
self.logger.info(
"Using v2 API of OpenStack... -> Using OS_TENANT_NAME")
project_id = creds.get("tenant_name")
- self.logger.debug("project id: %s" % project_id)
+ self.logger.debug("project id: %s", project_id)
vim_json = {
"name": "vim-instance",
}
}
- self.logger.debug("Registering VIM: %s" % vim_json)
+ self.logger.debug("Registering VIM: %s", vim_json)
self.main_agent.get_agent(
"vim",
nsd = {}
try:
- self.logger.info("sending: %s" % self.market_link)
+ 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)
+ self.logger.error(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")
+ self.logger.error("NSD not onboarded correctly")
try:
self.nsr = nsr_agent.create(nsd_id)
except NfvoException as e:
- self.step_failure(e.message)
+ self.logger.error(e.message)
if self.nsr.get('code') is not None:
self.logger.error(
- "vIMS cannot be deployed: %s -> %s" %
- (self.nsr.get('code'), self.nsr.get('message')))
- self.step_failure("vIMS cannot be deployed")
+ "vIMS cannot be deployed: %s -> %s",
+ self.nsr.get('code'),
+ self.nsr.get('message'))
+ self.logger.error("vIMS cannot be deployed")
i = 0
self.logger.info("Waiting for NSR to go to ACTIVE...")
"status") != 'ERROR':
i += 1
if i == 150:
- self.step_failure("After %s sec the NSR did not go to ACTIVE.."
- % 5 * i)
+ self.logger.error("INACTIVE NSR after %s sec..", 5 * i)
+
time.sleep(5)
self.nsr = json.loads(nsr_agent.find(self.nsr.get('id')))
else:
self.details["vnf"] = {'status': "FAIL", 'result': self.nsr}
self.logger.error(self.nsr)
- self.step_failure("Deploy VNF: ERROR")
+ self.logger.error("Deploy VNF: ERROR")
+ return False
+
self.ob_nsr_id = self.nsr.get("id")
self.logger.info(
"Sleep for 60s to ensure that all services are up and running...")
time.sleep(60)
- return self.details.get("vnf")
+ return True
def test_vnf(self):
# Adaptations probably needed
# code used for cloudify_ims
# ruby client on jumphost calling the vIMS on the SUT
self.logger.info(
- "Testing if %s works properly..." %
- self.nsr.get('name'))
+ "Testing if %s works properly...", self.nsr.get('name'))
for vnfr in self.nsr.get('vnfr'):
self.logger.info(
- "Checking ports %s of VNF %s" %
- (self.ims_conf.get(
- vnfr.get('name')).get('ports'),
- vnfr.get('name')))
+ "Checking ports %s of VNF %s",
+ self.ims_conf.get(vnfr.get('name')).get('ports'),
+ vnfr.get('name'))
for vdu in vnfr.get('vdu'):
for vnfci in vdu.get('vnfc_instance'):
self.logger.debug(
- "Checking ports of VNFC instance %s" %
+ "Checking ports of VNFC instance %s",
vnfci.get('hostname'))
for floatingIp in vnfci.get('floatingIps'):
self.logger.debug(
- "Testing %s:%s" %
- (vnfci.get('hostname'), floatingIp.get('ip')))
+ "Testing %s:%s",
+ vnfci.get('hostname'),
+ floatingIp.get('ip'))
for port in self.ims_conf.get(
vnfr.get('name')).get('ports'):
if servertest(floatingIp.get('ip'), port):
self.logger.info(
- "VNFC instance %s is reachable at %s:%s" %
- (vnfci.get('hostname'),
- floatingIp.get('ip'),
- port))
+ "VNFC instance %s is reachable at %s:%s",
+ vnfci.get('hostname'),
+ floatingIp.get('ip'),
+ port)
else:
self.logger.error(
"VNFC instance %s is not reachable "
- "at %s:%s" % (vnfci.get('hostname'),
- floatingIp.get('ip'), port))
+ "at %s:%s",
+ vnfci.get('hostname'),
+ floatingIp.get('ip'),
+ port)
self.details["test_vnf"] = {
'status': "FAIL", 'result': (
"Port %s of server %s -> %s is "
- "not reachable" %
- (port, vnfci.get('hostname'),
- floatingIp.get('ip')))}
- self.step_failure("Test VNF: ERROR")
+ "not reachable",
+ port,
+ vnfci.get('hostname'),
+ floatingIp.get('ip'))}
+ self.logger.error("Test VNF: ERROR")
+ return False
+
self.details["test_vnf"] = {
'status': "PASS",
'result': "All tests have been executed successfully"}
self.logger.info("Test VNF: OK")
- return self.details.get('test_vnf')
+ return True
def clean(self):
self.main_agent.get_agent(
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?
+ # question is the clean removing also the VM?
# I think so since is goinf to remove the tenant...
super(ImsVnf, self).clean()
-
- def main(self, **kwargs):
- self.logger.info("Orchestra IMS VNF onboarding test starting")
- self.execute()
- self.logger.info("Orchestra IMS VNF onboarding test executed")
- if self.result is "PASS":
- return self.EX_OK
- else:
- return self.EX_RUN_ERROR
-
- def run(self):
- kwargs = {}
- return self.main(**kwargs)
-
-
-if __name__ == '__main__':
- logging.basicConfig()
- test = ImsVnf()
- test.deploy_orchestrator()
- test.deploy_vnf()
- test.test_vnf()
- test.clean()
# http://www.apache.org/licenses/LICENSE-2.0
import logging
+import pkg_resources
import unittest
-from git.exc import NoSuchPathError
import mock
from functest.cli.commands import cli_env
return_value=False)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
def test_prepare_default(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/prepare_env.py start" %
- CONST.__getattribute__('dir_repo_functest'))
+ cmd = ("python %s start" % pkg_resources.resource_filename(
+ 'functest', 'ci/prepare_env.py'))
self.cli_environ.prepare()
mock_ft_utils.assert_called_with(cmd)
with mock.patch('__builtin__.raw_input', return_value="y"), \
mock.patch('functest.cli.commands.cli_testcase.os.remove') \
as mock_os_remove:
- cmd = ("python %s/functest/ci/prepare_env.py start" %
- CONST.__getattribute__('dir_repo_functest'))
+ cmd = ("python %s start" % pkg_resources.resource_filename(
+ 'functest', 'ci/prepare_env.py'))
self.cli_environ.prepare()
mock_os_remove.assert_called_once_with(
CONST.__getattribute__('env_active'))
mock_click_echo.assert_called_with(test_utils.
RegexMatch(reg_string))
- @mock.patch('functest.cli.commands.cli_env.git.Repo')
def test_show_missing_ci_installer_type(self, *args):
self._test_show_missing_env_var('INSTALLER_TYPE', *args)
- @mock.patch('functest.cli.commands.cli_env.git.Repo')
def test_show_missing_ci_installer_ip(self, *args):
self._test_show_missing_env_var('INSTALLER_IP', *args)
- @mock.patch('functest.cli.commands.cli_env.git.Repo')
def test_show_missing_ci_scenario(self, *args):
self._test_show_missing_env_var('SCENARIO', *args)
- @mock.patch('functest.cli.commands.cli_env.git.Repo')
def test_show_missing_ci_node(self, *args):
self._test_show_missing_env_var('NODE', *args)
- @mock.patch('functest.cli.commands.cli_env.git.Repo')
def test_show_missing_ci_build_tag(self, *args):
self._test_show_missing_env_var('BUILD_TAG', *args)
- @mock.patch('functest.cli.commands.cli_env.git.Repo')
def test_show_missing_ci_debug(self, *args):
self._test_show_missing_env_var('DEBUG', *args)
- @mock.patch('functest.cli.commands.cli_env.git.Repo')
@mock.patch('functest.cli.commands.cli_env.os.path.isfile',
return_value=False)
def test_show_missing_environment(self, *args):
self._test_show_missing_env_var('STATUS', *args)
- @mock.patch('functest.cli.commands.cli_env.os.path.exists',
- return_value=False)
- def test_show_missing_git_repo_dir(self, *args):
- CONST.__setattr__('dir_repo_functest', None)
- self.assertRaises(NoSuchPathError, lambda: self.cli_environ.show())
-
@mock.patch('functest.cli.commands.cli_env.click.echo')
@mock.patch('functest.cli.commands.cli_env.os.path.isfile',
return_value=True)
#
import logging
+import pkg_resources
import unittest
import os
import mock
from functest.cli.commands import cli_os
-from functest.utils.constants import CONST
class CliOpenStackTesting(unittest.TestCase):
self.installer_type = 'test_installer_type'
self.installer_ip = 'test_installer_ip'
self.openstack_creds = 'test_openstack_creds'
- self.dir_repo_functest = 'test_dir_repo_functest'
self.snapshot_file = 'test_snapshot_file'
self.cli_os = cli_os.CliOpenStack()
@mock.patch('functest.cli.commands.cli_os.ft_utils.execute_command')
def test_check(self, mock_ftutils_execute):
with mock.patch.object(self.cli_os, 'ping_endpoint'):
- CONST.__setattr__('dir_repo_functest', self.dir_repo_functest)
- cmd = os.path.join(CONST.__getattribute__('dir_repo_functest'),
- "functest/ci/check_os.sh")
self.cli_os.check()
- mock_ftutils_execute.assert_called_once_with(cmd, verbose=False)
+ mock_ftutils_execute.assert_called_once_with(
+ "sh %s" % pkg_resources.resource_filename(
+ 'functest', 'ci/check_os.sh'), verbose=False)
@mock.patch('functest.cli.commands.cli_os.os.path.isfile',
return_value=False)
import logging
+import pkg_resources
import unittest
import mock
from functest.cli.commands import cli_testcase
-from functest.utils.constants import CONST
class CliTestCasesTesting(unittest.TestCase):
return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
def test_run_default(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"-n -r ", self.testname))
self.cli_tests.run(self.testname, noclean=True, report=True)
mock_ft_utils.assert_called_with(cmd)
return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
def test_run_noclean_missing_report(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"-n ", self.testname))
self.cli_tests.run(self.testname, noclean=True, report=False)
mock_ft_utils.assert_called_with(cmd)
return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
def test_run_report_missing_noclean(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"-r ", self.testname))
self.cli_tests.run(self.testname, noclean=False, report=True)
mock_ft_utils.assert_called_with(cmd)
return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
def test_run_missing_noclean_report(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"", self.testname))
self.cli_tests.run(self.testname, noclean=False, report=False)
mock_ft_utils.assert_called_with(cmd)
import logging
+import pkg_resources
import unittest
import mock
from functest.cli.commands import cli_tier
-from functest.utils.constants import CONST
class CliTierTesting(unittest.TestCase):
return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
def test_run_default(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"-n -r ", self.tiername))
self.cli_tier.run(self.tiername, noclean=True, report=True)
mock_ft_utils.assert_called_with(cmd)
return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
def test_run_report_missing_noclean(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"-r ", self.tiername))
self.cli_tier.run(self.tiername, noclean=False, report=True)
mock_ft_utils.assert_called_with(cmd)
return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
def test_run_noclean_missing_report(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"-n ", self.tiername))
self.cli_tier.run(self.tiername, noclean=True, report=False)
mock_ft_utils.assert_called_with(cmd)
return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
def test_run_missing_noclean_report(self, mock_ft_utils, mock_os):
- cmd = ("python %s/functest/ci/run_tests.py "
- "%s -t %s" %
- (CONST.__getattribute__('dir_repo_functest'),
+ cmd = ("python %s %s -t %s" %
+ (pkg_resources.resource_filename(
+ 'functest', 'ci/run_tests.py'),
"", self.tiername))
self.cli_tier.run(self.tiername, noclean=False, report=False)
mock_ft_utils.assert_called_with(cmd)
self._test_run(result=mock_result,
status=testcase.TestCase.EX_RUN_ERROR)
self.assertEqual(self.psrunner.result, 0)
- self.assertEqual(self.psrunner.details, {'errors': [], 'failures': []})
+ self.assertEqual(self.psrunner.details,
+ {'errors': 0, 'failures': 0, 'stream': '',
+ 'testsRun': 0})
self.assertEqual(self.psrunner.is_successful(),
testcase.TestCase.EX_TESTCASE_FAILED)
self._test_run(result=mock_result)
self.assertEqual(self.psrunner.result, 96)
self.assertEqual(self.psrunner.details,
- {'errors': [('test1', 'error_msg1')],
- 'failures': [('test2', 'failure_msg1')]})
+ {'errors': 1, 'failures': 1, 'stream': '',
+ 'testsRun': 50})
self.assertEqual(self.psrunner.is_successful(),
testcase.TestCase.EX_TESTCASE_FAILED)
failures=[])
self._test_run(result=mock_result)
self.assertEqual(self.psrunner.result, 100)
- self.assertEqual(self.psrunner.details, {'errors': [], 'failures': []})
+ self.assertEqual(self.psrunner.details,
+ {'errors': 0, 'failures': 0, 'stream': '',
+ 'testsRun': 50})
self.assertEqual(self.psrunner.is_successful(),
testcase.TestCase.EX_OK)
# pylint: disable=missing-docstring
import logging
-import os
import unittest
import mock
from functest.core import vnf
from functest.core import testcase
+from functest.utils import constants
class VnfBaseTesting(unittest.TestCase):
+ """The class testing VNF."""
+ # pylint: disable=missing-docstring,too-many-public-methods
+
+ tenant_name = 'test_tenant_name'
+ tenant_description = 'description'
def setUp(self):
- self.test = vnf.VnfOnBoarding(
- project='functest', case_name='aaa')
- self.test.project = "functest"
- self.test.start_time = "1"
- self.test.stop_time = "5"
- self.test.result = ""
- self.test.details = {
- "orchestrator": {"status": "PASS", "result": "", "duration": 20},
- "vnf": {"status": "PASS", "result": "", "duration": 15},
- "test_vnf": {"status": "FAIL", "result": "", "duration": 5}}
- self.test.keystone_client = 'test_client'
- self.test.tenant_name = 'test_tenant_name'
-
- def test_execute_deploy_vnf_fail(self):
+ constants.CONST.__setattr__("vnf_foo_tenant_name", self.tenant_name)
+ constants.CONST.__setattr__(
+ "vnf_foo_tenant_description", self.tenant_description)
+ self.test = vnf.VnfOnBoarding(project='functest', case_name='foo')
+
+ def test_run_deploy_vnf_exc(self):
with mock.patch.object(self.test, 'prepare'),\
mock.patch.object(self.test, 'deploy_orchestrator',
return_value=None), \
mock.patch.object(self.test, 'deploy_vnf',
side_effect=Exception):
- self.assertEqual(self.test.execute(),
+ self.assertEqual(self.test.run(),
testcase.TestCase.EX_TESTCASE_FAILED)
- def test_execute_test_vnf_fail(self):
+ def test_run_test_vnf_exc(self):
with mock.patch.object(self.test, 'prepare'),\
mock.patch.object(self.test, 'deploy_orchestrator',
return_value=None), \
mock.patch.object(self.test, 'deploy_vnf'), \
mock.patch.object(self.test, 'test_vnf',
side_effect=Exception):
- self.assertEqual(self.test.execute(),
+ self.assertEqual(self.test.run(),
testcase.TestCase.EX_TESTCASE_FAILED)
- @mock.patch('functest.core.vnf.os_utils.get_tenant_id',
- return_value='test_tenant_id')
- @mock.patch('functest.core.vnf.os_utils.delete_tenant',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_user_id',
- return_value='test_user_id')
- @mock.patch('functest.core.vnf.os_utils.delete_user',
- return_value=True)
- def test_execute_default(self, *args):
+ def test_run_deploy_orch_ko(self):
with mock.patch.object(self.test, 'prepare'),\
mock.patch.object(self.test, 'deploy_orchestrator',
- return_value=None), \
- mock.patch.object(self.test, 'deploy_vnf'), \
- mock.patch.object(self.test, 'test_vnf'), \
- mock.patch.object(self.test, 'parse_results',
- return_value='ret_exit_code'), \
- mock.patch.object(self.test, 'log_results'):
- self.assertEqual(self.test.execute(),
- 'ret_exit_code')
-
- @mock.patch('functest.core.vnf.os_utils.get_credentials')
+ return_value=False), \
+ mock.patch.object(self.test, 'deploy_vnf',
+ return_value=True), \
+ mock.patch.object(self.test, 'test_vnf',
+ return_value=True):
+ self.assertEqual(self.test.run(),
+ testcase.TestCase.EX_TESTCASE_FAILED)
+
+ def test_run_vnf_deploy_ko(self):
+ with mock.patch.object(self.test, 'prepare'),\
+ mock.patch.object(self.test, 'deploy_orchestrator',
+ return_value=True), \
+ mock.patch.object(self.test, 'deploy_vnf',
+ return_value=False), \
+ mock.patch.object(self.test, 'test_vnf',
+ return_value=True):
+ self.assertEqual(self.test.run(),
+ testcase.TestCase.EX_TESTCASE_FAILED)
+
+ def test_run_vnf_test_ko(self):
+ with mock.patch.object(self.test, 'prepare'),\
+ mock.patch.object(self.test, 'deploy_orchestrator',
+ return_value=True), \
+ mock.patch.object(self.test, 'deploy_vnf',
+ return_value=True), \
+ mock.patch.object(self.test, 'test_vnf',
+ return_value=False):
+ self.assertEqual(self.test.run(),
+ testcase.TestCase.EX_TESTCASE_FAILED)
+
+ def test_run_default(self):
+ with mock.patch.object(self.test, 'prepare'),\
+ mock.patch.object(self.test, 'deploy_orchestrator',
+ return_value=True), \
+ mock.patch.object(self.test, 'deploy_vnf',
+ return_value=True), \
+ mock.patch.object(self.test, 'test_vnf',
+ return_value=True):
+ self.assertEqual(self.test.run(), testcase.TestCase.EX_OK)
+
+ def test_deploy_vnf_unimplemented(self):
+ with self.assertRaises(vnf.VnfDeploymentException):
+ self.test.deploy_vnf()
+
+ def test_test_vnf_unimplemented(self):
+ with self.assertRaises(vnf.VnfTestException):
+ self.test.test_vnf()
+
@mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.get_user_id', return_value='')
- def test_prepare_missing_userid(self, *args):
- with self.assertRaises(Exception):
- self.test.prepare()
+ @mock.patch('functest.core.vnf.os_utils.delete_user',
+ return_value=True)
+ def test_clean_user_set(self, *args):
+ self.test.user_created = True
+ self.test.clean()
+ args[0].assert_called_once_with(mock.ANY, self.tenant_name)
+ args[1].assert_called_once_with()
- @mock.patch('functest.core.vnf.os_utils.get_credentials')
@mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.get_user_id',
- return_value='test_roleid')
- @mock.patch('functest.core.vnf.os_utils.create_tenant',
- return_value='')
- def test_prepare_missing_tenantid(self, *args):
- with self.assertRaises(Exception):
- self.test.prepare()
+ @mock.patch('functest.core.vnf.os_utils.delete_user',
+ return_value=True)
+ def test_clean_user_unset(self, *args):
+ self.test.user_created = False
+ self.test.clean()
+ args[0].assert_not_called()
+ args[1].assert_called_once_with()
- @mock.patch('functest.core.vnf.os_utils.get_credentials')
@mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.get_user_id',
- return_value='test_roleid')
- @mock.patch('functest.core.vnf.os_utils.create_tenant',
- return_value='test_tenantid')
- @mock.patch('functest.core.vnf.os_utils.get_role_id',
- return_value='')
- def test_prepare_missing_roleid(self, *args):
- with self.assertRaises(Exception):
- self.test.prepare()
+ @mock.patch('functest.core.vnf.os_utils.delete_tenant',
+ return_value=True)
+ def test_clean_tenant_set(self, *args):
+ self.test.tenant_created = True
+ self.test.clean()
+ args[0].assert_called_once_with(mock.ANY, self.tenant_name)
+ args[1].assert_called_once_with()
- @mock.patch('functest.core.vnf.os_utils.get_credentials')
@mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.get_user_id',
- return_value='test_roleid')
- @mock.patch('functest.core.vnf.os_utils.create_tenant',
- return_value='test_tenantid')
- @mock.patch('functest.core.vnf.os_utils.get_role_id',
- return_value='test_roleid')
- @mock.patch('functest.core.vnf.os_utils.add_role_user',
- return_value='')
- def test_prepare_role_add_failure(self, *args):
- with self.assertRaises(Exception):
+ @mock.patch('functest.core.vnf.os_utils.delete_tenant',
+ return_value=True)
+ def test_clean_tenant_unset(self, *args):
+ self.test.tenant_created = False
+ self.test.clean()
+ args[0].assert_not_called()
+ args[1].assert_called_once_with()
+
+ def test_deploy_orch_unimplemented(self):
+ self.assertTrue(self.test.deploy_orchestrator())
+
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ return_value={'creds': 'test'})
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ return_value='test')
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ return_value=0)
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
+ return_value=0)
+ def test_prepare(self, *args):
+ self.assertEqual(self.test.prepare(),
+ testcase.TestCase.EX_OK)
+ args[0].assert_called_once_with('test', self.tenant_name)
+ args[1].assert_called_once_with(
+ 'test', self.tenant_name, self.tenant_description)
+ args[2].assert_called_once_with()
+ args[3].assert_called_once_with()
+
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ side_effect=Exception)
+ def test_prepare_admin_creds_ko(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
self.test.prepare()
+ args[0].assert_called_once_with()
+
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ return_value='creds')
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+ side_effect=Exception)
+ def test_prepare_keystone_client_ko(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
+ self.test.prepare()
+ args[0].assert_called_once_with()
+ args[1].assert_called_once_with()
- @mock.patch('functest.core.vnf.os_utils.get_credentials')
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ return_value='creds')
@mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.get_user_id',
- return_value='test_roleid')
- @mock.patch('functest.core.vnf.os_utils.create_tenant',
- return_value='test_tenantid')
- @mock.patch('functest.core.vnf.os_utils.get_role_id',
- return_value='test_roleid')
- @mock.patch('functest.core.vnf.os_utils.add_role_user')
- @mock.patch('functest.core.vnf.os_utils.create_user',
- return_value='')
- def test_create_user_failure(self, *args):
- with self.assertRaises(Exception):
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ side_effect=Exception)
+ def test_prepare_tenant_creation_ko(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
self.test.prepare()
+ args[0].assert_called_once_with(
+ mock.ANY, self.tenant_name, self.tenant_description)
+ args[1].assert_called_once_with()
+ args[2].assert_called_once_with()
- def test_log_results_default(self):
- with mock.patch('functest.core.vnf.'
- 'ft_utils.logger_test_results') \
- as mock_method:
- self.test.log_results()
- self.assertTrue(mock_method.called)
-
- def test_step_failures_default(self):
- with self.assertRaises(Exception):
- self.test.step_failure("error_msg")
-
- def test_deploy_vnf_unimplemented(self):
- with self.assertRaises(Exception) as context:
- self.test.deploy_vnf()
- self.assertIn('VNF not deployed', str(context.exception))
-
- def test_test_vnf_unimplemented(self):
- with self.assertRaises(Exception) as context:
- self.test.test_vnf()()
- self.assertIn('VNF not tested', str(context.exception))
-
- def test_parse_results_ex_ok(self):
- self.test.details['test_vnf']['status'] = 'PASS'
- self.assertEqual(self.test.parse_results(), os.EX_OK)
-
- def test_parse_results_ex_run_error(self):
- self.test.details['vnf']['status'] = 'FAIL'
- self.assertEqual(self.test.parse_results(), os.EX_SOFTWARE)
+ @mock.patch('functest.core.vnf.os_utils.get_credentials',
+ return_value='creds')
+ @mock.patch('functest.core.vnf.os_utils.get_keystone_client')
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
+ return_value=0)
+ @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
+ side_effect=Exception)
+ def test_prepare_user_creation_ko(self, *args):
+ with self.assertRaises(vnf.VnfPreparationException):
+ self.test.prepare()
+ args[0].assert_called_once_with(mock.ANY, self.tenant_name)
+ args[1].assert_called_once_with(
+ mock.ANY, self.tenant_name, self.tenant_description)
+ args[2].assert_called_once_with()
+ args[3].assert_called_once_with()
if __name__ == "__main__":
+++ /dev/null
-#!/usr/bin/env python
-
-# Copyright (c) 2017 Orange and others.
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-# http://www.apache.org/licenses/LICENSE-2.0
-
-# pylint: disable=missing-docstring
-
-import logging
-import sys
-import unittest
-
-import mock
-
-from functest.core import testcase
-with mock.patch('functest.utils.functest_utils.get_parameter_from_yaml'):
- from functest.opnfv_tests.features import barometer
-
-
-class BarometerTesting(unittest.TestCase):
-
- _case_name = "barometercollectd"
- _project_name = "barometer"
-
- def setUp(self):
- self.barometer = barometer.BarometerCollectd(
- case_name=self._case_name, project_name=self._project_name)
-
- def test_init(self):
- self.assertEqual(self.barometer.project_name, self._project_name)
- self.assertEqual(self.barometer.case_name, self._case_name)
-
- def test_run_ko(self):
- sys.modules['baro_tests'].collectd.main = mock.Mock(return_value=1)
- self.assertEqual(self.barometer.run(),
- testcase.TestCase.EX_RUN_ERROR)
-
- def test_run(self):
- sys.modules['baro_tests'].collectd.main = mock.Mock(return_value=0)
- self.assertEqual(self.barometer.run(), testcase.TestCase.EX_OK)
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
import logging
import mock
-import os
+import pkg_resources
import unittest
from functest.core import testcase
class OSRefstackClientTesting(unittest.TestCase):
- _config = os.path.join(
- CONST.__getattribute__('dir_functest_test'),
- CONST.__getattribute__('refstack_tempest_conf_path'))
- _testlist = os.path.join(
- CONST.__getattribute__('dir_functest_test'),
- CONST.__getattribute__('refstack_defcore_list'))
+ _config = pkg_resources.resource_filename(
+ 'functest',
+ 'opnfv_tests/openstack/refstack_client/refstack_tempest.conf')
+ _testlist = pkg_resources.resource_filename(
+ 'functest', 'opnfv_tests/openstack/refstack_client/defcore.txt')
def setUp(self):
self.defaultargs = {'config': self._config,
self.refstackclient = refstack_client.RefstackClient()
def test_source_venv(self):
- CONST.__setattr__('dir_refstack_client', 'test_repo_dir')
with mock.patch('functest.opnfv_tests.openstack.refstack_client.'
'refstack_client.ft_utils.execute_command') as m:
cmd = ("cd {0};"
mock.patch('__builtin__.open', mock.mock_open()), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.shutil.copyfile'):
- CONST.__setattr__('dir_functest_test', 'test_dir')
- CONST.__setattr__('refstack_tempest_conf_path', 'test_path')
conf_utils.configure_tempest_defcore('test_dep_dir',
img_flavor_dict)
mset.assert_any_call('compute', 'image_ref', 'test_image_id')
mock.patch('__builtin__.open', mock.mock_open()), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.backup_tempest_config'):
- CONST.__setattr__('dir_functest_test', 'test_dir')
CONST.__setattr__('OS_ENDPOINT_TYPE', None)
conf_utils.\
configure_tempest_update_params('test_conf_file',
# http://www.apache.org/licenses/LICENSE-2.0
import logging
+import pkg_resources
import os
import time
import unittest
-from git.exc import NoSuchPathError
import mock
import requests
from six.moves import urllib
self.testcase_dict = {'case_name': 'testname',
'criteria': self.criteria}
self.parameter = 'general.openstack.image_name'
- self.config_yaml = os.path.normpath(os.path.join(os.path.dirname(
- os.path.abspath(__file__)), '../../../ci/config_functest.yaml'))
+ self.config_yaml = pkg_resources.resource_filename(
+ 'functest', 'ci/config_functest.yaml')
self.db_url_env = 'http://foo/testdb'
self.testcases_yaml = "test_testcases_yaml"
self.file_yaml = {'general': {'openstack': {'image_name':
m.assert_called_once_with(dest, 'wb')
self.assertTrue(mock_sh.called)
- def test_get_git_branch(self):
- with mock.patch('functest.utils.functest_utils.Repo') as mock_repo:
- mock_obj2 = mock.Mock()
- attrs = {'name': 'test_branch'}
- mock_obj2.configure_mock(**attrs)
-
- mock_obj = mock.Mock()
- attrs = {'active_branch': mock_obj2}
- mock_obj.configure_mock(**attrs)
-
- mock_repo.return_value = mock_obj
- self.assertEqual(functest_utils.get_git_branch(self.repo_path),
- 'test_branch')
-
- @mock.patch('functest.utils.functest_utils.Repo',
- side_effect=NoSuchPathError)
- def test_get_git_branch_failed(self, mock_repo):
- self.assertRaises(NoSuchPathError,
- lambda: functest_utils.get_git_branch(self.repo_path
- ))
-
@mock.patch('functest.utils.functest_utils.logger.error')
def test_get_installer_type_failed(self, mock_logger_error):
with mock.patch.dict(os.environ,
def test_get_dict_by_test(self, mock_logger_error):
with mock.patch('six.moves.builtins.open', mock.mock_open()), \
mock.patch('functest.utils.functest_utils.yaml.safe_load') \
- as mock_yaml, \
- mock.patch('functest.utils.functest_utils.get_testcases_'
- 'file_dir'):
+ as mock_yaml:
mock_obj = mock.Mock()
attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
mock_obj.configure_mock(**attrs)
assert_called_once_with(self.parameter,
self.config_yaml)
- # TODO: merge_dicts
-
- @mock.patch('functest.utils.functest_utils.get_functest_config')
- def test_get_testcases_file_dir(self, mock_get_functest_config):
- mock_get_functest_config.return_value = self.testcases_yaml
- resp = functest_utils.get_testcases_file_dir()
- self.assertEqual(resp, self.testcases_yaml)
- mock_get_functest_config.assert_called_once_with(
- 'general.functest.testcases_yaml')
-
def test_get_functest_yaml(self):
with mock.patch('six.moves.builtins.open', mock.mock_open()), \
mock.patch('functest.utils.functest_utils.yaml.safe_load') \
None)
self.assertTrue(mock_logger_error.called)
+ def test_get_or_create_user_for_vnf_get(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'get_user_id',
+ return_value='user_id'), \
+ mock.patch('functest.utils.openstack_utils.get_tenant_id',
+ return_value='tenant_id'):
+ self.assertFalse(openstack_utils.
+ get_or_create_user_for_vnf(self.keystone_client,
+ 'my_vnf'))
+
+ def test_get_or_create_user_for_vnf_create(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'get_user_id',
+ return_value=None), \
+ mock.patch('functest.utils.openstack_utils.get_tenant_id',
+ return_value='tenant_id'):
+ self.assertTrue(openstack_utils.
+ get_or_create_user_for_vnf(self.keystone_client,
+ 'my_vnf'))
+
+ def test_get_or_create_user_for_vnf_error_get_user_id(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'get_user_id',
+ side_effect=Exception):
+ self.assertRaises(Exception)
+
+ def test_get_or_create_user_for_vnf_error_get_tenant_id(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'get_user_id',
+ return_value='user_id'), \
+ mock.patch('functest.utils.openstack_utils.get_tenant_id',
+ side_effect='Exception'):
+ self.assertRaises(Exception)
+
+ def test_get_or_create_tenant_for_vnf_get(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'get_tenant_id',
+ return_value='tenant_id'):
+ self.assertFalse(
+ openstack_utils.get_or_create_tenant_for_vnf(
+ self.keystone_client, 'tenant_name', 'tenant_description'))
+
+ def test_get_or_create_tenant_for_vnf_create(self):
+ with mock.patch('functest.utils.openstack_utils.get_tenant_id',
+ return_value=None):
+ self.assertTrue(
+ openstack_utils.get_or_create_tenant_for_vnf(
+ self.keystone_client, 'tenant_name', 'tenant_description'))
+
+ def test_get_or_create_tenant_for_vnf_error_get_tenant_id(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'get_tenant_id',
+ side_effect=Exception):
+ self.assertRaises(Exception)
+
+ def test_download_and_add_image_on_glance_image_creation_failure(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'os.makedirs'), \
+ mock.patch('functest.utils.openstack_utils.'
+ 'ft_utils.download_url',
+ return_value=True), \
+ mock.patch('functest.utils.openstack_utils.'
+ 'create_glance_image',
+ return_value=''):
+ resp = openstack_utils.download_and_add_image_on_glance(
+ self.glance_client,
+ 'image_name',
+ 'http://url',
+ 'data_dir')
+ self.assertEqual(resp, False)
+
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.'
'os_utils.get_image_id',
return_value=''), \
- mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.'
+ mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.os_utils.'
'download_and_add_image_on_glance') as m, \
self.assertRaises(Exception) as context:
self.ims_vnf.deploy_orchestrator()
{'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.'
- 'os.makedirs'), \
- mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.'
- 'ft_utils.download_url',
- return_value=False):
- resp = cloudify_ims.download_and_add_image_on_glance(self.
- glance_client,
- 'image_name',
- 'http://url',
- 'data_dir')
- self.assertEqual(resp, False)
-
- def test_download_and_add_image_on_glance_image_creation_failure(self):
- with mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.'
- 'os.makedirs'), \
- mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.'
- 'ft_utils.download_url',
- return_value=True), \
- mock.patch('functest.opnfv_tests.vnf.ims.cloudify_ims.'
- 'os_utils.create_glance_image',
- return_value=''):
- resp = cloudify_ims.download_and_add_image_on_glance(self.
- glance_client,
- 'image_name',
- 'http://url',
- 'data_dir')
- self.assertEqual(resp, False)
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
'cookies': ""}
self.mock_post_200.configure_mock(**attrs)
- def test_create_ellis_number_failure(self):
- with mock.patch('functest.opnfv_tests.vnf.ims.'
- 'clearwater_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__":
logging.disable(logging.CRITICAL)
unittest.main(verbosity=2)
#!/usr/bin/env python
+import pkg_resources
import os
import re
'BUILD_TAG': None,
'OS_ENDPOINT_TYPE': None,
'OS_AUTH_URL': None,
- 'CONFIG_FUNCTEST_YAML': os.path.normpath(os.path.join(os.path.dirname(
- os.path.abspath(__file__)), '../ci/config_functest.yaml'))
+ 'CONFIG_FUNCTEST_YAML': pkg_resources.resource_filename(
+ 'functest', 'ci/config_functest.yaml')
}
import json
import logging
import os
+import pkg_resources
import re
import shutil
import subprocess
import requests
from six.moves import urllib
import yaml
-from git import Repo
from functest.utils import constants
from functest.utils import decorators
# CI UTILS
#
# -----------------------------------------------------------
-def get_git_branch(repo_path):
- """
- Get git branch name
- """
- repo = Repo(repo_path)
- branch = repo.active_branch
- return branch.name
-
-
def get_installer_type():
"""
Get installer type (fuel, apex, joid, compass)
def get_dict_by_test(testname):
- with open(get_testcases_file_dir()) as f:
+ with open(pkg_resources.resource_filename(
+ 'functest', 'ci/testcases.yaml')) as f:
testcases_yaml = yaml.safe_load(f)
for dic_tier in testcases_yaml.get("tiers"):
yield (k, dict2[k])
-def get_testcases_file_dir():
- return get_functest_config('general.functest.testcases_yaml')
-
-
def get_functest_yaml():
with open(constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')) as f:
functest_yaml = yaml.safe_load(f)
return heatclient.Client(get_heat_client_version(), session=sess)
+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 = create_glance_image(
+ glance, image_name, dest_path + file_name)
+ if not image:
+ return False
+
+ return image
+
+
# *********************************************
# NOVA
# *********************************************
return tenant_id
+def get_or_create_tenant_for_vnf(keystone_client, tenant_name,
+ tenant_description):
+ """Get or Create a Tenant
+
+ Args:
+ keystone_client: keystone client reference
+ tenant_name: the name of the tenant
+ tenant_description: the description of the tenant
+
+ return False if tenant retrieved though get
+ return True if tenant created
+ raise Exception if error during processing
+ """
+ try:
+ tenant_id = get_tenant_id(keystone_client, tenant_name)
+ if not tenant_id:
+ tenant_id = create_tenant(keystone_client, tenant_name,
+ tenant_description)
+ return True
+ else:
+ return False
+ except:
+ raise Exception("Impossible to create a Tenant for the VNF {}".format(
+ tenant_name))
+
+
def create_user(keystone_client, user_name, user_password,
user_email, tenant_id):
try:
return user_id
+def get_or_create_user_for_vnf(keystone_client, vnf_ref):
+ """Get or Create user for VNF
+
+ Args:
+ keystone_client: keystone client reference
+ vnf_ref: VNF reference used as user name & password, tenant name
+
+ return False if user retrieved through get
+ return True if user created
+ raise Exception if error during processing
+ """
+ try:
+ user_id = get_user_id(keystone_client, vnf_ref)
+ tenant_id = get_tenant_id(keystone_client, vnf_ref)
+ if not user_id:
+ user_id = create_user(keystone_client, vnf_ref, vnf_ref,
+ "", tenant_id)
+ return True
+ else:
+ return False
+ add_role_user(keystone_client, user_id, 'admin', vnf_ref)
+ except:
+ raise Exception("Impossible to create a user for the VNF {}".format(
+ vnf_ref))
+
+
def add_role_user(keystone_client, user_id, role_id, tenant_id):
try:
if is_keystone_v3():
+++ /dev/null
-#
-#
-# 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
-#
-ddt==1.1.1
-oslosphinx==4.11.0
-oslotest==2.14.0
-pylint==1.4.5
-requests-mock==1.3.0
-tempest-lib==1.0.0
-testresources==2.0.1
-testscenarios==0.5.0
+++ /dev/null
-#
-#
-# 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
-#
-pyyaml==3.10
-gitpython==1.0.1
-python-openstackclient==2.3.0
-python-ceilometerclient==2.6.2
-python-heatclient==1.7.0
-python-keystoneclient==3.5.0
-python-neutronclient==6.0.0
-python-novaclient==6.0.0
-python-congressclient==1.5.0
-python-tackerclient==0.7.0
-pexpect==4.0
-requests==2.9.1
-robotframework==3.0.2
-robotframework-httplibrary==0.4.2
-robotframework-requests==0.4.7
-jmespath==0.9.2
-configObj==5.0.6
-Flask==0.10.1
-xmltodict==0.9.2
-scp==0.10.2
-paramiko==2.1.2
-shyaml
-dnspython
-Pillow==3.3.0
-click==6.6
-openbaton-cli==2.2.1-beta7
-mock==1.3.0
-iniparse==0.4
-PrettyTable>=0.7.1,<0.8 # BSD
-six>=1.9.0 # MIT
-git+https://gerrit.opnfv.org/gerrit/releng#egg=opnfv&subdirectory=modules
-git+https://gerrit.opnfv.org/gerrit/barometer#egg=baro_tests
-#
-#
-# 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
-#
-pyyaml==3.10
-gitpython==1.0.1
-python-openstackclient==2.3.0
-python-ceilometerclient==2.6.2
-python-heatclient==1.7.0
-python-keystoneclient==3.5.0
-python-neutronclient==6.0.0
-python-novaclient==6.0.0
-python-congressclient==1.5.0
-python-tackerclient==0.7.0
-pexpect==4.0
-requests==2.9.1
+pbr>=1.8 # Apache-2.0
+PyYAML>=3.10.0 # MIT
+GitPython>=1.0.1 # BSD License (3 clause)
+keystoneauth1>=2.18.0 # Apache-2.0
+python-cinderclient!=1.7.0,!=1.7.1,>=1.6.0 # Apache-2.0
+python-glanceclient>=2.5.0 # Apache-2.0
+python-heatclient>=1.6.1 # Apache-2.0
+python-keystoneclient>=3.8.0 # Apache-2.0
+python-neutronclient>=5.1.0 # Apache-2.0
+python-novaclient!=7.0.0,>=6.0.0 # Apache-2.0
+python-tackerclient>=0.8.0 # Apache-2.0
+pexpect!=3.3,>=3.1 # ISC License
+requests!=2.12.2,>=2.10.0 # Apache-2.0
robotframework==3.0.2
robotframework-httplibrary==0.4.2
robotframework-requests==0.4.7
-robotframework-sshlibrary==2.1.3
-jmespath==0.9.2
-configObj==5.0.6
-Flask==0.10.1
-xmltodict==0.9.2
+robotframework-sshlibrary==2.1.3;python_version=='2.7'
scp==0.10.2
-paramiko==2.1.2
-subprocess32
-shyaml
-dnspython
-Pillow==3.3.0
+subprocess32;python_version=='2.7'
+dnspython>=1.14.0;python_version=='2.7' # http://www.dnspython.org/LICENSE
+dnspython3!=1.13.0,!=1.14.0,>=1.12.0;python_version>='3.0' # http://www.dnspython.org/LICENSE
click==6.6
openbaton-cli==2.2.1-beta7
-mock==1.3.0
+mock>=2.0 # BSD
iniparse==0.4
-PrettyTable>=0.7.1,<0.8 # BSD
+PrettyTable<0.8,>=0.7.1 # BSD
six>=1.9.0 # MIT
-git+https://gerrit.opnfv.org/gerrit/releng#egg=opnfv&subdirectory=modules
-git+https://gerrit.opnfv.org/gerrit/barometer#egg=baro_tests
-git+https://gerrit.opnfv.org/gerrit/snaps#egg=snaps
+opnfv
+snaps
--- /dev/null
+[metadata]
+name = functest
+version = 5
+home-page = https://wiki.opnfv.org/display/functest
+
+[files]
+packages = functest
+scripts =
+ docker/docker_remote_api/enable_remote_api.sh
+ docker/add_images.sh
+ docker/config_install_env.sh
+
+[entry_points]
+console_scripts =
+ functest = functest.cli.cli_base:cli
-##############################################################################
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Orange and others.
+#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
-##############################################################################
-from setuptools import setup, find_packages
+# pylint: disable=missing-docstring
+
+import setuptools
+# In python < 2.7.4, a lazy loading of package `pbr` will break
+# setuptools if some other modules registered functions in `atexit`.
+# solution from: http://bugs.python.org/issue15881#msg170215
+try:
+ import multiprocessing # noqa
+except ImportError:
+ pass
-setup(
- name="functest",
- version="master",
- py_modules=['cli_base'],
- packages=find_packages(),
- include_package_data=True,
- package_data={
- },
- url="https://www.opnfv.org",
- entry_points={
- 'console_scripts': [
- 'functest=functest.cli.cli_base:cli'
- ],
- },
-)
+setuptools.setup(
+ setup_requires=['pbr>=1.8'],
+ pbr=True)
-#
-#
-# 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
-#
-coverage==4.1
-mock==1.3.0
-nose==1.3.7
-flake8>=2.5.4,<2.6.0 # MIT
+coverage>=4.0 # Apache-2.0
+mock>=2.0 # BSD
+nose # LGPL
+flake8<2.6.0,>=2.5.4 # MIT
pylint==1.4.5 # GPLv2
-sphinx!=1.6.1,>=1.5.1 # BSD
+sphinx!=1.3b1,<1.4,>=1.2.1 # BSD
deps =
-r{toxinidir}/requirements.txt
-r{toxinidir}/test-requirements.txt
+ git+https://gerrit.opnfv.org/gerrit/releng#egg=opnfv&subdirectory=modules
+ git+https://gerrit.opnfv.org/gerrit/snaps#egg=snaps
commands = nosetests --with-xunit \
--with-coverage \
--cover-tests \
[testenv:pep8]
basepython = python2.7
-deps =
- {[testenv]deps}
commands = flake8
[testenv:pylint]
basepython = python2.7
-deps =
- {[testenv]deps}
whitelist_externals = bash
modules =
- functest.core.feature
- functest.core.testcase
- functest.core.unit
+ functest.core
functest.opnfv_tests.sdn.odl
- functest.tests.unit.core.test_feature
- functest.tests.unit.core.test_testcase
- functest.tests.unit.core.test_unit
+ functest.tests.unit.core
functest.tests.unit.odl
functest.tests.unit.utils.test_decorators
functest.utils.decorators
functest/tests/unit/core
functest/tests/unit/odl
functest/tests/unit/utils/test_decorators.py
-deps =
- -r{toxinidir}/requirements.py3.txt
- -r{toxinidir}/test-requirements.txt
commands = nosetests {[testenv:py35]dirs}
Code Review / functest-xtesting.git / commitdiff