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

Wait until job1 exists in Redis before enqueueing job2

Browse Source
This commit is contained in:
Brian Tiemann 2024-11-21 09:51:09 -05:00
parent 09a0e579fa
commit fb7c3ebd26
1 changed files with 11 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,8 +1,11 @@
import time
from datetime import timedelta from datetime import timedelta
from redis import Redis
from django.test import TestCase from django.test import TestCase
from django.utils import timezone from django.utils import timezone
from django_rq import get_queue from django_rq import get_queue
from rq.job import Job as RQJob
from ..jobs import * from ..jobs import *
from core.models import DataSource, Job from core.models import DataSource, Job
@ -121,7 +124,15 @@ class EnqueueTest(JobRunnerTestCase):
def test_enqueue_once_after_enqueue(self): def test_enqueue_once_after_enqueue(self):
instance = DataSource() instance = DataSource()
redis = Redis()
job1 = TestJobRunner.enqueue(instance, schedule_at=self.get_schedule_at()) job1 = TestJobRunner.enqueue(instance, schedule_at=self.get_schedule_at())
job1_rq = None
max_sleep = 5
sleep_count = 0
while job1_rq is None and sleep_count < max_sleep:
job1_rq = RQJob.fetch(str(job1.job_id), connection=redis)
time.sleep(1)
sleep_count += 1
job2 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at(2)) job2 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at(2))
self.assertNotEqual(job1, job2) self.assertNotEqual(job1, job2)