/dist
.settings
*.DS_Store
+jenkins.war
--- /dev/null
+import os
+import time
+import shutil
+import tempfile
+import subprocess
+
+
+class Timeout(RuntimeError):
+ pass
+
+
+class JenkinsLauncher(object):
+
+ def __init__(self, timeout=10, update_war=False, launch=False):
+ self.timeout = timeout
+ self.directory = os.path.dirname(__file__)
+ if update_war:
+ self.update_war()
+ if launch:
+ self.launch()
+
+ def update_war(self):
+ os.chdir(self.directory)
+ subprocess.check_call('./get-jenkins-war.sh')
+
+ def launch(self):
+ '''
+ Launches jenkins and waits while it's ready.
+ '''
+ self.jenkins_home = tempfile.mkdtemp(prefix='jenkins-home-')
+ os.environ['JENKINS_HOME'] = self.jenkins_home
+ jenkins_command = 'java -jar jenkins.war'
+ self.jenkins_process = subprocess.Popen(
+ jenkins_command.split(), stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ start_time = time.time()
+ while time.time() - start_time < self.timeout:
+ line = self.jenkins_process.stderr.readline().strip()
+ if line == 'INFO: Jenkins is fully up and running':
+ return
+ raise Timeout('Timeout error occured while waiting for Jenkins start.')
+
+ def stop(self):
+ shutil.rmtree(self.jenkins_home)
+ self.jenkins_process.terminate()
+ self.jenkins_process.wait()
+
+
+launcher = None
+
+
+def setUpPackage():
+ global launcher
+ launcher = JenkinsLauncher(update_war=True, launch=True)
+
+
+def tearDownPackage():
+ launcher.stop()
--- /dev/null
+import unittest
+from jenkinsapi.jenkins import Jenkins
+
+
+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>
+'''
+
+
+class BaseSystemTest(unittest.TestCase):
+
+ def setUp(self):
+ self.jenkins = Jenkins('http://localhost:8080')
+
+ def tearDown(self):
+ self._delete_all_jobs()
+
+ def _delete_all_jobs(self):
+ self.jenkins.poll()
+ for name in self.jenkins.get_jobs_list():
+ self.jenkins.delete_job(name)
+
+ def _create_job(self, name='whatever', config=EMPTY_JOB_CONFIG):
+ job = self.jenkins.create_job(name, config)
+ self.jenkins.poll()
+ return job
+
+ def assertJobIsPresent(self, name):
+ self.jenkins.poll()
+ self.assertTrue(name in self.jenkins,
+ 'Job %r is absent in jenkins.' % name)
+
+ def assertJobIsAbsent(self, name):
+ self.jenkins.poll()
+ self.assertTrue(name not in self.jenkins,
+ 'Job %r is present in jenkins.' % name)
--- /dev/null
+#!/bin/sh
+
+JENKINS_WAR_URL="http://mirrors.jenkins-ci.org/war/latest/jenkins.war"
+
+if [ ! -e 'jenkins.war' ]; then
+ wget $JENKINS_WAR_URL
+fi
--- /dev/null
+'''
+System tests for `jenkinsapi.jenkins` module.
+'''
+from jenkinsapi_tests.systests.base import BaseSystemTest, EMPTY_JOB_CONFIG
+
+
+class JobTests(BaseSystemTest):
+
+ def test_create_job(self):
+ self.jenkins.create_job('whatever', EMPTY_JOB_CONFIG)
+ self.assertJobIsPresent('whatever')
+
+ def test_get_jobs_list(self):
+ self._create_job('job1')
+ self._create_job('job2')
+ job_list = self.jenkins.get_jobs_list()
+ self.assertEqual(['job1', 'job2'], job_list)
+
+ def test_delete_job(self):
+ self._create_job('job_to_delete')
+ self.jenkins.delete_job('job_to_delete')
+ self.assertJobIsAbsent('job_to_delete')
+
+ def test_rename_job(self):
+ self._create_job('job_to_rename')
+ self.jenkins.rename_job('job_to_rename', 'renamed_job')
+ self.assertJobIsAbsent('job_to_rename')
+ self.assertJobIsPresent('renamed_job')
+
+ def test_copy_job(self):
+ self._create_job('template_job')
+ self.jenkins.copy_job('template_job', 'copied_job')
+ self.assertJobIsPresent('template_job')
+ self.assertJobIsPresent('copied_job')
+
+
+class NodeTests(BaseSystemTest):
+
+ def test_get_node_dict(self):
+ self.assertEqual(self.jenkins.get_node_dict(), {
+ 'master': 'http://localhost:8080/computer/master/api/python/'})