+import json
import logging
import urlparse
-import urllib2
-import urllib
import xml.etree.ElementTree as ET
from collections import defaultdict
from time import sleep
def get_build_triggerurl(self):
return "%s/build" % self.baseurl
- def invoke(self, securitytoken=None, block=False, skip_if_running=False, invoke_pre_check_delay=3, invoke_block_delay=15, params=None, cause=None):
+ @staticmethod
+ def _mk_json_from_build_parameters(build_params):
+ """
+ Build parameters must be submitted in a particular format - Key-Value pairs would be
+ far too simple, no no! Watch and read on and behold!
+ """
+ assert isinstance(build_params, dict), 'Build parameters must be a dict'
+ return {'parameter':[
+ {'name':k, 'value':v} for k,v in build_params.iteritems()
+ ]}
+
+ @staticmethod
+ def mk_json_from_build_parameters(build_params):
+ to_json_structure = Job._mk_json_from_build_parameters(build_params)
+ return json.dumps(to_json_structure)
+
+ def invoke(self, securitytoken=None, block=False, skip_if_running=False, invoke_pre_check_delay=3, invoke_block_delay=15, build_params=None, cause=None):
assert isinstance( invoke_pre_check_delay, (int, float) )
assert isinstance( invoke_block_delay, (int, float) )
assert isinstance( block, bool )
assert isinstance( skip_if_running, bool )
# Either copy the params dict or make a new one.
- params = params and dict(params.items()) or {}
+ build_params = build_params and dict(build_params.items()) or {} # Via POSTed JSON
+ params = {} # Via Get string
if self.is_queued():
raise WillNotBuild('%s is already queued' % repr(self))
url = self.get_build_triggerurl()
if cause:
- params['cause'] = cause
+ build_params['cause'] = cause
if securitytoken:
- params['securitytoken'] = securitytoken
+ params['token'] = securitytoken
- response = self.jenkins.requester.post_url(url, params, data='')
+ response = self.jenkins.requester.post_and_confirm_status(
+ url,
+ data=self.mk_json_from_build_parameters(build_params), # See above - build params have to be JSON encoded & posted.
+ params=params
+ )
assert len( response.text ) > 0
if invoke_pre_check_delay > 0:
sleep( invoke_pre_check_delay )
if block:
total_wait = 0
+
while self.is_queued():
log.info( "Waited %is for %s to begin...", total_wait, self.name )
sleep( invoke_block_delay )
if self.is_running():
running_build = self.get_last_build()
running_build.block_until_complete( delay=invoke_pre_check_delay )
+
assert self.get_last_buildnumber() > original_build_no, "Job does not appear to have run."
else:
if self.is_queued():
return []
return upstream_jobs
+ def is_enabled(self):
+ self.poll()
+ return self._data["color"] != 'disabled'
+
def disable(self):
'''Disable job'''
- disableurl = urlparse.urljoin(self.baseurl, 'disable' )
- return self.post_data(disableurl, '')
+ url = "%s/disable" % self.baseurl
+ return self.get_jenkins_obj().requester.post_url(url, data='')
def enable(self):
'''Enable job'''
- enableurl = urlparse.urljoin(self.baseurl, 'enable' )
- return self.post_data(enableurl, '')
+ url = "%s/enable" % self.baseurl
+ return self.get_jenkins_obj().requester.post_url(url, data='')
def delete_from_queue(self):
"""
if not self.is_queued():
raise NotInQueue()
queue_id = self._data['queueItem']['id']
- cancelurl = urlparse.urljoin(self.get_jenkins_obj().get_queue().baseurl,
+ url = urlparse.urljoin(self.get_jenkins_obj().get_queue().baseurl,
'cancelItem?id=%s' % queue_id)
- try:
- self.post_data(cancelurl, '')
- except urllib2.HTTPError:
- # The request doesn't have a response, so it returns 404,
- # it's the expected behaviour
- pass
+ self.get_jenkins_obj().requester.post_and_confirm_status(url, data='')
return True
def get_params(self):
self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
self.assertJobIsPresent(job_name)
+ def test_enable_disable_job(self):
+ job_name = 'create_%s' % random_string()
+ self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
+ self.assertJobIsPresent(job_name)
+
+ j = self.jenkins[job_name]
+ j.invoke(block=True) # run this at least once
+
+ j.disable()
+ self.assertEquals(j.is_enabled(), False, 'A disabled job is reporting incorrectly')
+ j.enable()
+ self.assertEquals(j.is_enabled(), True, 'An enabled job is reporting incorrectly')
+
def test_get_job_and_update_config(self):
job_name = 'config_%s' % random_string()
self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
--- /dev/null
+'''
+System tests for `jenkinsapi.jenkins` module.
+'''
+import os
+import time
+import shutil
+import random
+import tempfile
+import unittest
+from jenkinsapi_tests.test_utils.random_strings import random_string
+from jenkinsapi_tests.systests.base import BaseSystemTest
+
+JOB_CONFIG = """
+<?xml version='1.0' encoding='UTF-8'?>
+<project>
+ <actions/>
+ <description>A build that explores the wonderous possibilities of parameterized builds.</description>
+ <keepDependencies>false</keepDependencies>
+ <properties>
+ <hudson.model.ParametersDefinitionProperty>
+ <parameterDefinitions>
+ <hudson.model.StringParameterDefinition>
+ <name>B</name>
+ <description>B, like buzzing B.</description>
+ <defaultValue></defaultValue>
+ </hudson.model.StringParameterDefinition>
+ </parameterDefinitions>
+ </hudson.model.ParametersDefinitionProperty>
+ </properties>
+ <scm class="hudson.scm.NullSCM"/>
+ <canRoam>true</canRoam>
+ <disabled>false</disabled>
+ <blockBuildWhenDownstreamBuilding>false</blockBuildWhenDownstreamBuilding>
+ <blockBuildWhenUpstreamBuilding>false</blockBuildWhenUpstreamBuilding>
+ <triggers class="vector"/>
+ <concurrentBuild>false</concurrentBuild>
+ <builders>
+ <hudson.tasks.Shell>
+ <command>ping -c 1 localhost | tee out.txt
+cat $A > a.txt
+cat $B > b.txt</command>
+ </hudson.tasks.Shell>
+ </builders>
+ <publishers>
+ <hudson.tasks.ArtifactArchiver>
+ <artifacts></artifacts>
+ <latestOnly>false</latestOnly>
+ </hudson.tasks.ArtifactArchiver>
+ <hudson.tasks.Fingerprinter>
+ <targets></targets>
+ <recordBuildArtifacts>false</recordBuildArtifacts>
+ </hudson.tasks.Fingerprinter>
+ </publishers>
+ <buildWrappers/>
+</project>""".strip()
+
+class TestParameterizedBuilds(BaseSystemTest):
+
+ def test_invoke_job_parameterized(self):
+
+ param_B = random_string()
+
+ job_name = 'create_%s' % random_string()
+ job = self.jenkins.create_job(job_name, JOB_CONFIG)
+ job.invoke(block=True, build_params={ 'B':param_B})
+
+ b = job.get_last_build()
+ while b.is_running():
+ time.sleep(0.25)
+
+ artifacts = b.get_artifact_dict()
+ self.assertIsInstance(artifacts, dict)
+
+ artA = artifacts['A.txt']
+ artB = artifacts['B.txt']
+
+ # TODO: Actually verify the download
+
+
+if __name__ == '__main__':
+ unittest.main()
projects / tools / python-jenkinsapi.git / commitdiff