vassili/seitime-frappe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

test: Wait for enqueued job to execute

Browse source
This commit is contained in:
Suraj Shetty 2021-03-10 10:29:07 +05:30
parent bcd3f7540d
commit 7c05fd2a17
1 changed files with 6 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
frappe/tests/test_background_jobs.py
View file

@ -5,24 +5,26 @@ from rq import Queue
import frappe
from frappe.core.page.background_jobs.background_jobs import remove_failed_jobs
from frappe.utils.background_jobs import get_redis_conn
import time
class TestBackgroundJobs(unittest.TestCase):
def test_remove_failed_jobs(self):
frappe.enqueue(method="frappe.tests.test_background_jobs.fail_function")
frappe.enqueue(method="frappe.tests.test_background_jobs.fail_function", queue="short")
# wait for enqueued job to execute
time.sleep(2)
conn = get_redis_conn()
queues = Queue.all(conn)
for queue in queues:
if queue.name == "default":
if queue.name == "short":
fail_registry = queue.failed_job_registry
self.assertGreater(fail_registry.count, 0)
remove_failed_jobs()
for queue in queues:
if queue.name == "default":
if queue.name == "short":
fail_registry = queue.failed_job_registry
self.assertEqual(fail_registry.count, 0)