/usr/share/jenkins-job-builder/jenkins_jobs/cmd.py is in jenkins-job-builder 0.9.0-0.2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 | #!/usr/bin/env python
# Copyright (C) 2012 OpenStack Foundation
#
# 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 ConfigParser
import logging
import os
import platform
import sys
import cStringIO
from jenkins_jobs.builder import Builder
from jenkins_jobs.errors import JenkinsJobsException
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
DEFAULT_CONF = """
[job_builder]
keep_descriptions=False
ignore_cache=False
recursive=False
[jenkins]
url=http://localhost:8080/
user=
password=
"""
def confirm(question):
answer = raw_input('%s (Y/N): ' % question).upper().strip()
if not answer == 'Y':
sys.exit('Aborted')
def recurse_path(root):
basepath = os.path.realpath(root)
pathlist = [basepath]
for root, dirs, files in os.walk(basepath, topdown=True):
pathlist.extend([os.path.join(root, path) for path in dirs])
return pathlist
def create_parser():
parser = argparse.ArgumentParser()
recursive_parser = argparse.ArgumentParser(add_help=False)
recursive_parser.add_argument('-r', '--recursive', action='store_true',
dest='recursive', default=False,
help='look for yaml files recursively')
subparser = parser.add_subparsers(help='update, test or delete job',
dest='command')
parser_update = subparser.add_parser('update', parents=[recursive_parser])
parser_update.add_argument('path', help='path to YAML file or directory')
parser_update.add_argument('names', help='name(s) of job(s)', nargs='*')
parser_update.add_argument('--delete-old', help='delete obsolete jobs',
action='store_true',
dest='delete_old', default=False,)
parser_test = subparser.add_parser('test', parents=[recursive_parser])
parser_test.add_argument('path', help='path to YAML file or directory',
nargs='?', default=sys.stdin)
parser_test.add_argument('-o', dest='output_dir', default=sys.stdout,
help='path to output XML')
parser_test.add_argument('name', help='name(s) of job(s)', nargs='*')
parser_delete = subparser.add_parser('delete')
parser_delete.add_argument('name', help='name of job', nargs='+')
parser_delete.add_argument('-p', '--path', default=None,
help='path to YAML file or directory')
subparser.add_parser('delete-all',
help='delete *ALL* jobs from Jenkins server, '
'including those not managed by Jenkins Job '
'Builder.')
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')
return parser
def main(argv=None):
# We default argv to None and assign to sys.argv[1:] below because having
# an argument default value be a mutable type in Python is a gotcha. See
# http://bit.ly/1o18Vff
if argv is None:
argv = sys.argv[1:]
parser = create_parser()
options = parser.parse_args(argv)
if (options.log_level is not None):
options.log_level = getattr(logging, options.log_level.upper(),
logger.getEffectiveLevel())
logger.setLevel(options.log_level)
config = setup_config_settings(options)
execute(options, config)
def setup_config_settings(options):
conf = '/etc/jenkins_jobs/jenkins_jobs.ini'
if options.conf:
conf = options.conf
else:
# Fallback to script directory
localconf = os.path.join(os.path.dirname(__file__),
'jenkins_jobs.ini')
if os.path.isfile(localconf):
conf = localconf
config = ConfigParser.ConfigParser()
## Load default config always
config.readfp(cStringIO.StringIO(DEFAULT_CONF))
if os.path.isfile(conf):
logger.debug("Reading config from {0}".format(conf))
conffp = open(conf, 'r')
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 when not run as a test"
"\n{0} is not a valid .ini file".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')
# workaround for python 2.6 interpolation error
# https://bugs.launchpad.net/openstack-ci/+bug/1259631
try:
user = config.get('jenkins', 'user')
except (TypeError, ConfigParser.NoOptionError):
user = None
try:
password = config.get('jenkins', 'password')
except (TypeError, ConfigParser.NoOptionError):
password = None
builder = Builder(config.get('jenkins', 'url'),
user,
password,
config,
ignore_cache=ignore_cache,
flush_cache=options.flush_cache)
if getattr(options, 'path', None):
if options.path == sys.stdin:
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)
# expand or convert options.path to a list
if (getattr(options, 'recursive', False)
or config.getboolean('job_builder', 'recursive')) and \
os.path.isdir(options.path):
options.path = recurse_path(options.path)
else:
options.path = [options.path]
if options.command == 'delete':
for job in options.name:
logger.info("Deleting jobs in [{0}]".format(job))
builder.delete_job(job, options.path)
elif options.command == 'delete-all':
confirm('Sure you want to delete *ALL* jobs from Jenkins server?\n'
'(including those not managed by Jenkins Job Builder)')
logger.info("Deleting all jobs")
builder.delete_all_jobs()
elif options.command == 'update':
logger.info("Updating jobs in {0} ({1})".format(
options.path, options.names))
jobs = builder.update_job(options.path, options.names)
if options.delete_old:
builder.delete_old_managed(keep=[x.name for x in jobs])
elif options.command == 'test':
builder.update_job(options.path, options.name,
output=options.output_dir)
if __name__ == '__main__':
sys.path.insert(0, '.')
main()
|