--- /dev/null
+"""
+Save this file as server.py
+>>> python server.py 0.0.0.0 8001
+serving on 0.0.0.0:8001
+
+or simply
+
+>>> python server.py
+Serving on localhost:8000
+
+You can use this to test GET and POST methods.
+
+"""
+
+import SimpleHTTPServer
+import SocketServer
+import logging
+import cgi
+
+import sys
+
+
+
+PORT = 8080
+I = "localhost"
+
+
+class ServerHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
+
+ def do_GET(self):
+ logging.warning("======= GET STARTED =======")
+ logging.warning(self.headers)
+ SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
+
+ def do_POST(self):
+ logging.warning("======= POST STARTED =======")
+ logging.warning(self.headers)
+ form = cgi.FieldStorage(
+ fp=self.rfile,
+ headers=self.headers,
+ environ={'REQUEST_METHOD':'POST',
+ 'CONTENT_TYPE':self.headers['Content-Type'],
+ })
+ logging.warning("======= POST VALUES =======")
+ for item in form.list:
+ logging.warning(item)
+ logging.warning("\n")
+ SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
+
+Handler = ServerHandler
+
+httpd = SocketServer.TCPServer(("", PORT), Handler)
+
+print "Serving at: http://%(interface)s:%(port)s" % dict(interface=I or "localhost", port=PORT)
+httpd.serve_forever()
\ No newline at end of file
self._element_tree = ET.fromstring(self._config)
return self._element_tree
- def get_build_triggerurl(self):
- if not self.has_params():
+ def get_build_triggerurl(self, files):
+ if files or (not self.has_params()):
+ # If job has file parameters - it must be triggered
+ # using "/build", not by "/buildWithParameters"
+ # "/buildWithParameters" will ignore non-file parameters
return "%s/build" % self.baseurl
return "%s/buildWithParameters" % self.baseurl
assert isinstance(
build_params, dict), 'Build parameters must be a dict'
- build_p = [{'name': k, 'value': v}
- for k, v in build_params.items()]
+ build_p = [{'name': k, 'value': str(v)}
+ for k, v in sorted(build_params.items())]
out = {'parameter': build_p}
if file_params:
file_p = [{'name': k, 'file': k}
for k in file_params.keys()]
out['parameter'].extend(file_p)
+ if len(out['parameter']) == 1:
+ out['parameter'] = out['parameter'][0]
+
return out
@staticmethod
def mk_json_from_build_parameters(build_params, file_params=None):
- to_json_structure = Job._mk_json_from_build_parameters(build_params,
- file_params)
- return json.dumps(to_json_structure)
+ json_structure = Job._mk_json_from_build_parameters(
+ build_params,
+ file_params
+ )
+ json_structure['statusCode'] = "303"
+ json_structure['redirectTo'] = "."
+ return json.dumps(json_structure)
def invoke(self, securitytoken=None, block=False, build_params=None, cause=None, files=None, delay=5):
assert isinstance(block, bool)
build_params = build_params and dict(
build_params.items()) or {} # Via POSTed JSON
- url = self.get_build_triggerurl()
+ url = self.get_build_triggerurl(files)
if cause:
build_params['cause'] = cause
-
+
+ # Build require params as form fields
+ # and as Json.
data = {'json': self.mk_json_from_build_parameters(
build_params, files)}
+ data.update(build_params)
response = self.jenkins.requester.post_url(
url,
redirect_url = response.headers['location']
- # It's possible that an error triggering the build will cause Jenkins
- # not build, the signal is that we will be redirected to something
- # other than a QueueItem URL.
if not redirect_url.startswith("%s/queue/item" % self.jenkins.baseurl):
raise ValueError("Not a Queue URL: %s" % redirect_url)
art_file = artifacts['file.txt']
self.assertTrue(art_file.get_data().strip(), file_data)
-# def test_invoke_job_parameterized(self):
-# param_B = random_string()
-#
-# job_name = 'create2_%s' % random_string()
-# job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
-# job.invoke(block=True, build_params={'B': param_B})
-#
-# build = job.get_last_build()
-# while build.is_running():
-# time.sleep(0.25)
-#
-# artifacts = build.get_artifact_dict()
-# self.assertIsInstance(artifacts, dict)
-# artB = artifacts['b.txt']
-# self.assertTrue(artB.get_data().strip(), param_B)
-#
-# self.assertIn(param_B, build.get_console())
-#
-# def test_parameterized_job_build_queuing(self):
-# """Accept multiple builds of parameterized jobs with unique
-# parameters."""
-# job_name = 'create_%s' % random_string()
-# job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
-#
-# for i in range(3):
-# param_B = random_string()
-# params = {'B': param_B}
-# job.invoke(build_params=params)
-# time.sleep(0.25)
-#
-# self.assertTrue(job.has_queued_build(params))
-#
-# while job.has_queued_build(params):
-# time.sleep(0.25)
-#
-# build = job.get_last_build()
-# while build.is_running():
-# time.sleep(0.25)
-#
-# artifacts = build.get_artifact_dict()
-# self.assertIsInstance(artifacts, dict)
-# artB = artifacts['b.txt']
-# self.assertTrue(artB.get_data().strip(), param_B)
-#
-# self.assertIn(param_B, build.get_console())
-#
-# def test_parameterized_multiple_builds_get_the_same_queue_item(self):
-# """Multiple attempts to run the same parameteized
-# build will get the same queue item."""
-# job_name = 'create_%s' % random_string()
-# job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
-#
-# for i in range(3):
-# params = {'B': random_string()}
-# qq0 = job.invoke(build_params=params)
-#
-#
-# qq1 = job.invoke(build_params=params)
-# self.assertEqual(qq0, qq1)
-#
-# def test_invoke_job_with_file_and_params(self):
-# file_data = random_string()
-# param_data = random_string()
-# param_file = StringIO(file_data)
-#
-# job_name = 'create_%s' % random_string()
-# job = self.jenkins.create_job(job_name, JOB_WITH_FILE_AND_PARAMS)
-# job.invoke(block=True, files={'file.txt': param_file},
-# build_params={'B': param_data})
-#
-# build = job.get_last_build()
-# while build.is_running():
-# time.sleep(0.25)
-#
-# artifacts = build.get_artifact_dict()
-# self.assertIsInstance(artifacts, dict)
-# art_file = artifacts['file.txt']
-# self.assertTrue(art_file.get_data().strip(), file_data)
-# art_param = artifacts['file1.txt']
-# self.assertTrue(art_param.get_data().strip(), param_data)
+ def test_invoke_job_parameterized(self):
+ param_B = random_string()
+
+ job_name = 'create2_%s' % random_string()
+ job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
+ job.invoke(block=True, build_params={'B': param_B})
+ build = job.get_last_build()
+
+ artifacts = build.get_artifact_dict()
+ artB = artifacts['b.txt']
+ self.assertEqual(
+ artB.get_data().strip(),
+ param_B,
+ )
+
+ self.assertIn(param_B, build.get_console())
+
+ def test_parameterized_job_build_queuing(self):
+ """Accept multiple builds of parameterized jobs with unique
+ parameters."""
+ job_name = 'create_%s' % random_string()
+ job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
+
+ for i in range(3):
+ param_B = random_string()
+ params = {'B': param_B}
+ job.invoke(build_params=params)
+ time.sleep(0.25)
+
+ self.assertTrue(job.has_queued_build(params))
+
+ while job.has_queued_build(params):
+ time.sleep(0.25)
+
+ build = job.get_last_build()
+ while build.is_running():
+ time.sleep(0.25)
+
+ artifacts = build.get_artifact_dict()
+ self.assertIsInstance(artifacts, dict)
+ artB = artifacts['b.txt']
+ self.assertTrue(artB.get_data().strip(), param_B)
+
+ self.assertIn(param_B, build.get_console())
+
+ def test_parameterized_multiple_builds_get_the_same_queue_item(self):
+ """Multiple attempts to run the same parameteized
+ build will get the same queue item."""
+ job_name = 'create_%s' % random_string()
+ job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
+
+ for i in range(3):
+ params = {'B': random_string()}
+ qq0 = job.invoke(build_params=params)
+
+
+ qq1 = job.invoke(build_params=params)
+ self.assertEqual(qq0, qq1)
+
+ def test_invoke_job_with_file_and_params(self):
+ file_data = random_string()
+ param_data = random_string()
+ param_file = StringIO(file_data)
+
+ job_name = 'create_%s' % random_string()
+ job = self.jenkins.create_job(job_name, JOB_WITH_FILE_AND_PARAMS)
+ job.invoke(
+ block=True,
+ files={'file.txt': param_file},
+ build_params={'B': param_data}
+ )
+ build = job.get_last_build()
+ artifacts = build.get_artifact_dict()
+ self.assertIsInstance(artifacts, dict)
+ art_file = artifacts['file.txt']
+ self.assertTrue(art_file.get_data().strip(), file_data)
+ art_param = artifacts['file1.txt']
+ self.assertTrue(art_param.get_data().strip(), param_data)
if __name__ == '__main__':
import mock
+import json
# To run unittests on python 2.6 please use unittest2 library
try:
import unittest2 as unittest
self.assertEquals(
str(ar.exception), 'Build parameters must be a dict')
- def test__mk_json_from_build_parameters(self):
- params = {'param1': 'value1', 'param2': 'value2'}
- ret = self.j.mk_json_from_build_parameters(build_params=params)
- self.assertTrue(isinstance(ret, str))
- try:
- self.assertItemsEqual(ret,
- '{"parameter": [{"name": "param2", "value": "value2"}, {"name": "param1", "value": "value1"}]}')
- except AttributeError:
- self.assertCountEqual(ret,
- '{"parameter": [{"name": "param2", "value": "value2"}, {"name": "param1", "value": "value1"}]}')
-
- def test_wrong_mk_json_from_build_parameters(self):
- with self.assertRaises(AssertionError) as ar:
- self.j.mk_json_from_build_parameters(build_params='bad parameter')
-
- self.assertEquals(
- str(ar.exception), 'Build parameters must be a dict')
-
@mock.patch.object(JenkinsBase, 'get_data', fakeGetData)
def test_wrong_field__build_id_for_type(self):
with self.assertRaises(AssertionError):
self.assertIsInstance(params, list)
self.assertEquals(len(params), 2)
self.assertEquals(params, ['param1', 'param2'])
-
+
def test_get_build(self):
buildnumber = 1
with mock.patch('jenkinsapi.job.Build') as build_mock:
build_mock.assert_called_with('http://halob:8080/job/foo/1/',
buildnumber, job=self.j, depth=0)
+ def assertJsonEqual(self, jsonA, jsonB, msg=None):
+ A = json.loads(jsonA)
+ B = json.loads(jsonB)
+ self.assertEqual(
+ A,
+ B,
+ msg
+ )
+
+ def test_get_json_for_single_param(self):
+ params = {"B": "one two three"}
+ expected = '{"parameter": {"name": "B", "value": "one two three"}, "statusCode": "303", "redirectTo": "."}'
+ self.assertJsonEqual(
+ Job.mk_json_from_build_parameters(params),
+ expected
+ )
+
+ def test_get_json_for_many_params(self):
+ params = {"B": "Honey", "A": "Boo", "C": 2}
+ expected = '{"parameter": [{"name": "A", "value": "Boo"}, {"name": "B", "value": "Honey"}, {"name": "C", "value": "2"}], "statusCode": "303", "redirectTo": "."}'
+
+ self.assertJsonEqual(
+ Job.mk_json_from_build_parameters(params),
+ expected
+ )
+
+ def test__mk_json_from_build_parameters(self):
+ params = {'param1': 'value1', 'param2': 'value2'}
+ result = self.j._mk_json_from_build_parameters(build_params=params)
+ self.assertTrue(isinstance(result, dict))
+
+ self.assertEquals(
+ result,
+ {"parameter": [{"name": "param1", "value": "value1"}, {
+ "name": "param2", "value": "value2"}]}
+ )
+
+ def test_wrong_mk_json_from_build_parameters(self):
+ with self.assertRaises(AssertionError) as ar:
+ self.j.mk_json_from_build_parameters(build_params='bad parameter')
+
+ self.assertEquals(
+ str(ar.exception), 'Build parameters must be a dict')
+
if __name__ == '__main__':
unittest.main()
projects / tools / python-jenkinsapi.git / commitdiff