ci: pass custom prefix for dragonfly

.github/workflows/test.yml

@@ -265,4 +265,4 @@ jobs:
       - name: Test with pytest
         run: |
           cd python
-          ./run_tests.sh
+          ./run_tests_dragonfly.sh

python/bullmq/flow_producer.py

@@ -29,6 +29,7 @@ class FlowProducer:
     Instantiate a FlowProducer object
     """
 
+    #pass only queueOpts, no need 2 parameters in next breaking change
     def __init__(self, redisOpts: dict | str = {}, opts: QueueBaseOptions = {}):
         """
         Initialize a connection

python/run_tests_dragonfly.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+BULLMQ_TEST_PREFIX="{b}"
+python3 -m unittest -v tests.bulk_tests
+python3 -m unittest -v tests.delay_tests
+python3 -m unittest -v tests.flow_tests
+python3 -m unittest -v tests.job_tests
+python3 -m unittest -v tests.queue_tests
+python3 -m unittest -v tests.worker_tests

python/tests/bulk_tests.py

@@ -5,26 +5,28 @@
 """
 
 import unittest
+import os
 
 from asyncio import Future
 from bullmq import Queue, Job, Worker
 from uuid import uuid4
 
 queueName = f"__test_queue__{uuid4().hex}"
+prefix = os.environ.get('BULLMQ_TEST_PREFIX') or "bull"
 
 class TestJob(unittest.IsolatedAsyncioTestCase):
 
     async def asyncSetUp(self):
         print("Setting up test queue")
         # Delete test queue
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         await queue.pause()
         await queue.obliterate()
         await queue.close()
 
     async def test_process_jobs(self):
         name = "test"
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
 
         async def process(job: Job, token: str):
             if job.data.get("idx") == 0:
@@ -34,7 +36,7 @@ async def process(job: Job, token: str):
                 self.assertEqual(job.data.get("foo"), "baz")
             return "done"
 
-        worker = Worker(queueName, process)
+        worker = Worker(queueName, process, {"prefix": prefix})
 
         completed_events = Future()
 

python/tests/delay_tests.py

@@ -6,19 +6,21 @@
 
 import unittest
 import time
+import os
 
 from asyncio import Future
 from bullmq import Queue, Job, Worker
 from uuid import uuid4
 
 queueName = f"__test_queue__{uuid4().hex}"
+prefix = os.environ.get('BULLMQ_TEST_PREFIX') or "bull"
 
 class TestJob(unittest.IsolatedAsyncioTestCase):
 
     async def asyncSetUp(self):
         print("Setting up test queue")
         # Delete test queue
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         await queue.pause()
         await queue.obliterate()
         await queue.close()
@@ -27,12 +29,12 @@ async def test_progress_delayed_job_only_after_delayed_time(self):
         delay = 1000
         margin = 1.2
         timestamp = round(time.time() * 1000)
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
 
         async def process(job: Job, token: str):
             return "done"
 
-        worker = Worker(queueName, process)
+        worker = Worker(queueName, process, {"prefix": prefix})
 
         completed_events = Future()
 

python/tests/flow_tests.py

@@ -5,20 +5,22 @@
 """
 
 from asyncio import Future
+import os
 
 from bullmq import Queue, Job, FlowProducer, Worker
 from uuid import uuid4
 
 import unittest
 
 queue_name = f"__test_queue__{uuid4().hex}"
+prefix = os.environ.get('BULLMQ_TEST_PREFIX') or "bull"
 
 class TestJob(unittest.IsolatedAsyncioTestCase):
 
     async def asyncSetUp(self):
         print("Setting up test queue")
         # Delete test queue
-        queue = Queue(queue_name)
+        queue = Queue(queue_name, {"prefix": prefix})
         await queue.pause()
         await queue.obliterate()
         await queue.close()
@@ -48,10 +50,10 @@ async def process2(job: Job, token: str):
             processing_parent.set_result(None)
             return 1
 
-        parent_worker = Worker(parent_queue_name, process2)
-        children_worker = Worker(queue_name, process1)
+        parent_worker = Worker(parent_queue_name, process2, {"prefix": prefix})
+        children_worker = Worker(queue_name, process1, {"prefix": prefix})
 
-        flow = FlowProducer()
+        flow = FlowProducer({}, {"prefix": prefix})
         await flow.add(
             {
                 "name": 'parent-job',
@@ -72,7 +74,7 @@ async def process2(job: Job, token: str):
         await children_worker.close()
         await flow.close()
 
-        parent_queue = Queue(parent_queue_name)
+        parent_queue = Queue(parent_queue_name, {"prefix": prefix})
         await parent_queue.pause()
         await parent_queue.obliterate()
         await parent_queue.close()
@@ -105,10 +107,10 @@ async def process2(job: Job, token: str):
                 processing_parents.set_result(None)
             return 1
 
-        parent_worker = Worker(parent_queue_name, process2)
-        children_worker = Worker(queue_name, process1)
+        parent_worker = Worker(parent_queue_name, process2, {"prefix": prefix})
+        children_worker = Worker(queue_name, process1, {"prefix": prefix})
 
-        flow = FlowProducer()
+        flow = FlowProducer({},{"prefix": prefix})
         await flow.addBulk([
             {
                 "name": 'parent-job-1',
@@ -135,7 +137,7 @@ async def process2(job: Job, token: str):
         await children_worker.close()
         await flow.close()
 
-        parent_queue = Queue(parent_queue_name)
+        parent_queue = Queue(parent_queue_name, {"prefix": prefix})
         await parent_queue.pause()
         await parent_queue.obliterate()
         await parent_queue.close()
@@ -166,10 +168,10 @@ async def process2(job: Job, token: str):
             processing_parent.set_result(children_values)
             return 1
 
-        parent_worker = Worker(parent_queue_name, process2)
-        children_worker = Worker(queue_name, process1)
+        parent_worker = Worker(parent_queue_name, process2, {"prefix": prefix})
+        children_worker = Worker(queue_name, process1, {"prefix": prefix})
 
-        flow = FlowProducer()
+        flow = FlowProducer({},{"prefix": prefix})
         await flow.add(
             {
                 "name": 'parent-job',
@@ -197,13 +199,13 @@ def on_parent_processed(future):
         await children_worker.close()
         await flow.close()
 
-        parent_queue = Queue(parent_queue_name)
+        parent_queue = Queue(parent_queue_name, {"prefix": prefix})
         await parent_queue.pause()
         await parent_queue.obliterate()
         await parent_queue.close()
 
     async def test_get_children_values_on_simple_jobs(self):
-        queue = Queue(queue_name)
+        queue = Queue(queue_name, {"prefix": prefix})
         job = await queue.add("test", {"foo": "bar"}, {"delay": 1500})
         children_values = await job.getChildrenValues()
         self.assertEqual(children_values, {})

python/tests/job_tests.py

@@ -5,24 +5,26 @@
 """
 
 import unittest
+import os
 
 from bullmq import Queue, Job
 from uuid import uuid4
 
 queueName = f"__test_queue__{uuid4().hex}"
+prefix = os.environ.get('BULLMQ_TEST_PREFIX') or "bull"
 
 class TestJob(unittest.IsolatedAsyncioTestCase):
 
     async def asyncSetUp(self):
         print("Setting up test queue")
         # Delete test queue
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         await queue.pause()
         await queue.obliterate()
         await queue.close()
 
     async def test_set_and_get_progress_as_number(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         job = await queue.add("test-job", {"foo": "bar"}, {})
         await job.updateProgress(42)
         stored_job = await Job.fromId(queue, job.id)
@@ -31,7 +33,7 @@ async def test_set_and_get_progress_as_number(self):
         await queue.close()
 
     async def test_set_and_get_progress_as_object(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         job = await queue.add("test-job", {"foo": "bar"}, {})
         await job.updateProgress({"total": 120, "completed": 40})
         stored_job = await Job.fromId(queue, job.id)
@@ -40,7 +42,7 @@ async def test_set_and_get_progress_as_object(self):
         await queue.close()
 
     async def test_get_job_state(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         job = await queue.add("test-job", {"foo": "bar"}, {})
         state = await job.getState()
 
@@ -49,7 +51,7 @@ async def test_get_job_state(self):
         await queue.close()
 
     async def test_job_log(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         firstLog = 'some log text 1'
         secondLog = 'some log text 2'
         job = await queue.add("test-job", {"foo": "bar"}, {})
@@ -63,7 +65,7 @@ async def test_job_log(self):
         await queue.close()
 
     async def test_update_job_data(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         job = await queue.add("test", {"foo": "bar"}, {})
         await job.updateData({"baz": "qux"})
         stored_job = await Job.fromId(queue, job.id)
@@ -73,15 +75,15 @@ async def test_update_job_data(self):
         await queue.close()
 
     async def test_job_data_json_compliant(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         job = await queue.add("test", {"foo": "bar"}, {})
         with self.assertRaises(ValueError):
             await job.updateData({"baz": float('nan')})
 
         await queue.close()
 
     async def test_update_job_data_when_is_removed(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         job = await queue.add("test", {"foo": "bar"}, {})
         await job.remove()
         with self.assertRaises(TypeError):
@@ -90,7 +92,7 @@ async def test_update_job_data_when_is_removed(self):
         await queue.close()
 
     async def test_promote_delayed_job(self):
-        queue = Queue(queueName)
+        queue = Queue(queueName, {"prefix": prefix})
         job = await queue.add("test", {"foo": "bar"}, {"delay": 1500})
         isDelayed = await job.isDelayed()
         self.assertEqual(isDelayed, True)