test_jobs.py 文件源码

def test_max_retries_job(self):
        """Test if the job will fail after max_retries limit is reached"""

        http_requests = setup_mock_redmine_server(max_failures=2)

        args = {
            'url': REDMINE_URL,
            'api_token': 'AAAA',
            'max_issues': 3
        }

        q = rq.Queue('queue', async=False)

        with self.assertRaises(requests.exceptions.HTTPError):
            job = q.enqueue(execute_perceval_job,
                            backend='redmine', backend_args=args,
                            qitems='items', task_id='mytask',
                            max_retries=1)
            self.assertEqual(job.is_failed, True)