From 6724c29ffb22492ac8c1fa6b2ffb5412012138f1 Mon Sep 17 00:00:00 2001 From: Martin Hauser Date: Mon, 1 Jun 2026 19:10:13 +0200 Subject: [PATCH] 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 | 20 ++++++++------------ netbox/core/tests/test_views.py | 16 ++-------------- netbox/netbox/tests/test_jobs.py | 11 ++--------- netbox/utilities/testing/mixins.py | 30 ++++++++++++++++++++++++++++++ 4 files changed, 42 insertions(+), 35 deletions(-) create mode 100644 netbox/utilities/testing/mixins.py diff --git a/netbox/core/tests/test_api.py b/netbox/core/tests/test_api.py index e3bbaf08d..2a159f6bd 100644 --- a/netbox/core/tests/test_api.py +++ b/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') diff --git a/netbox/core/tests/test_views.py b/netbox/core/tests/test_views.py index 40dc3b722..b763eaed2 100644 --- a/netbox/core/tests/test_views.py +++ b/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') diff --git a/netbox/netbox/tests/test_jobs.py b/netbox/netbox/tests/test_jobs.py index 71cbdce9a..3691592e2 100644 --- a/netbox/netbox/tests/test_jobs.py +++ b/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): diff --git a/netbox/utilities/testing/mixins.py b/netbox/utilities/testing/mixins.py new file mode 100644 index 000000000..408d5d644 --- /dev/null +++ b/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()