jenkins-job-builder/jenkins_jobs/config.py
Thanh Ha 5a98ebb8bb
Fix flake8 BLK100 and BLK999 issues
Resolves flake8 BLK related issues.

"BLK100 Black would make changes."
"BLK999 Unexpected exception: ..."

Change-Id: Ic6d4265e3d02b2f2a15ab851eaed98288c62730c
Signed-off-by: Thanh Ha <zxiiro@gmail.com>
2020-08-31 10:54:19 -04:00

383 lines
15 KiB
Python

#!/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):
"""Return a ConfigParser object from a file input."""
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):
"""Returns the value of a config in a config module.
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)