tempest/tests/cmd/test_run.py

# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import argparse
import atexit
import os
import shutil
import subprocess
import tempfile
from unittest import mock

import fixtures
import six

from tempest.cmd import run
from tempest.cmd import workspace
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.tests import base

DEVNULL = open(os.devnull, 'wb')
atexit.register(DEVNULL.close)

CONF = config.CONF


class TestTempestRun(base.TestCase):

    def setUp(self):
        super(TestTempestRun, self).setUp()
        self.run_cmd = run.TempestRun(None, None)

    def test__build_regex_default(self):
        args = mock.Mock(spec=argparse.Namespace)
        setattr(args, 'smoke', False)
        setattr(args, 'regex', '')
        self.assertIsNone(self.run_cmd._build_regex(args))

    def test__build_regex_smoke(self):
        args = mock.Mock(spec=argparse.Namespace)
        setattr(args, "smoke", True)
        setattr(args, 'regex', '')
        self.assertEqual(['smoke'], self.run_cmd._build_regex(args))

    def test__build_regex_regex(self):
        args = mock.Mock(spec=argparse.Namespace)
        setattr(args, 'smoke', False)
        setattr(args, "regex", 'i_am_a_fun_little_regex')
        self.assertEqual(['i_am_a_fun_little_regex'],
                         self.run_cmd._build_regex(args))

    def test__build_regex_smoke_regex(self):
        args = mock.Mock(spec=argparse.Namespace)
        setattr(args, "smoke", True)
        setattr(args, 'regex', 'i_am_a_fun_little_regex')
        self.assertEqual(['smoke'], self.run_cmd._build_regex(args))


class TestRunReturnCode(base.TestCase):
    def setUp(self):
        super(TestRunReturnCode, self).setUp()
        # Setup test dirs
        self.directory = tempfile.mkdtemp(prefix='tempest-unit')
        self.addCleanup(shutil.rmtree, self.directory)
        self.test_dir = os.path.join(self.directory, 'tests')
        os.mkdir(self.test_dir)
        # Setup Test files
        self.stestr_conf_file = os.path.join(self.directory, '.stestr.conf')
        self.setup_cfg_file = os.path.join(self.directory, 'setup.cfg')
        self.passing_file = os.path.join(self.test_dir, 'test_passing.py')
        self.failing_file = os.path.join(self.test_dir, 'test_failing.py')
        self.init_file = os.path.join(self.test_dir, '__init__.py')
        self.setup_py = os.path.join(self.directory, 'setup.py')
        shutil.copy('tempest/tests/files/testr-conf', self.stestr_conf_file)
        shutil.copy('tempest/tests/files/passing-tests', self.passing_file)
        shutil.copy('tempest/tests/files/failing-tests', self.failing_file)
        shutil.copy('setup.py', self.setup_py)
        shutil.copy('tempest/tests/files/setup.cfg', self.setup_cfg_file)
        shutil.copy('tempest/tests/files/__init__.py', self.init_file)
        # Change directory, run wrapper and check result
        self.addCleanup(os.chdir, os.path.abspath(os.curdir))
        os.chdir(self.directory)

    def assertRunExit(self, cmd, expected):
        p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)
        out, err = p.communicate()
        msg = ("Running %s got an unexpected returncode\n"
               "Stdout: %s\nStderr: %s" % (' '.join(cmd), out, err))
        self.assertEqual(p.returncode, expected, msg)
        return out, err

    def test_tempest_run_passes(self):
        self.assertRunExit(['tempest', 'run', '--regex', 'passing'], 0)

    def test_tempest_run_passes_with_stestr_repository(self):
        subprocess.call(['stestr', 'init'])
        self.assertRunExit(['tempest', 'run', '--regex', 'passing'], 0)

    def test_tempest_run_failing(self):
        self.assertRunExit(['tempest', 'run', '--regex', 'failing'], 1)

    def test_tempest_run_failing_with_stestr_repository(self):
        subprocess.call(['stestr', 'init'])
        self.assertRunExit(['tempest', 'run', '--regex', 'failing'], 1)

    def test_tempest_run_blackregex_failing(self):
        self.assertRunExit(['tempest', 'run', '--black-regex', 'failing'], 0)

    def test_tempest_run_blackregex_failing_with_stestr_repository(self):
        subprocess.call(['stestr', 'init'])
        self.assertRunExit(['tempest', 'run', '--black-regex', 'failing'], 0)

    def test_tempest_run_blackregex_passing(self):
        self.assertRunExit(['tempest', 'run', '--black-regex', 'passing'], 1)

    def test_tempest_run_blackregex_passing_with_stestr_repository(self):
        subprocess.call(['stestr', 'init'])
        self.assertRunExit(['tempest', 'run', '--black-regex', 'passing'], 1)

    def test_tempest_run_fails(self):
        self.assertRunExit(['tempest', 'run'], 1)

    def test_run_list(self):
        subprocess.call(['stestr', 'init'])
        out, err = self.assertRunExit(['tempest', 'run', '-l'], 0)
        tests = out.split()
        tests = sorted([six.text_type(x.rstrip()) for x in tests if x])
        result = [
            six.text_type('tests.test_failing.FakeTestClass.test_pass'),
            six.text_type('tests.test_failing.FakeTestClass.test_pass_list'),
            six.text_type('tests.test_passing.FakeTestClass.test_pass'),
            six.text_type('tests.test_passing.FakeTestClass.test_pass_list'),
        ]
        # NOTE(mtreinish): on python 3 the subprocess prints b'' around
        # stdout.
        result = ["b\'" + x + "\'" for x in result]
        self.assertEqual(result, tests)

    def test_tempest_run_with_worker_file(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        worker_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(worker_file.close)
        worker_file.write(
            '- worker:\n  - passing\n  concurrency: 3'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--worker-file=%s' % path], 0)

    def test_tempest_run_with_whitelist(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        whitelist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(whitelist_file.close)
        whitelist_file.write('passing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--whitelist-file=%s' % path], 0)

    def test_tempest_run_with_whitelist_regex_include_pass_check_fail(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        whitelist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(whitelist_file.close)
        whitelist_file.write('passing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--whitelist-file=%s' % path,
                            '--regex', 'fail'], 1)

    def test_tempest_run_with_whitelist_regex_include_pass_check_pass(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        whitelist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(whitelist_file.close)
        whitelist_file.write('passing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--whitelist-file=%s' % path,
                            '--regex', 'passing'], 0)

    def test_tempest_run_with_whitelist_regex_include_fail_check_pass(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        whitelist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(whitelist_file.close)
        whitelist_file.write('failing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--whitelist-file=%s' % path,
                            '--regex', 'pass'], 1)

    def test_tempest_run_passes_with_config_file(self):
        self.assertRunExit(['tempest', 'run',
                            '--config-file', self.stestr_conf_file,
                            '--regex', 'passing'], 0)

    def test_tempest_run_with_blacklist_failing(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        blacklist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(blacklist_file.close)
        blacklist_file.write('failing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--blacklist-file=%s' % path], 0)

    def test_tempest_run_with_blacklist_passing(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        blacklist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(blacklist_file.close)
        blacklist_file.write('passing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--blacklist-file=%s' % path], 1)

    def test_tempest_run_with_blacklist_regex_exclude_fail_check_pass(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        blacklist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(blacklist_file.close)
        blacklist_file.write('failing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--blacklist-file=%s' % path,
                            '--regex', 'pass'], 0)

    def test_tempest_run_with_blacklist_regex_exclude_pass_check_pass(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        blacklist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(blacklist_file.close)
        blacklist_file.write('passing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--blacklist-file=%s' % path,
                            '--regex', 'pass'], 1)

    def test_tempest_run_with_blacklist_regex_exclude_pass_check_fail(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        blacklist_file = os.fdopen(fd, 'wb', 0)
        self.addCleanup(blacklist_file.close)
        blacklist_file.write('passing'.encode('utf-8'))
        self.assertRunExit(['tempest', 'run', '--blacklist-file=%s' % path,
                            '--regex', 'fail'], 1)


class TestConfigPathCheck(base.TestCase):
    def setUp(self):
        super(TestConfigPathCheck, self).setUp()
        self.run_cmd = run.TempestRun(None, None)

    def test_tempest_run_set_config_path(self):
        # Note: (mbindlish) This test is created for the bug id: 1783751
        # Checking TEMPEST_CONFIG_DIR and TEMPEST_CONFIG is actually
        # getting set in os environment when some data has passed to
        # set the environment.

        _, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)

        self.run_cmd._set_env(path)
        self.assertEqual(path, CONF._path)
        self.assertIn('TEMPEST_CONFIG_DIR', os.environ)
        self.assertEqual(path, os.path.join(os.environ['TEMPEST_CONFIG_DIR'],
                                            os.environ['TEMPEST_CONFIG']))

    def test_tempest_run_set_config_no_exist_path(self):
        path = "fake/path"
        self.assertRaisesRegex(FileNotFoundError,
                               'Config file: .* doesn\'t exist',
                               self.run_cmd._set_env, path)

    def test_tempest_run_no_config_path(self):
        # Note: (mbindlish) This test is created for the bug id: 1783751
        # Checking TEMPEST_CONFIG_DIR and TEMPEST_CONFIG should have no value
        # in os environment when no data has passed to set the environment.

        self.run_cmd._set_env("")
        self.assertFalse(CONF._path)
        self.assertNotIn('TEMPEST_CONFIG_DIR', os.environ)
        self.assertNotIn('TEMPEST_CONFIG', os.environ)


class TestTakeAction(base.TestCase):
    def setUp(self):
        super(TestTakeAction, self).setUp()
        self.name = data_utils.rand_name('workspace')
        self.path = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.path, ignore_errors=True)
        store_dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, store_dir, ignore_errors=True)
        self.store_file = os.path.join(store_dir, 'workspace.yaml')
        self.workspace_manager = workspace.WorkspaceManager(
            path=self.store_file)
        self.workspace_manager.register_new_workspace(self.name, self.path)

    def _setup_test_dirs(self):
        self.directory = tempfile.mkdtemp(prefix='tempest-unit')
        self.addCleanup(shutil.rmtree, self.directory, ignore_errors=True)
        self.test_dir = os.path.join(self.directory, 'tests')
        os.mkdir(self.test_dir)
        # Change directory, run wrapper and check result
        self.addCleanup(os.chdir, os.path.abspath(os.curdir))
        os.chdir(self.directory)

    def test_workspace_not_registered(self):
        class Exception_(Exception):
            pass

        m_exit = self.useFixture(fixtures.MockPatch('sys.exit')).mock
        # sys.exit must not continue (or exit)
        m_exit.side_effect = Exception_

        workspace = self.getUniqueString()

        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()
        parsed_args.config_file = []

        # Override $HOME so that empty workspace gets created in temp dir.
        self.useFixture(fixtures.TempHomeDir())

        # Force use of the temporary home directory.
        parsed_args.workspace_path = None

        # Simulate --workspace argument.
        parsed_args.workspace = workspace

        self.assertRaises(Exception_, tempest_run.take_action, parsed_args)
        exit_msg = m_exit.call_args[0][0]
        self.assertIn(workspace, exit_msg)

    def test_config_file_specified(self):
        self._setup_test_dirs()
        _, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()

        parsed_args.workspace = None
        parsed_args.state = None
        parsed_args.list_tests = False
        parsed_args.config_file = path

        with mock.patch('stestr.commands.run_command') as m:
            m.return_value = 0
            self.assertEqual(0, tempest_run.take_action(parsed_args))
            m.assert_called()

    def test_no_config_file_no_workspace_no_state(self):
        self._setup_test_dirs()
        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()

        parsed_args.workspace = None
        parsed_args.state = None
        parsed_args.list_tests = False
        parsed_args.config_file = ''

        with mock.patch('stestr.commands.run_command'):
            self.assertRaises(SystemExit, tempest_run.take_action, parsed_args)

    def test_config_file_workspace_registered(self):
        self._setup_test_dirs()
        _, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()
        parsed_args.workspace = self.name
        parsed_args.workspace_path = self.store_file
        parsed_args.state = None
        parsed_args.list_tests = False
        parsed_args.config_file = path

        with mock.patch('stestr.commands.run_command') as m:
            m.return_value = 0
            self.assertEqual(0, tempest_run.take_action(parsed_args))
            m.assert_called()

    @mock.patch('tempest.cmd.run.TempestRun._init_state')
    def test_workspace_registered_no_config_no_state(self, mock_init_state):
        self._setup_test_dirs()
        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()
        parsed_args.workspace = self.name
        parsed_args.workspace_path = self.store_file
        parsed_args.state = None
        parsed_args.list_tests = False
        parsed_args.config_file = ''

        with mock.patch('stestr.commands.run_command') as m:
            m.return_value = 0
            self.assertEqual(0, tempest_run.take_action(parsed_args))
            m.assert_called()
        mock_init_state.assert_not_called()

    @mock.patch('tempest.cmd.run.TempestRun._init_state')
    def test_no_config_file_no_workspace_state_true(self, mock_init_state):
        self._setup_test_dirs()
        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()

        parsed_args.workspace = None
        parsed_args.state = True
        parsed_args.list_tests = False
        parsed_args.config_file = ''

        with mock.patch('stestr.commands.run_command'):
            self.assertRaises(SystemExit, tempest_run.take_action, parsed_args)
        mock_init_state.assert_not_called()

    @mock.patch('tempest.cmd.run.TempestRun._init_state')
    def test_workspace_registered_no_config_state_true(self, mock_init_state):
        self._setup_test_dirs()
        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()
        parsed_args.workspace = self.name
        parsed_args.workspace_path = self.store_file
        parsed_args.state = True
        parsed_args.list_tests = False
        parsed_args.config_file = ''

        with mock.patch('stestr.commands.run_command') as m:
            m.return_value = 0
            self.assertEqual(0, tempest_run.take_action(parsed_args))
            m.assert_called()
        mock_init_state.assert_called()

    @mock.patch('tempest.cmd.run.TempestRun._init_state')
    def test_no_workspace_config_file_state_true(self, mock_init_state):
        self._setup_test_dirs()
        _, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        tempest_run = run.TempestRun(app=mock.Mock(), app_args=mock.Mock())
        parsed_args = mock.Mock()
        parsed_args.workspace = None
        parsed_args.workspace_path = self.store_file
        parsed_args.state = True
        parsed_args.list_tests = False
        parsed_args.config_file = path

        with mock.patch('stestr.commands.run_command') as m:
            m.return_value = 0
            self.assertEqual(0, tempest_run.take_action(parsed_args))
            m.assert_called()
        mock_init_state.assert_called()