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
|
# simplejenkinsapi/api_tests.py -- unit test for simplejenkinsapi/api.py
#
# 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/>.
import httplib
import json
import StringIO
import unittest
import simplejenkinsapi
class DummyResponse(object):
def __init__(self, body, content_type):
self.status = httplib.OK
self.reason = 'yeah yeah'
self.body = body
self.content_type = content_type
def getheader(self, header):
if header == 'Content-Type':
return self.content_type
else:
return ''
def read(self):
return self.body
class DummyServer(object):
def __init__(self):
self.routes = {}
self.add_route('GET', '/api/json', self.toplevel)
self.add_route('POST', '/job/job-1/doDelete', self.delete_job_1)
self.add_route('POST', '/createItem?name=job-1', self.create_job_1)
self.add_route('GET', '/job/job-1/config.xml', self.config_job_1)
self.add_route('POST', '/job/job-1/config.xml', self.update_job_1)
self.jobs = {}
self.response = None
def add_route(self, method, path, callback):
key = (method, path)
assert key not in self.routes
self.routes[key] = callback
def toplevel(self, body, headers):
assert body == ''
assert headers == {}
result = []
for job_id in sorted(self.jobs.keys()):
result.append({
'name': job_id,
})
obj = { 'jobs': result }
return json.dumps(obj), 'application/json'
def create_job_1(self, body, headers):
assert 'job-1' not in self.jobs
self.jobs['job-1'] = body
return '', 'text/plain'
def update_job_1(self, body, headers):
assert 'job-1' in self.jobs
self.jobs['job-1'] = body
return '', 'text/plain'
def delete_job_1(self, body, headers):
del self.jobs['job-1']
return '', 'text/plain'
def config_job_1(self, body, headers):
return self.jobs['job-1'], 'application/xml'
def request(self, method, path, body=None, headers=None):
key = (method, path)
assert key in self.routes
self.response, self.response_type = self.routes[key](body, headers)
def getresponse(self):
return DummyResponse(self.response, self.response_type)
class JenkinsTests(unittest.TestCase):
def setUp(self):
self.url = 'http://does.not.exist:8080/'
self.jenkins = simplejenkinsapi.Jenkins(self.url)
self.dummy_server = DummyServer()
self.jenkins._connect = lambda: self.dummy_server
self.config_xml = '''
<?xml version="1.0" encoding="UTF-8"?>
<project>
</project>
'''
self.config_xml_2 = '''
<?xml version="1.0" encoding="UTF-8"?>
<project>
<foo/>
</project>
'''
def test_lists_no_jobs_when_there_are_none(self):
self.assertEqual(self.jenkins.list_jobs(), [])
def test_lists_single_job_when_one_exists(self):
self.dummy_server.jobs = { 'job-1': '' }
self.assertEqual(self.jenkins.list_jobs(), ['job-1'])
def test_deletes_existing_job(self):
self.dummy_server.jobs = { 'job-1': '' }
self.jenkins.delete_job('job-1')
self.assertEqual(self.jenkins.list_jobs(), [])
def test_creates_job(self):
self.jenkins.create_job('job-1', self.config_xml)
self.assertEqual(self.jenkins.list_jobs(), ['job-1'])
self.assertEqual(self.jenkins.get_job_config('job-1'), self.config_xml)
def test_updates_job(self):
self.jenkins.create_job('job-1', self.config_xml)
self.jenkins.update_job('job-1', self.config_xml_2)
self.assertEqual(self.jenkins.get_job_config('job-1'),
self.config_xml_2)
|
