path: root/jenkins_jobs/config.py

#!/usr/bin/env python
# Copyright (C) 2015 Wayne Warren
#
# 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.

# Manage JJB Configuration sources, defaults, and access.

from collections import defaultdict
import io
import logging
import os

from six.moves import configparser, StringIO
from six import PY2

from jenkins_jobs import builder
from jenkins_jobs.errors import JJBConfigException
from jenkins_jobs.errors import JenkinsJobsException

__all__ = [
    "JJBConfig"
]

logger = logging.getLogger(__name__)

DEFAULT_CONF = """
[job_builder]
keep_descriptions=False
ignore_cache=False
recursive=False
exclude=.*
allow_duplicates=False
allow_empty_variables=False
retain_anchors=False

# other named sections could be used in addition to the implicit [jenkins]
# if you have multiple jenkins servers.
[jenkins]
url=http://localhost:8080/
query_plugins_info=False
"""

CONFIG_REQUIRED_MESSAGE = ("A valid configuration file is required. "
                           "No configuration file passed.")
DEPRECATED_PLUGIN_CONFIG_SECTION_MESSAGE = (
    "Defining plugin configuration using a [{plugin}] section in your config"
    " file is deprecated. The recommended way to define plugins now is by"
    " using a [plugin \"{plugin}\"] section"
)
_NOTSET = object()


class JJBConfig(object):

    def __init__(self, config_filename=None,
                 config_file_required=False,
                 config_section='jenkins'):

        """
        The JJBConfig class is intended to encapsulate and resolve priority
        between all sources of configuration for the JJB library. This allows
        the various sources of configuration to provide a consistent accessor
        interface regardless of where they are used.

        It also allows users of JJB-as-an-API to create minimally valid
        configuration and easily make minor modifications to default values
        without strictly adhering to the confusing setup (see the _setup
        method, the behavior of which largely lived in the cmd.execute method
        previously) necessary for the jenkins-jobs command line tool.

        :arg str config_filename: Name of configuration file on which to base
            this config object.
        :arg bool config_file_required: Allows users of the JJBConfig class to
            decide whether or not it's really necessary for a config file to be
            passed in when creating an instance. This has two effects on the
            behavior of JJBConfig initialization:
            * It determines whether or not we try "local" and "global" config
              files.
            * It determines whether or not failure to read some config file
              will raise an exception or simply print a warning message
              indicating that no config file was found.
        """

        config_parser = self._init_defaults()

        global_conf = '/etc/jenkins_jobs/jenkins_jobs.ini'
        user_conf = os.path.join(os.path.expanduser('~'), '.config',
                                 'jenkins_jobs', 'jenkins_jobs.ini')
        local_conf = os.path.join(os.path.dirname(__file__),
                                  'jenkins_jobs.ini')
        conf = None
        if config_filename is not None:
            conf = config_filename
        else:
            if os.path.isfile(local_conf):
                conf = local_conf
            elif os.path.isfile(user_conf):
                conf = user_conf
            else:
                conf = global_conf

        if config_file_required and conf is None:
            raise JJBConfigException(CONFIG_REQUIRED_MESSAGE)

        config_fp = None
        if conf is not None:
            try:
                config_fp = self._read_config_file(conf)
            except JJBConfigException:
                if config_file_required:
                    raise JJBConfigException(CONFIG_REQUIRED_MESSAGE)
                else:
                    logger.warning("Config file, {0}, not found. Using "
                                   "default config values.".format(conf))

        if config_fp is not None:
            if PY2:
                config_parser.readfp(config_fp)
            else:
                config_parser.read_file(config_fp)

        self.config_parser = config_parser

        self._section = config_section
        self.print_job_urls = False

        self.jenkins = defaultdict(None)
        self.builder = defaultdict(None)
        self.yamlparser = defaultdict(None)

        self._setup()
        self._handle_deprecated_hipchat_config()

        if config_fp is not None:
            config_fp.close()

    def _init_defaults(self):
        """ Initialize default configuration values using DEFAULT_CONF
        """
        config = configparser.ConfigParser()
        # Load default config always
        if PY2:
            config.readfp(StringIO(DEFAULT_CONF))
        else:
            config.read_file(StringIO(DEFAULT_CONF))
        return config

    def _read_config_file(self, config_filename):
        """ Given path to configuration file, read it in as a ConfigParser
        object and return that object.
        """
        if os.path.isfile(config_filename):
            self.__config_file = config_filename  # remember file we read from
            logger.debug("Reading config from {0}".format(config_filename))
            config_fp = io.open(config_filename, 'r', encoding='utf-8')
        else:
            raise JJBConfigException(
                "A valid configuration file is required. "
                "\n{0} is not valid.".format(config_filename))

        return config_fp

    def _handle_deprecated_hipchat_config(self):
        config = self.config_parser

        if config.has_section('hipchat'):
            if config.has_section('plugin "hipchat"'):
                logger.warning(
                    "Both [hipchat] and [plugin \"hipchat\"] sections "
                    "defined, legacy [hipchat] section will be ignored."
                )
            else:
                logger.warning(
                    "[hipchat] section is deprecated and should be moved to a "
                    "[plugins \"hipchat\"] section instead as the [hipchat] "
                    "section will be ignored in the future."
                )
                config.add_section('plugin "hipchat"')
                for option in config.options("hipchat"):
                    config.set('plugin "hipchat"', option,
                               config.get("hipchat", option))

                config.remove_section("hipchat")

        # remove need to reference jenkins section when using hipchat plugin
        # moving to backports configparser would allow use of extended
        # interpolation to remove the need for plugins to need information
        # directly from the jenkins section within code and allow variables
        # in the config file to refer instead.
        if (config.has_section('plugin "hipchat"') and
                not config.has_option('plugin "hipchat"', 'url')):
            config.set('plugin "hipchat"', "url", config.get('jenkins', 'url'))

    def _setup(self):
        config = self.config_parser

        logger.debug("Config: {0}".format(config))

        # check the ignore_cache setting
        ignore_cache = False
        if config.has_option(self._section, 'ignore_cache'):
            logger.warning("ignore_cache option should be moved to the "
                           "[job_builder] section in the config file, the "
                           "one specified in the [jenkins] section will be "
                           "ignored in the future")
            ignore_cache = config.getboolean(self._section, 'ignore_cache')
        elif config.has_option('job_builder', 'ignore_cache'):
            ignore_cache = config.getboolean('job_builder', 'ignore_cache')
        self.builder['ignore_cache'] = ignore_cache

        # check the flush_cache setting
        flush_cache = False
        if config.has_option('job_builder', 'flush_cache'):
            flush_cache = config.getboolean('job_builder', 'flush_cache')
        self.builder['flush_cache'] = flush_cache

        # check the print_job_urls setting
        if config.has_option('job_builder', 'print_job_urls'):
            self.print_job_urls = config.getboolean('job_builder',
                                                    'print_job_urls')

        # Jenkins supports access as an anonymous user, which can be used to
        # ensure read-only behaviour when querying the version of plugins
        # installed for test mode to generate XML output matching what will be
        # uploaded. To enable must pass 'None' as the value for user and
        # password to python-jenkins
        #
        # catching 'TypeError' is a workaround for python 2.6 interpolation
        # error
        # https://bugs.launchpad.net/openstack-ci/+bug/1259631

        try:
            user = config.get(self._section, 'user')
        except (TypeError, configparser.NoOptionError):
            user = None
        self.jenkins['user'] = user

        try:
            password = config.get(self._section, 'password')
        except (TypeError, configparser.NoOptionError):
            password = None
        self.jenkins['password'] = password

        # None -- no timeout, blocking mode; same as setblocking(True)
        # 0.0 -- non-blocking mode; same as setblocking(False) <--- default
        # > 0 -- timeout mode; operations time out after timeout seconds
        # < 0 -- illegal; raises an exception
        # to retain the default must use
        # "timeout=jenkins_jobs.builder._DEFAULT_TIMEOUT" or not set timeout at
        # all.
        try:
            timeout = config.getfloat(self._section, 'timeout')
        except (ValueError):
            raise JenkinsJobsException("Jenkins timeout config is invalid")
        except (TypeError, configparser.NoOptionError):
            timeout = builder._DEFAULT_TIMEOUT
        self.jenkins['timeout'] = timeout

        plugins_info = None
        if (config.has_option(self._section, 'query_plugins_info') and
                not config.getboolean(self._section, "query_plugins_info")):
            logger.debug("Skipping plugin info retrieval")
            plugins_info = []
        self.builder['plugins_info'] = plugins_info

        self.recursive = config.getboolean('job_builder', 'recursive')
        self.excludes = config.get('job_builder', 'exclude').split(os.pathsep)

        # The way we want to do things moving forward:
        self.jenkins['url'] = config.get(self._section, 'url')
        self.builder['print_job_urls'] = self.print_job_urls

        # keep descriptions ? (used by yamlparser)
        keep_desc = False
        if (config and config.has_section('job_builder') and
                config.has_option('job_builder', 'keep_descriptions')):
            keep_desc = config.getboolean('job_builder',
                                          'keep_descriptions')
        self.yamlparser['keep_descriptions'] = keep_desc

        # figure out the include path (used by yamlparser)
        path = ["."]
        if (config and config.has_section('job_builder') and
                config.has_option('job_builder', 'include_path')):
            path = config.get('job_builder',
                              'include_path').split(':')
        self.yamlparser['include_path'] = path

        # allow duplicates?
        allow_duplicates = False
        if config and config.has_option('job_builder', 'allow_duplicates'):
            allow_duplicates = config.getboolean('job_builder',
                                                 'allow_duplicates')
        self.yamlparser['allow_duplicates'] = allow_duplicates

        # allow empty variables?
        self.yamlparser['allow_empty_variables'] = (
            config and config.has_section('job_builder') and
            config.has_option('job_builder', 'allow_empty_variables') and
            config.getboolean('job_builder', 'allow_empty_variables'))

        # retain anchors across files?
        retain_anchors = False
        if config and config.has_option('job_builder', 'retain_anchors'):
            retain_anchors = config.getboolean('job_builder',
                                               'retain_anchors')
        self.yamlparser['retain_anchors'] = retain_anchors

        update = None
        if (config and config.has_section('job_builder') and
                config.has_option('job_builder', 'update')):
            update = config.get('job_builder', 'update')
        self.builder['update'] = update

    def validate(self):
        # Inform the user as to what is likely to happen, as they may specify
        # a real jenkins instance in test mode to get the plugin info to check
        # the XML generated.
        if self.jenkins['user'] is None and self.jenkins['password'] is None:
            logger.info("Will use anonymous access to Jenkins if needed.")
        elif ((self.jenkins['user'] is not None and
               self.jenkins['password'] is None) or
              (self.jenkins['user'] is None and
               self.jenkins['password'] is not None)):
            raise JenkinsJobsException(
                "Cannot authenticate to Jenkins with only one of User and "
                "Password provided, please check your configuration."
            )

        if (self.builder['plugins_info'] is not None and
                not isinstance(self.builder['plugins_info'], list)):
            raise JenkinsJobsException("plugins_info must contain a list!")

    def get_module_config(self, section, key, default=None):
        """ Given a section name and a key value, return the value assigned to
        the key in the JJB .ini file if it exists, otherwise emit a warning
        indicating that the value is not set. Default value returned if no
        value is set in the file will be a blank string.
        """
        result = default
        try:
            result = self.config_parser.get(
                section, key
            )
        except (configparser.NoSectionError, configparser.NoOptionError,
                JenkinsJobsException) as e:
            # use of default ignores missing sections/options
            if result is None:
                logger.warning(
                    "You didn't set a %s neither in the yaml job definition "
                    "nor in the %s section, blank default value will be "
                    "applied:\n%s", key, section, e)
        return result

    def get_plugin_config(self, plugin, key, default=None):
        return self.get_module_config('plugin "{}"'.format(plugin), key,
                                      default)