From b0093397186c60cf76814beccd3523509e8323de Mon Sep 17 00:00:00 2001 From: salimfadhley Date: Sat, 15 Jun 2013 02:15:46 +0100 Subject: [PATCH] many small fixes --- examples/copy_a_job.py | 23 +++++ examples/example_download_war.py | 22 ++++ examples/login_with_auth.py | 11 ++ .../systests/test_downstream_upstream.py | 111 +++++++++++++++++++++ .../systests/test_parameterized_builds.py | 10 -- jenkinsapi_tests/test_utils/simple_post_logger.py | 31 ------ 6 files changed, 167 insertions(+), 41 deletions(-) create mode 100644 examples/copy_a_job.py create mode 100644 examples/example_download_war.py create mode 100644 examples/login_with_auth.py create mode 100644 jenkinsapi_tests/systests/test_downstream_upstream.py delete mode 100644 jenkinsapi_tests/test_utils/simple_post_logger.py diff --git a/examples/copy_a_job.py b/examples/copy_a_job.py new file mode 100644 index 0000000..3eef8b6 --- /dev/null +++ b/examples/copy_a_job.py @@ -0,0 +1,23 @@ +""" +A lower-level implementation of copying a job in Jenkins +""" + +import requests +from jenkinsapi.jenkins import Jenkins +from pkg_resources import resource_string +from jenkinsapi_tests.test_utils.random_strings import random_string + +J = Jenkins('http://localhost:8080') +jobName = random_string() +jobName2 = '%s_2' % jobName + +url = 'http://localhost:8080/createItem?from=%s&name=%s&mode=copy' % ( + jobName, jobName2) + +xml = resource_string('examples', 'addjob.xml') +j = J.create_job(jobname=jobName, config=xml) + + +h = {'Content-Type': 'application/x-www-form-urlencoded'} +response = requests.post(url, data='dysjsjsjs', headers=h) +print response.text.encode('UTF-8') diff --git a/examples/example_download_war.py b/examples/example_download_war.py new file mode 100644 index 0000000..f6b2121 --- /dev/null +++ b/examples/example_download_war.py @@ -0,0 +1,22 @@ +import io +import sys +import requests + +url = 'http://mirrors.jenkins-ci.org/war/latest/jenkins.war' +print url + + + +response = requests.get(url) +total_length = response.headers.get('content-length') + +print total_length + +with io.open('jenkins.war', 'wb') as jw: + + for i, data in enumerate(response.iter_content()): + jw.write(data) + + if i % 1024 == 0: + sys.stdout.write('.') + diff --git a/examples/login_with_auth.py b/examples/login_with_auth.py new file mode 100644 index 0000000..189be69 --- /dev/null +++ b/examples/login_with_auth.py @@ -0,0 +1,11 @@ +""" +A lower level example of how we login with authentication +""" + +from jenkinsapi import jenkins + + +J = jenkins.Jenkins("http://localhost:8080", username = "sal", password = "foobar") +J.poll() + +print J.items() diff --git a/jenkinsapi_tests/systests/test_downstream_upstream.py b/jenkinsapi_tests/systests/test_downstream_upstream.py new file mode 100644 index 0000000..0868a26 --- /dev/null +++ b/jenkinsapi_tests/systests/test_downstream_upstream.py @@ -0,0 +1,111 @@ +''' +System tests for `jenkinsapi.jenkins` module. +''' +import time +import logging +import unittest +from jenkinsapi_tests.systests.base import BaseSystemTest + +log = logging.getLogger(__name__) + +JOB_CONFIGS = { +'A':""" + + + + false + + + true + false + false + false + + false + + + + B + + SUCCESS + 0 + BLUE + + + + +""", + +'B':""" + + + + false + + + true + false + false + false + + false + + + + C + + SUCCESS + 0 + BLUE + + + + +""", + +'C':""" + + + + false + + + true + false + false + false + + false + + + +""" + +} + +class TestParameterizedBuilds(BaseSystemTest): + def test_invoke_job_parameterized(self): + for job_name, job_config in JOB_CONFIGS.items(): + self.jenkins.create_job(job_name, job_config) + + self.jenkins['A'].invoke() + + + for _ in range(0,10): + if not self.jenkins['C'].get_last_completed_buildnumber() > 0: + log.info('Waiting for the third test to complete') + time.sleep(2) + else: + break + else: + self.fail('Jenkins took too long to run these jobs') + + self.assertTrue(self.jenkins['C'].get_upstream_jobs(), self.jenkins['B']) + self.assertTrue(self.jenkins['B'].get_upstream_jobs(), self.jenkins['A']) + + self.assertTrue(self.jenkins['A'].get_downstream_jobs(), self.jenkins['B']) + self.assertTrue(self.jenkins['B'].get_downstream_jobs(), self.jenkins['C']) + +if __name__ == '__main__': + logging.basicConfig() + unittest.main() diff --git a/jenkinsapi_tests/systests/test_parameterized_builds.py b/jenkinsapi_tests/systests/test_parameterized_builds.py index 48a0c2f..ef58915 100644 --- a/jenkinsapi_tests/systests/test_parameterized_builds.py +++ b/jenkinsapi_tests/systests/test_parameterized_builds.py @@ -1,11 +1,7 @@ ''' 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 @@ -57,7 +53,6 @@ echo $B > b.txt class TestParameterizedBuilds(BaseSystemTest): def test_invoke_job_parameterized(self): - param_B = random_string() job_name = 'create_%s' % random_string() @@ -70,13 +65,8 @@ class TestParameterizedBuilds(BaseSystemTest): artifacts = b.get_artifact_dict() self.assertIsInstance(artifacts, dict) - artB = artifacts['b.txt'] - self.assertTrue(artB.get_data().strip(), param_B) - # TODO: Actually verify the download - - if __name__ == '__main__': unittest.main() diff --git a/jenkinsapi_tests/test_utils/simple_post_logger.py b/jenkinsapi_tests/test_utils/simple_post_logger.py deleted file mode 100644 index ef1050c..0000000 --- a/jenkinsapi_tests/test_utils/simple_post_logger.py +++ /dev/null @@ -1,31 +0,0 @@ -import SimpleHTTPServer -import SocketServer -import logging -import cgi - -PORT = 8000 - -class ServerHandler(SimpleHTTPServer.SimpleHTTPRequestHandler): - - def do_GET(self): - logging.error(self.headers) - SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self) - - def do_POST(self): - logging.error(self.headers) - form = cgi.FieldStorage( - fp=self.rfile, - headers=self.headers, - environ={'REQUEST_METHOD':'POST', - 'CONTENT_TYPE':self.headers['Content-Type'], - }) - for item in form.list: - logging.error(item) - SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self) - -Handler = ServerHandler - -httpd = SocketServer.TCPServer(("", PORT), Handler) - -print "serving at port", PORT -httpd.serve_forever() -- 2.7.4