From ecce4ba5a7dd73508c95a6ab5d72da61c6aee511 Mon Sep 17 00:00:00 2001 From: Ankush Menat Date: Tue, 9 May 2023 13:21:07 +0530 Subject: [PATCH] test: deduplication with unique job id --- frappe/core/doctype/rq_job/test_rq_job.py | 9 ++++++++- frappe/utils/background_jobs.py | 7 ++++--- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/frappe/core/doctype/rq_job/test_rq_job.py b/frappe/core/doctype/rq_job/test_rq_job.py index 3b8e2d5cc3..13d2fc37e7 100644 --- a/frappe/core/doctype/rq_job/test_rq_job.py +++ b/frappe/core/doctype/rq_job/test_rq_job.py @@ -11,7 +11,7 @@ import frappe from frappe.core.doctype.rq_job.rq_job import RQJob, remove_failed_jobs, stop_job from frappe.tests.utils import FrappeTestCase, timeout from frappe.utils import cstr, execute_in_shell -from frappe.utils.background_jobs import is_job_queued +from frappe.utils.background_jobs import is_job_enqueued, is_job_queued class TestRQJob(FrappeTestCase): @@ -107,6 +107,13 @@ class TestRQJob(FrappeTestCase): _, stderr = execute_in_shell("bench worker --queue short,default --burst", check_exit_code=True) self.assertIn("quitting", cstr(stderr)) + @timeout(20) + def test_job_id_dedup(self): + job_id = "test_dedup" + job = frappe.enqueue(self.BG_JOB, sleep=10, job_id=job_id) + self.assertTrue(is_job_enqueued(job_id)) + stop_job(job.id) + def test_func(fail=False, sleep=0): if fail: diff --git a/frappe/utils/background_jobs.py b/frappe/utils/background_jobs.py index e6ae283d41..e0fa455f60 100755 --- a/frappe/utils/background_jobs.py +++ b/frappe/utils/background_jobs.py @@ -19,7 +19,6 @@ import frappe.monitor from frappe import _ from frappe.utils import cstr, get_bench_id from frappe.utils.commands import log -from frappe.utils.deprecations import deprecation_warning from frappe.utils.redis_queue import RedisQueue if TYPE_CHECKING: @@ -452,9 +451,11 @@ def create_job_id(job_id: str) -> str: def is_job_enqueued(job_id: str) -> str: + from rq.job import Job + try: - job = Job.fetch(job_id, connection=get_redis_conn()) + job = Job.fetch(create_job_id(job_id), connection=get_redis_conn()) except NoSuchJobError: return False - return job.status in ("queued", "started") + return job.get_status() in ("queued", "started")