obj_jenkins = Jenkins(jenkinsurl, username=username, password=password)
obj_jobs = [obj_jenkins[jid] for jid in jobs]
- for time_left in xrange(maxwait, 0, -interval):
+ for time_left in range(maxwait, 0, -interval):
still_running = [j for j in obj_jobs if j.is_queued_or_running()]
if not still_running:
return
# It's probably a static file,
# we can get it from the static collection
staticurl = urlparse.urljoin(basestaticurl, artifactname)
- theartifact = Artifact(artifactname, staticurl)
+ theartifact = Artifact(artifactname, staticurl, None)
theartifact.save(destpath)
installed.append(destpath)
return installed
try:
with open(fspath, 'rb') as f:
for chunk in iter(lambda: f.read(chunksize), ''):
- md5.update(chunk)
+ if chunk:
+ md5.update(chunk)
+ else:
+ break
except:
raise
return md5.hexdigest()
Return the current state of the text console.
"""
url = "%s/consoleText" % self.baseurl
- return self.job.jenkins.requester.get_url(url).content
+ return self.job.jenkins.requester.get_url(url).content.decode('utf-8')
def stop(self):
"""
<concurrentBuild>false</concurrentBuild>
<builders>
<hudson.tasks.Shell>
- <command>ping -c 200 localhost</command>
+ <command>ping -c 50 localhost</command>
</hudson.tasks.Shell>
</builders>
<publishers/>
<concurrentBuild>false</concurrentBuild>
<builders>
<hudson.tasks.Shell>
- <command>ping -c 5 localhost | tee out.txt
+ <command>ping -c 10 localhost | tee out.txt
gzip < out.txt > out.gz</command>
</hudson.tasks.Shell>
</builders>
self.jenkins['A'].invoke()
- for _ in xrange(10):
+ for _ in range(10):
try:
self.jenkins['C'].get_last_completed_buildnumber() > 0
except NoBuildData:
class TestPingerJob(BaseSystemTest):
- def test_invoke_job(self):
+ def test_artefacts(self):
job_name = 'create_%s' % random_string()
job = self.jenkins.create_job(job_name, JOB_WITH_ARTIFACTS)
job.invoke(block=True)
try:
# Verify that we can handle text artifacts
text_artifact.save_to_dir(tempDir)
- readBackText = open(os.path.join(
- tempDir, text_artifact.filename), 'rb').read().strip()
+ readBackText = open(os.path.join(tempDir,
+ text_artifact.filename),
+ 'rb').read().strip()
+ readBackText = readBackText.decode('ascii')
self.assertTrue(re.match(r'^PING \S+ \(127.0.0.1\)', readBackText))
self.assertTrue(readBackText.endswith('ms'))
# Verify that we can hande binary artifacts
binary_artifact.save_to_dir(tempDir)
- readBackText = gzip.open(os.path.join(tempDir, binary_artifact.filename,), 'rb').read().strip()
+ readBackText = gzip.open(os.path.join(tempDir,
+ binary_artifact.filename,),
+ 'rb').read().strip()
+ readBackText = readBackText.decode('ascii')
self.assertTrue(re.match(r'^PING \S+ \(127.0.0.1\)', readBackText))
self.assertTrue(readBackText.endswith('ms'))
finally:
import unittest2 as unittest
except ImportError:
import unittest
-from StringIO import StringIO
+try:
+ from StringIO import StringIO
+except ImportError:
+ from io import StringIO
from jenkinsapi_tests.systests.base import BaseSystemTest
from jenkinsapi_tests.test_utils.random_strings import random_string
from jenkinsapi_tests.systests.job_configs import JOB_WITH_FILE
job = self.jenkins.create_job(job_name, JOB_WITH_FILE)
job.invoke(block=True, files={'file.txt': param_file})
- b = job.get_last_build()
- while b.is_running():
+ build = job.get_last_build()
+ while build.is_running():
time.sleep(0.25)
- artifacts = b.get_artifact_dict()
+ artifacts = build.get_artifact_dict()
self.assertIsInstance(artifacts, dict)
art_file = artifacts['file.txt']
self.assertTrue(art_file.get_data().strip(), file_data)
job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
job.invoke(block=True, build_params={'B': param_B})
- b = job.get_last_build()
- while b.is_running():
+ build = job.get_last_build()
+ while build.is_running():
time.sleep(0.25)
- artifacts = b.get_artifact_dict()
+ artifacts = build.get_artifact_dict()
self.assertIsInstance(artifacts, dict)
artB = artifacts['b.txt']
self.assertTrue(artB.get_data().strip(), param_B)
- self.assertIn(param_B, b.get_console())
+ self.assertIn(param_B, build.get_console())
def test_parameterized_job_build_queuing(self):
"""Accept multiple builds of parameterized jobs with unique
self.assertTrue(job.has_queued_build(params))
- while(job.has_queued_build(params)):
+ while job.has_queued_build(params):
time.sleep(0.25)
- b = job.get_last_build()
- while b.is_running():
+ build = job.get_last_build()
+ while build.is_running():
time.sleep(0.25)
- artifacts = b.get_artifact_dict()
+ artifacts = build.get_artifact_dict()
self.assertIsInstance(artifacts, dict)
artB = artifacts['b.txt']
self.assertTrue(artB.get_data().strip(), param_B)
- self.assertIn(param_B, b.get_console())
+ self.assertIn(param_B, build.get_console())
def test_parameterized_job_build_rejection(self):
"""Reject build of paramterized job when existing build with same