jjb/jenkins-job-builder
Code Issues Proposed changes

Use JJBConfig for arg/config initialization.

Browse Source 
* Create jenkins_jobs.config module with JJBConfig class.
* Move DEFAULT_CONF from jenkins_jobs.cmd into jenkins_jobs.config
* Move configuration initialization into JJBConfig
* Create method, "do_magical_things" to handle arbitration between
  config file and arguments as well as setting default values for
  config file settings if it doesn't contain the expected keys.
* Move JenkinsJobs.create_parser into its own module,
  jenkins_jobs.cli.parser, it can be used to provide default settings
  in the JJBConfig class when an argparse namespace object is not
  provided; this is primarily necessary because most of the original
  configuration initialization code relies on this being a namespace
  object (simple descendant of the object class).

At this point JJBConfig isn't much more than an object-oriented
version of the way configuration handling happened previously. Its
current form, however, is more amenable to the ultimate goal of the
2.0 refactorings--namely, being able to pass a single config object
around rather than breaking it up into apparently arbitrary settings
necessary to instantiate the Builder class and its delegate objects.

Change-Id: Ic0147e1dccbe620aaaba039a434e7cea6c670054
This commit is contained in:
Wayne Warren 2015-12-23 14:58:45 -08:00 committed by Darragh Bailey
parent 2bf92c3ed7
commit 2cf11f014c
10 changed files with 449 additions and 298 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
jenkins_jobs/cli/entry.py
View File

@ -13,21 +13,20 @@
# License for the specific language governing permissions and limitations
# under the License.
import argparse
import logging
import sys
from stevedore import extension
import jenkins_jobs.version
from jenkins_jobs import cmd
from jenkins_jobs import version
from jenkins_jobs.cli.parser import create_parser
from jenkins_jobs.config import JJBConfig
logging.basicConfig(level=logging.INFO)
def __version__():
return "Jenkins Job Builder version: %s" % \
jenkins_jobs.version.version_info.version_string()
version.version_info.version_string()
class JenkinsJobs(object):
@ -47,90 +46,27 @@ class JenkinsJobs(object):
various command line parameters.
"""
def __init__(self, args=None):
def __init__(self, args=None, **kwargs):
if args is None:
args = []
parser = self._create_parser()
self._options = parser.parse_args(args)
parser = create_parser()
options = parser.parse_args(args)
if not self._options.command:
self.jjb_config = JJBConfig(arguments=options, **kwargs)
self.jjb_config.do_magical_things()
if not options.command:
parser.error("Must specify a 'command' to be performed")
logger = logging.getLogger()
if (self._options.log_level is not None):
self._options.log_level = getattr(logging,
self._options.log_level.upper(),
logger.getEffectiveLevel())
logger.setLevel(self._options.log_level)
self._config_file_values = cmd.setup_config_settings(self._options)
def _create_parser(self):
parser = argparse.ArgumentParser()
parser.add_argument(
'--conf',
dest='conf',
help='''configuration file''')
parser.add_argument(
'-l',
'--log_level',
dest='log_level',
default='info',
help='''log level (default: %(default)s)''')
parser.add_argument(
'--ignore-cache',
action='store_true',
dest='ignore_cache',
default=False,
help='''ignore the cache and update the jobs anyhow (that will only
flush the specified jobs cache)''')
parser.add_argument(
'--flush-cache',
action='store_true',
dest='flush_cache',
default=False,
help='''flush all the cache entries before updating''')
parser.add_argument(
'--version',
dest='version',
action='version',
version=__version__(),
help='''show version''')
parser.add_argument(
'--allow-empty-variables',
action='store_true',
dest='allow_empty_variables',
default=None,
help='''Don\'t fail if any of the variables inside any string are
not defined, replace with empty string instead.''')
parser.add_argument(
'--user', '-u',
help='''The Jenkins user to use for authentication. This overrides
the user specified in the configuration file.''')
parser.add_argument(
'--password', '-p',
help='''Password or API token to use for authenticating towards
Jenkins. This overrides the password specified in the configuration
file.''')
subparser = parser.add_subparsers(
dest='command',
help='''update, test or delete job''')
extension_manager = extension.ExtensionManager(
namespace='jjb.cli.subcommands',
invoke_on_load=True,
)
def parse_subcommand_args(ext, subparser):
ext.obj.parse_args(subparser)
extension_manager.map(parse_subcommand_args, subparser)
return parser
if (options.log_level is not None):
options.log_level = getattr(logging,
options.log_level.upper(),
logger.getEffectiveLevel())
logger.setLevel(options.log_level)
def execute(self):
jenkins_jobs.cmd.execute(self._options, self._config_file_values)
cmd.execute(self.jjb_config)
def main():

92
jenkins_jobs/cli/parser.py Normal file
View File

@ -0,0 +1,92 @@
#!/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.
import argparse
import jenkins_jobs.version
from stevedore import extension
def __version__():
return "Jenkins Job Builder version: %s" % \
jenkins_jobs.version.version_info.version_string()
def create_parser():
""" Create an ArgumentParser object usable by JenkinsJobs.
"""
parser = argparse.ArgumentParser()
parser.add_argument(
'--conf',
dest='conf',
help='''configuration file''')
parser.add_argument(
'-l',
'--log_level',
dest='log_level',
default='info',
help='''log level (default: %(default)s)''')
parser.add_argument(
'--ignore-cache',
action='store_true',
dest='ignore_cache',
default=False,
help='''ignore the cache and update the jobs anyhow (that will only
flush the specified jobs cache)''')
parser.add_argument(
'--flush-cache',
action='store_true',
dest='flush_cache',
default=False,
help='''flush all the cache entries before updating''')
parser.add_argument(
'--version',
dest='version',
action='version',
version=__version__(),
help='''show version''')
parser.add_argument(
'--allow-empty-variables',
action='store_true',
dest='allow_empty_variables',
default=None,
help='''Don\'t fail if any of the variables inside any string are
not defined, replace with empty string instead.''')
parser.add_argument(
'--user', '-u',
help='''The Jenkins user to use for authentication. This overrides
the user specified in the configuration file.''')
parser.add_argument(
'--password', '-p',
help='''Password or API token to use for authenticating towards
Jenkins. This overrides the password specified in the configuration
file.''')
subparser = parser.add_subparsers(
dest='command',
help='''update, test or delete job''')
extension_manager = extension.ExtensionManager(
namespace='jjb.cli.subcommands',
invoke_on_load=True,
)
def parse_subcommand_args(ext, subparser):
ext.obj.parse_args(subparser)
extension_manager.map(parse_subcommand_args, subparser)
return parser

201
jenkins_jobs/cmd.py
View File

@ -13,25 +13,15 @@
# License for the specific language governing permissions and limitations
# under the License.
import fnmatch
import io
import logging
import os
import platform
import sys
import yaml
from six.moves import configparser
from six.moves import input
from six.moves import StringIO
from jenkins_jobs.builder import Builder
from jenkins_jobs.errors import JenkinsJobsException
import jenkins_jobs.version
import jenkins_jobs.cli
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
@ -63,191 +53,18 @@ def confirm(question):
sys.exit('Aborted')
def recurse_path(root, excludes=None):
if excludes is None:
excludes = []
basepath = os.path.realpath(root)
pathlist = [basepath]
patterns = [e for e in excludes if os.path.sep not in e]
absolute = [e for e in excludes if os.path.isabs(e)]
relative = [e for e in excludes if os.path.sep in e and
not os.path.isabs(e)]
for root, dirs, files in os.walk(basepath, topdown=True):
dirs[:] = [
d for d in dirs
if not any([fnmatch.fnmatch(d, pattern) for pattern in patterns])
if not any([fnmatch.fnmatch(os.path.abspath(os.path.join(root, d)),
path)
for path in absolute])
if not any([fnmatch.fnmatch(os.path.relpath(os.path.join(root, d)),
path)
for path in relative])
]
pathlist.extend([os.path.join(root, path) for path in dirs])
return pathlist
def get_config_file(options):
conf = '/etc/jenkins_jobs/jenkins_jobs.ini'
if options.conf:
conf = options.conf
else:
# Allow a script directory config to override.
localconf = os.path.join(os.path.dirname(__file__),
'jenkins_jobs.ini')
if os.path.isfile(localconf):
conf = localconf
# Allow a user directory config to override.
userconf = os.path.join(os.path.expanduser('~'), '.config',
'jenkins_jobs', 'jenkins_jobs.ini')
if os.path.isfile(userconf):
conf = userconf
return conf
def setup_config_settings(options):
conf = get_config_file(options)
config = configparser.ConfigParser()
# Load default config always
config.readfp(StringIO(DEFAULT_CONF))
if os.path.isfile(conf):
options.conf = conf # remember file we read from
logger.debug("Reading config from {0}".format(conf))
conffp = io.open(conf, 'r', encoding='utf-8')
config.readfp(conffp)
elif options.command == 'test':
logger.debug("Not requiring config for test output generation")
else:
raise JenkinsJobsException(
"A valid configuration file is required."
"\n{0} is not valid.".format(conf))
return config
def execute(options, config):
logger.debug("Config: {0}".format(config))
# check the ignore_cache setting: first from command line,
# if not present check from ini file
ignore_cache = False
if options.ignore_cache:
ignore_cache = options.ignore_cache
elif config.has_option('jenkins', 'ignore_cache'):
logging.warn('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('jenkins', 'ignore_cache')
elif config.has_option('job_builder', 'ignore_cache'):
ignore_cache = config.getboolean('job_builder', 'ignore_cache')
# 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
if options.user:
user = options.user
else:
try:
user = config.get('jenkins', 'user')
except (TypeError, configparser.NoOptionError):
user = None
if options.password:
password = options.password
else:
try:
password = config.get('jenkins', 'password')
except (TypeError, configparser.NoOptionError):
password = None
# 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 user is None and password is None:
logger.info("Will use anonymous access to Jenkins if needed.")
elif (user is not None and password is None) or (
user is None and password is not None):
raise JenkinsJobsException(
"Cannot authenticate to Jenkins with only one of User and "
"Password provided, please check your configuration."
)
# 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.
timeout = jenkins_jobs.builder._DEFAULT_TIMEOUT
try:
timeout = config.getfloat('jenkins', 'timeout')
except (ValueError):
raise JenkinsJobsException("Jenkins timeout config is invalid")
except (TypeError, configparser.NoOptionError):
pass
plugins_info = None
if getattr(options, 'plugins_info_path', None) is not None:
with io.open(options.plugins_info_path, 'r',
encoding='utf-8') as yaml_file:
plugins_info = yaml.load(yaml_file)
if not isinstance(plugins_info, list):
raise JenkinsJobsException("{0} must contain a Yaml list!"
.format(options.plugins_info_path))
elif (not options.conf or not
config.getboolean("jenkins", "query_plugins_info")):
logger.debug("Skipping plugin info retrieval")
plugins_info = {}
if options.allow_empty_variables is not None:
config.set('job_builder',
'allow_empty_variables',
str(options.allow_empty_variables))
def execute(jjb_config):
config = jjb_config.config_parser
options = jjb_config.arguments
builder = Builder(config.get('jenkins', 'url'),
user,
password,
config,
jenkins_timeout=timeout,
ignore_cache=ignore_cache,
jjb_config.user,
jjb_config.password,
jjb_config.config_parser,
jenkins_timeout=jjb_config.timeout,
ignore_cache=jjb_config.ignore_cache,
flush_cache=options.flush_cache,
plugins_list=plugins_info)
if getattr(options, 'path', None):
if hasattr(options.path, 'read'):
logger.debug("Input file is stdin")
if options.path.isatty():
key = 'CTRL+Z' if platform.system() == 'Windows' else 'CTRL+D'
logger.warn(
"Reading configuration from STDIN. Press %s to end input.",
key)
else:
# take list of paths
options.path = options.path.split(os.pathsep)
do_recurse = (getattr(options, 'recursive', False) or
config.getboolean('job_builder', 'recursive'))
excludes = [e for elist in options.exclude
for e in elist.split(os.pathsep)] or \
config.get('job_builder', 'exclude').split(os.pathsep)
paths = []
for path in options.path:
if do_recurse and os.path.isdir(path):
paths.extend(recurse_path(path, excludes))
else:
paths.append(path)
options.path = paths
plugins_list=jjb_config.plugins_info)
if options.command == 'delete':
for job in options.name:

279
jenkins_jobs/config.py Normal file
View File

@ -0,0 +1,279 @@
#!/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.
import io
import logging
import platform
import os
import yaml
from six.moves import configparser, StringIO
from jenkins_jobs import builder
from jenkins_jobs import utils
from jenkins_jobs.cli.parser import create_parser
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
[jenkins]
url=http://localhost:8080/
query_plugins_info=True
[hipchat]
authtoken=dummy
send-as=Jenkins
"""
class JJBConfigException(JenkinsJobsException):
pass
class JJBConfig(object):
def __init__(self, config_filename=None, arguments=None,
config_file_required=False):
"""
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
do_magical_things 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 dict arguments: A argparse.Namespace object as produced by the
jenkins-jobs argument parser.
: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()
if arguments is None:
jenkins_jobs_parser = create_parser()
arguments = jenkins_jobs_parser.parse_args(['test'])
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
elif hasattr(arguments, 'conf') and arguments.conf is not None:
conf = arguments.conf
elif config_file_required:
if os.path.isfile(local_conf):
conf = local_conf
elif os.path.isfile(user_conf):
conf = user_conf
else:
conf = global_conf
config_fp = None
if conf is not None:
try:
config_fp = self._read_config_file(conf)
except JJBConfigException as e:
if config_file_required:
raise e
else:
logger.warn("""Config file, {0}, not found. Using default
config values.""".format(conf))
if config_fp is not None:
config_parser.readfp(config_fp)
self.config_parser = config_parser
self.arguments = arguments
def _init_defaults(self):
""" Initialize default configuration values using DEFAULT_CONF
"""
config = configparser.ConfigParser()
# Load default config always
config.readfp(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 do_magical_things(self):
config = self.config_parser
options = self.arguments
logger.debug("Config: {0}".format(config))
# check the ignore_cache setting: first from command line,
# if not present check from ini file
self.ignore_cache = False
if options.ignore_cache:
self.ignore_cache = options.ignore_cache
elif config.has_option('jenkins', 'ignore_cache'):
logging.warn('''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''')
self.ignore_cache = config.getboolean('jenkins', 'ignore_cache')
elif config.has_option('job_builder', 'ignore_cache'):
self.ignore_cache = config.getboolean('job_builder',
'ignore_cache')
# 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
if options.user:
self.user = options.user
else:
try:
self.user = config.get('jenkins', 'user')
except (TypeError, configparser.NoOptionError):
self.user = None
if options.password:
self.password = options.password
else:
try:
self.password = config.get('jenkins', 'password')
except (TypeError, configparser.NoOptionError):
self.password = None
# 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.user is None and self.password is None:
logger.info("Will use anonymous access to Jenkins if needed.")
elif (self.user is not None and self.password is None) or (
self.user is None and self.password is not None):
raise JenkinsJobsException(
"Cannot authenticate to Jenkins with only one of User and "
"Password provided, please check your configuration."
)
# 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.
self.timeout = builder._DEFAULT_TIMEOUT
try:
self.timeout = config.getfloat('jenkins', 'timeout')
except (ValueError):
raise JenkinsJobsException("Jenkins timeout config is invalid")
except (TypeError, configparser.NoOptionError):
pass
self.plugins_info = None
if getattr(options, 'plugins_info_path', None) is not None:
with io.open(options.plugins_info_path, 'r',
encoding='utf-8') as yaml_file:
self.plugins_info = yaml.load(yaml_file)
if not isinstance(self.plugins_info, list):
raise JenkinsJobsException("{0} must contain a Yaml list!"
.format(options.plugins_info_path))
elif (not options.conf or not
config.getboolean("jenkins", "query_plugins_info")):
logger.debug("Skipping plugin info retrieval")
self.plugins_info = {}
if options.allow_empty_variables is not None:
config.set('job_builder',
'allow_empty_variables',
str(options.allow_empty_variables))
if getattr(options, 'path', None):
if hasattr(options.path, 'read'):
logger.debug("Input file is stdin")
if options.path.isatty():
if platform.system() == 'Windows':
key = 'CTRL+Z'
else:
key = 'CTRL+D'
logger.warn("""Reading configuration from STDIN. Press %s
to end input.""", key)
else:
# take list of paths
options.path = options.path.split(os.pathsep)
do_recurse = (getattr(options, 'recursive', False) or
config.getboolean('job_builder', 'recursive'))
excludes = [e for elist in options.exclude
for e in elist.split(os.pathsep)] or \
config.get('job_builder', 'exclude').split(os.pathsep)
paths = []
for path in options.path:
if do_recurse and os.path.isdir(path):
paths.extend(utils.recurse_path(path, excludes))
else:
paths.append(path)
options.path = paths
self.config_parser = config
self.arguments = options

29
jenkins_jobs/utils.py
View File

@ -16,7 +16,9 @@
# functions that don't fit in well elsewhere
import codecs
import fnmatch
import locale
import os.path
def wrap_stream(stream, encoding='utf-8'):
@ -33,3 +35,30 @@ def wrap_stream(stream, encoding='utf-8'):
return stream
return codecs.EncodedFile(stream, encoding, stream_enc)
def recurse_path(root, excludes=None):
if excludes is None:
excludes = []
basepath = os.path.realpath(root)
pathlist = [basepath]
patterns = [e for e in excludes if os.path.sep not in e]
absolute = [e for e in excludes if os.path.isabs(e)]
relative = [e for e in excludes if os.path.sep in e and
not os.path.isabs(e)]
for root, dirs, files in os.walk(basepath, topdown=True):
dirs[:] = [
d for d in dirs
if not any([fnmatch.fnmatch(d, pattern) for pattern in patterns])
if not any([fnmatch.fnmatch(os.path.abspath(os.path.join(root, d)),
path)
for path in absolute])
if not any([fnmatch.fnmatch(os.path.relpath(os.path.join(root, d)),
path)
for path in relative])
]
pathlist.extend([os.path.join(root, path) for path in dirs])
return pathlist

12
tests/base.py
View File

@ -27,13 +27,12 @@ import re
import xml.etree.ElementTree as XML
import fixtures
from six.moves import configparser
from six.moves import StringIO
import testtools
from testtools.content import text_content
from yaml import safe_dump
from jenkins_jobs.cmd import DEFAULT_CONF
from jenkins_jobs.config import JJBConfig
import jenkins_jobs.local_yaml as yaml
from jenkins_jobs.modules import project_externaljob
from jenkins_jobs.modules import project_flow
@ -131,12 +130,9 @@ class BaseTestCase(LoggingFixture):
return yaml_content
def _get_config(self):
config = configparser.ConfigParser()
config.readfp(StringIO(DEFAULT_CONF))
if self.conf_filename is not None:
with io.open(self.conf_filename, 'r', encoding='utf-8') as cf:
config.readfp(cf)
return config
jjb_config = JJBConfig(self.conf_filename)
return jjb_config.config_parser
def test_yaml_snippet(self):
if not self.in_filename:

4
tests/cmd/subcommands/test_test.py
View File

@ -152,8 +152,8 @@ class TestTests(CmdTestsBase):
os.path.join(self.fixtures_path, 'cmd-001.yaml')]
with mock.patch('sys.stdout'):
jenkins_jobs = entry.JenkinsJobs(args)
e = self.assertRaises(JenkinsJobsException, jenkins_jobs.execute)
e = self.assertRaises(JenkinsJobsException, entry.JenkinsJobs,
args)
self.assertIn("must contain a Yaml list", str(e))

2
tests/cmd/test_cmd.py
View File

@ -31,7 +31,7 @@ class CmdTestsBase(LoggingFixture, testtools.TestCase):
jenkins_jobs.execute()
class CmdTests(CmdTestsBase):
class TestCmd(CmdTestsBase):
def test_with_empty_args(self):
"""

10
tests/cmd/test_config.py
View File

@ -13,6 +13,8 @@ class TestConfigs(CmdTestsBase):
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')
def test_use_global_config(self):
"""
@ -24,15 +26,14 @@ class TestConfigs(CmdTestsBase):
with patch('os.path.isfile', return_value=True) as m_isfile:
def side_effect(path):
if path == self.user_conf:
return False
if path == self.global_conf:
return True
return False
m_isfile.side_effect = side_effect
with patch('io.open', return_value=conffp) as m_open:
entry.JenkinsJobs(args)
entry.JenkinsJobs(args, config_file_required=True)
m_open.assert_called_with(self.global_conf, 'r',
encoding='utf-8')
@ -48,10 +49,11 @@ class TestConfigs(CmdTestsBase):
def side_effect(path):
if path == self.user_conf:
return True
return False
m_isfile.side_effect = side_effect
with patch('io.open', return_value=conffp) as m_open:
entry.JenkinsJobs(args)
entry.JenkinsJobs(args, config_file_required=True)
m_open.assert_called_with(self.user_conf, 'r',
encoding='utf-8')

20
tests/cmd/test_recurse_path.py
View File

@ -3,7 +3,7 @@ import os
from tests.base import mock
import testtools
from jenkins_jobs import cmd
from jenkins_jobs import utils
def fake_os_walk(paths):
@ -25,14 +25,14 @@ def fake_os_walk(paths):
return os_walk
# Testing the cmd module can sometimes result in the CacheStorage class
# Testing the utils module can sometimes result in the CacheStorage class
# attempting to create the cache directory multiple times as the tests
# are run in parallel. Stub out the CacheStorage to ensure that each
# test can safely create the object without effect.
@mock.patch('jenkins_jobs.builder.CacheStorage', mock.MagicMock)
class CmdRecursePath(testtools.TestCase):
@mock.patch('jenkins_jobs.cmd.os.walk')
@mock.patch('jenkins_jobs.utils.os.walk')
def test_recursive_path_option_exclude_pattern(self, oswalk_mock):
"""
Test paths returned by the recursive processing when using pattern
@ -60,9 +60,9 @@ class CmdRecursePath(testtools.TestCase):
paths = [k for k, v in os_walk_paths if v is not None]
oswalk_mock.side_effect = fake_os_walk(os_walk_paths)
self.assertEqual(paths, cmd.recurse_path('/jjb_configs', ['test*']))
self.assertEqual(paths, utils.recurse_path('/jjb_configs', ['test*']))
@mock.patch('jenkins_jobs.cmd.os.walk')
@mock.patch('jenkins_jobs.utils.os.walk')
def test_recursive_path_option_exclude_absolute(self, oswalk_mock):
"""
Test paths returned by the recursive processing when using absolute
@ -93,10 +93,10 @@ class CmdRecursePath(testtools.TestCase):
oswalk_mock.side_effect = fake_os_walk(os_walk_paths)
self.assertEqual(paths, cmd.recurse_path('/jjb_configs',
['/jjb_configs/dir1']))
self.assertEqual(paths, utils.recurse_path('/jjb_configs',
['/jjb_configs/dir1']))
@mock.patch('jenkins_jobs.cmd.os.walk')
@mock.patch('jenkins_jobs.utils.os.walk')
def test_recursive_path_option_exclude_relative(self, oswalk_mock):
"""
Test paths returned by the recursive processing when using relative
@ -132,5 +132,5 @@ class CmdRecursePath(testtools.TestCase):
oswalk_mock.side_effect = fake_os_walk(rel_os_walk_paths)
self.assertEqual(paths, cmd.recurse_path('jjb_configs',
['jjb_configs/test3/bar']))
self.assertEqual(paths, utils.recurse_path('jjb_configs',
['jjb_configs/test3/bar']))