"""
pass
+
class NotFound(JenkinsAPIException):
"""
Resource cannot be found
Module for MutableJenkinsThing
"""
+
class MutableJenkinsThing(object):
"""
A mixin for certain mutable objects which can be renamed and deleted.
Module for jenkinsapi Plugin
"""
+
class Plugin(object):
"""
Plugin class
Module for jenkinsapi Result
"""
+
class Result(object):
"""
Result class
import os
from jenkinsapi_utils.jenkins_launcher import JenkinsLancher
-state={}
+state = {}
# Extra plugins required by the systests
PLUGIN_DEPENDENCIES = ["http://updates.jenkins-ci.org/latest/git.hpi",
"http://updates.jenkins-ci.org/latest/git-client.hpi",
"https://updates.jenkins-ci.org/latest/nested-view.hpi"]
-
+
+
def setUpPackage():
systests_dir, _ = os.path.split(__file__)
- war_path = os.path.join(systests_dir, 'jenkins.war' )
+ war_path = os.path.join(systests_dir, 'jenkins.war')
state['launcher'] = JenkinsLancher(war_path, PLUGIN_DEPENDENCIES)
state['launcher'].start()
+
def tearDownPackage():
state['launcher'].stop()
ii = job.invoke()
ii.block(until='completed')
build_number = ii.get_build_number()
- self.assertEquals(build_number, invocation+1)
+ self.assertEquals(build_number, invocation +1)
if __name__ == '__main__':
from jenkinsapi_tests.test_utils.random_strings import random_string
-
class JobTests(BaseSystemTest):
def test_create_job(self):
self.assertJobIsPresent(job_name)
j = self.jenkins[job_name]
- j.invoke(block=True) # run this at least once
+ j.invoke(block=True) # run this at least once
j.disable()
self.assertEquals(j.is_enabled(), False, 'A disabled job is reporting incorrectly')
# Verify that we can hande binary artifacts
binary_artifact.save_to_dir(tempDir)
readBackText = gzip.open(os.path.join(
- tempDir,
+ tempDir,
binary_artifact.filename,
- ), 'rb' ).read().strip()
+ ), 'rb').read().strip()
self.assertTrue(re.match(r'^PING \S+ \(127.0.0.1\)', readBackText))
self.assertTrue(readBackText.endswith('ms'))
finally:
m = re.search(u'\xbb (.*) #\\d+$', r.name)
self.assertIsNotNone(m)
s.add(m.group(1))
- self.assertEqual(s, {'one','two','three'})
+ self.assertEqual(s, {'one', 'two', 'three'})
if __name__ == '__main__':
unittest.main()
Can we flip the online / offline state of the master node.
"""
# Master node name should be case insensitive
- #mn0 = self.jenkins.get_node('MaStEr')
+ # mn0 = self.jenkins.get_node('MaStEr')
mn = self.jenkins.get_node('master')
- #self.assertEquals(mn, mn0)
+ # self.assertEquals(mn, mn0)
- mn.set_online() # It should already be online, hence no-op
+ mn.set_online() # It should already be online, hence no-op
self.assertTrue(mn.is_online())
- mn.set_offline() # We switch that suckah off
- mn.set_offline() # This should be a no-op
+ mn.set_offline() # We switch that suckah off
+ mn.set_offline() # This should be a no-op
self.assertFalse(mn.is_online())
- mn.set_online() # Switch it back on
+ mn.set_online() # Switch it back on
self.assertTrue(mn.is_online())
log = logging.getLogger(__name__)
+
class TestQueue(BaseSystemTest):
"""
All kinds of testing on Jenkins Queues
"""
- #TODO: Test timeout behavior
+ # TODO: Test timeout behavior
def test_get_queue(self):
qq = self.jenkins.get_queue()
self.assertEquals(len(queue), 0)
-
def test_start_and_stop_long_running_job(self):
job_name = random_string()
j = self.jenkins.create_job(job_name, LONG_RUNNING_JOB)
# if __name__ == '__main__':
# unittest.main()
-
import random
import string
+
def random_string(length=10):
- return ''.join( random.choice(string.ascii_lowercase) for i in range(length) )
+ return ''.join(random.choice(string.ascii_lowercase) for i in range(length))
if __name__ == '__main__':
print random_string()
import unittest
from jenkinsapi.build import Build
+
class test_build(unittest.TestCase):
DATA = {'actions': [{'causes': [{'shortDescription': 'Started by an SCM change'}]},
'timestamp': 1372553675652,
'url': 'http://localhost:8080/job/git_yssrtigfds/3/'}
-
@mock.patch.object(Build, '_poll')
def setUp(self, _poll):
_poll.return_value = self.DATA
self.assertEquals(self.b.get_revision(), '7def9ed6e92580f37d00e4980c36c4d36e68f702')
if __name__ == '__main__':
- unittest.main()
\ No newline at end of file
+ unittest.main()
username='foouser', password='foopassword')
job = J.create_job('job_one', None)
self.assertTrue(isinstance(job, Job))
- self.assertTrue(job.baseurl=='http://localhost:8080/job_one')
- self.assertTrue(job.name=='job_one')
+ self.assertTrue(job.baseurl == 'http://localhost:8080/job_one')
+ self.assertTrue(job.name == 'job_one')
# Here we're going to test function, which is going to modify
# Jenkins internal data. It calls for data once to check
job = J.create_job('job_new', None)
self.assertTrue(isinstance(job, Job))
- self.assertTrue(job.baseurl=='http://localhost:8080/job_new')
- self.assertTrue(job.name=='job_new')
+ self.assertTrue(job.baseurl == 'http://localhost:8080/job_new')
+ self.assertTrue(job.name == 'job_new')
@mock.patch.object(JenkinsBase, '_poll')
@mock.patch.object(Jenkins, '_poll')
def test_has_plugin(self, _p_poll, _poll):
_poll.return_value = {}
_p_poll.return_value = {'plugins': [
- {'deleted': False, 'hasUpdate': True, 'downgradable': False,
- 'dependencies': [{}, {}, {}, {}],
- 'longName': 'Jenkins Subversion Plug-in', 'active': True,
- 'shortName': 'subversion', 'backupVersion': None,
+ {'deleted': False, 'hasUpdate': True, 'downgradable': False,
+ 'dependencies': [{}, {}, {}, {}],
+ 'longName': 'Jenkins Subversion Plug-in', 'active': True,
+ 'shortName': 'subversion', 'backupVersion': None,
'url': 'http://wiki.jenkins-ci.org/display/JENKINS/Subversion+Plugin',
- 'enabled': True, 'pinned': False, 'version': '1.45',
+ 'enabled': True, 'pinned': False, 'version': '1.45',
'supportsDynamicLoad': 'MAYBE', 'bundled': True}]}
J = Jenkins('http://localhost:8080/',
class TestJenkinsBaseMixin(object):
- """
- Tests which apply to all or most Jenkins objects
- """
- pass
\ No newline at end of file
+ """
+ Tests which apply to all or most Jenkins objects
+ """
+ pass
"iconUrl": "health-80plus.png", "score": 100}],
"inQueue": False,
"keepDependencies": False,
- "lastBuild": {"number": 4, "url": "http://halob:8080/job/foo/4/"}, # build running
+ "lastBuild": {"number": 4, "url": "http://halob:8080/job/foo/4/"}, # build running
"lastCompletedBuild": {"number": 3, "url": "http://halob:8080/job/foo/3/"},
"lastFailedBuild": None,
"lastStableBuild": {"number": 3, "url": "http://halob:8080/job/foo/3/"},
"scm": {},
"upstreamProjects": []}
- URL_DATA = {'http://halob:8080/job/foo/%s' % config.JENKINS_API:JOB_DATA}
+ URL_DATA = {'http://halob:8080/job/foo/%s' % config.JENKINS_API: JOB_DATA}
def fakeGetData(self, url, *args):
try:
"iconUrl": "health-80plus.png", "score": 100}],
"inQueue": False,
"keepDependencies": False,
- "lastBuild": {"number": 4, "url": "http://halob:8080/job/foo/4/"}, # build running
+ "lastBuild": {"number": 4, "url": "http://halob:8080/job/foo/4/"}, # build running
"lastCompletedBuild": {"number": 3, "url": "http://halob:8080/job/foo/3/"},
"lastFailedBuild": None,
"lastStableBuild": {"number": 3, "url": "http://halob:8080/job/foo/3/"},
"iconUrl": "health-80plus.png", "score": 100}],
"inQueue": False,
"keepDependencies": False,
- "lastBuild": {"number": 4, "url": "http://halob:8080/job/fullfoo/4/"}, # build running
+ "lastBuild": {"number": 4, "url": "http://halob:8080/job/fullfoo/4/"}, # build running
"lastCompletedBuild": {"number": 3, "url": "http://halob:8080/job/fullfoo/3/"},
"lastFailedBuild": None,
"lastStableBuild": {"number": 3, "url": "http://halob:8080/job/fullfoo/3/"},
(JOB3_API_URL, str({'tree': 'allBuilds[number,url]'})): JOB3_ALL_BUILDS_DATA,
}
-
def fakeGetData(self, url, params=None):
TestJobGetAllBuilds.__get_data_call_count += 1
if params is None:
'views': [{'name': 'All', 'url': 'http://halob:8080/'},
{'name': 'FodFanFo', 'url': 'http://halob:8080/view/FodFanFo/'}]}
-
DATA1 = {'busyExecutors': 0,
'computer': [{'actions': [],
'displayName': 'master',
'oneOffExecutors': [],
'temporarilyOffline': False}
- DATA3= { 'actions': [],
+ DATA3 = {'actions': [],
'displayName': 'halob',
'executors': [{}],
'icon': 'computer-x.png',
'oneOffExecutors': [],
'temporarilyOffline': False}
-
@mock.patch.object(Jenkins, '_poll')
@mock.patch.object(Nodes, '_poll')
def setUp(self, _poll_nodes, _poll_jenkins):
self.J = Jenkins('http://localhost:8080')
self.ns = self.J.get_nodes()
- #self.ns = Nodes('http://localhost:8080/computer', 'bobnit', self.J)
+ # self.ns = Nodes('http://localhost:8080/computer', 'bobnit', self.J)
def testRepr(self):
# Can we produce a repr string for this object
class TestPlugins(unittest.TestCase):
DATA = {'plugins': [
- {'deleted': False, 'hasUpdate': True, 'downgradable': False,
- 'dependencies': [{}, {}, {}, {}],
- 'longName': 'Jenkins Subversion Plug-in', 'active': True,
- 'shortName': 'subversion', 'backupVersion': None,
+ {'deleted': False, 'hasUpdate': True, 'downgradable': False,
+ 'dependencies': [{}, {}, {}, {}],
+ 'longName': 'Jenkins Subversion Plug-in', 'active': True,
+ 'shortName': 'subversion', 'backupVersion': None,
'url': 'http://wiki.jenkins-ci.org/display/JENKINS/Subversion+Plugin',
- 'enabled': True, 'pinned': False, 'version': '1.45',
- 'supportsDynamicLoad': 'MAYBE', 'bundled': True},
- {'deleted': False, 'hasUpdate': True, 'downgradable': False,
- 'dependencies': [{}, {}], 'longName': 'Maven Integration plugin',
- 'active': True, 'shortName': 'maven-plugin', 'backupVersion': None,
- 'url': 'http://wiki.jenkins-ci.org/display/JENKINS/Maven+Project+Plugin',
- 'enabled': True, 'pinned': False, 'version': '1.521',
+ 'enabled': True, 'pinned': False, 'version': '1.45',
+ 'supportsDynamicLoad': 'MAYBE', 'bundled': True},
+ {'deleted': False, 'hasUpdate': True, 'downgradable': False,
+ 'dependencies': [{}, {}], 'longName': 'Maven Integration plugin',
+ 'active': True, 'shortName': 'maven-plugin', 'backupVersion': None,
+ 'url': 'http://wiki.jenkins-ci.org/display/JENKINS/Maven+Project+Plugin',
+ 'enabled': True, 'pinned': False, 'version': '1.521',
'supportsDynamicLoad': 'MAYBE', 'bundled': True}
]}
def test_plugins_values(self, _poll_plugins):
_poll_plugins.return_value = self.DATA
- p = Plugin({'deleted': False, 'hasUpdate': True, 'downgradable': False,
- 'dependencies': [{}, {}, {}, {}],
- 'longName': 'Jenkins Subversion Plug-in', 'active': True,
- 'shortName': 'subversion', 'backupVersion': None,
+ p = Plugin({'deleted': False, 'hasUpdate': True, 'downgradable': False,
+ 'dependencies': [{}, {}, {}, {}],
+ 'longName': 'Jenkins Subversion Plug-in', 'active': True,
+ 'shortName': 'subversion', 'backupVersion': None,
'url': 'http://wiki.jenkins-ci.org/display/JENKINS/Subversion+Plugin',
- 'enabled': True, 'pinned': False, 'version': '1.45',
+ 'enabled': True, 'pinned': False, 'version': '1.45',
'supportsDynamicLoad': 'MAYBE', 'bundled': True})
plugins = self.J.get_plugins().values()
def test_plugin_get_by_name(self, _poll_plugins):
_poll_plugins.return_value = self.DATA
- p = Plugin({'deleted': False, 'hasUpdate': True, 'downgradable': False,
- 'dependencies': [{}, {}, {}, {}],
- 'longName': 'Jenkins Subversion Plug-in', 'active': True,
- 'shortName': 'subversion', 'backupVersion': None,
+ p = Plugin({'deleted': False, 'hasUpdate': True, 'downgradable': False,
+ 'dependencies': [{}, {}, {}, {}],
+ 'longName': 'Jenkins Subversion Plug-in', 'active': True,
+ 'shortName': 'subversion', 'backupVersion': None,
'url': 'http://wiki.jenkins-ci.org/display/JENKINS/Subversion+Plugin',
- 'enabled': True, 'pinned': False, 'version': '1.45',
+ 'enabled': True, 'pinned': False, 'version': '1.45',
'supportsDynamicLoad': 'MAYBE', 'bundled': True})
plugin = self.J.get_plugins()['subversion']
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi.job import Job
+
class FourOhFourError(Exception):
"""
Missing fake data
"""
+
class TestQueue(unittest.TestCase):
@classmethod
URL_DATA = {}
URL_DATA['http://localhost:8080/%s' % config.JENKINS_API] = \
- {'jobs':[
- {'name':'utmebvpxrw',
- 'url':'http://localhost/job/utmebvpxrw'}
+ {'jobs': [
+ {'name': 'utmebvpxrw',
+ 'url': 'http://localhost/job/utmebvpxrw'}
]
}
URL_DATA['http://localhost/job/utmebvpxrw/%s' % config.JENKINS_API] = \
{}
-
URL_DATA['http://localhost:8080/queue/%s' % config.JENKINS_API] = \
{'items': [{'actions': [{'causes': [{'shortDescription': 'Started by user anonymous',
'userId': None,
'url': 'http://localhost:8080/job/utmebvpxrw/'},
'why': 'Waiting for next available executor'}]}
-
@mock.patch.object(JenkinsBase, 'get_data', mockGetData)
def setUp(self):
self.J = Jenkins('http://localhost:8080') # Jenkins
self.assertEquals(len(self.q), 3)
def test_list_items(self):
- self.assertEquals(set(self.q.keys()), set([40,41,42]))
+ self.assertEquals(set(self.q.keys()), set([40, 41, 42]))
def test_getitem(self):
item40 = self.q[40]
from jenkinsapi.jenkins import Requester
from jenkinsapi.custom_exceptions import JenkinsAPIException
+
class TestQueue(unittest.TestCase):
def test_get_request_dict_auth(self):
headers=None)
self.assertTrue(isinstance(req_return, dict))
self.assertTrue(req_return.get('auth'))
- self.assertTrue(req_return['auth']==('foo', 'bar'))
+ self.assertTrue(req_return['auth'] == ('foo', 'bar'))
def test_get_request_dict_wrong_params(self):
req = Requester('foo', 'bar')
with self.assertRaises(AssertionError) as na:
- req_return = req.get_request_dict(url='http://', params='wrong',
+ req_return = req.get_request_dict(url='http://', params='wrong',
data=None, headers=None)
self.assertTrue(
- na.exception.message=="Params must be a dict, got 'wrong'")
+ na.exception.message == "Params must be a dict, got 'wrong'")
def test_get_request_dict_correct_params(self):
req = Requester('foo', 'bar')
- req_return = req.get_request_dict(url='http://',
+ req_return = req.get_request_dict(url='http://',
params={'param': 'value'},
data=None, headers=None)
self.assertTrue(isinstance(req_return, dict))
self.assertTrue(req_return.get('params'))
- self.assertTrue(req_return['params']=={'param': 'value'})
+ self.assertTrue(req_return['params'] == {'param': 'value'})
def test_get_request_dict_wrong_headers(self):
req = Requester('foo', 'bar')
with self.assertRaises(AssertionError) as na:
- req_return = req.get_request_dict(url='http://', params={},
+ req_return = req.get_request_dict(url='http://', params={},
data=None, headers='wrong')
self.assertTrue(
- na.exception.message=="headers must be a dict, got 'wrong'")
+ na.exception.message == "headers must be a dict, got 'wrong'")
def test_get_request_dict_correct_headers(self):
req = Requester('foo', 'bar')
- req_return = req.get_request_dict(url='http://',
+ req_return = req.get_request_dict(url='http://',
params={'param': 'value'},
data=None, headers={'header': 'value'})
self.assertTrue(isinstance(req_return, dict))
self.assertTrue(req_return.get('headers'))
- self.assertTrue(req_return['headers']=={'header': 'value'})
+ self.assertTrue(req_return['headers'] == {'header': 'value'})
def test_get_request_dict_data_passed(self):
req = Requester('foo', 'bar')
- req_return = req.get_request_dict(url='http://',
+ req_return = req.get_request_dict(url='http://',
params={'param': 'value'},
data='some data', headers={'header': 'value'})
self.assertTrue(isinstance(req_return, dict))
print req_return.get('data')
self.assertTrue(req_return.get('data'))
- self.assertTrue(req_return['data']=='some data')
+ self.assertTrue(req_return['data'] == 'some data')
def test_get_request_dict_data_not_passed(self):
req = Requester('foo', 'bar')
- req_return = req.get_request_dict(url='http://',
+ req_return = req.get_request_dict(url='http://',
params={'param': 'value'},
data=None, headers={'header': 'value'})
def test_get_url(self, _get):
_get.return_value = 'SUCCESS'
req = Requester('foo', 'bar')
- self.assertTrue(req.get_url('http://dummy',
+ self.assertTrue(req.get_url('http://dummy',
params={'param': 'value'},
- headers=None)=='SUCCESS')
-
+ headers=None) == 'SUCCESS')
+
@mock.patch.object(requests, 'post')
def test_get_url(self, _post):
_post.return_value = 'SUCCESS'
req = Requester('foo', 'bar')
- self.assertTrue(req.post_url('http://dummy',
+ self.assertTrue(req.post_url('http://dummy',
params={'param': 'value'},
- headers=None)=='SUCCESS')
-
+ headers=None) == 'SUCCESS')
+
@mock.patch.object(requests, 'post')
def test_post_xml_and_confirm_status_empty_xml(self, _post):
_post.return_value = 'SUCCESS'
params={'param': 'value'},
data=None)
- self.assertTrue(ae.exception.message=="Unexpected type of parameter 'data': <type 'NoneType'>. Expected (str, dict)")
+ self.assertTrue(ae.exception.message == "Unexpected type of parameter 'data': <type 'NoneType'>. Expected (str, dict)")
@mock.patch.object(requests, 'post')
def test_post_xml_and_confirm_status_some_xml(self, _post):
params={'param': 'value'},
data=None)
- self.assertTrue(ae.exception.message=="Unexpected type of parameter 'data': <type 'NoneType'>. Expected (str, dict)")
+ self.assertTrue(ae.exception.message == "Unexpected type of parameter 'data': <type 'NoneType'>. Expected (str, dict)")
@mock.patch.object(requests, 'post')
def test_post_and_confirm_status_some_data(self, _post):
data='some data')
print ae.exception.message
- self.assertTrue(ae.exception.message=="Operation failed. url=None, data=some data, headers={'Content-Type': 'application/x-www-form-urlencoded'}, status=500, text=")
+ self.assertTrue(ae.exception.message == "Operation failed. url=None, data=some data, headers={'Content-Type': 'application/x-www-form-urlencoded'}, status=500, text=")
@mock.patch.object(requests, 'get')
def test_get_and_confirm_status(self, _get):
params={'param': 'value'})
print ae.exception.message
- self.assertTrue(ae.exception.message=="Operation failed. url=None, headers=None, status=500, text=")
-
+ self.assertTrue(ae.exception.message == "Operation failed. url=None, headers=None, status=500, text=")
from jenkinsapi.result_set import ResultSet
from jenkinsapi.result import Result
+
class TestResultSet(unittest.TestCase):
DATA = {'duration': 0.0,
from jenkinsapi.view import View
from jenkinsapi.job import Job
+
class TestView(unittest.TestCase):
DATA = {'description': 'Important Shizz',
result = v.add_job('bar')
self.assertTrue(result)
-
class SelfPatchJenkins(object):
def has_job(self, job_name):
return False
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi.utils.requester import Requester
-class TestDataMissing(Exception): pass
+
+class TestDataMissing(Exception):
+ pass
+
class TestViews(unittest.TestCase):
@mock.patch.object(Jenkins, '_poll')
def mockGetData(JJ, url):
DATA = {}
DATA['http://localhost:8080/%s' % config.JENKINS_API] = \
- {'views':[dict(name='NewView', url='http://xxxxx/yyyy')]}
+ {'views': [dict(name='NewView', url='http://xxxxx/yyyy')]}
DATA['http://xxxxx/yyyy/%s' % config.JENKINS_API] = \
{}
if __name__ == '__main__':
unittest.main()
-
def update_config(self):
config_dest = os.path.join(self.jenkins_home, 'config.xml')
- config_dest_file = open(config_dest,'w')
+ config_dest_file = open(config_dest, 'w')
config_source = pkg_resources.resource_string('jenkinsapi_tests.systests', 'config.xml')
config_dest_file.write(config_source.encode('UTF-8'))
plugin_dir = os.path.join(self.jenkins_home, 'plugins')
if not os.path.exists(plugin_dir):
os.mkdir(plugin_dir)
-
+
log.info("Downloading %s", hpi_url)
log.info("Plugins will be installed in '%s'" % plugin_dir)
# FIXME: This is kinda ugly but works
plugin_path = os.path.join(plugin_dir, filename)
with open(plugin_path, 'wb') as h:
request = requests.get(hpi_url)
- h.write(request.content)
+ h.write(request.content)
def stop(self):
log.info("Shutting down jenkins.")
jl = JenkinsLancher(
'/home/sal/workspace/jenkinsapi/src/jenkinsapi_tests/systests/jenkins.war'
- )
+ )
jl.start()
log.info("Jenkins was launched...")
PORT = 8080
+
class ServerHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
def do_GET(self):
form = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
- environ={'REQUEST_METHOD':'POST',
- 'CONTENT_TYPE':self.headers['Content-Type'],
+ environ={'REQUEST_METHOD': 'POST',
+ 'CONTENT_TYPE': self.headers['Content-Type'],
})
for item in form.list:
logging.error(item)
projects / tools / python-jenkinsapi.git / commitdiff