diff options
| -rwxr-xr-x | ci/run_tests.py | 42 | ||||
| -rw-r--r-- | ci/testcases.yaml | 3 | ||||
| -rw-r--r-- | core/TestCasesBase.py | 46 | ||||
| -rwxr-xr-x | core/TestCasesBaseTesting.py | 80 | ||||
| -rw-r--r-- | core/__init__.py | 0 | ||||
| -rw-r--r-- | docker/Dockerfile | 1 | ||||
| -rwxr-xr-x | testcases/Controllers/ODL/OpenDaylightTesting.py | 158 | ||||
| -rwxr-xr-x | testcases/Controllers/ODL/OpenDaylightUnitTesting.py | 341 | ||||
| -rw-r--r-- | utils/functest_utils.py | 11 | ||||
| -rwxr-xr-x | utils/openstack_utils.py | 12 |
10 files changed, 609 insertions, 85 deletions
diff --git a/ci/run_tests.py b/ci/run_tests.py index 638a6edf..f30062f7 100755 --- a/ci/run_tests.py +++ b/ci/run_tests.py @@ -9,6 +9,7 @@ # import datetime +import importlib import os import re import sys @@ -17,12 +18,13 @@ import argparse import functest.ci.generate_report as generate_report import functest.ci.tier_builder as tb +import functest.core.TestCasesBase as TestCasesBase import functest.utils.functest_logger as ft_logger 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.testcases.Controllers.ODL.OpenDaylightTesting import ODLTestCases + parser = argparse.ArgumentParser() parser.add_argument("-t", "--test", dest="test", action='store', @@ -82,6 +84,19 @@ def update_test_info(test_name, result, duration): "duration": duration}) +def get_run_dict_if_defined(testname): + try: + dict = ft_utils.get_dict_by_test(testname) + if not dict: + logger.error("Cannot get {}'s config options".format(testname)) + elif 'run' in dict: + return dict['run'] + return None + except Exception: + logger.exception("Cannot get {}'s config options".format(testname)) + return None + + def run_test(test, tier_name): global OVERALL_RESULT, EXECUTED_TEST_CASES result_str = "PASS" @@ -100,14 +115,27 @@ def run_test(test, tier_name): if REPORT_FLAG: flags += " -r" - if test_name == 'odl': - result = ODLTestCases.functest_run() - if result and REPORT_FLAG: - result = ODLTestCases.push_to_db() - result = not result + result = TestCasesBase.TestCasesBase.EX_RUN_ERROR + run_dict = get_run_dict_if_defined(test_name) + if run_dict: + try: + module = importlib.import_module(run_dict['module']) + cls = getattr(module, run_dict['class']) + test_case = cls() + result = test_case.run() + if result == TestCasesBase.TestCasesBase.EX_OK and REPORT_FLAG: + result = test_case.push_to_db() + except ImportError: + logger.exception("Cannot import module {}".format( + run_dict['module'])) + except AttributeError: + logger.exception("Cannot get class {}".format( + run_dict['class'])) else: cmd = ("%s%s" % (EXEC_SCRIPT, flags)) - logger.debug("Executing command '%s'" % cmd) + logger.info("Executing command {} because {} " + "doesn't implement the new framework".format( + cmd, test_name)) result = ft_utils.execute_command(cmd) if CLEAN_FLAG: diff --git a/ci/testcases.yaml b/ci/testcases.yaml index 02b69ec8..22509e61 100644 --- a/ci/testcases.yaml +++ b/ci/testcases.yaml @@ -92,6 +92,9 @@ tiers: dependencies: installer: '' scenario: 'odl' + run: + module: 'functest.testcases.Controllers.ODL.OpenDaylightTesting' + class: 'ODLTestCases' - name: onos diff --git a/core/TestCasesBase.py b/core/TestCasesBase.py new file mode 100644 index 00000000..8e68fa68 --- /dev/null +++ b/core/TestCasesBase.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python + +import os + +import functest.utils.functest_logger as ft_logger +import functest.utils.functest_utils as ft_utils + + +class TestCasesBase(object): + + EX_OK = os.EX_OK + EX_RUN_ERROR = os.EX_SOFTWARE + EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1 + + logger = ft_logger.Logger(__name__).getLogger() + + project = "functest" + + def __init__(self): + self.details = {} + self.case_name = "" + self.criteria = "" + self.start_time = "" + self.stop_time = "" + + def run(self, **kwargs): + self.logger.error("Run must be implemented") + return TestCasesBase.EX_RUN_ERROR + + def push_to_db(self): + try: + assert self.case_name + assert self.criteria + assert self.start_time + assert self.stop_time + if ft_utils.push_results_to_db( + TestCasesBase.project, self.case_name, self.start_time, + self.stop_time, self.criteria, self.details): + self.logger.info("The results were successfully pushed to DB") + return TestCasesBase.EX_OK + else: + self.logger.error("The results cannot be pushed to DB") + return TestCasesBase.EX_PUSH_TO_DB_ERROR + except Exception: + self.logger.exception("The results cannot be pushed to DB") + return TestCasesBase.EX_PUSH_TO_DB_ERROR diff --git a/core/TestCasesBaseTesting.py b/core/TestCasesBaseTesting.py new file mode 100755 index 00000000..9658e8dc --- /dev/null +++ b/core/TestCasesBaseTesting.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python + +import logging +import mock +import unittest + +import TestCasesBase + + +class TestCasesBaseTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + def setUp(self): + self.test = TestCasesBase.TestCasesBase() + self.test.project = "functest" + self.test.case_name = "base" + self.test.start_time = "1" + self.test.stop_time = "2" + self.test.criteria = "100" + self.test.details = {"Hello": "World"} + + def test_run_unimplemented(self): + self.assertEqual(self.test.run(), + TestCasesBase.TestCasesBase.EX_RUN_ERROR) + + @mock.patch('functest.utils.functest_utils.push_results_to_db', + return_value=False) + def _test_missing_attribute(self, mock_function): + self.assertEqual(self.test.push_to_db(), + TestCasesBase.TestCasesBase.EX_PUSH_TO_DB_ERROR) + mock_function.assert_not_called() + + def test_missing_case_name(self): + self.test.case_name = None + self._test_missing_attribute() + + def test_missing_criteria(self): + self.test.criteria = None + self._test_missing_attribute() + + def test_missing_start_time(self): + self.test.start_time = None + self._test_missing_attribute() + + def test_missing_stop_time(self): + self.test.stop_time = None + self._test_missing_attribute() + + @mock.patch('functest.utils.functest_utils.push_results_to_db', + return_value=True) + def test_missing_details(self, mock_function): + self.test.details = None + self.assertEqual(self.test.push_to_db(), + TestCasesBase.TestCasesBase.EX_OK) + mock_function.assert_called_once_with( + self.test.project, self.test.case_name, self.test.start_time, + self.test.stop_time, self.test.criteria, self.test.details) + + @mock.patch('functest.utils.functest_utils.push_results_to_db', + return_value=False) + def test_push_to_db_failed(self, mock_function): + self.assertEqual(self.test.push_to_db(), + TestCasesBase.TestCasesBase.EX_PUSH_TO_DB_ERROR) + mock_function.assert_called_once_with( + self.test.project, self.test.case_name, self.test.start_time, + self.test.stop_time, self.test.criteria, self.test.details) + + @mock.patch('functest.utils.functest_utils.push_results_to_db', + return_value=True) + def test_push_to_db(self, mock_function): + self.assertEqual(self.test.push_to_db(), + TestCasesBase.TestCasesBase.EX_OK) + mock_function.assert_called_once_with( + self.test.project, self.test.case_name, self.test.start_time, + self.test.stop_time, self.test.criteria, self.test.details) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/core/__init__.py b/core/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/core/__init__.py diff --git a/docker/Dockerfile b/docker/Dockerfile index 2d325eaa..9ff0f4f6 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -53,6 +53,7 @@ git \ gcc \ wget \ python-dev \ +python-mock \ python-pip \ bundler \ postgresql \ diff --git a/testcases/Controllers/ODL/OpenDaylightTesting.py b/testcases/Controllers/ODL/OpenDaylightTesting.py index 35a2ee11..e302b569 100755 --- a/testcases/Controllers/ODL/OpenDaylightTesting.py +++ b/testcases/Controllers/ODL/OpenDaylightTesting.py @@ -1,6 +1,7 @@ #!/usr/bin/python import argparse +import errno import fileinput import os import re @@ -8,13 +9,13 @@ import shutil import sys import urlparse -from robot import run from robot.api import ExecutionResult, ResultVisitor from robot.errors import RobotError +import robot.run from robot.utils.robottime import timestamp_to_secs +from functest.core import TestCasesBase import functest.utils.functest_logger as ft_logger -import functest.utils.functest_utils as ft_utils import functest.utils.openstack_utils as op_utils @@ -39,7 +40,7 @@ class ODLResultVisitor(ResultVisitor): return self._data -class ODLTestCases: +class ODLTestCases(TestCasesBase.TestCasesBase): repos = "/home/opnfv/repos/" odl_test_repo = repos + "odl_test/" @@ -48,13 +49,16 @@ class ODLTestCases: res_dir = '/home/opnfv/functest/results/odl/' logger = ft_logger.Logger("opendaylight").getLogger() + def __init__(self): + self.case_name = "odl" + @classmethod def copy_opnf_testcases(cls): opnfv_testcases_dir = (os.path.dirname(os.path.abspath(__file__)) + "/custom_tests/neutron/") - file = opnfv_testcases_dir + "001__reachability.robot" + f = opnfv_testcases_dir + "001__reachability.robot" try: - shutil.copy(file, cls.neutron_suite_dir) + shutil.copy(f, cls.neutron_suite_dir) except IOError as e: cls.logger.error( "Cannot copy OPNFV's testcase to ODL directory: %s" % str(e)) @@ -76,9 +80,19 @@ class ODLTestCases: cls.logger.error("Cannot set ODL creds: %s" % str(e)) return False - @classmethod - def run(cls, **kwargs): - dirs = [cls.basic_suite_dir, cls.neutron_suite_dir] + def parse_results(self): + result = ExecutionResult(self.res_dir + 'output.xml') + visitor = ODLResultVisitor() + result.visit(visitor) + self.criteria = result.suite.status + self.start_time = timestamp_to_secs(result.suite.starttime) + self.stop_time = timestamp_to_secs(result.suite.endtime) + self.details = {} + self.details['description'] = result.suite.name + self.details['tests'] = visitor.get_data() + + def main(self, **kwargs): + dirs = [self.basic_suite_dir, self.neutron_suite_dir] try: odlusername = kwargs['odlusername'] odlpassword = kwargs['odlpassword'] @@ -91,49 +105,60 @@ class ODLTestCases: 'PORT:' + kwargs['odlwebport'], 'RESTCONFPORT:' + kwargs['odlrestconfport']] except KeyError as e: - cls.logger.error("Cannot run ODL testcases. Please check " - "%s" % str(e)) - return False - if (cls.copy_opnf_testcases() and - cls.set_robotframework_vars(odlusername, odlpassword)): + self.logger.error("Cannot run ODL testcases. Please check " + "%s" % str(e)) + return self.EX_RUN_ERROR + if (self.copy_opnf_testcases() and + self.set_robotframework_vars(odlusername, odlpassword)): try: - os.makedirs(cls.res_dir) - except OSError: - pass - stdout_file = cls.res_dir + 'stdout.txt' + os.makedirs(self.res_dir) + except OSError as e: + if e.errno != errno.EEXIST: + self.logger.exception( + "Cannot create {}".format(self.res_dir)) + return self.EX_RUN_ERROR + stdout_file = self.res_dir + 'stdout.txt' with open(stdout_file, 'w+') as stdout: - run(*dirs, variable=variables, - output=cls.res_dir + 'output.xml', - log='NONE', - report='NONE', - stdout=stdout) + robot.run(*dirs, variable=variables, + output=self.res_dir + 'output.xml', + log='NONE', + report='NONE', + stdout=stdout) stdout.seek(0, 0) - cls.logger.info("\n" + stdout.read()) - cls.logger.info("ODL results were successfully generated") + self.logger.info("\n" + stdout.read()) + self.logger.info("ODL results were successfully generated") + try: + self.parse_results() + self.logger.info("ODL results were successfully parsed") + except RobotError as e: + self.logger.error("Run tests before publishing: %s" % + e.message) + return self.EX_RUN_ERROR try: os.remove(stdout_file) except OSError: - pass - return True + self.logger.warning("Cannot remove {}".format(stdout_file)) + return self.EX_OK else: - return False + return self.EX_RUN_ERROR - @classmethod - def functest_run(cls): - kclient = op_utils.get_keystone_client() - keystone_url = kclient.service_catalog.url_for( - service_type='identity', endpoint_type='publicURL') - neutron_url = kclient.service_catalog.url_for( - service_type='network', endpoint_type='publicURL') - kwargs = {'keystoneip': urlparse.urlparse(keystone_url).hostname} - kwargs['neutronip'] = urlparse.urlparse(neutron_url).hostname - kwargs['odlip'] = kwargs['neutronip'] - kwargs['odlwebport'] = '8080' - kwargs['odlrestconfport'] = '8181' - kwargs['odlusername'] = 'admin' - kwargs['odlpassword'] = 'admin' + def run(self): try: - installer_type = os.environ['INSTALLER_TYPE'] + kclient = op_utils.get_keystone_client() + keystone_url = kclient.service_catalog.url_for( + service_type='identity', endpoint_type='publicURL') + neutron_url = kclient.service_catalog.url_for( + service_type='network', endpoint_type='publicURL') + kwargs = {'keystoneip': urlparse.urlparse(keystone_url).hostname} + kwargs['neutronip'] = urlparse.urlparse(neutron_url).hostname + kwargs['odlip'] = kwargs['neutronip'] + kwargs['odlwebport'] = '8080' + kwargs['odlrestconfport'] = '8181' + kwargs['odlusername'] = 'admin' + kwargs['odlpassword'] = 'admin' + installer_type = None + if 'INSTALLER_TYPE' in os.environ: + installer_type = os.environ['INSTALLER_TYPE'] kwargs['osusername'] = os.environ['OS_USERNAME'] kwargs['ostenantname'] = os.environ['OS_TENANT_NAME'] kwargs['ospassword'] = os.environ['OS_PASSWORD'] @@ -149,34 +174,15 @@ class ODLTestCases: else: kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP'] except KeyError as e: - cls.logger.error("Cannot run ODL testcases. Please check env var: " - "%s" % str(e)) - return False - - return cls.run(**kwargs) - - @classmethod - def push_to_db(cls): - try: - result = ExecutionResult(cls.res_dir + 'output.xml') - visitor = ODLResultVisitor() - result.visit(visitor) - start_time = timestamp_to_secs(result.suite.starttime) - stop_time = timestamp_to_secs(result.suite.endtime) - details = {} - details['description'] = result.suite.name - details['tests'] = visitor.get_data() - if not ft_utils.push_results_to_db( - "functest", "odl", start_time, stop_time, - result.suite.status, details): - cls.logger.error("Cannot push ODL results to DB") - return False - else: - cls.logger.info("ODL results were successfully pushed to DB") - return True - except RobotError as e: - cls.logger.error("Run tests before publishing: %s" % e.message) + self.logger.error("Cannot run ODL testcases. " + "Please check env var: " + "%s" % str(e)) + return self.EX_RUN_ERROR + except Exception: + self.logger.exception("Cannot run ODL testcases.") + return self.EX_RUN_ERROR + return self.main(**kwargs) if __name__ == '__main__': parser = argparse.ArgumentParser() @@ -215,8 +221,12 @@ if __name__ == '__main__': action='store_true') args = vars(parser.parse_args()) - if not ODLTestCases.run(**args): - sys.exit(os.EX_SOFTWARE) - if args['pushtodb']: - sys.exit(not ODLTestCases.push_to_db()) - sys.exit(os.EX_OK) + odl = ODLTestCases() + try: + result = odl.main(**args) + if result != TestCasesBase.TestCasesBase.EX_OK: + sys.exit(result) + if args['pushtodb']: + sys.exit(odl.push_to_db()) + except Exception: + sys.exit(TestCasesBase.TestCasesBase.EX_RUN_ERROR) diff --git a/testcases/Controllers/ODL/OpenDaylightUnitTesting.py b/testcases/Controllers/ODL/OpenDaylightUnitTesting.py new file mode 100755 index 00000000..500888b2 --- /dev/null +++ b/testcases/Controllers/ODL/OpenDaylightUnitTesting.py @@ -0,0 +1,341 @@ +#!/usr/bin/env python + +import errno +import logging +import mock +import os +import unittest + +from robot.errors import RobotError + +from functest.core import TestCasesBase +from functest.testcases.Controllers.ODL import OpenDaylightTesting + + +class ODLTestCasesTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + _keystone_ip = "127.0.0.1" + _neutron_ip = "127.0.0.2" + _sdn_controller_ip = "127.0.0.3" + _os_tenantname = "admin" + _os_username = "admin" + _os_password = "admin" + _odl_webport = "8080" + _odl_restconfport = "8181" + _odl_username = "admin" + _odl_password = "admin" + + def setUp(self): + if "INSTALLER_TYPE" in os.environ: + del os.environ["INSTALLER_TYPE"] + os.environ["OS_USERNAME"] = self._os_username + os.environ["OS_PASSWORD"] = self._os_password + os.environ["OS_TENANT_NAME"] = self._os_tenantname + self.test = OpenDaylightTesting.ODLTestCases() + + @mock.patch('shutil.copy', side_effect=Exception()) + def test_copy_opnf_testcases_exception(self, *args): + with self.assertRaises(Exception): + self.test.copy_opnf_testcases() + + @mock.patch('shutil.copy', side_effect=IOError()) + def test_copy_opnf_testcases_ioerror(self, *args): + self.assertFalse(self.test.copy_opnf_testcases()) + + @mock.patch('shutil.copy') + def test_copy_opnf_testcases(self, *args): + self.assertTrue(self.test.copy_opnf_testcases()) + + @mock.patch('fileinput.input', side_effect=Exception()) + def test_set_robotframework_vars_failed(self, *args): + self.assertFalse(self.test.set_robotframework_vars()) + + @mock.patch('fileinput.input', return_value=[]) + def test_set_robotframework_vars(self, args): + self.assertTrue(self.test.set_robotframework_vars()) + + def _test_run_missing_env_var(self, var): + del os.environ[var] + self.assertEqual(self.test.run(), + TestCasesBase.TestCasesBase.EX_RUN_ERROR) + + @classmethod + def _fake_url_for(cls, service_type='identity', **kwargs): + if service_type == 'identity': + return "http://{}:5000/v2.0".format( + ODLTestCasesTesting._keystone_ip) + elif service_type == 'network': + return "http://{}:9696".format(ODLTestCasesTesting._neutron_ip) + else: + return None + + @classmethod + def _get_fake_keystone_client(cls): + kclient = mock.Mock() + kclient.service_catalog = mock.Mock() + kclient.service_catalog.url_for = mock.Mock( + side_effect=cls._fake_url_for) + return kclient + + def _get_main_kwargs(self, key=None): + kwargs = {'odlusername': self._odl_username, + 'odlpassword': self._odl_password, + 'keystoneip': self._keystone_ip, + 'neutronip': self._neutron_ip, + 'osusername': self._os_username, + 'ostenantname': self._os_tenantname, + 'ospassword': self._os_password, + 'odlip': self._sdn_controller_ip, + 'odlwebport': self._odl_webport, + 'odlrestconfport': self._odl_restconfport} + if key: + del kwargs[key] + return kwargs + + def _test_main(self, status, *args): + kwargs = self._get_main_kwargs() + self.assertEqual(self.test.main(**kwargs), status) + if len(args) > 0: + args[0].assert_called_once_with( + OpenDaylightTesting.ODLTestCases.res_dir) + if len(args) > 1: + variable = ['KEYSTONE:{}'.format(self._keystone_ip), + 'NEUTRON:{}'.format(self._neutron_ip), + 'OSUSERNAME:"{}"'.format(self._os_username), + 'OSTENANTNAME:"{}"'.format(self._os_tenantname), + 'OSPASSWORD:"{}"'.format(self._os_password), + 'ODL_SYSTEM_IP:{}'.format(self._sdn_controller_ip), + 'PORT:{}'.format(self._odl_webport), + 'RESTCONFPORT:{}'.format(self._odl_restconfport)] + args[1].assert_called_once_with( + OpenDaylightTesting.ODLTestCases.basic_suite_dir, + OpenDaylightTesting.ODLTestCases.neutron_suite_dir, + log='NONE', + output=OpenDaylightTesting.ODLTestCases.res_dir + 'output.xml', + report='NONE', + stdout=mock.ANY, + variable=variable) + if len(args) > 2: + args[2].assert_called_with( + OpenDaylightTesting.ODLTestCases.res_dir + 'stdout.txt') + + def _test_main_missing_keyword(self, key): + kwargs = self._get_main_kwargs(key) + self.assertEqual(self.test.main(**kwargs), + TestCasesBase.TestCasesBase.EX_RUN_ERROR) + + def test_main_missing_odlusername(self): + self._test_main_missing_keyword('odlusername') + + def test_main_missing_odlpassword(self): + self._test_main_missing_keyword('odlpassword') + + def test_main_missing_keystoneip(self): + self._test_main_missing_keyword('keystoneip') + + def test_main_missing_neutronip(self): + self._test_main_missing_keyword('neutronip') + + def test_main_missing_osusername(self): + self._test_main_missing_keyword('osusername') + + def test_main_missing_ostenantname(self): + self._test_main_missing_keyword('ostenantname') + + def test_main_missing_ospassword(self): + self._test_main_missing_keyword('ospassword') + + def test_main_missing_odlip(self): + self._test_main_missing_keyword('odlip') + + def test_main_missing_odlwebport(self): + self._test_main_missing_keyword('odlwebport') + + def test_main_missing_odlrestconfport(self): + self._test_main_missing_keyword('odlrestconfport') + + def test_main_copy_opnf_testcases_failed(self): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=False): + self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR) + self.test.copy_opnf_testcases.assert_called_once_with() + + def test_main_set_robotframework_vars_failed(self): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=False): + self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR) + self.test.set_robotframework_vars.assert_called_once_with( + self._odl_username, self._odl_password) + + @mock.patch('os.makedirs', side_effect=Exception) + def test_main_makedirs_exception(self, mock_method): + with mock.patch.object(self.test, + 'copy_opnf_testcases', return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + self.assertRaises(Exception): + self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR, + mock_method) + + @mock.patch('os.makedirs', side_effect=OSError) + def test_main_makedirs_oserror(self, mock_method): + with mock.patch.object(self.test, + 'copy_opnf_testcases', return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True): + self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR, + mock_method) + + @mock.patch('robot.run', side_effect=RobotError) + @mock.patch('os.makedirs') + def test_main_robot_run_failed(self, *args): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + self.assertRaises(RobotError): + self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR, *args) + + @mock.patch('robot.run') + @mock.patch('os.makedirs') + def test_main_parse_results_failed(self, *args): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + mock.patch.object(self.test, 'parse_results', + side_effect=RobotError): + self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR, *args) + + @mock.patch('os.remove', side_effect=Exception) + @mock.patch('robot.run') + @mock.patch('os.makedirs') + def test_main_remove_exception(self, *args): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + mock.patch.object(self.test, 'parse_results'), \ + self.assertRaises(Exception): + self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args) + + @mock.patch('os.remove') + @mock.patch('robot.run') + @mock.patch('os.makedirs') + def test_main(self, *args): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + mock.patch.object(self.test, 'parse_results'): + self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args) + + @mock.patch('os.remove') + @mock.patch('robot.run') + @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, '')) + def test_main_makedirs_oserror17(self, *args): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + mock.patch.object(self.test, 'parse_results'): + self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args) + + @mock.patch('os.remove') + @mock.patch('robot.run', return_value=1) + @mock.patch('os.makedirs') + def test_main_testcases_in_failure(self, *args): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + mock.patch.object(self.test, 'parse_results'): + self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args) + + @mock.patch('os.remove', side_effect=OSError) + @mock.patch('robot.run') + @mock.patch('os.makedirs') + def test_main_remove_oserror(self, *args): + with mock.patch.object(self.test, 'copy_opnf_testcases', + return_value=True), \ + mock.patch.object(self.test, 'set_robotframework_vars', + return_value=True), \ + mock.patch.object(self.test, 'parse_results'): + self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args) + + def _test_run(self, status=TestCasesBase.TestCasesBase.EX_OK, + exception=None, odlip="127.0.0.3", odlwebport="8080"): + with mock.patch('functest.utils.openstack_utils.get_keystone_client', + return_value=self._get_fake_keystone_client()): + if exception: + self.test.main = mock.Mock(side_effect=exception) + else: + self.test.main = mock.Mock(return_value=status) + self.assertEqual(self.test.run(), status) + self.test.main.assert_called_once_with( + keystoneip=self._keystone_ip, neutronip=self._neutron_ip, + odlip=odlip, odlpassword=self._odl_password, + odlrestconfport=self._odl_restconfport, + odlusername=self._odl_username, odlwebport=odlwebport, + ospassword=self._os_password, ostenantname=self._os_tenantname, + osusername=self._os_username) + + def test_run_missing_os_username(self): + self._test_run_missing_env_var("OS_USERNAME") + + def test_run_missing_os_password(self): + self._test_run_missing_env_var("OS_PASSWORD") + + def test_run_missing_os_tenant_name(self): + self._test_run_missing_env_var("OS_TENANT_NAME") + + def test_run_main_false(self): + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip + self._test_run(TestCasesBase.TestCasesBase.EX_RUN_ERROR, + odlip=self._sdn_controller_ip, + odlwebport=self._odl_webport) + + def test_run_main_exception(self): + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip + with self.assertRaises(Exception): + self._test_run(status=TestCasesBase.TestCasesBase.EX_RUN_ERROR, + exception=Exception(), + odlip=self._sdn_controller_ip, + odlwebport=self._odl_webport) + + def test_run_without_installer_type(self): + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip + self._test_run(TestCasesBase.TestCasesBase.EX_OK, + odlip=self._sdn_controller_ip, + odlwebport=self._odl_webport) + + def test_run_fuel(self): + os.environ["INSTALLER_TYPE"] = "fuel" + self._test_run(TestCasesBase.TestCasesBase.EX_OK, + odlip=self._neutron_ip, odlwebport='8282') + + def test_run_apex(self): + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip + os.environ["INSTALLER_TYPE"] = "apex" + self._test_run(TestCasesBase.TestCasesBase.EX_OK, + odlip=self._sdn_controller_ip, odlwebport='8181') + + def test_run_joid(self): + os.environ["SDN_CONTROLLER"] = self._sdn_controller_ip + os.environ["INSTALLER_TYPE"] = "joid" + self._test_run(TestCasesBase.TestCasesBase.EX_OK, + odlip=self._sdn_controller_ip, + odlwebport=self._odl_webport) + + def test_run_compass(self, *args): + os.environ["INSTALLER_TYPE"] = "compass" + self._test_run(TestCasesBase.TestCasesBase.EX_OK, + odlip=self._neutron_ip, odlwebport='8181') + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/utils/functest_utils.py b/utils/functest_utils.py index ef865bed..41b6485d 100644 --- a/utils/functest_utils.py +++ b/utils/functest_utils.py @@ -341,19 +341,26 @@ def get_deployment_dir(): return deployment_dir -def get_criteria_by_test(testname): +def get_dict_by_test(testname): with open(get_testcases_file()) as f: testcases_yaml = yaml.safe_load(f) for dic_tier in testcases_yaml.get("tiers"): for dic_testcase in dic_tier['testcases']: if dic_testcase['name'] == testname: - return dic_testcase['criteria'] + return dic_testcase logger.error('Project %s is not defined in testcases.yaml' % testname) return None +def get_criteria_by_test(testname): + dict = get_dict_by_test(testname) + if dict: + return dict['criteria'] + return None + + # ---------------------------------------------------------- # # YAML UTILS diff --git a/utils/openstack_utils.py b/utils/openstack_utils.py index e0da7d97..39594a2a 100755 --- a/utils/openstack_utils.py +++ b/utils/openstack_utils.py @@ -28,6 +28,15 @@ logger = ft_logger.Logger("openstack_utils").getLogger() # ********************************************* # CREDENTIALS # ********************************************* +class MissingEnvVar(Exception): + + def __init__(self, var): + self.var = var + + def __str__(self): + return str.format("Please set the mandatory env var: {}", self.var) + + def check_credentials(): """ Check if the OpenStack credentials (openrc) are sourced @@ -51,8 +60,7 @@ def get_credentials(service): envvars = ('OS_USERNAME', 'OS_PASSWORD', 'OS_AUTH_URL', 'OS_TENANT_NAME') for envvar in envvars: if os.getenv(envvar) is None: - logger.error("'%s' is not exported as an env variable." % envvar) - exit(-1) + raise MissingEnvVar(envvar) # Unfortunately, each of the OpenStack client will request slightly # different entries in their credentials dict. |