A command-line tool to run a job on Jenkins remotely.

jenkins.py

#!/usr/bin/env python3
# coding: utf-8

from collections import namedtuple


Crumb = namedtuple('Crumb', ('key', 'value'))


def run(args):
    auth = encode_auth(args.user, args.password)

    try:
        crumb = get_crumb(
            args.server,
            auth
        )
    except Exception as exception:
        fail_with('Failed to get crumb: {}'.format(exception))

    try:
        request_build(
            args.server,
            auth,
            crumb,
            args.job,
            args.commit_reference
        )
    except Exception as exception:
        fail_with('Failed to request build: {}'.format(exception))


def get_crumb(server, auth):
    from urllib import request
    from xml.etree import ElementTree

    r = request.Request(
        'http://{}/crumbIssuer/api/xml'.format(
            server
        ),
        method='GET'
    )

    r.add_header('Authorization', auth)

    response = request.urlopen(r)
    body = response.read()

    tree = ElementTree.fromstring(body.decode())
    key = tree.find('crumbRequestField').text
    value = tree.find('crumb').text

    return Crumb(key, value)


def request_build(server, auth, crumb, job, commit_reference):
    import json
    from urllib import parse
    from urllib import request

    parameter = json.dumps({
        'parameter': [
            {
                'name': 'GIT_COMMIT_REFERENCE',
                'value': commit_reference
            }
        ]
    })

    data = parse.urlencode({'json': parameter})

    r = request.Request(
        'http://{}/job/{}/build'.format(
            server,
            job
        ),
        method='POST',
        data=data.encode()
    )

    r.add_header(crumb.key, crumb.value)
    r.add_header('Authorization', auth)

    request.urlopen(r)


def encode_auth(user, password):
    import base64
    return 'Basic {}'.format(
        base64.encodebytes(
            '{}:{}'.format(user, password).encode()
        ).decode()
    ).rstrip('\n')


def parse(raw_args):
    from argparse import ArgumentParser

    parser = ArgumentParser()
    subparsers = parser.add_subparsers(dest='subcommand')
    subparsers.required = True

    run_command_name = 'run'

    actions = {
        run_command_name: run,
    }

    run_command = subparsers.add_parser(
        name=run_command_name,
        help='send request to run a job',
    )

    run_command.add_argument(
        '--user',
        '-u',
        type=str,
        required=True,
        help='The ID of user',
    )

    run_command.add_argument(
        '--password',
        '-p',
        type=str,
        required=True,
        help='The password of user',
    )

    run_command.add_argument(
        '--server',
        '-s',
        type=str,
        required=True,
        help='The localtion of server (example: \'128.0.0.1:8080\')',
    )

    run_command.add_argument(
        '--job',
        '-j',
        type=str,
        required=True,
        help='The name of job',
    )

    run_command.add_argument(
        '--commit-reference',
        '-r',
        type=str,
        required=True,
        help='The reference of a commit to build',
    )

    args = parser.parse_args(raw_args)

    actions[args.subcommand](args)


def fail_with(message):
    import sys
    print('error: {}'.format(message), file=sys.stderr)
    sys.exit(128)


if __name__ == '__main__':
    import sys
    parse(sys.argv[1:])