traditional.py 文件源码

def test_traditional_task_cancel_in_queue(self, params):
        # Fill up queue
        blockers = []
        for _ in range(0, multiprocessing.cpu_count()):
            blockers .append(cancelable.delay(sleep_interval=0.1))

        jobModel = self.model('job', 'jobs')
        job = jobModel.createJob(
            title='test_traditional_task_cancel',
            type='worker', handler='worker_handler',
            user=self.getCurrentUser(), public=False, args=(self.girder_worker_run_cancelable,),
            kwargs={'inputs': {},
                    'outputs': {}})

        job['kwargs']['jobInfo'] = utils.jobInfoSpec(job)

        jobModel.save(job)
        jobModel.scheduleJob(job)
        jobModel.cancelJob(job)

        # Now clean up the blockers
        for blocker in blockers:
            blocker.revoke()

        return job