xtesting/tests/unit/ci/test_run_tests.py

   1 #!/usr/bin/env python
   2
   3 # All rights reserved. This program and the accompanying materials
   4 # are made available under the terms of the Apache License, Version 2.0
   5 # which accompanies this distribution, and is available at
   6 # http://www.apache.org/licenses/LICENSE-2.0
   7
   8 # pylint: disable=missing-docstring
   9
  10 import logging
  11 import unittest
  12 import os
  13
  14 import mock
  15
  16 from xtesting.ci import run_tests
  17 from xtesting.core.testcase import TestCase
  18
  19
  20 class FakeModule(TestCase):
  21
  22     def run(self, **kwargs):
  23         return TestCase.EX_OK
  24
  25
  26 class RunTestsTesting(unittest.TestCase):
  27
  28     def setUp(self):
  29         self.runner = run_tests.Runner()
  30         mock_test_case = mock.Mock()
  31         mock_test_case.is_successful.return_value = TestCase.EX_OK
  32         self.runner.executed_test_cases['test1'] = mock_test_case
  33         self.runner.executed_test_cases['test2'] = mock_test_case
  34         self.sep = 'test_sep'
  35         self.creds = {'OS_AUTH_URL': 'http://test_ip:test_port/v2.0',
  36                       'OS_USERNAME': 'test_os_username',
  37                       'OS_TENANT_NAME': 'test_tenant',
  38                       'OS_PASSWORD': 'test_password'}
  39         self.test = {'test_name': 'test_name'}
  40         self.tier = mock.Mock()
  41         test1 = mock.Mock()
  42         test1.get_name.return_value = 'test1'
  43         test2 = mock.Mock()
  44         test2.get_name.return_value = 'test2'
  45         attrs = {'get_name.return_value': 'test_tier',
  46                  'get_tests.return_value': [test1, test2],
  47                  'get_ci_loop.return_value': 'test_ci_loop',
  48                  'get_test_names.return_value': ['test1', 'test2']}
  49         self.tier.configure_mock(**attrs)
  50
  51         self.tiers = mock.Mock()
  52         attrs = {'get_tiers.return_value': [self.tier]}
  53         self.tiers.configure_mock(**attrs)
  54
  55         self.run_tests_parser = run_tests.RunTestsParser()
  56
  57     @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test')
  58     def test_get_run_dict(self, *args):
  59         retval = {'run': mock.Mock()}
  60         args[0].return_value = retval
  61         self.assertEqual(self.runner.get_run_dict('test_name'), retval['run'])
  62         args[0].assert_called_once_with('test_name')
  63
  64     @mock.patch('xtesting.ci.run_tests.LOGGER.error')
  65     @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test',
  66                 return_value=None)
  67     def test_get_run_dict_config_ko(self, *args):
  68         testname = 'test_name'
  69         self.assertEqual(self.runner.get_run_dict(testname), None)
  70         args[0].return_value = {}
  71         self.assertEqual(self.runner.get_run_dict(testname), None)
  72         calls = [mock.call(testname), mock.call(testname)]
  73         args[0].assert_has_calls(calls)
  74         calls = [mock.call("Cannot get %s's config options", testname),
  75                  mock.call("Cannot get %s's config options", testname)]
  76         args[1].assert_has_calls(calls)
  77
  78     @mock.patch('xtesting.ci.run_tests.LOGGER.exception')
  79     @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test',
  80                 side_effect=Exception)
  81     def test_get_run_dict_exception(self, *args):
  82         testname = 'test_name'
  83         self.assertEqual(self.runner.get_run_dict(testname), None)
  84         args[1].assert_called_once_with(
  85             "Cannot get %s's config options", testname)
  86
  87     def _test_source_envfile(self, msg, key='OS_TENANT_NAME', value='admin'):
  88         try:
  89             del os.environ[key]
  90         except Exception:  # pylint: disable=broad-except
  91             pass
  92         envfile = 'rc_file'
  93         with mock.patch('six.moves.builtins.open',
  94                         mock.mock_open(read_data=msg)) as mock_method,\
  95                 mock.patch('os.path.isfile', return_value=True):
  96             mock_method.return_value.__iter__ = lambda self: iter(
  97                 self.readline, '')
  98             self.runner.source_envfile(envfile)
  99             mock_method.assert_called_once_with(envfile, 'r')
 100             self.assertEqual(os.environ[key], value)
 101
 102     def test_source_envfile(self):
 103         self._test_source_envfile('OS_TENANT_NAME=admin')
 104         self._test_source_envfile('OS_TENANT_NAME= admin')
 105         self._test_source_envfile('OS_TENANT_NAME = admin')
 106         self._test_source_envfile('OS_TENANT_NAME = "admin"')
 107         self._test_source_envfile('export OS_TENANT_NAME=admin')
 108         self._test_source_envfile('export OS_TENANT_NAME =admin')
 109         self._test_source_envfile('export OS_TENANT_NAME = admin')
 110         self._test_source_envfile('export OS_TENANT_NAME = "admin"')
 111         # This test will fail as soon as rc_file is fixed
 112         self._test_source_envfile(
 113             'export "\'OS_TENANT_NAME\'" = "\'admin\'"')
 114
 115     def test_get_dict_by_test(self):
 116         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 117                 mock.patch('yaml.safe_load') as mock_yaml:
 118             mock_obj = mock.Mock()
 119             testcase_dict = {'case_name': 'testname',
 120                              'criteria': 50}
 121             attrs = {'get.return_value': [{'testcases': [testcase_dict]}]}
 122             mock_obj.configure_mock(**attrs)
 123             mock_yaml.return_value = mock_obj
 124             self.assertDictEqual(
 125                 run_tests.Runner.get_dict_by_test('testname'),
 126                 testcase_dict)
 127
 128     @mock.patch('xtesting.ci.run_tests.Runner.get_run_dict',
 129                 return_value=None)
 130     def test_run_tests_import_exception(self, *args):
 131         mock_test = mock.Mock()
 132         kwargs = {'get_name.return_value': 'test_name',
 133                   'is_skipped.return_value': False,
 134                   'is_enabled.return_value': True,
 135                   'needs_clean.return_value': False}
 136         mock_test.configure_mock(**kwargs)
 137         with self.assertRaises(Exception) as context:
 138             self.runner.run_test(mock_test)
 139         args[0].assert_called_with('test_name')
 140         msg = "Cannot import the class for the test case."
 141         self.assertTrue(msg in str(context.exception))
 142
 143     @mock.patch('importlib.import_module', name="module",
 144                 return_value=mock.Mock(test_class=mock.Mock(
 145                     side_effect=FakeModule)))
 146     @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test')
 147     def test_run_tests_default(self, *args):
 148         mock_test = mock.Mock()
 149         kwargs = {'get_name.return_value': 'test_name',
 150                   'is_skipped.return_value': False,
 151                   'is_enabled.return_value': True,
 152                   'needs_clean.return_value': True}
 153         mock_test.configure_mock(**kwargs)
 154         test_run_dict = {'module': 'test_module',
 155                          'class': 'test_class'}
 156         with mock.patch('xtesting.ci.run_tests.Runner.get_run_dict',
 157                         return_value=test_run_dict):
 158             self.runner.clean_flag = True
 159             self.runner.run_test(mock_test)
 160         args[0].assert_called_with('test_name')
 161         args[1].assert_called_with('test_module')
 162         self.assertEqual(self.runner.overall_result,
 163                          run_tests.Result.EX_OK)
 164
 165     @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test')
 166     def test_run_tests_disabled(self, *args):
 167         mock_test = mock.Mock()
 168         kwargs = {'get_name.return_value': 'test_name',
 169                   'is_skipped.return_value': False,
 170                   'is_enabled.return_value': False,
 171                   'needs_clean.return_value': True}
 172         mock_test.configure_mock(**kwargs)
 173         test_run_dict = {'module': 'test_module',
 174                          'class': 'test_class'}
 175         with mock.patch('xtesting.ci.run_tests.Runner.get_run_dict',
 176                         return_value=test_run_dict):
 177             self.runner.clean_flag = True
 178             self.runner.run_test(mock_test)
 179         args[0].assert_not_called()
 180         self.assertEqual(self.runner.overall_result,
 181                          run_tests.Result.EX_OK)
 182
 183     @mock.patch('xtesting.ci.run_tests.Runner.get_dict_by_test')
 184     def test_run_tests_skipped(self, *args):
 185         mock_test = mock.Mock()
 186         kwargs = {'get_name.return_value': 'test_name',
 187                   'is_skipped.return_value': True,
 188                   'is_enabled.return_value': True,
 189                   'needs_clean.return_value': True}
 190         mock_test.configure_mock(**kwargs)
 191         test_run_dict = {'module': 'test_module',
 192                          'class': 'test_class'}
 193         with mock.patch('xtesting.ci.run_tests.Runner.get_run_dict',
 194                         return_value=test_run_dict):
 195             self.runner.clean_flag = True
 196             self.runner.run_test(mock_test)
 197         args[0].assert_not_called()
 198         self.assertEqual(self.runner.overall_result,
 199                          run_tests.Result.EX_OK)
 200
 201     @mock.patch('xtesting.ci.run_tests.Runner.run_test',
 202                 return_value=TestCase.EX_OK)
 203     def test_run_tier_default(self, *mock_methods):
 204         self.assertEqual(self.runner.run_tier(self.tier),
 205                          run_tests.Result.EX_OK)
 206         mock_methods[0].assert_called_with(mock.ANY)
 207
 208     @mock.patch('xtesting.ci.run_tests.LOGGER.info')
 209     def test_run_tier_missing_test(self, mock_logger_info):
 210         self.tier.get_tests.return_value = None
 211         self.assertEqual(self.runner.run_tier(self.tier),
 212                          run_tests.Result.EX_ERROR)
 213         self.assertTrue(mock_logger_info.called)
 214
 215     @mock.patch('xtesting.ci.run_tests.LOGGER.info')
 216     @mock.patch('xtesting.ci.run_tests.Runner.run_tier')
 217     @mock.patch('xtesting.ci.run_tests.Runner.summary')
 218     def test_run_all_default(self, *mock_methods):
 219         os.environ['CI_LOOP'] = 'test_ci_loop'
 220         self.runner.run_all()
 221         mock_methods[1].assert_not_called()
 222         self.assertTrue(mock_methods[2].called)
 223
 224     @mock.patch('xtesting.ci.run_tests.LOGGER.info')
 225     @mock.patch('xtesting.ci.run_tests.Runner.summary')
 226     def test_run_all_missing_tier(self, *mock_methods):
 227         os.environ['CI_LOOP'] = 'loop_re_not_available'
 228         self.runner.run_all()
 229         self.assertTrue(mock_methods[1].called)
 230
 231     @mock.patch('xtesting.ci.run_tests.Runner.source_envfile',
 232                 side_effect=Exception)
 233     @mock.patch('xtesting.ci.run_tests.Runner.summary')
 234     def test_main_failed(self, *mock_methods):
 235         kwargs = {'test': 'test_name', 'noclean': True, 'report': True}
 236         args = {'get_tier.return_value': False,
 237                 'get_test.return_value': False}
 238         self.runner.tiers = mock.Mock()
 239         self.runner.tiers.configure_mock(**args)
 240         self.assertEqual(self.runner.main(**kwargs),
 241                          run_tests.Result.EX_ERROR)
 242         mock_methods[1].assert_called_once_with()
 243
 244     @mock.patch('xtesting.ci.run_tests.Runner.source_envfile')
 245     @mock.patch('xtesting.ci.run_tests.Runner.run_test',
 246                 return_value=TestCase.EX_OK)
 247     @mock.patch('xtesting.ci.run_tests.Runner.summary')
 248     def test_main_tier(self, *mock_methods):
 249         mock_tier = mock.Mock()
 250         test_mock = mock.Mock()
 251         test_mock.get_name.return_value = 'test1'
 252         args = {'get_name.return_value': 'tier_name',
 253                 'get_tests.return_value': [test_mock]}
 254         mock_tier.configure_mock(**args)
 255         kwargs = {'test': 'tier_name', 'noclean': True, 'report': True}
 256         args = {'get_tier.return_value': mock_tier,
 257                 'get_test.return_value': None}
 258         self.runner.tiers = mock.Mock()
 259         self.runner.tiers.configure_mock(**args)
 260         self.assertEqual(self.runner.main(**kwargs),
 261                          run_tests.Result.EX_OK)
 262         mock_methods[1].assert_called()
 263
 264     @mock.patch('xtesting.ci.run_tests.Runner.source_envfile')
 265     @mock.patch('xtesting.ci.run_tests.Runner.run_test',
 266                 return_value=TestCase.EX_OK)
 267     def test_main_test(self, *mock_methods):
 268         kwargs = {'test': 'test_name', 'noclean': True, 'report': True}
 269         args = {'get_tier.return_value': None,
 270                 'get_test.return_value': 'test_name'}
 271         self.runner.tiers = mock.Mock()
 272         mock_methods[1].return_value = self.creds
 273         self.runner.tiers.configure_mock(**args)
 274         self.assertEqual(self.runner.main(**kwargs),
 275                          run_tests.Result.EX_OK)
 276         mock_methods[0].assert_called_once_with('test_name')
 277
 278     @mock.patch('xtesting.ci.run_tests.Runner.source_envfile')
 279     @mock.patch('xtesting.ci.run_tests.Runner.run_all')
 280     @mock.patch('xtesting.ci.run_tests.Runner.summary')
 281     def test_main_all_tier(self, *args):
 282         kwargs = {'get_tier.return_value': None,
 283                   'get_test.return_value': None}
 284         self.runner.tiers = mock.Mock()
 285         self.runner.tiers.configure_mock(**kwargs)
 286         self.assertEqual(
 287             self.runner.main(test='all', noclean=True, report=True),
 288             run_tests.Result.EX_OK)
 289         args[0].assert_called_once_with(None)
 290         args[1].assert_called_once_with()
 291         args[2].assert_called_once_with()
 292
 293     @mock.patch('xtesting.ci.run_tests.Runner.source_envfile')
 294     def test_main_any_tier_test_ko(self, *args):
 295         kwargs = {'get_tier.return_value': None,
 296                   'get_test.return_value': None}
 297         self.runner.tiers = mock.Mock()
 298         self.runner.tiers.configure_mock(**kwargs)
 299         self.assertEqual(
 300             self.runner.main(test='any', noclean=True, report=True),
 301             run_tests.Result.EX_ERROR)
 302         args[0].assert_called_once_with()
 303
 304
 305 if __name__ == "__main__":
 306     logging.disable(logging.CRITICAL)
 307     unittest.main(verbosity=2)