NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-29 11:56:25 -06:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Catch InvalidJobOperation and use as trigger for retry

Browse Source
This commit is contained in:
Brian Tiemann 2024-11-21 10:02:34 -05:00
parent aa9d4a1f1b
commit c14542c8cc
1 changed files with 8 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
netbox/netbox/tests/test_jobs.py
View File

@ -5,7 +5,7 @@ from redis import Redis
from django.test import TestCase
from django.utils import timezone
from django_rq import get_queue
from rq.job import Job as RQJob
from rq.job import Job as RQJob, InvalidJobOperation
from ..jobs import *
from core.models import DataSource, Job
@ -127,9 +127,13 @@ class EnqueueTest(JobRunnerTestCase):
redis = Redis()
job1 = TestJobRunner.enqueue(instance, schedule_at=self.get_schedule_at())
job1_rq = RQJob.fetch(str(job1.job_id), connection=redis)
job1_status = None
max_sleep = 5
sleep_count = 0
while not job1_rq.get_status() and sleep_count < max_sleep:
while job1_status is None and sleep_count < max_sleep:
try:
job1_status = job1_rq.get_status()
except InvalidJobOperation:
time.sleep(1)
sleep_count += 1
job2 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at(2))