diff --git a/AGENTS.md b/AGENTS.md
index 73ea3502..8946588b 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -136,8 +136,9 @@ Check these integration details when changing the affected paths:
   15 and 14, and nothing else isolates it. A committing test otherwise enqueues
   jobs that a live worker would run with test primary keys. Logical database
   numbers do not make a shared server safe from `FLUSHALL`.
-- **Do not use `RQQueueTestMixin` to clear queues.** It calls server-wide
-  `flushall()`.
+- **`--parallel` workers share the test Redis database.** A test class that
+  reaches Redis lists NetBox's `RQQueueTestMixin` first. See the
+  [testing conventions](./docs/development/conventions.md#testing).
 - **Migration fixtures need real temporary source storage.** `ManagedFile.storage`
   and the built-in loader do not necessarily use the same backend instance.
 - **`has_perm(perm)` without an object is not a target-object check.** It
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index fc73da24..69e72fdc 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -99,9 +99,10 @@ services matching that configuration. The database account must be able to creat
 the test database.
 
 **Use isolated development services, never production.** The test configuration
-uses Redis databases 15 for tasks and 14 for caching. No worker serving a different
-NetBox database may consume those test queues. Redis database numbers do not
-protect against commands that clear the whole server.
+uses Redis databases 15 for tasks and 14 for caching. The suite clears the tasks
+database during a run, whichever one the configuration names. No worker serving a
+different NetBox database may consume those test queues. Redis database numbers do
+not protect against commands that clear the whole server.
 
 These exports apply only inside the subshell:
 
diff --git a/docs/development/conventions.md b/docs/development/conventions.md
index 098b71e3..fee8531a 100644
--- a/docs/development/conventions.md
+++ b/docs/development/conventions.md
@@ -137,6 +137,14 @@ and APIs through NetBox's test client rather than mocking persistence. Use
 transactional tests for concurrency and transaction boundaries. A sequential
 `TestCase` alone does not demonstrate that an interleaving is safe.
 
+Parallel test workers get their own databases but share the test Redis
+database. A test class that reaches Redis lists NetBox's `RQQueueTestMixin`,
+from `utilities.testing.mixins`, as its first base and calls `super().setUp()`
+first. A class reaches Redis when it reads or clears a queue, writes an RQ job,
+or lets an unpatched enqueue run, either by committing or through
+`captureOnCommitCallbacks(execute=True)`. The mixin keeps such classes from
+running at the same time and clears the task database around each test.
+
 Keep tests in the same commit as the implementation they cover. Report the
 NetBox ref, command, configuration and result, including skipped or unrun checks.
 Distinguish actual integration tests from checks that replace a framework or
diff --git a/netbox_scripts/tests/api/test_locking.py b/netbox_scripts/tests/api/test_locking.py
index 9da79329..8f27fd73 100644
--- a/netbox_scripts/tests/api/test_locking.py
+++ b/netbox_scripts/tests/api/test_locking.py
@@ -13,7 +13,6 @@
 
 from unittest import mock
 
-import django_rq
 from django.test import TransactionTestCase, override_settings
 from django.urls import reverse
 from rest_framework.test import APIClient
@@ -31,9 +30,6 @@ class ConditionalWriteLockOrderTestCase(ObjectPermissionTestMixin, TransactionTe
 
     def setUp(self):
         self.enterContext(override_settings(STORAGES=IN_MEMORY_STORAGES))
-        # This case commits, so a deletion really enqueues its cleanup. The queue is isolated by
-        # testing/configuration.py, and draining it leaves nothing behind even there.
-        self.addCleanup(django_rq.get_queue('default').empty)
         self.project = ScriptProject.objects.create(name='Deploy Devices', key='deploy-devices')
         self.user = User.objects.create_user(username='testuser')
         self.token = Token.objects.create(user=self.user)
diff --git a/netbox_scripts/tests/migration/test_cutover.py b/netbox_scripts/tests/migration/test_cutover.py
index b5122037..d58eb1f3 100644
--- a/netbox_scripts/tests/migration/test_cutover.py
+++ b/netbox_scripts/tests/migration/test_cutover.py
@@ -25,6 +25,7 @@
 from netbox_scripts.tests.migration.test_staging import LEGACY_SCRIPT as SYNCED_SCRIPT
 from netbox_scripts.tests.migration.test_staging import LegacySourceMixin
 from users.models import ObjectPermission
+from utilities.testing.mixins import RQQueueTestMixin
 
 LEGACY_SCRIPT = b"""from extras.scripts import Script
 
@@ -35,7 +36,7 @@ def run(self, data, commit):
 """
 
 
-class CutoverTestCase(TestCase):
+class CutoverTestCase(RQQueueTestMixin, TestCase):
     """Crossing the fence: what it refuses, what it records, and what it closes."""
 
     @classmethod
@@ -58,6 +59,7 @@ def setUpClass(cls):
         )
 
     def setUp(self):
+        super().setUp()
         self.migration = MigrationRun.objects.create(state=MigrationStateChoices.STAGING)
         self.source = DataSource.objects.create(name='Automation', type='local', source_url='file:///tmp/a')
         self.data_file = DataFile.objects.create(
@@ -149,13 +151,8 @@ def rq_task(self, job, kwargs):
             timeout=600,
         )
         task.save()
-        self.addCleanup(self.discard_task, queue, str(job.job_id))
         return task
 
-    @staticmethod
-    def discard_task(queue, task_id):
-        queue.connection.delete(f'rq:job:{task_id}')
-
     def report_module(self, *, auto_sync_enabled=False):
         """Create a REPORTS-root module, which this migration does not cover."""
         data_file = DataFile.objects.create(
diff --git a/netbox_scripts/tests/migration/test_history.py b/netbox_scripts/tests/migration/test_history.py
index 5b48a8db..91dd25ab 100644
--- a/netbox_scripts/tests/migration/test_history.py
+++ b/netbox_scripts/tests/migration/test_history.py
@@ -20,6 +20,7 @@
 from netbox_scripts.runtime.exceptions import ScriptFileImportError
 from netbox_scripts.tests.migration.test_references import ReferenceMigrationMixin
 from users.models import ObjectPermission
+from utilities.testing.mixins import RQQueueTestMixin
 
 # A legacy script with a variable, so the journal's primary keys have something to resolve back to.
 VARIABLE_SCRIPT = b"""from extras.scripts import ObjectVar, Script
@@ -49,7 +50,7 @@ def run(self, data, commit):
 """
 
 
-class LegacyJobMixin(ReferenceMigrationMixin):
+class LegacyJobMixin(RQQueueTestMixin, ReferenceMigrationMixin):
     """The built-in content plus a Job history and a queue to read schedules out of."""
 
     def setUp(self):
@@ -58,8 +59,6 @@ def setUp(self):
         self.variable_script = Script.objects.get(module=self.variable_module, name='Retire')
         self.site = Site.objects.create(name='HQ', slug='hq')
         self.queue = django_rq.get_queue('default')
-        self.queue.empty()
-        self.addCleanup(self.queue.empty)
 
     def legacy_job(self, *, script=None, status=JobStatusChoices.STATUS_COMPLETED, **fields):
         """One built-in Custom Script Job row, in whatever state the case needs."""
@@ -89,7 +88,6 @@ def legacy_schedule(self, *, script=None, data=None, commit=True, scheduled=None
             timeout=timeout,
         )
         task.save()
-        self.addCleanup(self.queue.connection.delete, f'rq:job:{job.job_id}')
         return job
 
     def plugin_script(self, class_name='Deploy'):
diff --git a/netbox_scripts/tests/migration/test_verification.py b/netbox_scripts/tests/migration/test_verification.py
index f4109479..7409c575 100644
--- a/netbox_scripts/tests/migration/test_verification.py
+++ b/netbox_scripts/tests/migration/test_verification.py
@@ -13,9 +13,10 @@
 from netbox_scripts.models import MigrationRun, NetBoxScript, ScriptProject
 from netbox_scripts.tests.migration.test_cleanup import CleanupMixin
 from users.models import ObjectPermission
+from utilities.testing.mixins import RQQueueTestMixin
 
 
-class VerificationMixin(CleanupMixin):
+class VerificationMixin(RQQueueTestMixin, CleanupMixin):
     """A migration taken as far as each case needs, and a way to read one check out of the report."""
 
     @staticmethod
@@ -34,7 +35,6 @@ def rq_task(self, job):
             timeout=600,
         )
         task.save()
-        self.addCleanup(queue.connection.delete, f'rq:job:{job.job_id}')
         return task
 
     def waiting_schedule(self, run, *, name='nightly deploy', status=JobStatusChoices.STATUS_SCHEDULED):
diff --git a/netbox_scripts/tests/storage/test_concurrency.py b/netbox_scripts/tests/storage/test_concurrency.py
index 7174cbd4..4ce787ca 100644
--- a/netbox_scripts/tests/storage/test_concurrency.py
+++ b/netbox_scripts/tests/storage/test_concurrency.py
@@ -17,7 +17,6 @@
 import uuid
 from unittest import mock
 
-import django_rq
 from django.db.models.signals import post_save, pre_save
 from django.test import TransactionTestCase, override_settings
 
@@ -30,6 +29,7 @@
 from netbox_scripts.storage.manifest import compute_digest
 from netbox_scripts.storage.paths import revision_prefix
 from netbox_scripts.tests.plugin_testing import IN_MEMORY_STORAGES, SecondSession
+from utilities.testing.mixins import RQQueueTestMixin
 
 SOURCE = {'hello.py': b'print("hi")\n'}
 
@@ -42,15 +42,12 @@ def manifest_for(files):
     )
 
 
-class SerializationTestCase(TransactionTestCase):
+class SerializationTestCase(RQQueueTestMixin, TransactionTestCase):
     """Shared fixture: one upload project whose content can be staged and reclaimed."""
 
     def setUp(self):
+        super().setUp()
         self.enterContext(override_settings(STORAGES=IN_MEMORY_STORAGES))
-        # This case commits, so the revision-deletion signal's on_commit callback really enqueues a
-        # cleanup job. The queue is isolated by testing/configuration.py, and draining it here means
-        # the suite leaves nothing behind even there.
-        self.addCleanup(django_rq.get_queue('default').empty)
         self.storage = config.get_storage()
         self.project = ScriptProject.objects.create(name='Deploy Devices', key='deploy-devices')
 
diff --git a/netbox_scripts/tests/test_event_rules.py b/netbox_scripts/tests/test_event_rules.py
index d8195578..3388a634 100644
--- a/netbox_scripts/tests/test_event_rules.py
+++ b/netbox_scripts/tests/test_event_rules.py
@@ -11,6 +11,7 @@
 from netbox_scripts.event_rules import RunNetBoxScriptAction
 from netbox_scripts.models import NetBoxScript
 from netbox_scripts.tests.test_execution import MAKES_A_TAG, ScriptJobTestMixin
+from utilities.testing.mixins import RQQueueTestMixin
 
 SLUG = 'netbox_scripts.run'
 
@@ -183,7 +184,7 @@ def test_a_script_in_another_project_does_not_resolve(self):
             self.action.resolve_import_object(f'other:{self.netbox_script.full_name}')
 
 
-class ActionInputTestCase(ScriptJobTestMixin, TestCase):
+class ActionInputTestCase(RQQueueTestMixin, ScriptJobTestMixin, TestCase):
     """That the rule's action_data reaches the script as its input, verbatim."""
 
     # enqueue_run keeps input off the Job row on purpose, because a variable resolves to a model
@@ -191,11 +192,7 @@ class ActionInputTestCase(ScriptJobTestMixin, TestCase):
 
     def setUp(self):
         super().setUp()
-        # The queue is isolated by testing/configuration.py, so this only separates one test from
-        # the next. Never clear with NetBox's RQQueueTestMixin, which uses a server-wide flushall().
         self.queue = django_rq.get_queue('default')
-        self.queue.empty()
-        self.addCleanup(self.queue.empty)
         self.publish({'deploy.py': MAKES_A_TAG})
         self.netbox_script = self.script()
         self.action = get_event_rule_action(SLUG)
diff --git a/netbox_scripts/tests/test_event_sources.py b/netbox_scripts/tests/test_event_sources.py
index 32c25c17..a13beffa 100644
--- a/netbox_scripts/tests/test_event_sources.py
+++ b/netbox_scripts/tests/test_event_sources.py
@@ -13,6 +13,7 @@
     ScriptProject,
     ScriptProjectRevision,
 )
+from utilities.testing.mixins import RQQueueTestMixin
 
 EVERY_MODEL = (ScriptProject, ScriptProjectRevision, ScriptFile, NetBoxScript)
 
@@ -83,16 +84,12 @@ def test_every_model_serializes(self):
                 self.assertEqual(serialize_for_event(instance)['id'], instance.pk)
 
 
-class EventDispatchTestCase(TestCase):
+class EventDispatchTestCase(RQQueueTestMixin, TestCase):
     """That a rule matching a plugin object reaches the queue."""
 
     def setUp(self):
         super().setUp()
-        # The queue is isolated by testing/configuration.py, so this only separates one test from
-        # the next. Never clear with NetBox's RQQueueTestMixin, which uses a server-wide flushall().
         self.queue = django_rq.get_queue('default')
-        self.queue.empty()
-        self.addCleanup(self.queue.empty)
         self.project = ScriptProject.objects.create(name='Dispatched', key='dispatched')
         self.object_type = ObjectType.objects.get_for_model(ScriptProject)
         webhook = Webhook.objects.create(name='Receiver', payload_url='http://localhost/hook/')
diff --git a/netbox_scripts/tests/test_execution.py b/netbox_scripts/tests/test_execution.py
index 9546da28..b18cfcd5 100644
--- a/netbox_scripts/tests/test_execution.py
+++ b/netbox_scripts/tests/test_execution.py
@@ -39,6 +39,7 @@
 from utilities.datetime import local_now
 from utilities.exceptions import AbortScript as LegacyAbortScript
 from utilities.request import NetBoxFakeRequest
+from utilities.testing.mixins import RQQueueTestMixin
 
 
 class Creating(Script):
@@ -477,7 +478,7 @@ def test_relative_imports_remain_available_during_construction_and_form_building
             self.assertIsNotNone(instance.as_form())
 
 
-class EnqueueRunTestCase(ScriptJobTestMixin, TestCase):
+class EnqueueRunTestCase(RQQueueTestMixin, ScriptJobTestMixin, TestCase):
     def test_enqueue_pins_the_active_revision_on_the_job(self):
         revision = self.publish({'deploy.py': MAKES_A_TAG})
         script = self.script()
@@ -554,8 +555,6 @@ def test_a_run_no_event_drove_records_none(self):
     def test_an_immediate_run_never_reaches_the_queue(self):
         self.publish({'deploy.py': MAKES_A_TAG})
         queue = django_rq.get_queue('default')
-        queue.empty()
-        self.addCleanup(queue.empty)
 
         # Core queues every non-immediate job from a transaction.on_commit hook, so an
         # implementation reaching that branch would hand the finished run to a worker as well.