diff options
Diffstat (limited to 'xtesting/tests/unit')
| -rw-r--r-- | xtesting/tests/unit/__init__.py | 0 | ||||
| -rw-r--r-- | xtesting/tests/unit/ci/__init__.py | 0 | ||||
| -rw-r--r-- | xtesting/tests/unit/ci/test_run_tests.py | 267 | ||||
| -rw-r--r-- | xtesting/tests/unit/ci/test_tier_builder.py | 93 | ||||
| -rw-r--r-- | xtesting/tests/unit/ci/test_tier_handler.py | 139 | ||||
| -rw-r--r-- | xtesting/tests/unit/core/__init__.py | 0 | ||||
| -rw-r--r-- | xtesting/tests/unit/core/test_feature.py | 117 | ||||
| -rw-r--r-- | xtesting/tests/unit/core/test_robotframework.py | 199 | ||||
| -rw-r--r-- | xtesting/tests/unit/core/test_testcase.py | 277 | ||||
| -rw-r--r-- | xtesting/tests/unit/core/test_unit.py | 98 | ||||
| -rw-r--r-- | xtesting/tests/unit/core/test_vnf.py | 187 | ||||
| -rw-r--r-- | xtesting/tests/unit/energy/__init__.py | 0 | ||||
| -rw-r--r-- | xtesting/tests/unit/energy/test_functest_energy.py | 371 | ||||
| -rw-r--r-- | xtesting/tests/unit/utils/__init__.py | 0 | ||||
| -rw-r--r-- | xtesting/tests/unit/utils/test_decorators.py | 125 | ||||
| -rw-r--r-- | xtesting/tests/unit/utils/test_env.py | 57 |
16 files changed, 1930 insertions, 0 deletions
diff --git a/xtesting/tests/unit/__init__.py b/xtesting/tests/unit/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/xtesting/tests/unit/__init__.py diff --git a/xtesting/tests/unit/ci/__init__.py b/xtesting/tests/unit/ci/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/xtesting/tests/unit/ci/__init__.py diff --git a/xtesting/tests/unit/ci/test_run_tests.py b/xtesting/tests/unit/ci/test_run_tests.py new file mode 100644 index 00000000..de2af66d --- /dev/null +++ b/xtesting/tests/unit/ci/test_run_tests.py @@ -0,0 +1,267 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +# pylint: disable=missing-docstring + +import logging +import unittest +import os + +import mock + +from xtesting.ci import run_tests +from xtesting.core.testcase import TestCase + + +class FakeModule(TestCase): + + def run(self, **kwargs): + return TestCase.EX_OK + + +class RunTestsTesting(unittest.TestCase): + + def setUp(self): + self.runner = run_tests.Runner() + mock_test_case = mock.Mock() + mock_test_case.is_successful.return_value = TestCase.EX_OK + self.runner.executed_test_cases['test1'] = mock_test_case + self.runner.executed_test_cases['test2'] = mock_test_case + self.sep = 'test_sep' + self.creds = {'OS_AUTH_URL': 'http://test_ip:test_port/v2.0', + 'OS_USERNAME': 'test_os_username', + 'OS_TENANT_NAME': 'test_tenant', + '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_ci_loop.return_value': 'test_ci_loop', + 'get_test_names.return_value': ['test1', 'test2']} + self.tier.configure_mock(**attrs) + + self.tiers = mock.Mock() + attrs = {'get_tiers.return_value': [self.tier]} + self.tiers.configure_mock(**attrs) + + self.run_tests_parser = run_tests.RunTestsParser() + + @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test') + def test_get_run_dict(self, *args): + retval = {'run': mock.Mock()} + args[0].return_value = retval + self.assertEqual(self.runner.get_run_dict('test_name'), retval['run']) + args[0].assert_called_once_with('test_name') + + @mock.patch('xtesting.ci.run_tests.LOGGER.error') + @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test', + return_value=None) + def test_get_run_dict_config_ko(self, *args): + testname = 'test_name' + self.assertEqual(self.runner.get_run_dict(testname), None) + args[0].return_value = {} + self.assertEqual(self.runner.get_run_dict(testname), None) + calls = [mock.call(testname), mock.call(testname)] + args[0].assert_has_calls(calls) + calls = [mock.call("Cannot get %s's config options", testname), + mock.call("Cannot get %s's config options", testname)] + args[1].assert_has_calls(calls) + + @mock.patch('xtesting.ci.run_tests.LOGGER.exception') + @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test', + side_effect=Exception) + def test_get_run_dict_exception(self, *args): + testname = 'test_name' + self.assertEqual(self.runner.get_run_dict(testname), None) + args[1].assert_called_once_with( + "Cannot get %s's config options", testname) + + def _test_source_envfile(self, msg, key='OS_TENANT_NAME', value='admin'): + try: + del os.environ[key] + except Exception: # pylint: disable=broad-except + pass + envfile = 'rc_file' + with mock.patch('six.moves.builtins.open', + mock.mock_open(read_data=msg)) as mock_method,\ + mock.patch('os.path.isfile', return_value=True): + mock_method.return_value.__iter__ = lambda self: iter( + self.readline, '') + self.runner.source_envfile(envfile) + mock_method.assert_called_once_with(envfile, 'r') + self.assertEqual(os.environ[key], value) + + def test_source_envfile(self): + self._test_source_envfile('OS_TENANT_NAME=admin') + self._test_source_envfile('OS_TENANT_NAME= admin') + self._test_source_envfile('OS_TENANT_NAME = admin') + self._test_source_envfile('OS_TENANT_NAME = "admin"') + self._test_source_envfile('export OS_TENANT_NAME=admin') + self._test_source_envfile('export OS_TENANT_NAME =admin') + self._test_source_envfile('export OS_TENANT_NAME = admin') + self._test_source_envfile('export OS_TENANT_NAME = "admin"') + # This test will fail as soon as rc_file is fixed + self._test_source_envfile( + 'export "\'OS_TENANT_NAME\'" = "\'admin\'"') + + def test_get_dict_by_test(self): + with mock.patch('six.moves.builtins.open', mock.mock_open()), \ + mock.patch('yaml.safe_load') as mock_yaml: + mock_obj = mock.Mock() + testcase_dict = {'case_name': 'testname', + 'criteria': 50} + attrs = {'get.return_value': [{'testcases': [testcase_dict]}]} + mock_obj.configure_mock(**attrs) + mock_yaml.return_value = mock_obj + self.assertDictEqual( + run_tests.Runner.get_dict_by_test('testname'), + testcase_dict) + + @mock.patch('xtesting.ci.run_tests.Runner.get_run_dict', + return_value=None) + def test_run_tests_import_exception(self, *args): + mock_test = mock.Mock() + kwargs = {'get_name.return_value': 'test_name', + 'needs_clean.return_value': False} + mock_test.configure_mock(**kwargs) + with self.assertRaises(Exception) as context: + self.runner.run_test(mock_test) + args[0].assert_called_with('test_name') + msg = "Cannot import the class for the test case." + self.assertTrue(msg in str(context.exception)) + + @mock.patch('importlib.import_module', name="module", + return_value=mock.Mock(test_class=mock.Mock( + side_effect=FakeModule))) + @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test') + def test_run_tests_default(self, *args): + mock_test = mock.Mock() + kwargs = {'get_name.return_value': 'test_name', + 'needs_clean.return_value': True} + mock_test.configure_mock(**kwargs) + test_run_dict = {'module': 'test_module', + 'class': 'test_class'} + with mock.patch('xtesting.ci.run_tests.Runner.get_run_dict', + return_value=test_run_dict): + self.runner.clean_flag = True + self.runner.run_test(mock_test) + args[0].assert_called_with('test_name') + args[1].assert_called_with('test_module') + self.assertEqual(self.runner.overall_result, + run_tests.Result.EX_OK) + + @mock.patch('xtesting.ci.run_tests.Runner.run_test', + return_value=TestCase.EX_OK) + def test_run_tier_default(self, *mock_methods): + self.assertEqual(self.runner.run_tier(self.tier), + run_tests.Result.EX_OK) + mock_methods[0].assert_called_with(mock.ANY) + + @mock.patch('xtesting.ci.run_tests.LOGGER.info') + def test_run_tier_missing_test(self, mock_logger_info): + self.tier.get_tests.return_value = None + self.assertEqual(self.runner.run_tier(self.tier), + run_tests.Result.EX_ERROR) + self.assertTrue(mock_logger_info.called) + + @mock.patch('xtesting.ci.run_tests.LOGGER.info') + @mock.patch('xtesting.ci.run_tests.Runner.run_tier') + @mock.patch('xtesting.ci.run_tests.Runner.summary') + def test_run_all_default(self, *mock_methods): + os.environ['CI_LOOP'] = 'test_ci_loop' + self.runner.run_all() + mock_methods[1].assert_not_called() + self.assertTrue(mock_methods[2].called) + + @mock.patch('xtesting.ci.run_tests.LOGGER.info') + @mock.patch('xtesting.ci.run_tests.Runner.summary') + def test_run_all_missing_tier(self, *mock_methods): + os.environ['CI_LOOP'] = 'loop_re_not_available' + self.runner.run_all() + self.assertTrue(mock_methods[1].called) + + @mock.patch('xtesting.ci.run_tests.Runner.source_envfile', + side_effect=Exception) + @mock.patch('xtesting.ci.run_tests.Runner.summary') + def test_main_failed(self, *mock_methods): + kwargs = {'test': 'test_name', 'noclean': True, 'report': True} + args = {'get_tier.return_value': False, + 'get_test.return_value': False} + self.runner.tiers = mock.Mock() + self.runner.tiers.configure_mock(**args) + self.assertEqual(self.runner.main(**kwargs), + run_tests.Result.EX_ERROR) + mock_methods[1].assert_called_once_with() + + @mock.patch('xtesting.ci.run_tests.Runner.source_envfile') + @mock.patch('xtesting.ci.run_tests.Runner.run_test', + return_value=TestCase.EX_OK) + @mock.patch('xtesting.ci.run_tests.Runner.summary') + def test_main_tier(self, *mock_methods): + mock_tier = mock.Mock() + test_mock = mock.Mock() + test_mock.get_name.return_value = 'test1' + args = {'get_name.return_value': 'tier_name', + 'get_tests.return_value': [test_mock]} + mock_tier.configure_mock(**args) + kwargs = {'test': 'tier_name', 'noclean': True, 'report': True} + args = {'get_tier.return_value': mock_tier, + 'get_test.return_value': None} + self.runner.tiers = mock.Mock() + self.runner.tiers.configure_mock(**args) + self.assertEqual(self.runner.main(**kwargs), + run_tests.Result.EX_OK) + mock_methods[1].assert_called() + + @mock.patch('xtesting.ci.run_tests.Runner.source_envfile') + @mock.patch('xtesting.ci.run_tests.Runner.run_test', + return_value=TestCase.EX_OK) + def test_main_test(self, *mock_methods): + kwargs = {'test': 'test_name', 'noclean': True, 'report': True} + args = {'get_tier.return_value': None, + 'get_test.return_value': 'test_name'} + self.runner.tiers = mock.Mock() + mock_methods[1].return_value = self.creds + self.runner.tiers.configure_mock(**args) + self.assertEqual(self.runner.main(**kwargs), + run_tests.Result.EX_OK) + mock_methods[0].assert_called_once_with('test_name') + + @mock.patch('xtesting.ci.run_tests.Runner.source_envfile') + @mock.patch('xtesting.ci.run_tests.Runner.run_all') + @mock.patch('xtesting.ci.run_tests.Runner.summary') + def test_main_all_tier(self, *args): + kwargs = {'get_tier.return_value': None, + 'get_test.return_value': None} + self.runner.tiers = mock.Mock() + self.runner.tiers.configure_mock(**kwargs) + self.assertEqual( + self.runner.main(test='all', noclean=True, report=True), + run_tests.Result.EX_OK) + args[0].assert_called_once_with(None) + args[1].assert_called_once_with() + args[2].assert_called_once_with() + + @mock.patch('xtesting.ci.run_tests.Runner.source_envfile') + def test_main_any_tier_test_ko(self, *args): + kwargs = {'get_tier.return_value': None, + 'get_test.return_value': None} + self.runner.tiers = mock.Mock() + self.runner.tiers.configure_mock(**kwargs) + self.assertEqual( + self.runner.main(test='any', noclean=True, report=True), + run_tests.Result.EX_ERROR) + args[0].assert_called_once_with() + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/ci/test_tier_builder.py b/xtesting/tests/unit/ci/test_tier_builder.py new file mode 100644 index 00000000..22a44a58 --- /dev/null +++ b/xtesting/tests/unit/ci/test_tier_builder.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +# pylint: disable=missing-docstring + +import logging +import unittest + +import mock + +from xtesting.ci import tier_builder + + +class TierBuilderTesting(unittest.TestCase): + + def setUp(self): + self.dependency = { + 'installer': 'test_installer', 'scenario': 'test_scenario'} + self.testcase = { + 'dependencies': self.dependency, 'enabled': 'true', + 'case_name': 'test_name', 'criteria': 'test_criteria', + 'blocking': 'test_blocking', 'description': 'test_desc', + 'project_name': 'project_name'} + self.dic_tier = { + 'name': 'test_tier', 'order': 'test_order', + 'ci_loop': 'test_ci_loop', 'description': 'test_desc', + 'testcases': [self.testcase]} + self.mock_yaml = mock.Mock() + attrs = {'get.return_value': [self.dic_tier]} + self.mock_yaml.configure_mock(**attrs) + + with mock.patch('xtesting.ci.tier_builder.yaml.safe_load', + return_value=self.mock_yaml), \ + mock.patch('six.moves.builtins.open', mock.mock_open()): + self.tierbuilder = tier_builder.TierBuilder( + 'test_installer', 'test_scenario', 'testcases_file') + self.tier_obj = self.tierbuilder.tier_objects[0] + + def test_get_tiers(self): + self.assertEqual(self.tierbuilder.get_tiers(), + [self.tier_obj]) + + def test_get_tier_names(self): + self.assertEqual(self.tierbuilder.get_tier_names(), + ['test_tier']) + + def test_get_tier_present_tier(self): + self.assertEqual(self.tierbuilder.get_tier('test_tier'), + self.tier_obj) + + def test_get_tier_missing_tier(self): + self.assertEqual(self.tierbuilder.get_tier('test_tier2'), + None) + + def test_get_test_present_test(self): + self.assertEqual(self.tierbuilder.get_test('test_name'), + self.tier_obj.get_test('test_name')) + + def test_get_test_missing_test(self): + self.assertEqual(self.tierbuilder.get_test('test_name2'), + None) + + def test_get_tests_present_tier(self): + self.assertEqual(self.tierbuilder.get_tests('test_tier'), + self.tier_obj.tests_array) + + def test_get_tests_missing_tier(self): + self.assertEqual(self.tierbuilder.get_tests('test_tier2'), + None) + + def test_get_tier_name_ok(self): + self.assertEqual(self.tierbuilder.get_tier_name('test_name'), + 'test_tier') + + def test_get_tier_name_ko(self): + self.assertEqual(self.tierbuilder.get_tier_name('test_name2'), None) + + def test_str(self): + message = str(self.tierbuilder) + self.assertTrue('test_tier' in message) + self.assertTrue('test_order' in message) + self.assertTrue('test_ci_loop' in message) + self.assertTrue('test_desc' in message) + self.assertTrue('test_name' in message) + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/ci/test_tier_handler.py b/xtesting/tests/unit/ci/test_tier_handler.py new file mode 100644 index 00000000..d1900103 --- /dev/null +++ b/xtesting/tests/unit/ci/test_tier_handler.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +# pylint: disable=missing-docstring + +import logging +import unittest + +import mock + +from xtesting.ci import tier_handler + + +class TierHandlerTesting(unittest.TestCase): + # pylint: disable=too-many-public-methods + + def setUp(self): + self.test = mock.Mock() + attrs = {'get_name.return_value': 'test_name'} + self.test.configure_mock(**attrs) + self.mock_depend = mock.Mock() + attrs = {'get_scenario.return_value': 'test_scenario', + 'get_installer.return_value': 'test_installer'} + self.mock_depend.configure_mock(**attrs) + self.tier = tier_handler.Tier( + 'test_tier', 'test_order', 'test_ci_loop', description='test_desc') + self.testcase = tier_handler.TestCase( + 'test_name', 'true', self.mock_depend, 'test_criteria', + True, description='test_desc', project='project_name') + self.dependency = tier_handler.Dependency( + 'test_installer', 'test_scenario') + self.testcase.str = self.testcase.__str__() + self.dependency.str = self.dependency.__str__() + self.tier.str = self.tier.__str__() + + def test_split_text(self): + test_str = 'this is for testing' + self.assertEqual(tier_handler.split_text(test_str, 10), + ['this is ', 'for ', 'testing ']) + + def test_add_test(self): + self.tier.add_test(self.test) + self.assertEqual(self.tier.tests_array, [self.test]) + + def test_get_skipped_test1(self): + self.assertEqual(self.tier.get_skipped_test(), []) + + def test_get_skipped_test2(self): + self.tier.skip_test(self.test) + self.assertEqual(self.tier.get_skipped_test(), [self.test]) + + def test_get_tests(self): + self.tier.tests_array = [self.test] + self.assertEqual(self.tier.get_tests(), [self.test]) + + def test_get_test_names(self): + self.tier.tests_array = [self.test] + self.assertEqual(self.tier.get_test_names(), ['test_name']) + + def test_get_test(self): + self.tier.tests_array = [self.test] + with mock.patch.object(self.tier, 'is_test', return_value=True): + self.assertEqual(self.tier.get_test('test_name'), self.test) + + def test_get_test_missing_test(self): + self.tier.tests_array = [self.test] + with mock.patch.object(self.tier, 'is_test', return_value=False): + self.assertEqual(self.tier.get_test('test_name'), None) + + def test_get_name(self): + self.assertEqual(self.tier.get_name(), 'test_tier') + + def test_get_order(self): + self.assertEqual(self.tier.get_order(), 'test_order') + + def test_get_ci_loop(self): + self.assertEqual(self.tier.get_ci_loop(), 'test_ci_loop') + + def test_testcase_is_none_in_item(self): + self.assertEqual(tier_handler.TestCase.is_none("item"), False) + + def test_testcase_is_none_no_item(self): + self.assertEqual(tier_handler.TestCase.is_none(None), True) + + def test_testcase_is_compatible(self): + self.assertEqual( + self.testcase.is_compatible('test_installer', 'test_scenario'), + True) + + def test_testcase_is_compatible_2(self): + self.assertEqual( + self.testcase.is_compatible('missing_installer', 'test_scenario'), + False) + self.assertEqual( + self.testcase.is_compatible('test_installer', 'missing_scenario'), + False) + + @mock.patch('re.search', side_effect=TypeError) + def test_testcase_is_compatible3(self, *args): + self.assertEqual( + self.testcase.is_compatible('test_installer', 'test_scenario'), + False) + args[0].assert_called_once_with('test_installer', 'test_installer') + + def test_testcase_get_name(self): + self.assertEqual(self.tier.get_name(), 'test_tier') + + def test_testcase_is_enabled(self): + self.assertEqual(self.testcase.is_enabled(), 'true') + + def test_testcase_get_criteria(self): + self.assertEqual(self.testcase.get_criteria(), 'test_criteria') + + def test_testcase_is_blocking(self): + self.assertTrue(self.testcase.is_blocking()) + + def test_testcase_get_project(self): + self.assertEqual(self.testcase.get_project(), 'project_name') + + def test_testcase_get_order(self): + self.assertEqual(self.tier.get_order(), 'test_order') + + def test_testcase_get_ci_loop(self): + self.assertEqual(self.tier.get_ci_loop(), 'test_ci_loop') + + def test_dependency_get_installer(self): + self.assertEqual(self.dependency.get_installer(), 'test_installer') + + def test_dependency_get_scenario(self): + self.assertEqual(self.dependency.get_scenario(), 'test_scenario') + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/core/__init__.py b/xtesting/tests/unit/core/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/xtesting/tests/unit/core/__init__.py diff --git a/xtesting/tests/unit/core/test_feature.py b/xtesting/tests/unit/core/test_feature.py new file mode 100644 index 00000000..9bbe5331 --- /dev/null +++ b/xtesting/tests/unit/core/test_feature.py @@ -0,0 +1,117 @@ +#!/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 unittest + +import mock + +from xtesting.core import feature +from xtesting.core import testcase + + +class FeatureTestingBase(unittest.TestCase): + + _case_name = "foo" + _project_name = "bar" + _repo = "dir_repo_bar" + _cmd = "run_bar_tests.py" + _output_file = '/home/opnfv/xtesting/results/foo.log' + feature = None + + @mock.patch('time.time', side_effect=[1, 2]) + def _test_run(self, status, mock_method=None): + self.assertEqual(self.feature.run(cmd=self._cmd), status) + if status == testcase.TestCase.EX_OK: + self.assertEqual(self.feature.result, 100) + else: + self.assertEqual(self.feature.result, 0) + mock_method.assert_has_calls([mock.call(), mock.call()]) + self.assertEqual(self.feature.start_time, 1) + self.assertEqual(self.feature.stop_time, 2) + + def test_logger_module_ko(self): + with mock.patch('six.moves.builtins.open'): + self.feature = feature.Feature( + project_name=self._project_name, case_name=self._case_name) + self.assertEqual(self.feature.logger.name, self._case_name) + + def test_logger_module(self): + with mock.patch('six.moves.builtins.open'): + self.feature = feature.Feature( + project_name=self._project_name, case_name=self._case_name, + run={'module': 'bar'}) + self.assertEqual(self.feature.logger.name, 'bar') + + +class FeatureTesting(FeatureTestingBase): + + def setUp(self): + # logging must be disabled else it calls time.time() + # what will break these unit tests. + logging.disable(logging.CRITICAL) + with mock.patch('six.moves.builtins.open'): + self.feature = feature.Feature( + project_name=self._project_name, case_name=self._case_name) + + def test_run_exc(self): + # pylint: disable=bad-continuation + with mock.patch.object( + self.feature, 'execute', + side_effect=Exception) as mock_method: + self._test_run(testcase.TestCase.EX_RUN_ERROR) + mock_method.assert_called_once_with(cmd=self._cmd) + + def test_run(self): + self._test_run(testcase.TestCase.EX_RUN_ERROR) + + +class BashFeatureTesting(FeatureTestingBase): + + def setUp(self): + # logging must be disabled else it calls time.time() + # what will break these unit tests. + logging.disable(logging.CRITICAL) + with mock.patch('six.moves.builtins.open'): + self.feature = feature.BashFeature( + project_name=self._project_name, case_name=self._case_name) + + @mock.patch('subprocess.Popen') + def test_run_no_cmd(self, mock_subproc): + self.assertEqual( + self.feature.run(), testcase.TestCase.EX_RUN_ERROR) + mock_subproc.assert_not_called() + + @mock.patch('subprocess.Popen') + def test_run_ko(self, mock_subproc): + with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen: + mock_obj = mock.Mock() + attrs = {'wait.return_value': 1} + mock_obj.configure_mock(**attrs) + + mock_subproc.return_value = mock_obj + self._test_run(testcase.TestCase.EX_RUN_ERROR) + mopen.assert_called_once_with(self._output_file, "w+") + + @mock.patch('subprocess.Popen') + def test_run(self, mock_subproc): + with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen: + mock_obj = mock.Mock() + attrs = {'wait.return_value': 0} + mock_obj.configure_mock(**attrs) + + mock_subproc.return_value = mock_obj + self._test_run(testcase.TestCase.EX_OK) + mopen.assert_called_once_with(self._output_file, "w+") + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/core/test_robotframework.py b/xtesting/tests/unit/core/test_robotframework.py new file mode 100644 index 00000000..7131b7e2 --- /dev/null +++ b/xtesting/tests/unit/core/test_robotframework.py @@ -0,0 +1,199 @@ +#!/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 + +"""Define the classes required to fully cover robot.""" + +import errno +import logging +import os +import unittest + +import mock +from robot.errors import DataError, RobotError +from robot.result import model +from robot.utils.robottime import timestamp_to_secs + +from xtesting.core import robotframework + +__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" + + +class ResultVisitorTesting(unittest.TestCase): + + """The class testing ResultVisitor.""" + # pylint: disable=missing-docstring + + def setUp(self): + self.visitor = robotframework.ResultVisitor() + + def test_empty(self): + self.assertFalse(self.visitor.get_data()) + + def test_ok(self): + data = {'name': 'foo', + 'parent': 'bar', + 'status': 'PASS', + 'starttime': "20161216 16:00:00.000", + 'endtime': "20161216 16:00:01.000", + 'elapsedtime': 1000, + 'text': 'Hello, World!', + 'critical': True} + test = model.TestCase( + name=data['name'], status=data['status'], message=data['text'], + starttime=data['starttime'], endtime=data['endtime']) + test.parent = mock.Mock() + config = {'name': data['parent'], + 'criticality.test_is_critical.return_value': data[ + 'critical']} + test.parent.configure_mock(**config) + self.visitor.visit_test(test) + self.assertEqual(self.visitor.get_data(), [data]) + + +class ParseResultTesting(unittest.TestCase): + + """The class testing RobotFramework.parse_results().""" + # pylint: disable=missing-docstring + + _config = {'name': 'dummy', 'starttime': '20161216 16:00:00.000', + 'endtime': '20161216 16:00:01.000'} + + def setUp(self): + self.test = robotframework.RobotFramework( + case_name='robot', project_name='xtesting') + + @mock.patch('robot.api.ExecutionResult', side_effect=DataError) + def test_raises_exc(self, mock_method): + with self.assertRaises(DataError): + self.test.parse_results() + mock_method.assert_called_once_with( + os.path.join(self.test.res_dir, 'output.xml')) + + def _test_result(self, config, result): + suite = mock.Mock() + suite.configure_mock(**config) + with mock.patch('robot.api.ExecutionResult', + return_value=mock.Mock(suite=suite)): + self.test.parse_results() + self.assertEqual(self.test.result, result) + self.assertEqual(self.test.start_time, + timestamp_to_secs(config['starttime'])) + self.assertEqual(self.test.stop_time, + timestamp_to_secs(config['endtime'])) + self.assertEqual(self.test.details, + {'description': config['name'], 'tests': []}) + + def test_null_passed(self): + self._config.update({'statistics.critical.passed': 0, + 'statistics.critical.total': 20}) + self._test_result(self._config, 0) + + def test_no_test(self): + self._config.update({'statistics.critical.passed': 20, + 'statistics.critical.total': 0}) + self._test_result(self._config, 0) + + def test_half_success(self): + self._config.update({'statistics.critical.passed': 10, + 'statistics.critical.total': 20}) + self._test_result(self._config, 50) + + def test_success(self): + self._config.update({'statistics.critical.passed': 20, + 'statistics.critical.total': 20}) + self._test_result(self._config, 100) + + +class RunTesting(unittest.TestCase): + + """The class testing RobotFramework.run().""" + # pylint: disable=missing-docstring + + suites = ["foo"] + variable = [] + variablefile = [] + + def setUp(self): + self.test = robotframework.RobotFramework( + case_name='robot', project_name='xtesting') + + def test_exc_key_error(self): + self.assertEqual(self.test.run(), self.test.EX_RUN_ERROR) + + @mock.patch('robot.run') + def _test_makedirs_exc(self, *args): + with mock.patch.object(self.test, 'parse_results') as mock_method: + self.assertEqual( + self.test.run( + suites=self.suites, variable=self.variable, + variablefile=self.variablefile), + self.test.EX_RUN_ERROR) + args[0].assert_not_called() + mock_method.asser_not_called() + + @mock.patch('os.makedirs', side_effect=Exception) + def test_makedirs_exc(self, *args): + self._test_makedirs_exc() + args[0].assert_called_once_with(self.test.res_dir) + + @mock.patch('os.makedirs', side_effect=OSError) + def test_makedirs_oserror(self, *args): + self._test_makedirs_exc() + args[0].assert_called_once_with(self.test.res_dir) + + @mock.patch('robot.run') + def _test_makedirs(self, *args): + with mock.patch.object(self.test, 'parse_results') as mock_method: + self.assertEqual( + self.test.run(suites=self.suites, variable=self.variable), + self.test.EX_OK) + args[0].assert_called_once_with( + *self.suites, log='NONE', output=self.test.xml_file, + report='NONE', stdout=mock.ANY, variable=self.variable, + variablefile=self.variablefile) + mock_method.assert_called_once_with() + + @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, '')) + def test_makedirs_oserror17(self, *args): + self._test_makedirs() + args[0].assert_called_once_with(self.test.res_dir) + + @mock.patch('os.makedirs') + def test_makedirs(self, *args): + self._test_makedirs() + args[0].assert_called_once_with(self.test.res_dir) + + @mock.patch('robot.run') + def _test_parse_results(self, status, *args): + self.assertEqual( + self.test.run( + suites=self.suites, variable=self.variable, + variablefile=self.variablefile), + status) + args[0].assert_called_once_with( + *self.suites, log='NONE', output=self.test.xml_file, + report='NONE', stdout=mock.ANY, variable=self.variable, + variablefile=self.variablefile) + + def test_parse_results_exc(self): + with mock.patch.object(self.test, 'parse_results', + side_effect=Exception) as mock_method: + self._test_parse_results(self.test.EX_RUN_ERROR) + mock_method.assert_called_once_with() + + def test_parse_results_robot_error(self): + with mock.patch.object(self.test, 'parse_results', + side_effect=RobotError('foo')) as mock_method: + self._test_parse_results(self.test.EX_RUN_ERROR) + mock_method.assert_called_once_with() + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/core/test_testcase.py b/xtesting/tests/unit/core/test_testcase.py new file mode 100644 index 00000000..e2f56f8f --- /dev/null +++ b/xtesting/tests/unit/core/test_testcase.py @@ -0,0 +1,277 @@ +#!/usr/bin/env python + +# Copyright (c) 2016 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 + +"""Define the class required to fully cover testcase.""" + +from datetime import datetime +import json +import logging +import os +import unittest + +from xtesting.core import testcase + +import mock +import requests + + +__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" + + +class TestCaseTesting(unittest.TestCase): + """The class testing TestCase.""" + + # pylint: disable=missing-docstring,too-many-public-methods + + _case_name = "base" + _project_name = "xtesting" + _published_result = "PASS" + _test_db_url = "http://testresults.opnfv.org/test/api/v1/results" + _headers = {'Content-Type': 'application/json'} + + def setUp(self): + self.test = testcase.TestCase(case_name=self._case_name, + project_name=self._project_name) + self.test.start_time = 1 + self.test.stop_time = 2 + self.test.result = 100 + self.test.details = {"Hello": "World"} + os.environ['TEST_DB_URL'] = TestCaseTesting._test_db_url + os.environ['INSTALLER_TYPE'] = "installer_type" + os.environ['DEPLOY_SCENARIO'] = "scenario" + os.environ['NODE_NAME'] = "node_name" + os.environ['BUILD_TAG'] = "foo-daily-master-bar" + + def test_run_unimplemented(self): + self.assertEqual(self.test.run(), + testcase.TestCase.EX_RUN_ERROR) + + def _test_pushdb_missing_attribute(self): + self.assertEqual(self.test.push_to_db(), + testcase.TestCase.EX_PUSH_TO_DB_ERROR) + + def test_pushdb_no_project_name(self): + self.test.project_name = None + self._test_pushdb_missing_attribute() + + def test_pushdb_no_case_name(self): + self.test.case_name = None + self._test_pushdb_missing_attribute() + + def test_pushdb_no_start_time(self): + self.test.start_time = None + self._test_pushdb_missing_attribute() + + def test_pushdb_no_stop_time(self): + self.test.stop_time = None + self._test_pushdb_missing_attribute() + + def _test_pushdb_missing_env(self, var): + del os.environ[var] + self.assertEqual(self.test.push_to_db(), + testcase.TestCase.EX_PUSH_TO_DB_ERROR) + + def test_pushdb_no_db_url(self): + self._test_pushdb_missing_env('TEST_DB_URL') + + def test_pushdb_no_installer_type(self): + self._test_pushdb_missing_env('INSTALLER_TYPE') + + def test_pushdb_no_deploy_scenario(self): + self._test_pushdb_missing_env('DEPLOY_SCENARIO') + + def test_pushdb_no_node_name(self): + self._test_pushdb_missing_env('NODE_NAME') + + def test_pushdb_no_build_tag(self): + self._test_pushdb_missing_env('BUILD_TAG') + + @mock.patch('requests.post') + def test_pushdb_bad_start_time(self, mock_function=None): + self.test.start_time = "1" + self.assertEqual( + self.test.push_to_db(), + testcase.TestCase.EX_PUSH_TO_DB_ERROR) + mock_function.assert_not_called() + + @mock.patch('requests.post') + def test_pushdb_bad_end_time(self, mock_function=None): + self.test.stop_time = "2" + self.assertEqual( + self.test.push_to_db(), + testcase.TestCase.EX_PUSH_TO_DB_ERROR) + mock_function.assert_not_called() + + def _get_data(self): + return { + "build_tag": os.environ['BUILD_TAG'], + "case_name": self._case_name, + "criteria": 'PASS' if self.test.is_successful( + ) == self.test.EX_OK else 'FAIL', + "details": self.test.details, + "installer": os.environ['INSTALLER_TYPE'], + "pod_name": os.environ['NODE_NAME'], + "project_name": self.test.project_name, + "scenario": os.environ['DEPLOY_SCENARIO'], + "start_date": datetime.fromtimestamp( + self.test.start_time).strftime('%Y-%m-%d %H:%M:%S'), + "stop_date": datetime.fromtimestamp( + self.test.stop_time).strftime('%Y-%m-%d %H:%M:%S'), + "version": "master"} + + @mock.patch('requests.post') + def _test_pushdb_version(self, mock_function=None, **kwargs): + payload = self._get_data() + payload["version"] = kwargs.get("version", "unknown") + self.assertEqual(self.test.push_to_db(), testcase.TestCase.EX_OK) + mock_function.assert_called_once_with( + os.environ['TEST_DB_URL'], + data=json.dumps(payload, sort_keys=True), + headers=self._headers) + + def test_pushdb_daily_job(self): + self._test_pushdb_version(version="master") + + def test_pushdb_weekly_job(self): + os.environ['BUILD_TAG'] = 'foo-weekly-master-bar' + self._test_pushdb_version(version="master") + + def test_pushdb_random_build_tag(self): + os.environ['BUILD_TAG'] = 'whatever' + self._test_pushdb_version(version="unknown") + + @mock.patch('requests.post', return_value=mock.Mock( + raise_for_status=mock.Mock( + side_effect=requests.exceptions.HTTPError))) + def test_pushdb_http_errors(self, mock_function=None): + self.assertEqual( + self.test.push_to_db(), + testcase.TestCase.EX_PUSH_TO_DB_ERROR) + mock_function.assert_called_once_with( + os.environ['TEST_DB_URL'], + data=json.dumps(self._get_data(), sort_keys=True), + headers=self._headers) + + def test_check_criteria_missing(self): + self.test.criteria = None + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + + def test_check_result_missing(self): + self.test.result = None + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + + def test_check_result_failed(self): + # Backward compatibility + # It must be removed as soon as TestCase subclasses + # stop setting result = 'PASS' or 'FAIL'. + self.test.result = 'FAIL' + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + + def test_check_result_pass(self): + # Backward compatibility + # It must be removed as soon as TestCase subclasses + # stop setting result = 'PASS' or 'FAIL'. + self.test.result = 'PASS' + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_OK) + + def test_check_result_lt(self): + self.test.result = 50 + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + + def test_check_result_eq(self): + self.test.result = 100 + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_OK) + + def test_check_result_gt(self): + self.test.criteria = 50 + self.test.result = 100 + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_OK) + + def test_check_result_zero(self): + self.test.criteria = 0 + self.test.result = 0 + self.assertEqual(self.test.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + + def test_get_duration_start_ko(self): + self.test.start_time = None + self.assertEqual(self.test.get_duration(), "XX:XX") + self.test.start_time = 0 + self.assertEqual(self.test.get_duration(), "XX:XX") + + def test_get_duration_end_ko(self): + self.test.stop_time = None + self.assertEqual(self.test.get_duration(), "XX:XX") + self.test.stop_time = 0 + self.assertEqual(self.test.get_duration(), "XX:XX") + + def test_get_invalid_duration(self): + self.test.start_time = 2 + self.test.stop_time = 1 + self.assertEqual(self.test.get_duration(), "XX:XX") + + def test_get_zero_duration(self): + self.test.start_time = 2 + self.test.stop_time = 2 + self.assertEqual(self.test.get_duration(), "00:00") + + def test_get_duration(self): + self.test.start_time = 1 + self.test.stop_time = 180 + self.assertEqual(self.test.get_duration(), "02:59") + + def test_str_project_name_ko(self): + self.test.project_name = None + self.assertIn("<xtesting.core.testcase.TestCase object at", + str(self.test)) + + def test_str_case_name_ko(self): + self.test.case_name = None + self.assertIn("<xtesting.core.testcase.TestCase object at", + str(self.test)) + + def test_str_pass(self): + duration = '01:01' + with mock.patch.object(self.test, 'get_duration', + return_value=duration), \ + mock.patch.object(self.test, 'is_successful', + return_value=testcase.TestCase.EX_OK): + message = str(self.test) + self.assertIn(self._project_name, message) + self.assertIn(self._case_name, message) + self.assertIn(duration, message) + self.assertIn('PASS', message) + + def test_str_fail(self): + duration = '00:59' + with mock.patch.object(self.test, 'get_duration', + return_value=duration), \ + mock.patch.object( + self.test, 'is_successful', + return_value=testcase.TestCase.EX_TESTCASE_FAILED): + message = str(self.test) + self.assertIn(self._project_name, message) + self.assertIn(self._case_name, message) + self.assertIn(duration, message) + self.assertIn('FAIL', message) + + def test_clean(self): + self.assertEqual(self.test.clean(), None) + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/core/test_unit.py b/xtesting/tests/unit/core/test_unit.py new file mode 100644 index 00000000..8afe0bde --- /dev/null +++ b/xtesting/tests/unit/core/test_unit.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +# pylint: disable=missing-docstring + +import logging +import unittest + +import mock + +from xtesting.core import unit +from xtesting.core import testcase + + +class PyTestSuiteRunnerTesting(unittest.TestCase): + + def setUp(self): + self.psrunner = unit.Suite() + self.psrunner.suite = "foo" + + @mock.patch('unittest.TestLoader') + def _test_run(self, mock_class=None, result=mock.Mock(), + status=testcase.TestCase.EX_OK): + with mock.patch('xtesting.core.unit.unittest.TextTestRunner.run', + return_value=result): + self.assertEqual(self.psrunner.run(), status) + mock_class.assert_not_called() + + def test_check_suite_null(self): + self.assertEqual(unit.Suite().suite, None) + self.psrunner.suite = None + self._test_run(result=mock.Mock(), + status=testcase.TestCase.EX_RUN_ERROR) + + def test_run_no_ut(self): + mock_result = mock.Mock(testsRun=0, errors=[], failures=[]) + self._test_run(result=mock_result, + status=testcase.TestCase.EX_RUN_ERROR) + self.assertEqual(self.psrunner.result, 0) + self.assertEqual(self.psrunner.details, + {'errors': 0, 'failures': 0, 'stream': '', + 'testsRun': 0}) + self.assertEqual(self.psrunner.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + + def test_run_result_ko(self): + self.psrunner.criteria = 100 + mock_result = mock.Mock(testsRun=50, errors=[('test1', 'error_msg1')], + failures=[('test2', 'failure_msg1')]) + self._test_run(result=mock_result) + self.assertEqual(self.psrunner.result, 96) + self.assertEqual(self.psrunner.details, + {'errors': 1, 'failures': 1, 'stream': '', + 'testsRun': 50}) + self.assertEqual(self.psrunner.is_successful(), + testcase.TestCase.EX_TESTCASE_FAILED) + + def test_run_result_ok(self): + mock_result = mock.Mock(testsRun=50, errors=[], + failures=[]) + self._test_run(result=mock_result) + self.assertEqual(self.psrunner.result, 100) + self.assertEqual(self.psrunner.details, + {'errors': 0, 'failures': 0, 'stream': '', + 'testsRun': 50}) + self.assertEqual(self.psrunner.is_successful(), + testcase.TestCase.EX_OK) + + @mock.patch('unittest.TestLoader') + def test_run_name_exc(self, mock_class=None): + mock_obj = mock.Mock(side_effect=ImportError) + mock_class.side_effect = mock_obj + self.assertEqual(self.psrunner.run(name='foo'), + testcase.TestCase.EX_RUN_ERROR) + mock_class.assert_called_once_with() + mock_obj.assert_called_once_with() + + @mock.patch('unittest.TestLoader') + def test_run_name(self, mock_class=None): + mock_result = mock.Mock(testsRun=50, errors=[], + failures=[]) + mock_obj = mock.Mock() + mock_class.side_effect = mock_obj + with mock.patch('xtesting.core.unit.unittest.TextTestRunner.run', + return_value=mock_result): + self.assertEqual(self.psrunner.run(name='foo'), + testcase.TestCase.EX_OK) + mock_class.assert_called_once_with() + mock_obj.assert_called_once_with() + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/core/test_vnf.py b/xtesting/tests/unit/core/test_vnf.py new file mode 100644 index 00000000..ec8a783e --- /dev/null +++ b/xtesting/tests/unit/core/test_vnf.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python + +# Copyright (c) 2016 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 unittest + +import mock + +from xtesting.core import vnf +from xtesting.core import testcase +from xtesting.utils import constants + +from snaps.openstack.os_credentials import OSCreds + + +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='xtesting', case_name='foo') + + def test_run_deploy_orch_exc(self): + with mock.patch.object(self.test, 'prepare'), \ + mock.patch.object(self.test, 'deploy_orchestrator', + side_effect=Exception) as mock_method, \ + 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) + mock_method.assert_called_with() + + def test_run_deploy_vnf_exc(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', + side_effect=Exception) as mock_method: + self.assertEqual(self.test.run(), + testcase.TestCase.EX_TESTCASE_FAILED) + mock_method.assert_called_with() + + def test_run_test_vnf_exc(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', + side_effect=Exception) as mock_method: + self.assertEqual(self.test.run(), + testcase.TestCase.EX_TESTCASE_FAILED) + mock_method.assert_called_with() + + def test_run_deploy_orch_ko(self): + with mock.patch.object(self.test, 'prepare'),\ + mock.patch.object(self.test, 'deploy_orchestrator', + 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) + + @mock.patch('xtesting.core.vnf.OpenStackUser') + @mock.patch('xtesting.core.vnf.OpenStackProject') + @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials', + side_effect=Exception) + def test_prepare_exc1(self, *args): + with self.assertRaises(Exception): + self.test.prepare() + args[0].assert_called_with(os_env_file=constants.ENV_FILE) + args[1].assert_not_called() + args[2].assert_not_called() + + @mock.patch('xtesting.core.vnf.OpenStackUser') + @mock.patch('xtesting.core.vnf.OpenStackProject', side_effect=Exception) + @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials') + def test_prepare_exc2(self, *args): + with self.assertRaises(Exception): + self.test.prepare() + args[0].assert_called_with(os_env_file=constants.ENV_FILE) + args[1].assert_called_with(mock.ANY, mock.ANY) + args[2].assert_not_called() + + @mock.patch('xtesting.core.vnf.OpenStackUser', side_effect=Exception) + @mock.patch('xtesting.core.vnf.OpenStackProject') + @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials') + def test_prepare_exc3(self, *args): + with self.assertRaises(Exception): + self.test.prepare() + args[0].assert_called_with(os_env_file=constants.ENV_FILE) + args[1].assert_called_with(mock.ANY, mock.ANY) + args[2].assert_called_with(mock.ANY, mock.ANY) + + @mock.patch('xtesting.core.vnf.OpenStackUser') + @mock.patch('xtesting.core.vnf.OpenStackProject') + @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials') + def test_prepare_default(self, *args): + self.assertEqual(self.test.prepare(), testcase.TestCase.EX_OK) + args[0].assert_called_with(os_env_file=constants.ENV_FILE) + args[1].assert_called_with(mock.ANY, mock.ANY) + args[2].assert_called_with(mock.ANY, mock.ANY) + + 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() + + def test_deploy_orch_unimplemented(self): + self.assertTrue(self.test.deploy_orchestrator()) + + @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials', + return_value=OSCreds( + username='user', password='pass', + auth_url='http://foo.com:5000/v3', project_name='bar'), + side_effect=Exception) + def test_prepare_keystone_client_ko(self, *args): + with self.assertRaises(vnf.VnfPreparationException): + self.test.prepare() + args[0].assert_called_once() + + def test_vnf_clean_exc(self): + obj = mock.Mock() + obj.clean.side_effect = Exception + self.test.created_object = [obj] + self.test.clean() + obj.clean.assert_called_with() + + def test_vnf_clean(self): + obj = mock.Mock() + self.test.created_object = [obj] + self.test.clean() + obj.clean.assert_called_with() + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/energy/__init__.py b/xtesting/tests/unit/energy/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/xtesting/tests/unit/energy/__init__.py diff --git a/xtesting/tests/unit/energy/test_functest_energy.py b/xtesting/tests/unit/energy/test_functest_energy.py new file mode 100644 index 00000000..ea83c1ea --- /dev/null +++ b/xtesting/tests/unit/energy/test_functest_energy.py @@ -0,0 +1,371 @@ +#!/usr/bin/env python +# -*- coding: UTF-8 -*- + +# 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 + +"""Unitary test for energy module.""" +# pylint: disable=unused-argument +import logging +import os +import unittest + +import mock +import requests + +from xtesting.energy.energy import EnergyRecorder +import xtesting.energy.energy as energy + +CASE_NAME = "UNIT_TEST_CASE" +STEP_NAME = "UNIT_TEST_STEP" + +PREVIOUS_SCENARIO = "previous_scenario" +PREVIOUS_STEP = "previous_step" + + +class MockHttpResponse(object): # pylint: disable=too-few-public-methods + """Mock response for Energy recorder API.""" + + def __init__(self, text, status_code): + """Create an instance of MockHttpResponse.""" + self.text = text + self.status_code = status_code + + +API_OK = MockHttpResponse( + '{"status": "OK"}', + 200 +) +API_KO = MockHttpResponse( + '{"message": "API-KO"}', + 500 +) + +RECORDER_OK = MockHttpResponse( + '{"environment": "UNIT_TEST",' + ' "step": "string",' + ' "scenario": "' + CASE_NAME + '"}', + 200 +) +RECORDER_KO = MockHttpResponse( + '{"message": "An unhandled API exception occurred (MOCK)"}', + 500 +) +RECORDER_NOT_FOUND = MockHttpResponse( + '{"message": "Recorder not found (MOCK)"}', + 404 +) + + +# pylint: disable=too-many-public-methods +class EnergyRecorderTest(unittest.TestCase): + """Energy module unitary test suite.""" + + case_name = CASE_NAME + request_headers = {'content-type': 'application/json'} + returned_value_to_preserve = "value" + exception_message_to_preserve = "exception_message" + + @staticmethod + def _set_env_creds(): + """Set config values.""" + os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888" + os.environ["ENERGY_RECORDER_API_USER"] = "user" + os.environ["ENERGY_RECORDER_API_PASSWORD"] = "password" + + @staticmethod + def _set_env_nocreds(): + """Set config values.""" + os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888" + del os.environ["ENERGY_RECORDER_API_USER"] + del os.environ["ENERGY_RECORDER_API_PASSWORD"] + + @mock.patch('xtesting.energy.energy.requests.post', + return_value=RECORDER_OK) + def test_start(self, post_mock=None, get_mock=None): + """EnergyRecorder.start method (regular case).""" + self.test_load_config() + self.assertTrue(EnergyRecorder.start(self.case_name)) + post_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"], + auth=EnergyRecorder.energy_recorder_api["auth"], + data=mock.ANY, + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.requests.post', + side_effect=Exception("Internal execution error (MOCK)")) + def test_start_error(self, post_mock=None): + """EnergyRecorder.start method (error in method).""" + self.test_load_config() + self.assertFalse(EnergyRecorder.start(self.case_name)) + post_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"], + auth=EnergyRecorder.energy_recorder_api["auth"], + data=mock.ANY, + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.EnergyRecorder.load_config', + side_effect=Exception("Internal execution error (MOCK)")) + def test_start_exception(self, conf_loader_mock=None): + """EnergyRecorder.start test with exception during execution.""" + start_status = EnergyRecorder.start(CASE_NAME) + self.assertFalse(start_status) + + @mock.patch('xtesting.energy.energy.requests.post', + return_value=RECORDER_KO) + def test_start_api_error(self, post_mock=None): + """EnergyRecorder.start method (API error).""" + self.test_load_config() + self.assertFalse(EnergyRecorder.start(self.case_name)) + post_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"], + auth=EnergyRecorder.energy_recorder_api["auth"], + data=mock.ANY, + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.requests.post', + return_value=RECORDER_OK) + def test_set_step(self, post_mock=None): + """EnergyRecorder.set_step method (regular case).""" + self.test_load_config() + self.assertTrue(EnergyRecorder.set_step(STEP_NAME)) + post_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"] + "/step", + auth=EnergyRecorder.energy_recorder_api["auth"], + data=mock.ANY, + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.requests.post', + return_value=RECORDER_KO) + def test_set_step_api_error(self, post_mock=None): + """EnergyRecorder.set_step method (API error).""" + self.test_load_config() + self.assertFalse(EnergyRecorder.set_step(STEP_NAME)) + post_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"] + "/step", + auth=EnergyRecorder.energy_recorder_api["auth"], + data=mock.ANY, + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.requests.post', + side_effect=Exception("Internal execution error (MOCK)")) + def test_set_step_error(self, post_mock=None): + """EnergyRecorder.set_step method (method error).""" + self.test_load_config() + self.assertFalse(EnergyRecorder.set_step(STEP_NAME)) + post_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"] + "/step", + auth=EnergyRecorder.energy_recorder_api["auth"], + data=mock.ANY, + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.EnergyRecorder.load_config', + side_effect=requests.exceptions.ConnectionError()) + def test_set_step_connection_error(self, conf_loader_mock=None): + """EnergyRecorder.start test with exception during execution.""" + step_status = EnergyRecorder.set_step(STEP_NAME) + self.assertFalse(step_status) + + @mock.patch('xtesting.energy.energy.requests.delete', + return_value=RECORDER_OK) + def test_stop(self, delete_mock=None): + """EnergyRecorder.stop method (regular case).""" + self.test_load_config() + self.assertTrue(EnergyRecorder.stop()) + delete_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"], + auth=EnergyRecorder.energy_recorder_api["auth"], + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.requests.delete', + return_value=RECORDER_KO) + def test_stop_api_error(self, delete_mock=None): + """EnergyRecorder.stop method (API Error).""" + self.test_load_config() + self.assertFalse(EnergyRecorder.stop()) + delete_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"], + auth=EnergyRecorder.energy_recorder_api["auth"], + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @mock.patch('xtesting.energy.energy.requests.delete', + side_effect=Exception("Internal execution error (MOCK)")) + def test_stop_error(self, delete_mock=None): + """EnergyRecorder.stop method (method error).""" + self.test_load_config() + self.assertFalse(EnergyRecorder.stop()) + delete_mock.assert_called_once_with( + EnergyRecorder.energy_recorder_api["uri"], + auth=EnergyRecorder.energy_recorder_api["auth"], + headers=self.request_headers, + timeout=EnergyRecorder.CONNECTION_TIMEOUT + ) + + @energy.enable_recording + def __decorated_method(self): + """Call with to energy recorder decorators.""" + return self.returned_value_to_preserve + + @energy.enable_recording + def __decorated_method_with_ex(self): + """Call with to energy recorder decorators.""" + raise Exception(self.exception_message_to_preserve) + + @mock.patch("xtesting.energy.energy.EnergyRecorder.get_current_scenario", + return_value=None) + @mock.patch("xtesting.energy.energy.EnergyRecorder") + def test_decorators(self, + recorder_mock=None, + cur_scenario_mock=None): + """Test energy module decorators.""" + self.__decorated_method() + calls = [mock.call.start(self.case_name), + mock.call.stop()] + recorder_mock.assert_has_calls(calls) + + @mock.patch("xtesting.energy.energy.EnergyRecorder.get_current_scenario", + return_value={"scenario": PREVIOUS_SCENARIO, + "step": PREVIOUS_STEP}) + @mock.patch("xtesting.energy.energy.EnergyRecorder") + def test_decorators_with_previous(self, + recorder_mock=None, + cur_scenario_mock=None): + """Test energy module decorators.""" + os.environ['NODE_NAME'] = 'MOCK_POD' + self._set_env_creds() + self.__decorated_method() + calls = [mock.call.start(self.case_name), + mock.call.submit_scenario(PREVIOUS_SCENARIO, + PREVIOUS_STEP)] + recorder_mock.assert_has_calls(calls, True) + + def test_decorator_preserve_return(self): + """Test that decorator preserve method returned value.""" + self.test_load_config() + self.assertTrue( + self.__decorated_method() == self.returned_value_to_preserve + ) + + @mock.patch( + "xtesting.energy.energy.finish_session") + def test_decorator_preserve_ex(self, finish_mock=None): + """Test that decorator preserve method exceptions.""" + self.test_load_config() + with self.assertRaises(Exception) as context: + self.__decorated_method_with_ex() + self.assertTrue( + self.exception_message_to_preserve in str(context.exception) + ) + self.assertTrue(finish_mock.called) + + @mock.patch("xtesting.energy.energy.requests.get", + return_value=API_OK) + def test_load_config(self, loader_mock=None, get_mock=None): + """Test load config.""" + os.environ['NODE_NAME'] = 'MOCK_POD' + self._set_env_creds() + EnergyRecorder.energy_recorder_api = None + EnergyRecorder.load_config() + + self.assertEquals( + EnergyRecorder.energy_recorder_api["auth"], + ("user", "password") + ) + self.assertEquals( + EnergyRecorder.energy_recorder_api["uri"], + "http://pod-uri:8888/recorders/environment/MOCK_POD" + ) + + @mock.patch("xtesting.energy.energy.requests.get", + return_value=API_OK) + def test_load_config_no_creds(self, loader_mock=None, get_mock=None): + """Test load config without creds.""" + os.environ['NODE_NAME'] = 'MOCK_POD' + self._set_env_nocreds() + EnergyRecorder.energy_recorder_api = None + EnergyRecorder.load_config() + self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None) + self.assertEquals( + EnergyRecorder.energy_recorder_api["uri"], + "http://pod-uri:8888/recorders/environment/MOCK_POD" + ) + + @mock.patch("xtesting.energy.energy.requests.get", + return_value=API_OK) + def test_load_config_ex(self, loader_mock=None, get_mock=None): + """Test load config with exception.""" + for key in ['NODE_NAME', 'ENERGY_RECORDER_API_URL']: + os.environ[key] = '' + with self.assertRaises(AssertionError): + EnergyRecorder.energy_recorder_api = None + EnergyRecorder.load_config() + self.assertEquals(EnergyRecorder.energy_recorder_api, None) + + @mock.patch("xtesting.energy.energy.requests.get", + return_value=API_KO) + def test_load_config_api_ko(self, loader_mock=None, get_mock=None): + """Test load config with API unavailable.""" + os.environ['NODE_NAME'] = 'MOCK_POD' + self._set_env_creds() + EnergyRecorder.energy_recorder_api = None + EnergyRecorder.load_config() + self.assertEquals(EnergyRecorder.energy_recorder_api["available"], + False) + + @mock.patch('xtesting.energy.energy.requests.get', + return_value=RECORDER_OK) + def test_get_current_scenario(self, loader_mock=None, get_mock=None): + """Test get_current_scenario.""" + os.environ['NODE_NAME'] = 'MOCK_POD' + self.test_load_config() + scenario = EnergyRecorder.get_current_scenario() + self.assertTrue(scenario is not None) + + @mock.patch('xtesting.energy.energy.requests.get', + return_value=RECORDER_NOT_FOUND) + def test_current_scenario_not_found(self, get_mock=None): + """Test get current scenario not existing.""" + os.environ['NODE_NAME'] = 'MOCK_POD' + self.test_load_config() + scenario = EnergyRecorder.get_current_scenario() + self.assertTrue(scenario is None) + + @mock.patch('xtesting.energy.energy.requests.get', + return_value=RECORDER_KO) + def test_current_scenario_api_error(self, get_mock=None): + """Test get current scenario with API error.""" + os.environ['NODE_NAME'] = 'MOCK_POD' + self.test_load_config() + scenario = EnergyRecorder.get_current_scenario() + self.assertTrue(scenario is None) + + @mock.patch('xtesting.energy.energy.EnergyRecorder.load_config', + side_effect=Exception("Internal execution error (MOCK)")) + def test_current_scenario_exception(self, get_mock=None): + """Test get current scenario with exception.""" + scenario = EnergyRecorder.get_current_scenario() + self.assertTrue(scenario is None) + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/utils/__init__.py b/xtesting/tests/unit/utils/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/xtesting/tests/unit/utils/__init__.py diff --git a/xtesting/tests/unit/utils/test_decorators.py b/xtesting/tests/unit/utils/test_decorators.py new file mode 100644 index 00000000..83b182a8 --- /dev/null +++ b/xtesting/tests/unit/utils/test_decorators.py @@ -0,0 +1,125 @@ +#!/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 + +"""Define the class required to fully cover decorators.""" + +from datetime import datetime +import errno +import json +import logging +import os +import unittest + +import mock + +from xtesting.core import testcase +from xtesting.utils import decorators + +__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" + +DIR = '/dev' +FILE = '{}/null'.format(DIR) +URL = 'file://{}'.format(FILE) + + +class DecoratorsTesting(unittest.TestCase): + # pylint: disable=missing-docstring + + _case_name = 'base' + _project_name = 'xtesting' + _start_time = 1.0 + _stop_time = 2.0 + _result = 'PASS' + _version = 'unknown' + _build_tag = 'none' + _node_name = 'bar' + _deploy_scenario = 'foo' + _installer_type = 'debian' + + def setUp(self): + os.environ['INSTALLER_TYPE'] = self._installer_type + os.environ['DEPLOY_SCENARIO'] = self._deploy_scenario + os.environ['NODE_NAME'] = self._node_name + os.environ['BUILD_TAG'] = self._build_tag + + def test_wraps(self): + self.assertEqual(testcase.TestCase.push_to_db.__name__, + "push_to_db") + + def _get_json(self): + stop_time = datetime.fromtimestamp(self._stop_time).strftime( + '%Y-%m-%d %H:%M:%S') + start_time = datetime.fromtimestamp(self._start_time).strftime( + '%Y-%m-%d %H:%M:%S') + data = {'project_name': self._project_name, + 'stop_date': stop_time, 'start_date': start_time, + 'case_name': self._case_name, 'build_tag': self._build_tag, + 'pod_name': self._node_name, 'installer': self._installer_type, + 'scenario': self._deploy_scenario, 'version': self._version, + 'details': {}, 'criteria': self._result} + return json.dumps(data, sort_keys=True) + + def _get_testcase(self): + test = testcase.TestCase( + project_name=self._project_name, case_name=self._case_name) + test.start_time = self._start_time + test.stop_time = self._stop_time + test.result = 100 + test.details = {} + return test + + @mock.patch('requests.post') + def test_http_shema(self, *args): + os.environ['TEST_DB_URL'] = 'http://127.0.0.1' + test = self._get_testcase() + self.assertEqual(test.push_to_db(), testcase.TestCase.EX_OK) + args[0].assert_called_once_with( + 'http://127.0.0.1', data=self._get_json(), + headers={'Content-Type': 'application/json'}) + + def test_wrong_shema(self): + os.environ['TEST_DB_URL'] = '/dev/null' + test = self._get_testcase() + self.assertEqual( + test.push_to_db(), testcase.TestCase.EX_PUSH_TO_DB_ERROR) + + def _test_dump(self): + os.environ['TEST_DB_URL'] = URL + with mock.patch.object(decorators, 'open', mock.mock_open(), + create=True) as mock_open: + test = self._get_testcase() + self.assertEqual(test.push_to_db(), testcase.TestCase.EX_OK) + mock_open.assert_called_once_with(FILE, 'a') + handle = mock_open() + call_args, _ = handle.write.call_args + self.assertIn('POST', call_args[0]) + self.assertIn(self._get_json(), call_args[0]) + + @mock.patch('os.makedirs') + def test_default_dump(self, mock_method=None): + self._test_dump() + mock_method.assert_called_once_with(DIR) + + @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, '')) + def test_makedirs_dir_exists(self, mock_method=None): + self._test_dump() + mock_method.assert_called_once_with(DIR) + + @mock.patch('os.makedirs', side_effect=OSError) + def test_makedirs_exc(self, *args): + os.environ['TEST_DB_URL'] = URL + test = self._get_testcase() + self.assertEqual( + test.push_to_db(), testcase.TestCase.EX_PUSH_TO_DB_ERROR) + args[0].assert_called_once_with(DIR) + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) diff --git a/xtesting/tests/unit/utils/test_env.py b/xtesting/tests/unit/utils/test_env.py new file mode 100644 index 00000000..08601fa5 --- /dev/null +++ b/xtesting/tests/unit/utils/test_env.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python + +# Copyright (c) 2018 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 os +import unittest + +from six.moves import reload_module + +from xtesting.utils import env + + +class EnvTesting(unittest.TestCase): + # pylint: disable=missing-docstring + + def setUp(self): + os.environ['FOO'] = 'foo' + os.environ['BUILD_TAG'] = 'master' + os.environ['CI_LOOP'] = 'weekly' + + def test_get_unset_unknown_env(self): + del os.environ['FOO'] + self.assertEqual(env.get('FOO'), None) + + def test_get_unknown_env(self): + self.assertEqual(env.get('FOO'), 'foo') + reload_module(env) + + def test_get_unset_env(self): + del os.environ['CI_LOOP'] + self.assertEqual( + env.get('CI_LOOP'), env.INPUTS['CI_LOOP']) + + def test_get_env(self): + self.assertEqual( + env.get('CI_LOOP'), 'weekly') + + def test_get_unset_env2(self): + del os.environ['BUILD_TAG'] + self.assertEqual( + env.get('BUILD_TAG'), env.INPUTS['BUILD_TAG']) + + def test_get_env2(self): + self.assertEqual(env.get('BUILD_TAG'), 'master') + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) |