--- /dev/null
+class Invocation(object):
+ """
+ Represents the state and consequences of a single attempt to start a job.
+ This class provides a context manager which is intended to watch the state of the job
+ before and after the invoke. It will detect whether a process got queued, launched
+ or whether nothing at all happened.
+
+ An instance of this object will be returned by job.invoke()
+ """
+
+ def __init__(self, job):
+ self.job = job
+
+
+ def __enter__(self):
+ """
+ Start watching the job
+ """
+
+ def __exit__(self, type, value, traceback):
+ """
+ Finish watching the job - it will track which new queue items or builds have
+ been created as a consequence of invoking the job.
+ """
+
+ def block(self, until='completed'):
+ """
+ Block this item until a condition is met.
+ Setting until to 'running' blocks the item until it is running (i.e. it's no longer queued)
+ """
+
+ def stop(self):
+ """
+ Stop this item, whether it is on the queue or blocked.
+ """
+
+ def is_queued(self):
+ """
+ Returns True if this item is on the queue
+ """
+
+ def is_running(self):
+ """
+ Returns True if this item is executing now
+ """
+
+ def is_queued_or_running(self):
+ return self.is_queued() or self.is_running()
+
+ def get_queue_item(self):
+ """
+ If the item is queued it will return that QueueItem, otherwise it will
+ raise an exception.
+ """
+
+ def get_build(self):
+ """
+ If the item is building it will return a Build object, otherwise it will
+ raise an exception.
+ """
+++ /dev/null
-class Invocation(object):
- """
- Represents the state and consequences of a single attempt to start a job.
- This class provides a context manager which is intended to watch the state of the job
- before and after the invoke. It will detect whether a process got queued, launched
- or whether nothing at all happened.
-
- An instance of this object will be returned by job.invoke()
- """
-
- def __init__(self, job):
- self.job = job
-
-
- def __enter__(self):
- """
- Start watching the job
- """
-
- def __exit__(self):
- """
- Finish watching the job - it will track which new queue items or builds have
- been created as a consequence of invoking the job.
- """
-
- def block(self, until='completed'):
- """
- Block this item until a condition is met.
- Setting until to 'running' blocks the item until it is running (i.e. it's no longer queued)
- """
-
- def stop(self):
- """
- Stop this item, whether it is on the queue or blocked.
- """
-
- def is_queued(self):
- """
- Returns True if this item is on the queue
- """
-
- def is_running(self):
- """
- Returns True if this item is executing now
- """
-
- def is_queued_or_running(self):
- return self.is_queued() or self.is_running()
-
- def get_queue_item(self):
- """
- If the item is queued it will return that QueueItem, otherwise it will
- raise an exception.
- """
-
- def get_build(self):
- """
- If the item is building it will return a Build object, otherwise it will
- raise an exception.
- """
from collections import defaultdict
from time import sleep
from jenkinsapi.build import Build
+from jenkinsapi.invocation import Invocation
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi import exceptions
from jenkinsapi.mutable_jenkins_thing import MutableJenkinsThing
assert isinstance(block, bool)
assert isinstance(skip_if_running, bool)
+ # Create a new invocation instance
+ invocation = Invocation(self)
+
# Either copy the params dict or make a new one.
build_params = build_params and dict(
build_params.items()) or {} # Via POSTed JSON
params = {} # Via Get string
- if self.is_queued():
- raise WillNotBuild('%s is already queued' % repr(self))
- elif self.is_running():
- if skip_if_running:
- log.warn(
- "Will not request new build because %s is already running", self.name)
- else:
- log.warn(
- "Will re-schedule %s even though it is already running", self.name)
+ with invocation:
+ if self.is_queued():
+ raise WillNotBuild('%s is already queued' % repr(self))
- log.info("Attempting to start %s on %s", self.name, repr(
- self.get_jenkins_obj()))
+ elif self.is_running():
+ if skip_if_running:
+ log.warn(
+ "Will not request new build because %s is already running", self.name)
+ else:
+ log.warn(
+ "Will re-schedule %s even though it is already running", self.name)
- url = self.get_build_triggerurl()
+ log.info("Attempting to start %s on %s", self.name, repr(
+ self.get_jenkins_obj()))
- if cause:
- build_params['cause'] = cause
+ url = self.get_build_triggerurl()
- if securitytoken:
- params['token'] = securitytoken
+ if cause:
+ build_params['cause'] = cause
- response = self.jenkins.requester.post_and_confirm_status(
- url,
- data={'json': self.mk_json_from_build_parameters(
- build_params)}, # See above - build params have to be JSON encoded & posted.
- params=params,
- valid=[200, 201]
- )
- if invoke_pre_check_delay > 0:
- log.info(
- "Waiting for %is to allow Jenkins to catch up", invoke_pre_check_delay)
- sleep(invoke_pre_check_delay)
- if block:
- total_wait = 0
+ if securitytoken:
+ params['token'] = securitytoken
- while self.is_queued():
+ response = self.jenkins.requester.post_and_confirm_status(
+ url,
+ data={'json': self.mk_json_from_build_parameters(
+ build_params)}, # See above - build params have to be JSON encoded & posted.
+ params=params,
+ valid=[200, 201]
+ )
+ if invoke_pre_check_delay > 0:
log.info(
- "Waited %is for %s to begin...", total_wait, self.name)
- sleep(invoke_block_delay)
- total_wait += invoke_block_delay
- if self.is_running():
- running_build = self.get_last_build()
- running_build.block_until_complete(
- delay=invoke_pre_check_delay)
+ "Waiting for %is to allow Jenkins to catch up", invoke_pre_check_delay)
+ sleep(invoke_pre_check_delay)
+ if block:
+ total_wait = 0
+
+ while self.is_queued():
+ log.info(
+ "Waited %is for %s to begin...", total_wait, self.name)
+ sleep(invoke_block_delay)
+ total_wait += invoke_block_delay
+ if self.is_running():
+ running_build = self.get_last_build()
+ running_build.block_until_complete(
+ delay=invoke_pre_check_delay)
+ return invocation
def _buildid_for_type(self, buildtype):
self.poll()
System tests for `jenkinsapi.jenkins` module.
'''
import unittest
+from jenkinsapi.invocation import Invocation
from jenkinsapi_tests.test_utils.random_strings import random_string
from jenkinsapi_tests.systests.base import BaseSystemTest, EMPTY_JOB_CONFIG
job = self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
job.invoke()
+ def test_invocation_object(self):
+ job_name = 'create_%s' % random_string()
+ job = self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
+ ii = job.invoke()
+ self.assertIsInstance(ii, Invocation)
+
def test_get_jobs_list(self):
job1_name = 'first_%s' % random_string()
job2_name = 'second_%s' % random_string()
self.assertJobIsPresent(copied_job_name)
if __name__ == '__main__':
- unittest.main()
\ No newline at end of file
+ unittest.main()
'''
import logging
import unittest
-from jenkinsapi.jenkins import Jenkins
from jenkinsapi_tests.systests.base import BaseSystemTest
from jenkinsapi_tests.test_utils.random_strings import random_string
'''
import time
import unittest
-from jenkinsapi_tests.test_utils.random_strings import random_string
from jenkinsapi_tests.systests.base import BaseSystemTest
+from jenkinsapi_tests.test_utils.random_strings import random_string
+
JOB_CONFIG = """
<?xml version='1.0' encoding='UTF-8'?>
projects / tools / python-jenkinsapi.git / commitdiff