import time
import logging
import unittest
+from jenkinsapi.exceptions import NoBuildData
from jenkinsapi_tests.systests.base import BaseSystemTest
log = logging.getLogger(__name__)
JOB_CONFIGS = {
-'A':"""<?xml version='1.0' encoding='UTF-8'?>
+ 'A': """<?xml version='1.0' encoding='UTF-8'?>
<project>
<actions/>
<description></description>
<buildWrappers/>
</project>""",
-'B':"""<?xml version='1.0' encoding='UTF-8'?>
+ 'B': """<?xml version='1.0' encoding='UTF-8'?>
<project>
<actions/>
<description></description>
<buildWrappers/>
</project>""",
-'C':"""<?xml version='1.0' encoding='UTF-8'?>
+ 'C': """<?xml version='1.0' encoding='UTF-8'?>
<project>
<actions/>
<description></description>
}
+
class TestDownstreamUpstream(BaseSystemTest):
- def test_invoke_job_parameterized(self):
+ DELAY = 2
+
+ def test_stream_relationship(self):
+ """
+ Can we keep track of the relationships between upstream & downstream jobs?
+ """
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)
+ for _ in xrange(10):
+ try:
+ self.jenkins['C'].get_last_completed_buildnumber() > 0
+ except NoBuildData:
+ log.info("Waiting %i seconds for until the final job has run", self.DELAY)
+ time.sleep(self.DELAY)
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[
+ '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'])
+ 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()
zip_safe=True,
include_package_data=False,
setup_requires=['nose'],
- install_requires=['requests==1.2.3', 'pytz'],
+ install_requires=['requests>=1.2.3', 'pytz'],
test_suite='jenkinsapi_tests',
tests_require=['mock', 'coverage'],
- # extras_require={
- # 'kerberos': ['kerberos']
- #},
entry_points=GLOBAL_ENTRY_POINTS,
url=PROJECT_URL,
description=SHORT_DESCRIPTION,