+import logging
+logging.basicConfig()
+
from jenkinsapi.jenkins import Jenkins
+from pkg_resources import resource_string
J = Jenkins('http://localhost:8080')
+jobName = 'foo_job2'
+xml = resource_string('examples', 'addjob.xml')
-EMPTY_JOB_CONFIG = '''\
-<?xml version='1.0' encoding='UTF-8'?>
-<project>
- <actions/>
- <description></description>
- <keepDependencies>false</keepDependencies>
- <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/>
- <publishers/>
- <buildWrappers/>
-</project>
-'''
+print xml
-new_job = J.create_job(jobname='foo_job', config=EMPTY_JOB_CONFIG)
+j = J.create_job(jobname=jobName, config=xml)
-j= J['foo_job']
-print j
\ No newline at end of file
+j2= J[jobName]
+print j
:param nodename: string, hostname
:return: boolean
"""
+ self.poll()
return nodename in self.get_node_dict()
def delete_node(self, nodename):
assert self.has_node(nodename), "This node: %s is not registered as a slave" % nodename
assert nodename != "master", "you cannot delete the master node"
url = "%s/doDelete" % self.get_node_url(nodename)
- fn_urlopen = self.get_jenkins_obj().get_opener()
- try:
- fn_urlopen(url).read()
- except urllib2.HTTPError, e:
- log.debug("Error reading %s" % url)
- log.exception(e)
- raise
- return not self.has_node(nodename)
+ self.requester.get_and_confirm_status(url)
def create_node(self, name, num_executors=2, node_description=None,
remote_fs='/var/lib/jenkins', labels=None, exclusive=False):
pprint.pprint(self._data)
def __str__(self):
- raise NotImplemented
+ raise NotImplementedError
def __init__(self, baseurl, poll=True):
"""
Initialize a jenkins connection
"""
+ self._data = None
self.baseurl = self.strip_trailing_slash(baseurl)
if poll:
- try:
- self.poll()
- except urllib2.HTTPError, hte: #TODO: Wrong exception
- log.exception(hte)
- log.warn( "Failed to connect to %s" % baseurl )
- raise
+ self.poll()
def __eq__(self, other):
"""
try:
return eval(response.text)
except Exception:
- log.exception('Inappropriate content found at %s' % url)
+ log.exception('Inappropriate content found at %s', url)
raise JenkinsAPIException('Cannot parse %s' % url)
@classmethod
import os
from jenkinsapi_utils.jenkins_launcher import JenkinsLancher
-systests_dir, _ = os.path.split(__file__)
-war_path = os.path.join(systests_dir, 'jenkins.war' )
-launcher = JenkinsLancher(war_path)
+state={}
def setUpPackage():
- launcher.start()
+ systests_dir, _ = os.path.split(__file__)
+ war_path = os.path.join(systests_dir, 'jenkins.war' )
+ state['launcher'] = JenkinsLancher(war_path)
+
+ state['launcher'].start()
def tearDownPackage():
- launcher.stop()
+ state['launcher'].stop()
System tests for `jenkinsapi.jenkins` module.
'''
import unittest
-from jenkinsapi.build import Build
from jenkinsapi_tests.test_utils.random_strings import random_string
from jenkinsapi_tests.systests.base import BaseSystemTest, EMPTY_JOB_CONFIG
import unittest
from jenkinsapi.jenkins import Jenkins
from jenkinsapi_tests.systests.base import BaseSystemTest
+from jenkinsapi_tests.test_utils.random_strings import random_string
log = logging.getLogger(__name__)
class TestNodes(BaseSystemTest):
def test_invoke_job_parameterized(self):
-
+ node_name = random_string()
J = Jenkins('http://localhost:8080')
- J.create_node('test')
+ J.create_node(node_name)
+ self.assertTrue(J.has_node(node_name))
+
+ N = J.get_node(node_name)
+ self.assertEquals(N.baseurl, J.get_node_url(node_name))
+
+ J.delete_node(node_name)
+ self.assertFalse(J.has_node(node_name))
if __name__ == '__main__':
logging.basicConfig()