Fixes #22466: Fix test failure against SSL-enabled PosgtreSQL

netbox/core/tests/test_api.py

@@ -321,9 +321,8 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
         # Enqueue & run a job that will fail
         queue = get_queue('default')
         job = queue.enqueue(self.dummy_job_failing)
-        worker = get_worker('default')
         with disable_logging():
-            worker.work(burst=True)
+            self.run_rq_jobs('default')
         self.assertTrue(job.is_failed)
         url = reverse('core-api:rqtask-requeue', args=[job.id])
 

netbox/core/tests/test_views.py

@@ -386,9 +386,8 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
 
         # Enqueue & run a job that will fail
         job = queue.enqueue(self.dummy_job_failing)
-        worker = get_worker('default')
         with disable_logging():
-            worker.work(burst=True)
+            self.run_rq_jobs('default')
         self.assertTrue(job.is_failed)
 
         # Re-enqueue the failed job and check that its status has been reset

netbox/extras/tests/test_event_rules.py

@@ -27,9 +27,10 @@ from extras.signals import process_job_end_event_rules
 from extras.webhooks import generate_signature, send_webhook
 from netbox.context_managers import event_tracking
 from utilities.testing import APITestCase, create_test_device
+from utilities.testing.mixins import RQQueueTestMixin
 
 
-class EventRuleTestCase(APITestCase):
+class EventRuleTestCase(RQQueueTestMixin, APITestCase):
 
     def setUp(self):
         super().setUp()
@@ -741,7 +742,7 @@ class EventRuleTestCase(APITestCase):
 
             # silence rqworker (cleaner output) and trigger job execution
             logging.getLogger('rq.worker').setLevel(logging.ERROR)
-            django_rq.get_worker().work(burst=True)
+            self.run_rq_jobs('default')
 
         # Assert that our script was executed without any errors
         script_job.refresh_from_db()

netbox/utilities/testing/mixins.py

@@ -1,4 +1,6 @@
 from django_rq import get_queue
+from django_rq.workers import get_worker
+from rq import SimpleWorker
 
 __all__ = (
     'RQQueueTestMixin',
@@ -16,6 +18,19 @@ class RQQueueTestMixin:
         for queue_name in cls.rq_queue_names:
             get_queue(queue_name).connection.flushall()
 
+    def run_rq_jobs(self, *queue_names, burst=True):
+        """
+        Process queued RQ jobs synchronously for the given queue(s) (defaulting to 'default').
+
+        Uses a non-forking SimpleWorker: the default RQ worker forks a work horse which would
+        inherit the test's open database connection. Two processes sharing one connection
+        corrupts it — on an SSL-encrypted connection this surfaces as "bad record mac" and
+        closes the connection for every subsequent test. SimpleWorker runs jobs in-process,
+        so the connection is never shared.
+        """
+        worker = get_worker(*(queue_names or ('default',)), worker_class=SimpleWorker)
+        worker.work(burst=burst)
+
     def setUp(self):
         super().setUp()