RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/netready ${REPOS_DIR}/netready
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/fds ${REPOS_DIR}/fds
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/sfc ${REPOS_DIR}/sfc
-RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/snaps ${REPOS_DIR}/snaps
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/securityscanning ${REPOS_DIR}/securityscanning
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/opera ${REPOS_DIR}/opera
RUN gpg --keyserver hkp://p80.pool.sks-keyservers.net:80 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3
RUN curl -L https://get.rvm.io | bash -s stable
-# SNAPS integration
-RUN pip install -e ${REPOS_DIR}/snaps/
-
# SFC integration
RUN /bin/bash -c ". ${REPOS_DIR}/sfc/sfc/tests/functest/setup_scripts/tacker_client_install.sh"
RUN cd ${REPOS_DIR}/sfc && pip install -e .
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/doctor ${REPOS_DIR}/doctor
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/promise ${REPOS_DIR}/promise
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/sfc ${REPOS_DIR}/sfc
-RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/snaps ${REPOS_DIR}/snaps
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/securityscanning ${REPOS_DIR}/securityscanning
RUN git clone --depth 1 https://gerrit.opnfv.org/gerrit/opera ${REPOS_DIR}/opera
RUN gpg --keyserver hkp://p80.pool.sks-keyservers.net:80 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3
RUN curl -L https://get.rvm.io | bash -s stable
-# SNAPS integration
-RUN pip install -e ${REPOS_DIR}/snaps/
-
RUN pip install ${REPOS_DIR}/opera
# SFC integration
### methods
-| Method | Purpose |
-|-------------------|--------------------------------------------|
-| run(**kwargs) | run the test case |
-| is_successful() | interpret the results of the test case |
-| get_duration() | return the duration of the test case |
-| push_to_db() | push the results of the test case to the DB|
+| Method | Purpose |
+|-------------------|------------------------------|
+| run(**kwargs) | run the test case |
+| is_successful() | interpret the results |
+| get_duration() | return the duration |
+| push_to_db() | push the results to the DB |
+| create_snapshot() | save the testing environment |
+| clean() | clean the resources |
### run(**kwargs)
project_name: functest
criteria: 100
blocking: true
-clean_flag: false
description: ''
dependencies:
installer: ''
project_name: functest
criteria: 100
blocking: true
-clean_flag: false
description: ''
dependencies:
installer: ''
project_name: functest
criteria: 100
blocking: true
-clean_flag: false
description: ''
dependencies:
installer: ''
project_name: functest
criteria: 100
blocking: true
-clean_flag: false
description: ''
dependencies:
installer: ''
all the OpenStack resources (images, networks, volumes, security groups, tenants,
users) so that an eventual cleanup does not remove any of these defaults.
-It is also called before running a test except if it is disabled by configuration
-in the testcases.yaml file (clean_flag=false). This flag has been added as some
-upstream tests already include their own cleaning mechanism (e.g. Rally).
-
The script **openstack_clean.py** which is located in
-*$REPOS_DIR/functest/functest/utils/* is normally called after a test execution. It is
-in charge of cleaning the OpenStack resources that are not specified in the
-defaults file generated previously which is stored in
+*$REPOS_DIR/functest/functest/utils/* is in charge of cleaning the OpenStack resources
+that are not specified in the defaults file generated previously which is stored in
*/home/opnfv/functest/conf/openstack_snapshot.yaml* in the Functest docker container.
It is important to mention that if there are new OpenStack resources created
* the name of the test case
* the criteria (experimental): a criteria used to declare the test case as PASS or FAIL
* blocking: if set to true, if the test is failed, the execution of the following tests is canceled
- * clean_flag: shall the functect internal mechanism be invoked after the test
* the description of the test case
* the dependencies: a combination of 2 regex on the scenario and the installer name
* run: In Danube we introduced the notion of abstract class in order to harmonize the way to run internal, feature or vnf tests
CONST.__setattr__('OS_PASSWORD', value)
-def patch_config_file():
+def update_config_file():
patch_file(CONFIG_PATCH_PATH)
if pod_arch and pod_arch in arch_filter:
patch_file(CONFIG_AARCH64_PATCH_PATH)
+ if "TEST_DB_URL" in os.environ:
+ update_db_url()
+
def patch_file(patch_file_path):
logger.debug('Updating file: %s', patch_file_path)
os.remove(CONFIG_FUNCTEST_PATH)
with open(CONFIG_FUNCTEST_PATH, "w") as f:
f.write(yaml.dump(new_functest_yaml, default_style='"'))
- f.close()
+
+
+def update_db_url():
+ with open(CONFIG_FUNCTEST_PATH) as f:
+ functest_yaml = yaml.safe_load(f)
+
+ with open(CONFIG_FUNCTEST_PATH, "w") as f:
+ functest_yaml["results"]["test_db_url"] = os.environ.get('TEST_DB_URL')
+ f.write(yaml.dump(functest_yaml, default_style='"'))
def verify_deployment():
get_deployment_handler()
create_directories()
source_rc_file()
- patch_config_file()
+ update_config_file()
verify_deployment()
install_rally()
install_tempest()
import functest.ci.tier_builder as tb
import functest.core.testcase as testcase
import functest.utils.functest_utils as ft_utils
-import functest.utils.openstack_clean as os_clean
-import functest.utils.openstack_snapshot as os_snapshot
import functest.utils.openstack_utils as os_utils
from functest.utils.constants import CONST
elif key == 'OS_PASSWORD':
CONST.__setattr__('OS_PASSWORD', value)
- @staticmethod
- def generate_os_snapshot():
- os_snapshot.main()
-
- @staticmethod
- def cleanup():
- os_clean.main()
-
@staticmethod
def get_run_dict(testname):
try:
"The test case {} is not enabled".format(test.get_name()))
logger.info("\n") # blank line
self.print_separator("=")
- logger.info("Running test case '%s'..." % test.get_name())
+ logger.info("Running test case '%s'...", test.get_name())
self.print_separator("=")
logger.debug("\n%s" % test)
self.source_rc_file()
- if test.needs_clean() and self.clean_flag:
- self.generate_os_snapshot()
-
flags = " -t %s" % test.get_name()
if self.report_flag:
flags += " -r"
test_dict = ft_utils.get_dict_by_test(test.get_name())
test_case = cls(**test_dict)
self.executed_test_cases.append(test_case)
+ if self.clean_flag:
+ if test_case.create_snapshot() != test_case.EX_OK:
+ return result
try:
kwargs = run_dict['args']
result = test_case.run(**kwargs)
test_case.push_to_db()
result = test_case.is_successful()
logger.info("Test result:\n\n%s\n", test_case)
+ if self.clean_flag:
+ test_case.clean()
except ImportError:
logger.exception("Cannot import module {}".format(
run_dict['module']))
else:
raise Exception("Cannot import the class for the test case.")
- if test.needs_clean() and self.clean_flag:
- self.cleanup()
- if result != testcase.TestCase.EX_OK:
- logger.error("The test case '%s' failed. " % test.get_name())
- self.overall_result = Result.EX_ERROR
- if test.is_blocking():
- raise BlockingTestFailed(
- "The test case {} failed and is blocking".format(
- test.get_name()))
+ return result
def run_tier(self, tier):
tier_name = tier.get_name()
self.print_separator("#")
logger.debug("\n%s" % tier)
for test in tests:
- self.run_test(test, tier_name)
+ result = self.run_test(test, tier_name)
+ if result != testcase.TestCase.EX_OK:
+ logger.error("The test case '%s' failed.", test.get_name())
+ self.overall_result = Result.EX_ERROR
+ if test.is_blocking():
+ raise BlockingTestFailed(
+ "The test case {} failed and is blocking".format(
+ test.get_name()))
def run_all(self, tiers):
summary = ""
if _tiers.get_tier(kwargs['test']):
self.run_tier(_tiers.get_tier(kwargs['test']))
elif _tiers.get_test(kwargs['test']):
- self.run_test(_tiers.get_test(kwargs['test']),
- _tiers.get_tier_name(kwargs['test']),
- kwargs['test'])
+ result = self.run_test(
+ _tiers.get_test(kwargs['test']),
+ _tiers.get_tier_name(kwargs['test']),
+ kwargs['test'])
+ if result != testcase.TestCase.EX_OK:
+ logger.error("The test case '%s' failed.",
+ kwargs['test'])
+ self.overall_result = Result.EX_ERROR
elif kwargs['test'] == "all":
self.run_all(_tiers)
else:
project_name: functest
criteria: 100
blocking: true
- clean_flag: false
description: >-
This test case verifies the retrieval of OpenStack clients:
Keystone, Glance, Neutron and Nova and may perform some
project_name: functest
criteria: 100
blocking: true
- clean_flag: false
description: >-
This test case verifies the retrieval of OpenStack clients:
Keystone, Glance, Neutron and Nova and may perform some
project_name: functest
criteria: 100
blocking: true
- clean_flag: false
description: >-
This test case creates executes the SimpleHealthCheck
Python test class which creates an, image, flavor, network,
project_name: functest
criteria: 100
blocking: true
- clean_flag: true
description: >-
This test case verifies: 1) SSH to an instance using floating
IPs over the public network. 2) Connectivity between 2 instances
project_name: functest
criteria: 100
blocking: true
- clean_flag: true
description: >-
This test case verifies: 1) Boot a VM with given userdata.
2) Connectivity between 2 instances over a private network.
project_name: functest
criteria: 100
blocking: false
- clean_flag: true
description: >-
This test case runs the smoke subset of the OpenStack
Tempest suite. The list of test cases is generated by
project_name: functest
criteria: 100
blocking: false
- clean_flag: false
description: >-
This test case runs a sub group of tests of the OpenStack
Rally suite in smoke mode.
project_name: functest
criteria: 100
blocking: false
- clean_flag: true
description: >-
This test case runs a sub group of tests of the OpenStack
Defcore testcases by using refstack client.
project_name: functest
criteria: 100
blocking: true
- clean_flag: false
description: >-
Test Suite for the OpenDaylight SDN Controller. It
integrates some test suites from upstream using
project_name: functest
criteria: 100
blocking: false
- clean_flag: false
description: >-
Test Suite for the OpenDaylight SDN Controller when
the NetVirt features are installed. It integrates
project_name: functest
criteria: 100
blocking: false
- clean_flag: false
description: >-
Test Suite for the OpenDaylight SDN Controller when GBP features are
installed. It integrates some test suites from upstream using
project_name: functest
criteria: 100
blocking: true
- clean_flag: true
description: >-
Test Suite for the ONOS SDN Controller. It integrates
some test suites from upstream using TestON as the test
project_name: functest
criteria: 100
blocking: false
- clean_flag: false
description: >-
This test case contains tests that setup and destroy
environments with VMs with and without Floating IPs
project_name: promise
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite from Promise project.
dependencies:
project_name: doctor
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite from Doctor project.
dependencies:
project_name: sdnvpn
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite from SDNVPN project.
dependencies:
project_name: securityscanning
criteria: 100
blocking: false
- clean_flag: true
description: >-
Simple Security Scan
dependencies:
project_name: copper
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite for policy management based on OpenStack Congress
dependencies:
project_name: multisite
criteria: 100
blocking: false
- clean_flag: false
description: >-
Test suite from kingbird
dependencies:
project_name: sfc
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite for odl-sfc to test two chains and two SFs
dependencies:
project_name: functest
criteria: 100
blocking: true
- clean_flag: true
description: >-
Test Suite for onos-sfc to test sfc function.
dependencies:
project_name: parser
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite from Parser project.
dependencies:
project_name: domino
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite from Domino project.
dependencies:
project_name: netready
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite from Netready project.
dependencies:
project_name: barometer
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite for the Barometer project. Separate tests verify the
proper configuration and functionality of the following
project_name: functest
criteria: 80
blocking: false
- clean_flag: true
description: >-
The list of test cases is generated by
Tempest automatically and depends on the parameters of
project_name: functest
criteria: 100
blocking: false
- clean_flag: true
description: >-
The test case allows running a customized list of tempest
test cases defined in a file under
project_name: functest
criteria: 90
blocking: false
- clean_flag: false
description: >-
This test case runs the full suite of scenarios of the OpenStack
Rally suite using several threads and iterations.
project_name: functest
criteria: 100
blocking: false
- clean_flag: true
description: >-
This test case deploys an OpenSource vIMS solution from Clearwater
using the Cloudify orchestrator. It also runs some signaling traffic.
project_name: functest
criteria: 100
blocking: false
- clean_flag: true
description: >-
Test suite from Parser project.
dependencies:
project_name: functest
criteria: 100
blocking: false
- clean_flag: true
description: >-
VNF deployment with OpenBaton (Orchestra)
dependencies:
project_name: opera
criteria: 100
blocking: false
- clean_flag: true
description: >-
VNF deployment with OPEN-O
dependencies:
project_name: functest
criteria: 100
blocking: false
- clean_flag: true
description: >-
This test case is vRouter testing.
dependencies:
dependency=dep,
criteria=dic_testcase['criteria'],
blocking=dic_testcase['blocking'],
- clean_flag=dic_testcase['clean_flag'],
description=dic_testcase['description'])
if (testcase.is_compatible(self.ci_installer,
self.ci_scenario) and
dependency,
criteria,
blocking,
- clean_flag,
description=""):
self.name = name
self.enabled = enabled
self.dependency = dependency
self.criteria = criteria
self.blocking = blocking
- self.clean_flag = clean_flag
self.description = description
@staticmethod
def is_blocking(self):
return self.blocking
- def needs_clean(self):
- return self.clean_flag
-
def __str__(self):
lines = split_text(self.description, LINE_LENGTH - 6)
import prettytable
import functest.utils.functest_utils as ft_utils
+import functest.utils.openstack_clean as os_clean
+import functest.utils.openstack_snapshot as os_snapshot
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
except Exception: # pylint: disable=broad-except
self.__logger.exception("The results cannot be pushed to DB")
return TestCase.EX_PUSH_TO_DB_ERROR
+
+ def create_snapshot(self): # pylint: disable=no-self-use
+ """Save the testing environment before running test.
+
+ It can be overriden if resources must be listed running the
+ test case.
+
+ Returns:
+ TestCase.EX_OK
+ """
+ return TestCase.EX_OK
+
+ def clean(self):
+ """Clean the resources.
+
+ It can be overriden if resources must be deleted after
+ running the test case.
+ """
+
+
+class OSGCTestCase(TestCase):
+ """Model for single test case which requires an OpenStack Garbage
+ Collector."""
+
+ __logger = logging.getLogger(__name__)
+
+ def create_snapshot(self):
+ """Create a snapshot listing the OpenStack resources.
+
+ Returns:
+ TestCase.EX_OK if os_snapshot.main() returns 0.
+ TestCase.EX_RUN_ERROR otherwise.
+ """
+ try:
+ assert os_snapshot.main() == 0
+ self.__logger.info("OpenStack resources snapshot created")
+ return TestCase.EX_OK
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Cannot create the snapshot")
+ return TestCase.EX_RUN_ERROR
+
+ def clean(self):
+ """Clean the OpenStack resources."""
+ try:
+ assert os_clean.main() == 0
+ self.__logger.info("OpenStack resources cleaned")
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Cannot clean the OpenStack resources")
logger = logging.getLogger(__name__)
-class RallyBase(testcase.TestCase):
+class RallyBase(testcase.OSGCTestCase):
TESTS = ['authenticate', 'glance', 'cinder', 'heat', 'keystone',
'neutron', 'nova', 'quotas', 'requests', 'vm', 'all']
GLANCE_IMAGE_NAME = CONST.__getattribute__('openstack_image_name')
logger = logging.getLogger(__name__)
-class RefstackClient(testcase.TestCase):
+class RefstackClient(testcase.OSGCTestCase):
def __init__(self, **kwargs):
if "case_name" not in kwargs:
-# Copyright (c) 2015 All rights reserved
+# Copyright (c) 2017 Cable Television Laboratories, Inc. and others.
+#
# 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
self.suite = unittest.TestSuite()
+ def run(self, **kwargs):
+ """
+ Builds the test suite then calls super.run()
+ :param kwargs: the arguments to pass on
+ :return:
+ """
test_suite_builder.add_openstack_client_tests(
suite=self.suite,
os_creds=self.os_creds,
ext_net_name=self.ext_net_name,
use_keystone=self.use_keystone)
+ return super(self.__class__, self).run()
-# Copyright (c) 2015 All rights reserved
+# Copyright (c) 2017 Cable Television Laboratories, Inc. and others.
+#
# 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
super(SnapsTestRunner, self).__init__(**kwargs)
self.logger = logging.getLogger(__name__)
- self.os_creds = openstack_tests.get_credentials(
- os_env_file=CONST.__getattribute__('openstack_creds'),
- proxy_settings_str=None, ssh_proxy_cmd=None)
+ if 'os_creds' in kwargs:
+ self.os_creds = kwargs['os_creds']
+ else:
+ self.os_creds = openstack_tests.get_credentials(
+ os_env_file=CONST.__getattribute__('openstack_creds'),
+ proxy_settings_str=None, ssh_proxy_cmd=None)
+
+ if 'ext_net_name' in kwargs:
+ self.ext_net_name = kwargs['ext_net_name']
+ else:
+ self.ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
- self.ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
self.use_fip = CONST.__getattribute__('snaps_use_floating_ips')
self.use_keystone = CONST.__getattribute__('snaps_use_keystone')
scenario = functest_utils.get_scenario()
- self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_ANY
+ self.flavor_metadata = None
if 'ovs' in scenario or 'fdio' in scenario:
self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
logger = logging.getLogger(__name__)
-class TempestCommon(testcase.TestCase):
+class TempestCommon(testcase.OSGCTestCase):
def __init__(self, **kwargs):
super(TempestCommon, self).__init__(**kwargs)
import time
import uuid
-from functest.core.testcase import TestCase
+from functest.core import testcase
from functest.utils import functest_utils
from functest.utils.constants import CONST
from snaps.openstack.utils import deploy_utils, nova_utils
-class VPingBase(TestCase):
+class VPingBase(testcase.OSGCTestCase):
"""
Base class for vPing tests that check connectivity between two VMs shared
scenario = functest_utils.get_scenario()
- self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_ANY
+ self.flavor_metadata = None
if 'ovs' in scenario or 'fdio' in scenario:
self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
else:
raise Exception('VMs never became active')
- if result == TestCase.EX_RUN_ERROR:
- return TestCase.EX_RUN_ERROR
+ if result == testcase.TestCase.EX_RUN_ERROR:
+ return testcase.TestCase.EX_RUN_ERROR
self.stop_time = time.time()
self.result = 100
- return TestCase.EX_OK
+ return testcase.TestCase.EX_OK
def _cleanup(self):
"""
from six.moves import urllib
from functest.core import testcase
+from functest.utils import constants
import functest.utils.openstack_utils as op_utils
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
class ODLTests(testcase.TestCase):
"""ODL test runner."""
- repos = "/home/opnfv/repos/"
- odl_test_repo = os.path.join(repos, "odl_test")
+ odl_test_repo = os.path.join(
+ constants.CONST.__getattribute__('dir_repos'), 'odl_test')
neutron_suite_dir = os.path.join(odl_test_repo,
"csit/suites/openstack/neutron")
basic_suite_dir = os.path.join(odl_test_repo,
"csit/suites/integration/basic")
default_suites = [basic_suite_dir, neutron_suite_dir]
- res_dir = '/home/opnfv/functest/results/odl/'
+ res_dir = os.path.join(
+ constants.CONST.__getattribute__('dir_results'), 'odl')
__logger = logging.getLogger(__name__)
@classmethod
import time
from functest.utils.constants import CONST
-import functest.utils.functest_utils as ft_utils
class Foundation(object):
"""
Get Default Parameters value
"""
- self.Result_DB = ft_utils.get_db_url()
+ self.Result_DB = CONST.__getattribute__("results_test_db_url")
self.masterusername = CONST.__getattribute__('ONOS_onosbench_username')
self.masterpassword = CONST.__getattribute__('ONOS_onosbench_password')
self.agentusername = CONST.__getattribute__('ONOS_onoscli_username')
def setUp(self):
self.prepare_envparser = prepare_env.PrepareEnvParser()
+ self.db_url_env = 'http://foo/testdb'
@mock.patch('functest.ci.prepare_env.logger.info')
def test_print_separator(self, mock_logger_info):
prepare_env.patch_file('test_file')
self.assertTrue(m.called)
+ @mock.patch('functest.ci.prepare_env.ft_utils.get_functest_yaml',
+ return_value={'tkey1': 'tvalue1'})
+ @mock.patch('functest.ci.prepare_env.yaml.safe_load',
+ return_value={'test_scenario': {'tkey': 'tvalue'}})
+ @mock.patch('functest.ci.prepare_env.update_db_url')
+ def test_update_db_url(self, mock_db_url, mock_safe_load,
+ mock_get_functest_yaml):
+ CONST.__setattr__('DEPLOY_SCENARIO', 'default_scenario')
+ with mock.patch("__builtin__.open", mock.mock_open()), \
+ mock.patch('functest.ci.prepare_env.yaml.dump'), \
+ mock.patch.dict('functest.ci.prepare_env.os.environ',
+ {'TEST_DB_URL': self.db_url_env},
+ clear=True):
+ prepare_env.update_config_file()
+ self.assertTrue(mock_db_url.called)
+
@mock.patch('functest.ci.prepare_env.logger.info')
def test_verify_deployment_error(self, mock_logger_error):
mock_popen = mock.Mock()
@mock.patch('functest.ci.prepare_env.install_tempest')
@mock.patch('functest.ci.prepare_env.install_rally')
@mock.patch('functest.ci.prepare_env.verify_deployment')
- @mock.patch('functest.ci.prepare_env.patch_config_file')
+ @mock.patch('functest.ci.prepare_env.update_config_file')
@mock.patch('functest.ci.prepare_env.source_rc_file')
@mock.patch('functest.ci.prepare_env.create_directories')
@mock.patch('functest.ci.prepare_env.get_deployment_handler')
@mock.patch('functest.ci.prepare_env.check_env_variables')
@mock.patch('functest.ci.prepare_env.logger.info')
def test_main_start(self, mock_logger_info, mock_env_var, mock_dep_handler,
- mock_create_dir, mock_source_rc, mock_patch_config,
+ mock_create_dir, mock_source_rc, mock_update_config,
mock_verify_depl, mock_install_rally,
mock_install_temp, mock_create_flavor,
mock_check_env, mock_print_info):
self.assertTrue(mock_dep_handler.called)
self.assertTrue(mock_create_dir.called)
self.assertTrue(mock_source_rc.called)
- self.assertTrue(mock_patch_config.called)
+ self.assertTrue(mock_update_config.called)
self.assertTrue(mock_verify_depl.called)
self.assertTrue(mock_install_rally.called)
self.assertTrue(mock_install_temp.called)
'OS_PASSWORD': 'test_password'}
self.test = {'test_name': 'test_name'}
self.tier = mock.Mock()
+ test1 = mock.Mock()
+ test1.get_name.return_value = 'test1'
+ test2 = mock.Mock()
+ test2.get_name.return_value = 'test2'
attrs = {'get_name.return_value': 'test_tier',
- 'get_tests.return_value': ['test1', 'test2'],
+ 'get_tests.return_value': [test1, test2],
'get_ci_loop.return_value': 'test_ci_loop',
'get_test_names.return_value': ['test1', 'test2']}
self.tier.configure_mock(**attrs)
return_value=self.creds):
self.runner.source_rc_file()
- @mock.patch('functest.ci.run_tests.os_snapshot.main')
- def test_generate_os_snapshot(self, mock_os_snap):
- self.runner.generate_os_snapshot()
- self.assertTrue(mock_os_snap.called)
-
- @mock.patch('functest.ci.run_tests.os_clean.main')
- def test_cleanup(self, mock_os_clean):
- self.runner.cleanup()
- self.assertTrue(mock_os_clean.called)
-
def test_get_run_dict_if_defined_default(self):
mock_obj = mock.Mock()
with mock.patch('functest.ci.run_tests.'
@mock.patch('functest.ci.run_tests.Runner.print_separator')
@mock.patch('functest.ci.run_tests.Runner.source_rc_file')
- @mock.patch('functest.ci.run_tests.Runner.generate_os_snapshot')
- @mock.patch('functest.ci.run_tests.Runner.cleanup')
@mock.patch('importlib.import_module', name="module",
return_value=mock.Mock(test_class=mock.Mock(
side_effect=FakeModule)))
def test_run_tier_default(self, mock_logger_info):
with mock.patch('functest.ci.run_tests.Runner.print_separator'), \
mock.patch(
- 'functest.ci.run_tests.Runner.run_test') as mock_method:
+ 'functest.ci.run_tests.Runner.run_test',
+ return_value=TestCase.EX_OK) as mock_method:
self.runner.run_tier(self.tier)
- mock_method.assert_any_call('test1', 'test_tier')
- mock_method.assert_any_call('test2', 'test_tier')
+ mock_method.assert_any_call(mock.ANY, 'test_tier')
self.assertTrue(mock_logger_info.called)
@mock.patch('functest.ci.run_tests.logger.info')
with mock.patch('functest.ci.run_tests.tb.TierBuilder',
return_value=mock_obj), \
mock.patch('functest.ci.run_tests.Runner.source_rc_file'), \
- mock.patch('functest.ci.run_tests.Runner.run_test') as m:
+ mock.patch('functest.ci.run_tests.Runner.run_test',
+ return_value=TestCase.EX_OK) as m:
self.assertEqual(self.runner.main(**kwargs),
run_tests.Result.EX_OK)
self.assertTrue(m.called)
'case_name': 'test_name',
'criteria': 'test_criteria',
'blocking': 'test_blocking',
- 'clean_flag': 'test_clean_flag',
'description': 'test_desc'}
self.dic_tier = {'name': 'test_tier',
self.mock_depend,
'test_criteria',
'test_blocking',
- 'test_clean_flag',
description='test_desc')
self.dependency = tier_handler.Dependency('test_installer',
self.assertIn(duration, message)
self.assertIn('FAIL', message)
+ def test_create_snapshot(self):
+ self.assertEqual(self.test.create_snapshot(),
+ testcase.TestCase.EX_OK)
+
+ def test_clean(self):
+ self.assertEqual(self.test.clean(), None)
+
+
+class OSGCTestCaseTesting(unittest.TestCase):
+ """The class testing OSGCTestCase."""
+ # pylint: disable=missing-docstring
+
+ def setUp(self):
+ self.test = testcase.OSGCTestCase()
+
+ @mock.patch('functest.utils.openstack_snapshot.main',
+ side_effect=Exception)
+ def test_create_snapshot_exc(self, mock_method=None):
+ self.assertEqual(self.test.create_snapshot(),
+ testcase.TestCase.EX_RUN_ERROR)
+ mock_method.assert_called_once_with()
+
+ @mock.patch('functest.utils.openstack_snapshot.main', return_value=-1)
+ def test_create_snapshot_ko(self, mock_method=None):
+ self.assertEqual(self.test.create_snapshot(),
+ testcase.TestCase.EX_RUN_ERROR)
+ mock_method.assert_called_once_with()
+
+ @mock.patch('functest.utils.openstack_snapshot.main', return_value=0)
+ def test_create_snapshot_env(self, mock_method=None):
+ self.assertEqual(self.test.create_snapshot(),
+ testcase.TestCase.EX_OK)
+ mock_method.assert_called_once_with()
+
+ @mock.patch('functest.utils.openstack_clean.main', side_effect=Exception)
+ def test_clean_exc(self, mock_method=None):
+ self.assertEqual(self.test.clean(), None)
+ mock_method.assert_called_once_with()
+
+ @mock.patch('functest.utils.openstack_clean.main', return_value=-1)
+ def test_clean_ko(self, mock_method=None):
+ self.assertEqual(self.test.clean(), None)
+ mock_method.assert_called_once_with()
+
+ @mock.patch('functest.utils.openstack_clean.main', return_value=0)
+ def test_clean(self, mock_method=None):
+ self.assertEqual(self.test.clean(), None)
+ mock_method.assert_called_once_with()
+
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Cable Television Laboratories, Inc. and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+
+import mock
+import unittest
+
+from snaps.openstack.os_credentials import OSCreds
+
+from functest.core.testcase import TestCase
+from functest.opnfv_tests.openstack.snaps import connection_check
+
+
+class ConnectionCheckTesting(unittest.TestCase):
+ """
+ Ensures the VPingUserdata class can run in Functest. This test does not
+ actually connect with an OpenStack pod.
+ """
+
+ def setUp(self):
+ self.os_creds = OSCreds(
+ username='user', password='pass',
+ auth_url='http://foo.com:5000/v3', project_name='bar')
+
+ self.connection_check = connection_check.ConnectionCheck(
+ os_creds=self.os_creds, ext_net_name='foo')
+
+ @mock.patch('functest.opnfv_tests.openstack.snaps.connection_check.'
+ 'ConnectionCheck')
+ @mock.patch('snaps.test_suite_builder.add_openstack_client_tests')
+ def test_run_success(self, mock_test, add_os_client_tests):
+ result = mock.MagicMock(name='unittest.TextTestResult')
+ result.testsRun = 100
+ result.failures = []
+ result.errors = []
+ with mock.patch('unittest.TextTestRunner.run', return_value=result):
+ self.assertEquals(TestCase.EX_OK, self.connection_check.run())
+ self.assertEquals(TestCase.EX_OK,
+ self.connection_check.is_successful())
+
+ @mock.patch('functest.opnfv_tests.openstack.snaps.connection_check.'
+ 'ConnectionCheck')
+ @mock.patch('snaps.test_suite_builder.add_openstack_client_tests')
+ def test_run_1_of_100_failures(self, mock_test, add_os_client_tests):
+ result = mock.MagicMock(name='unittest.TextTestResult')
+ result.testsRun = 100
+ result.failures = ['foo']
+ result.errors = []
+ with mock.patch('unittest.TextTestRunner.run', return_value=result):
+ self.assertEquals(TestCase.EX_OK, self.connection_check.run())
+ self.assertEquals(TestCase.EX_TESTCASE_FAILED,
+ self.connection_check.is_successful())
+
+ @mock.patch('functest.opnfv_tests.openstack.snaps.connection_check.'
+ 'ConnectionCheck')
+ @mock.patch('snaps.test_suite_builder.add_openstack_client_tests')
+ def test_run_1_of_100_failures_within_criteria(self, mock_test,
+ add_os_client_tests):
+ self.connection_check.criteria = 90
+ result = mock.MagicMock(name='unittest.TextTestResult')
+ result.testsRun = 100
+ result.failures = ['foo']
+ result.errors = []
+ with mock.patch('unittest.TextTestRunner.run', return_value=result):
+ self.assertEquals(TestCase.EX_OK, self.connection_check.run())
+ self.assertEquals(TestCase.EX_OK,
+ self.connection_check.is_successful())
from functest.utils import decorators
from functest.utils import functest_utils
+from functest.utils.constants import CONST
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
'details': {}, 'criteria': self._result}
return json.dumps(data, sort_keys=True)
- @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
- return_value='http://127.0.0.1')
@mock.patch('{}.get_version'.format(functest_utils.__name__),
return_value=VERSION)
@mock.patch('requests.post')
def test_http_shema(self, *args):
+ CONST.__setattr__('results_test_db_url', 'http://127.0.0.1')
self.assertTrue(functest_utils.push_results_to_db(
self._project_name, self._case_name, self._start_time,
self._stop_time, self._result, {}))
args[1].assert_called_once_with()
- args[2].assert_called_once_with()
args[0].assert_called_once_with(
'http://127.0.0.1', data=self._get_json(),
headers={'Content-Type': 'application/json'})
- @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
- return_value="/dev/null")
- def test_wrong_shema(self, mock_method=None):
+ def test_wrong_shema(self):
+ CONST.__setattr__('results_test_db_url', '/dev/null')
self.assertFalse(functest_utils.push_results_to_db(
self._project_name, self._case_name, self._start_time,
self._stop_time, self._result, {}))
- mock_method.assert_called_once_with()
@mock.patch('{}.get_version'.format(functest_utils.__name__),
return_value=VERSION)
- @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
- return_value=URL)
def _test_dump(self, *args):
+ CONST.__setattr__('results_test_db_url', URL)
with mock.patch.object(decorators, 'open', mock.mock_open(),
create=True) as mock_open:
self.assertTrue(functest_utils.push_results_to_db(
self.assertIn('POST', call_args[0])
self.assertIn(self._get_json(), call_args[0])
args[0].assert_called_once_with()
- args[1].assert_called_once_with()
@mock.patch('os.makedirs')
def test_default_dump(self, mock_method=None):
self._test_dump()
mock_method.assert_called_once_with(DIR)
- @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
- return_value=URL)
@mock.patch('os.makedirs', side_effect=OSError)
def test_makedirs_exc(self, *args):
+ CONST.__setattr__('results_test_db_url', URL)
self.assertFalse(
functest_utils.push_results_to_db(
self._project_name, self._case_name, self._start_time,
self._stop_time, self._result, {}))
args[0].assert_called_once_with(DIR)
- args[1].assert_called_once_with()
if __name__ == "__main__":
from functest.tests.unit import test_utils
from functest.utils import functest_utils
+from functest.utils.constants import CONST
class FunctestUtilsTesting(unittest.TestCase):
self.config_yaml = os.path.normpath(os.path.join(os.path.dirname(
os.path.abspath(__file__)), '../../../ci/config_functest.yaml'))
self.db_url_env = 'http://foo/testdb'
+ self.testcases_yaml = "test_testcases_yaml"
self.file_yaml = {'general': {'openstack': {'image_name':
'test_image_name'}}}
self.assertEqual(functest_utils.get_build_tag(),
self.build_tag)
- def test_get_db_url_env_var(self):
- with mock.patch.dict(os.environ,
- {'TEST_DB_URL': self.db_url_env,
- 'CONFIG_FUNCTEST_YAML':
- "./functest/ci/config_functest.yaml"},
- clear=True):
- self.assertEqual(functest_utils.get_db_url(),
- self.db_url_env)
-
- @mock.patch('functest.utils.functest_utils.get_functest_config')
- def test_get_db_url_default(self, mock_get_functest_config):
- mock_get_functest_config.return_value = self.db_url
- self.assertEqual(functest_utils.get_db_url(), self.db_url)
- mock_get_functest_config.assert_called_once_with('results.test_db_url')
-
@mock.patch('functest.utils.functest_utils.logger.info')
def test_logger_test_results(self, mock_logger_info):
+ CONST.__setattr__('results_test_db_url', self.db_url)
with mock.patch('functest.utils.functest_utils.get_pod_name',
return_value=self.node_name), \
mock.patch('functest.utils.functest_utils.get_scenario',
mock.patch('functest.utils.functest_utils.get_version',
return_value=self.version), \
mock.patch('functest.utils.functest_utils.get_build_tag',
- return_value=self.build_tag), \
- mock.patch('functest.utils.functest_utils.get_db_url',
- return_value=self.db_url):
+ return_value=self.build_tag):
functest_utils.logger_test_results(self.project, self.case_name,
self.status, self.details)
mock_logger_info.assert_called_once_with(
"details:\t%(d)s\n"
% {'p': self.project,
'n': self.case_name,
- 'db': self.db_url,
+ 'db': CONST.__getattribute__('results_test_db_url'),
'pod': self.node_name,
'v': self.version,
's': self.scenario,
def _test_push_results_to_db_missing_env(self, env_var):
dic = self._get_env_dict(env_var)
- with mock.patch('functest.utils.functest_utils.get_db_url',
- return_value=self.db_url), \
- mock.patch.dict(os.environ,
- dic,
- clear=True), \
+ CONST.__setattr__('results_test_db_url', self.db_url)
+ with mock.patch.dict(os.environ,
+ dic,
+ clear=True), \
mock.patch('functest.utils.functest_utils.logger.error') \
as mock_logger_error:
functest_utils.push_results_to_db(self.project, self.case_name,
def test_push_results_to_db_request_post_failed(self):
dic = self._get_env_dict(None)
- with mock.patch('functest.utils.functest_utils.get_db_url',
- return_value=self.db_url), \
- mock.patch.dict(os.environ,
- dic,
- clear=True), \
+ CONST.__setattr__('results_test_db_url', self.db_url)
+ with mock.patch.dict(os.environ,
+ dic,
+ clear=True), \
mock.patch('functest.utils.functest_utils.logger.error') \
as mock_logger_error, \
mock.patch('functest.utils.functest_utils.requests.post',
def test_push_results_to_db_request_post_exception(self):
dic = self._get_env_dict(None)
- with mock.patch('functest.utils.functest_utils.get_db_url',
- return_value=self.db_url), \
- mock.patch.dict(os.environ,
- dic,
- clear=True), \
+ CONST.__setattr__('results_test_db_url', self.db_url)
+ with mock.patch.dict(os.environ,
+ dic,
+ clear=True), \
mock.patch('functest.utils.functest_utils.logger.error') \
as mock_logger_error, \
mock.patch('functest.utils.functest_utils.requests.post',
def test_push_results_to_db_default(self):
dic = self._get_env_dict(None)
- with mock.patch('functest.utils.functest_utils.get_db_url',
- return_value=self.db_url), \
- mock.patch.dict(os.environ,
- dic,
- clear=True), \
+ CONST.__setattr__('results_test_db_url', self.db_url)
+ with mock.patch.dict(os.environ,
+ dic,
+ clear=True), \
mock.patch('functest.utils.functest_utils.requests.post'):
self.assertTrue(functest_utils.
push_results_to_db(self.project, self.case_name,
# TODO: merge_dicts
- def test_get_testcases_file_dir(self):
+ @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,
- "/home/opnfv/repos/functest/"
- "functest/ci/testcases.yaml")
+ 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()), \
from functest.utils import constants
from functest.utils import decorators
+from functest.utils.constants import CONST
logger = logging.getLogger(__name__)
return build_tag
-def get_db_url():
+def logger_test_results(project, case_name, status, details):
"""
- Returns DB URL
+ Format test case results for the logger
"""
- # TODO use CONST mechanism
- try:
- # if TEST_DB_URL declared in env variable, use it!
- db_url = os.environ['TEST_DB_URL']
- except KeyError:
- db_url = get_functest_config('results.test_db_url')
- return db_url
-
-
-def logger_test_results(project, case_name, status, details):
pod_name = get_pod_name()
scenario = get_scenario()
version = get_version()
build_tag = get_build_tag()
+ db_url = CONST.__getattribute__("results_test_db_url")
logger.info(
"\n"
"details:\t%(d)s\n"
% {'p': project,
'n': case_name,
- 'db': get_db_url(),
+ 'db': db_url,
'pod': pod_name,
'v': version,
's': scenario,
POST results to the Result target DB
"""
# Retrieve params from CI and conf
- url = get_db_url()
+ url = CONST.__getattribute__("results_test_db_url")
try:
installer = os.environ['INSTALLER_TYPE']
# http://www.apache.org/licenses/LICENSE-2.0
#
-import time
-
import logging
+import sys
+import time
import yaml
import functest.utils.openstack_utils as os_utils
except Exception:
logger.info("The file %s does not exist. The OpenStack snapshot must"
" be created first. Aborting cleanup." % OS_SNAPSHOT_FILE)
- exit(0)
+ return 0
default_images = snapshot_yaml.get('images')
default_instances = snapshot_yaml.get('instances')
if not os_utils.check_credentials():
logger.error("Please source the openrc credentials and run "
"the script again.")
- exit(-1)
+ return -1
remove_instances(nova_client, default_instances)
separator()
if __name__ == '__main__':
logging.basicConfig()
- main()
+ sys.exit(main())
import logging
import yaml
+import sys
import functest.utils.openstack_utils as os_utils
from functest.utils.constants import CONST
if not os_utils.check_credentials():
logger.error("Please source the openrc credentials and run the" +
"script again.")
- exit(-1)
+ return -1
snapshot = {}
snapshot.update(get_instances(nova_client))
if __name__ == '__main__':
logging.basicConfig()
- main()
+ sys.exit(main())
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
deps =
{[testenv]deps}
whitelist_externals = bash
-commands = bash -c "\
- pylint -f parseable --disable=locally-disabled functest | \
- tee pylint.out | sed -ne '/Raw metrics/,//p'"
+modules =
+ functest.core.feature
+ functest.core.testcase
+ functest.core.unit
+ 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.odl
+ functest.tests.unit.utils.test_decorators
+ functest.utils.decorators
+commands =
+ bash -c "\
+ pylint -f parseable --disable=locally-disabled functest | \
+ tee pylint.out | sed -ne '/Raw metrics/,//p'"
+ pylint --disable=locally-disabled --reports=n {[testenv:pylint]modules}
[testenv:py35]
dirs =
Code Review / functest.git / commitdiff