test(core): Clear RQ queues before and after tests (#22320)

Add RQQueueTestMixin to centralize RQ queue cleanup for test cases that
interact with background jobs. The mixin clears all RQ queues in setUp()
and tearDown(), preventing jobs created by one test from leaking into
later unrelated test runs.

Replace duplicate queue cleanup logic in core and netbox tests with the
shared mixin for better maintainability.

Fixes #22318

netbox/core/tests/test_api.py

@@ -11,8 +11,9 @@ from rq.job import JobStatus
 from rq.registry import FailedJobRegistry, StartedJobRegistry
 
 from users.constants import TOKEN_PREFIX
-from users.models import Token, User
+from users.models import Token
 from utilities.testing import APITestCase, APIViewTestCases, TestCase
+from utilities.testing.mixins import RQQueueTestMixin
 from utilities.testing.utils import disable_logging
 
 from ..models import *
@@ -168,7 +169,7 @@ class JobTestCase(
         )
 
 
-class BackgroundTaskTestCase(TestCase):
+class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
     user_permissions = ()
 
     @staticmethod
@@ -180,19 +181,14 @@ class BackgroundTaskTestCase(TestCase):
         raise Exception("Job failed")
 
     def setUp(self):
-        """
-        Create a user and token for API calls.
-        """
-        # Create the test user and assign permissions
-        self.user = User.objects.create_user(username='testuser', is_active=True)
+        super().setUp()
+
+        # The base TestCase creates self.user; make it active and create a token for API calls.
+        self.user.is_active = True
+        self.user.save()
         self.token = Token.objects.create(user=self.user)
         self.header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{self.token.key}.{self.token.token}'}
 
-        # Clear all queues prior to running each test
-        get_queue('default').connection.flushall()
-        get_queue('high').connection.flushall()
-        get_queue('low').connection.flushall()
-
     def test_background_queue_list(self):
         url = reverse('core-api:rqqueue-list')
 

netbox/core/tests/test_views.py

@@ -18,6 +18,7 @@ from core.models import *
 from dcim.models import Site
 from users.models import User
 from utilities.testing import TestCase, ViewTestCases, create_tags, disable_logging
+from utilities.testing.mixins import RQQueueTestMixin
 
 
 class DataSourceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
@@ -188,7 +189,7 @@ class ObjectChangeTestCase(TestCase):
         self.assertHttpStatus(response, 200)
 
 
-class BackgroundTaskTestCase(TestCase):
+class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
     user_permissions = ()
 
     # Dummy worker functions
@@ -210,19 +211,6 @@ class BackgroundTaskTestCase(TestCase):
         self.user.is_active = True
         self.user.save()
 
-        # Clear all queues prior to running each test
-        get_queue('default').connection.flushall()
-        get_queue('high').connection.flushall()
-        get_queue('low').connection.flushall()
-
-    def tearDown(self):
-        super().tearDown()
-
-        # Clear all queues after each test so no leftover jobs leak into the next test suite
-        get_queue('default').connection.flushall()
-        get_queue('high').connection.flushall()
-        get_queue('low').connection.flushall()
-
     def test_background_queue_list(self):
         url = reverse('core:background_queue_list')
 

netbox/netbox/tests/test_jobs.py

@@ -4,12 +4,12 @@ from unittest.mock import patch
 
 from django.test import TestCase
 from django.utils import timezone
-from django_rq import get_queue
 
 from core.choices import JobStatusChoices
 from core.exceptions import JobFailed
 from core.models import DataSource, Job
 from utilities.testing import disable_warnings
+from utilities.testing.mixins import RQQueueTestMixin
 
 from ..jobs import *
 from ..jobs import _INSTALL_ROOT
@@ -33,14 +33,7 @@ class TestSystemJobRunner(JobRunner):
         pass
 
 
-class BaseJobRunnerTestCase(TestCase):
-    def tearDown(self):
-        super().tearDown()
-
-        # Clear all queues after running each test
-        get_queue('default').connection.flushall()
-        get_queue('high').connection.flushall()
-        get_queue('low').connection.flushall()
+class BaseJobRunnerTestCase(RQQueueTestMixin, TestCase):
 
     @staticmethod
     def get_schedule_at(offset=1):

netbox/utilities/testing/mixins.py

@@ -0,0 +1,30 @@
+from django_rq import get_queue
+
+__all__ = (
+    'RQQueueTestMixin',
+)
+
+
+class RQQueueTestMixin:
+    """
+    Clear RQ queues before and after each test.
+    """
+    rq_queue_names = ('default', 'high', 'low')
+
+    @classmethod
+    def clear_rq_queues(cls):
+        for queue_name in cls.rq_queue_names:
+            get_queue(queue_name).connection.flushall()
+
+    def setUp(self):
+        super().setUp()
+
+        # Clear all queues before running each test
+        self.clear_rq_queues()
+
+    def tearDown(self):
+        try:
+            # Clear all queues after each test so no leftover jobs leak into the next test suite
+            self.clear_rq_queues()
+        finally:
+            super().tearDown()