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
|
# simplejenkinsapi/api.py -- simple job management in Jenkins
#
# Copyright 2012 Lars Wirzenius
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
'''Simple job management in Jenkins.
This module provides tools for listing, creating, updating, and
removing jobs in a running Jenkins instance, using its HTTP API.
'''
import httplib
import json
import logging
import urlparse
class Jenkins(object):
'''Access a running Jenkins server.'''
def __init__(self, url): # pragma: no cover
self._url = url
def _connect(self): # pragma: no cover
'''Connect to HTTP server, return httplib.HTTPConnection.'''
(scheme, netloc, path, params, query,
fragment) = urlparse.urlparse(self._url)
assert path == '/'
assert params == ''
assert fragment == ''
if ':' in netloc:
host, port = netloc.split(':', 1)
port = int(port)
else:
host = netloc
port = None
if scheme == 'https':
return httplib.HTTPSConnection(host, port)
else:
return httplib.HTTPConnection(host, port)
def _do(self, method, path, body='', headers={}):
'''Make an HTTP request, return result.
If server returned JSON, return it as a Python object.
Otherwise, return as string.
'''
logging.debug('HTTP method: %s' % repr(method))
logging.debug('HTTP path: %s' % repr(path))
logging.debug('HTTP body: %s' % repr(body))
logging.debug('HTTP headers: %s' % repr(headers))
conn = self._connect()
conn.request(method, path, body, headers)
resp = conn.getresponse()
text = resp.read()
logging.debug('HTTP status: %d' % resp.status)
logging.debug('HTTP reason: %s' % resp.reason)
logging.debug('HTTP text: %s' % repr(text))
ok = [httplib.OK, httplib.FOUND]
if resp.status not in ok: # pragma: no cover
raise httplib.HTTPException('Error %d (%s) from server:\n%s' %
(resp.status, resp.reason, text))
typ = resp.getheader('Content-Type') or 'application/octet-stream'
if typ == 'application/json' or typ.startswith('application/json;'):
return json.loads(text)
else:
return text
def list_jobs(self):
'''Return list of job ids on the server.'''
obj = self._do('GET', '/api/json')
logging.debug('obj: %s' % repr(obj))
return [job['name'] for job in obj['jobs']]
def delete_job(self, job_id):
'''Delete a job from Jenkins server, given job id.'''
self._do('POST', '/job/%s/doDelete' % job_id)
def create_job(self, job_id, config_xml):
'''Create a new job.
``config_xml`` is a string containing the XML format configuration
of the job.
'''
self._do('POST', '/createItem?name=%s' % job_id, body=config_xml,
headers={'Content-Type': 'text/xml'})
def update_job(self, job_id, config_xml):
'''Update the configuration of a new job.
``config_xml`` is a string containing the XML format configuration
of the job.
'''
self._do('POST', '/job/%s/config.xml' % job_id, body=config_xml,
headers={ 'Content-Type': 'text/xml' })
def get_job_config(self, job_id):
'''Return the XML configuration of a job.'''
return self._do('GET', '/job/%s/config.xml' % job_id)
def run_job(self, job_id): # pragma: no cover
'''Run an existing job.'''
self._do('POST', '/job/%s/build?delay=0sec' % job_id)
def get_latest_build_number(self, job_id): # pragma: no cover
'''Return number of latest finished build for a job.'''
obj = self._do('GET', '/job/%s/api/json' % job_id)
last = obj.get('lastBuild', None) or {}
return last.get('number', None)
def get_build_info(self, job_id, build_number): # pragma: no cover
'''Return information about a specific build for a job.'''
return self._do('GET', '/job/%s/%s/api/json' % (job_id, build_number))
|
