diff options
Diffstat (limited to 'xtesting/tests/unit/ci')
| -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 |
4 files changed, 499 insertions, 0 deletions
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) |