mirror of
https://github.com/netbox-community/netbox.git
synced 2026-02-04 06:16:23 -06:00
Add queue name to Job
This commit is contained in:
@@ -0,0 +1,18 @@
|
|||||||
|
# Generated by Django 5.2.9 on 2026-01-27 00:39
|
||||||
|
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('core', '0020_owner'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='job',
|
||||||
|
name='queue_name',
|
||||||
|
field=models.CharField(blank=True, max_length=100),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -112,6 +112,12 @@ class Job(models.Model):
|
|||||||
verbose_name=_('job ID'),
|
verbose_name=_('job ID'),
|
||||||
unique=True
|
unique=True
|
||||||
)
|
)
|
||||||
|
queue_name = models.CharField(
|
||||||
|
verbose_name=_('queue name'),
|
||||||
|
max_length=100,
|
||||||
|
blank=True,
|
||||||
|
help_text=_('Name of the queue in which this job was enqueued')
|
||||||
|
)
|
||||||
log_entries = ArrayField(
|
log_entries = ArrayField(
|
||||||
verbose_name=_('log entries'),
|
verbose_name=_('log entries'),
|
||||||
base_field=models.JSONField(
|
base_field=models.JSONField(
|
||||||
@@ -181,7 +187,8 @@ class Job(models.Model):
|
|||||||
def delete(self, *args, **kwargs):
|
def delete(self, *args, **kwargs):
|
||||||
super().delete(*args, **kwargs)
|
super().delete(*args, **kwargs)
|
||||||
|
|
||||||
rq_queue_name = get_queue_for_model(self.object_type.model if self.object_type else None)
|
# Use the stored queue name, or fall back to get_queue_for_model for legacy jobs
|
||||||
|
rq_queue_name = self.queue_name or get_queue_for_model(self.object_type.model if self.object_type else None)
|
||||||
queue = django_rq.get_queue(rq_queue_name)
|
queue = django_rq.get_queue(rq_queue_name)
|
||||||
job = queue.fetch_job(str(self.job_id))
|
job = queue.fetch_job(str(self.job_id))
|
||||||
|
|
||||||
@@ -288,7 +295,8 @@ class Job(models.Model):
|
|||||||
scheduled=schedule_at,
|
scheduled=schedule_at,
|
||||||
interval=interval,
|
interval=interval,
|
||||||
user=user,
|
user=user,
|
||||||
job_id=uuid.uuid4()
|
job_id=uuid.uuid4(),
|
||||||
|
queue_name=rq_queue_name
|
||||||
)
|
)
|
||||||
job.full_clean()
|
job.full_clean()
|
||||||
job.save()
|
job.save()
|
||||||
|
|||||||
@@ -1,8 +1,10 @@
|
|||||||
|
from unittest.mock import patch, MagicMock
|
||||||
|
|
||||||
from django.contrib.contenttypes.models import ContentType
|
from django.contrib.contenttypes.models import ContentType
|
||||||
from django.core.exceptions import ObjectDoesNotExist
|
from django.core.exceptions import ObjectDoesNotExist
|
||||||
from django.test import TestCase
|
from django.test import TestCase
|
||||||
|
|
||||||
from core.models import DataSource, ObjectType
|
from core.models import DataSource, Job, ObjectType
|
||||||
from core.choices import ObjectChangeActionChoices
|
from core.choices import ObjectChangeActionChoices
|
||||||
from dcim.models import Site, Location, Device
|
from dcim.models import Site, Location, Device
|
||||||
from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED
|
from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED
|
||||||
@@ -200,3 +202,61 @@ class ObjectTypeTest(TestCase):
|
|||||||
bookmarks_ots = ObjectType.objects.with_feature('bookmarks')
|
bookmarks_ots = ObjectType.objects.with_feature('bookmarks')
|
||||||
self.assertIn(ObjectType.objects.get_by_natural_key('dcim', 'site'), bookmarks_ots)
|
self.assertIn(ObjectType.objects.get_by_natural_key('dcim', 'site'), bookmarks_ots)
|
||||||
self.assertNotIn(ObjectType.objects.get_by_natural_key('dcim', 'cabletermination'), bookmarks_ots)
|
self.assertNotIn(ObjectType.objects.get_by_natural_key('dcim', 'cabletermination'), bookmarks_ots)
|
||||||
|
|
||||||
|
|
||||||
|
class JobTest(TestCase):
|
||||||
|
|
||||||
|
@patch('core.models.jobs.django_rq.get_queue')
|
||||||
|
def test_enqueue_with_custom_queue_name(self, mock_get_queue):
|
||||||
|
"""
|
||||||
|
Test that when a job is enqueued with a custom queue_name, the queue_name is stored in the Job instance.
|
||||||
|
"""
|
||||||
|
mock_queue = MagicMock()
|
||||||
|
mock_get_queue.return_value = mock_queue
|
||||||
|
|
||||||
|
def dummy_func(**kwargs):
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Enqueue a job with a custom queue name
|
||||||
|
custom_queue = 'my_custom_queue'
|
||||||
|
job = Job.enqueue(
|
||||||
|
func=dummy_func,
|
||||||
|
name='Test Job',
|
||||||
|
queue_name=custom_queue
|
||||||
|
)
|
||||||
|
|
||||||
|
# Verify the queue_name was stored
|
||||||
|
self.assertEqual(job.queue_name, custom_queue)
|
||||||
|
mock_get_queue.assert_called_with(custom_queue)
|
||||||
|
|
||||||
|
@patch('core.models.jobs.django_rq.get_queue')
|
||||||
|
def test_delete_cancels_job_from_correct_queue(self, mock_get_queue):
|
||||||
|
"""
|
||||||
|
Test that when a job is deleted, it's canceled from the correct queue (the one stored in queue_name).
|
||||||
|
"""
|
||||||
|
mock_queue = MagicMock()
|
||||||
|
mock_rq_job = MagicMock()
|
||||||
|
mock_queue.fetch_job.return_value = mock_rq_job
|
||||||
|
mock_get_queue.return_value = mock_queue
|
||||||
|
|
||||||
|
def dummy_func(**kwargs):
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Enqueue a job with a custom queue name
|
||||||
|
custom_queue = 'my_custom_queue'
|
||||||
|
job = Job.enqueue(
|
||||||
|
func=dummy_func,
|
||||||
|
name='Test Job',
|
||||||
|
queue_name=custom_queue
|
||||||
|
)
|
||||||
|
|
||||||
|
# Reset mock to clear enqueue call
|
||||||
|
mock_get_queue.reset_mock()
|
||||||
|
|
||||||
|
# Delete the job
|
||||||
|
job.delete()
|
||||||
|
|
||||||
|
# Verify the correct queue was used for cancellation
|
||||||
|
mock_get_queue.assert_called_with(custom_queue)
|
||||||
|
mock_queue.fetch_job.assert_called_with(str(job.job_id))
|
||||||
|
mock_rq_job.cancel.assert_called_once()
|
||||||
|
|||||||
Reference in New Issue
Block a user