jenkins-job-builder/tests/cmd/subcommands/test_test.py

# Joint copyright:
#  - 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.

# The goal of these tests is to check that given a particular set of flags to
# Jenkins Job Builder's command line tools it will result in a particular set
# of actions by the JJB library, usually through interaction with the
# python-jenkins library.

import difflib
import filecmp
import io
import os
import shutil
import tempfile
import yaml

import jenkins
from six.moves import StringIO
import testtools

from jenkins_jobs.cli import entry
from tests.base import mock
from tests.cmd.test_cmd import CmdTestsBase


@mock.patch('jenkins_jobs.builder.Jenkins.get_plugins_info', mock.MagicMock)
class TestTests(CmdTestsBase):

    def test_non_existing_job(self):
        """
        Run test mode and pass a non-existing job name
        (probably better to fail here)
        """
        args = ['--conf', self.default_config_file, 'test',
                os.path.join(self.fixtures_path,
                             'cmd-001.yaml'),
                'invalid']
        self.execute_jenkins_jobs_with_args(args)

    def test_valid_job(self):
        """
        Run test mode and pass a valid job name
        """
        args = ['--conf', self.default_config_file, 'test',
                os.path.join(self.fixtures_path,
                             'cmd-001.yaml'),
                'foo-job']
        self.execute_jenkins_jobs_with_args(args)

    def test_console_output(self):
        """
        Run test mode and verify that resulting XML gets sent to the console.
        """

        console_out = io.BytesIO()
        with mock.patch('sys.stdout', console_out):
            args = ['--conf', self.default_config_file, 'test',
                    os.path.join(self.fixtures_path, 'cmd-001.yaml')]
            self.execute_jenkins_jobs_with_args(args)
        xml_content = io.open(os.path.join(self.fixtures_path, 'cmd-001.xml'),
                              'r', encoding='utf-8').read()
        self.assertEqual(console_out.getvalue().decode('utf-8'), xml_content)

    def test_stream_input_output_utf8_encoding(self):
        """
        Run test mode simulating using pipes for input and output using
        utf-8 encoding
        """
        console_out = io.BytesIO()

        input_file = os.path.join(self.fixtures_path, 'cmd-001.yaml')
        with io.open(input_file, 'r') as f:
            with mock.patch('sys.stdout', console_out):
                with mock.patch('sys.stdin', f):
                    args = ['--conf', self.default_config_file, 'test']
                    self.execute_jenkins_jobs_with_args(args)

        xml_content = io.open(os.path.join(self.fixtures_path, 'cmd-001.xml'),
                              'r', encoding='utf-8').read()
        value = console_out.getvalue().decode('utf-8')
        self.assertEqual(value, xml_content)

    def test_stream_input_output_ascii_encoding(self):
        """
        Run test mode simulating using pipes for input and output using
        ascii encoding with unicode input
        """
        console_out = io.BytesIO()
        console_out.encoding = 'ascii'

        input_file = os.path.join(self.fixtures_path, 'cmd-001.yaml')
        with io.open(input_file, 'r') as f:
            with mock.patch('sys.stdout', console_out):
                with mock.patch('sys.stdin', f):
                    args = ['--conf', self.default_config_file, 'test']
                    self.execute_jenkins_jobs_with_args(args)

        xml_content = io.open(os.path.join(self.fixtures_path, 'cmd-001.xml'),
                              'r', encoding='utf-8').read()
        value = console_out.getvalue().decode('ascii')
        self.assertEqual(value, xml_content)

    def test_stream_output_ascii_encoding_invalid_char(self):
        """
        Run test mode simulating using pipes for input and output using
        ascii encoding for output with include containing a character
        that cannot be converted.
        """
        console_out = io.BytesIO()
        console_out.encoding = 'ascii'

        input_file = os.path.join(self.fixtures_path, 'unicode001.yaml')
        with io.open(input_file, 'r', encoding='utf-8') as f:
            with mock.patch('sys.stdout', console_out):
                with mock.patch('sys.stdin', f):
                    args = ['--conf', self.default_config_file, 'test']
                    jenkins_jobs = entry.JenkinsJobs(args)
                    e = self.assertRaises(UnicodeError, jenkins_jobs.execute)
        self.assertIn("'ascii' codec can't encode character", str(e))

    @mock.patch('jenkins_jobs.cli.subcommand.update.YamlParser.generateXML')
    @mock.patch('jenkins_jobs.parser.ModuleRegistry')
    def test_plugins_info_stub_option(self, registry_mock, generateXML_mock):
        """
        Test handling of plugins_info stub option.
        """
        plugins_info_stub_yaml_file = os.path.join(self.fixtures_path,
                                                   'plugins-info.yaml')
        args = ['--conf',
                os.path.join(self.fixtures_path, 'cmd-001.conf'),
                'test',
                '-p',
                plugins_info_stub_yaml_file,
                os.path.join(self.fixtures_path, 'cmd-001.yaml')]

        self.execute_jenkins_jobs_with_args(args)

        with io.open(plugins_info_stub_yaml_file,
                     'r', encoding='utf-8') as yaml_file:
            plugins_info_list = yaml.load(yaml_file)

        registry_mock.assert_called_with(mock.ANY,
                                         plugins_info_list)

    @mock.patch('jenkins_jobs.cli.subcommand.update.YamlParser.generateXML')
    @mock.patch('jenkins_jobs.parser.ModuleRegistry')
    def test_bogus_plugins_info_stub_option(self, registry_mock,
                                            generateXML_mock):
        """
        Verify that a JenkinsJobException is raised if the plugins_info stub
        file does not yield a list as its top-level object.
        """
        plugins_info_stub_yaml_file = os.path.join(self.fixtures_path,
                                                   'bogus-plugins-info.yaml')
        args = ['--conf',
                os.path.join(self.fixtures_path, 'cmd-001.conf'),
                'test',
                '-p',
                plugins_info_stub_yaml_file,
                os.path.join(self.fixtures_path, 'cmd-001.yaml')]

        stderr = StringIO()
        with mock.patch('sys.stderr', stderr):
            self.assertRaises(SystemExit, entry.JenkinsJobs, args)
        self.assertIn("must contain a Yaml list",
                      stderr.getvalue())


class TestJenkinsGetPluginInfoError(CmdTestsBase):
    """ This test class is used for testing the 'test' subcommand when we want
    to validate its behavior without mocking
    jenkins_jobs.builder.Jenkins.get_plugins_info
    """

    @mock.patch('jenkins.Jenkins.get_plugins_info')
    def test_console_output_jenkins_connection_failure_warning(
            self, get_plugins_info_mock):
        """
        Run test mode and verify that failed Jenkins connection attempt
        exception does not bubble out of cmd.main. Ideally, we would also test
        that an appropriate message is logged to stderr but it's somewhat
        difficult to figure out how to actually enable stderr in this test
        suite.
        """

        get_plugins_info_mock.side_effect = \
            jenkins.JenkinsException("Connection refused")
        with mock.patch('sys.stdout'):
            try:
                args = ['--conf', self.default_config_file, 'test',
                        os.path.join(self.fixtures_path, 'cmd-001.yaml')]
                self.execute_jenkins_jobs_with_args(args)
            except jenkins.JenkinsException:
                self.fail("jenkins.JenkinsException propagated to main")
            except:
                pass  # only care about jenkins.JenkinsException for now

    @mock.patch('jenkins.Jenkins.get_plugins_info')
    def test_skip_plugin_retrieval_if_no_config_provided(
            self, get_plugins_info_mock):
        """
        Verify that retrieval of information from Jenkins instance about its
        plugins will be skipped when run if no config file provided.
        """
        with mock.patch('sys.stdout', new_callable=io.BytesIO):
            args = ['--conf', self.default_config_file, 'test',
                    os.path.join(self.fixtures_path, 'cmd-001.yaml')]
            entry.JenkinsJobs(args)
        self.assertFalse(get_plugins_info_mock.called)

    @mock.patch('jenkins.Jenkins.get_plugins_info')
    def test_skip_plugin_retrieval_if_disabled(self, get_plugins_info_mock):
        """
        Verify that retrieval of information from Jenkins instance about its
        plugins will be skipped when run if a config file provided and disables
        querying through a config option.
        """
        with mock.patch('sys.stdout', new_callable=io.BytesIO):
            args = ['--conf',
                    os.path.join(self.fixtures_path,
                                 'disable-query-plugins.conf'),
                    'test',
                    os.path.join(self.fixtures_path, 'cmd-001.yaml')]
            entry.JenkinsJobs(args)
        self.assertFalse(get_plugins_info_mock.called)


class MatchesDirMissingFilesMismatch(object):
    def __init__(self, left_directory, right_directory):
        self.left_directory = left_directory
        self.right_directory = right_directory

    def describe(self):
        return "{0} and {1} contain different files".format(
            self.left_directory,
            self.right_directory)

    def get_details(self):
        return {}


class MatchesDirFileContentsMismatch(object):
    def __init__(self, left_file, right_file):
        self.left_file = left_file
        self.right_file = right_file

    def describe(self):
        left_contents = open(self.left_file).readlines()
        right_contents = open(self.right_file).readlines()

        return "{0} is not equal to {1}:\n{2}".format(
            difflib.unified_diff(left_contents, right_contents,
                                 fromfile=self.left_file,
                                 tofile=self.right_file),
            self.left_file,
            self.right_file)

    def get_details(self):
        return {}


class MatchesDir(object):
    def __init__(self, directory):
        self.__directory = directory
        self.__files = self.__get_files(directory)

    def __get_files(self, directory):
        for root, _, files in os.walk(directory):
            return files

    def __str__(self,):
        return "MatchesDir({0})".format(self.__dirname)

    def match(self, other_directory):
        other_files = self.__get_files(other_directory)

        self.__files.sort()
        other_files.sort()

        if self.__files != other_files:
            return MatchesDirMissingFilesMismatch(self.__directory,
                                                  other_directory)

        for i, file in enumerate(self.__files):
            my_file = os.path.join(self.__directory, file)
            other_file = os.path.join(other_directory, other_files[i])
            if not filecmp.cmp(my_file, other_file):
                return MatchesDirFileContentsMismatch(my_file, other_file)

        return None


@mock.patch('jenkins_jobs.builder.Jenkins.get_plugins_info', mock.MagicMock)
class TestTestsMultiPath(CmdTestsBase):

    def setUp(self):
        super(TestTestsMultiPath, self).setUp()

        path_list = [os.path.join(self.fixtures_path,
                                  'multi-path/yamldirs/', p)
                     for p in ['dir1', 'dir2']]
        self.multipath = os.pathsep.join(path_list)
        self.output_dir = tempfile.mkdtemp()

    def check_dirs_match(self, expected_dir):
        try:
            self.assertThat(self.output_dir, MatchesDir(expected_dir))
        except testtools.matchers.MismatchError as e:
            raise e
        else:
            shutil.rmtree(self.output_dir)

    def test_multi_path(self):
        """
        Run test mode and pass multiple paths.
        """
        args = ['--conf', self.default_config_file, 'test',
                '-o', self.output_dir, self.multipath]

        self.execute_jenkins_jobs_with_args(args)
        self.check_dirs_match(os.path.join(self.fixtures_path,
                                           'multi-path/output_simple'))

    def test_recursive_multi_path_command_line(self):
        """
        Run test mode and pass multiple paths with recursive path option.
        """
        args = ['--conf', self.default_config_file, 'test',
                '-o', self.output_dir, '-r', self.multipath]

        self.execute_jenkins_jobs_with_args(args)
        self.check_dirs_match(os.path.join(self.fixtures_path,
                                           'multi-path/output_recursive'))

    def test_recursive_multi_path_config_file(self):
        # test recursive set in configuration file
        args = ['--conf', os.path.join(self.fixtures_path,
                                       'multi-path/builder-recursive.ini'),
                'test', '-o', self.output_dir, self.multipath]
        self.execute_jenkins_jobs_with_args(args)
        self.check_dirs_match(os.path.join(self.fixtures_path,
                                           'multi-path/output_recursive'))

    def test_recursive_multi_path_with_excludes(self):
        """
        Run test mode and pass multiple paths with recursive path option.
        """
        exclude_path = os.path.join(self.fixtures_path,
                                    'multi-path/yamldirs/dir2/dir1')
        args = ['--conf', self.default_config_file, 'test',
                '-x', exclude_path,
                '-o', self.output_dir,
                '-r', self.multipath]

        self.execute_jenkins_jobs_with_args(args)
        self.check_dirs_match(
            os.path.join(self.fixtures_path,
                         'multi-path/output_recursive_with_excludes'))