diff --git a/queue_job/job.py b/queue_job/job.py
index 457df97ecd..af0bafb943 100644
--- a/queue_job/job.py
+++ b/queue_job/job.py
@@ -300,7 +300,7 @@ def job_record_with_same_identity_key(self):
             .search(
                 [
                     ("identity_key", "=", self.identity_key),
-                    ("state", "in", [PENDING, ENQUEUED]),
+                    ("state", "in", [WAIT_DEPENDENCIES, PENDING, ENQUEUED]),
                 ],
                 limit=1,
             )
diff --git a/queue_job/tests/common.py b/queue_job/tests/common.py
index b6ebf96f71..aa8ea73d28 100644
--- a/queue_job/tests/common.py
+++ b/queue_job/tests/common.py
@@ -233,6 +233,7 @@ def by_graph(job):
                 self._perform_graph_jobs(jobs)
             else:
                 self._perform_single_jobs(jobs)
+        self.enqueued_jobs = []
 
     def _perform_single_jobs(self, jobs):
         # we probably don't want to replicate a perfect order here, but at
@@ -252,11 +253,14 @@ def _perform_graph_jobs(self, jobs):
 
     def _add_job(self, *args, **kwargs):
         job = Job(*args, **kwargs)
-        self.enqueued_jobs.append(job)
-
-        patcher = mock.patch.object(job, "store")
-        self._store_patchers.append(patcher)
-        patcher.start()
+        if not job.identity_key or all(
+            j.identity_key != job.identity_key for j in self.enqueued_jobs
+        ):
+            self.enqueued_jobs.append(job)
+
+            patcher = mock.patch.object(job, "store")
+            self._store_patchers.append(patcher)
+            patcher.start()
 
         job_args = kwargs.pop("args", None) or ()
         job_kwargs = kwargs.pop("kwargs", None) or {}
diff --git a/test_queue_job/tests/test_delay_mocks.py b/test_queue_job/tests/test_delay_mocks.py
index 3238f15d5a..d513a90797 100644
--- a/test_queue_job/tests/test_delay_mocks.py
+++ b/test_queue_job/tests/test_delay_mocks.py
@@ -82,6 +82,38 @@ def test_trap_jobs_on_with_delay_recordset_partial_properties(self):
                 ),
             )
 
+    def test_trap_with_identity_key(self):
+        with trap_jobs() as trap:
+            self.env["test.queue.job"].button_that_uses_with_delay()
+            trap.assert_jobs_count(1)
+            trap.assert_jobs_count(1, only=self.env["test.queue.job"].testing_method)
+
+            trap.assert_enqueued_job(
+                self.env["test.queue.job"].testing_method,
+                args=(1,),
+                kwargs={"foo": 2},
+                properties=dict(
+                    channel="root.test",
+                    description="Test",
+                    eta=15,
+                    identity_key=identity_exact,
+                    max_retries=1,
+                    priority=15,
+                ),
+            )
+
+            # Should not enqueue again
+            self.env["test.queue.job"].button_that_uses_with_delay()
+            trap.assert_jobs_count(1)
+
+            trap.perform_enqueued_jobs()
+            # Should no longer be enqueued
+            trap.assert_jobs_count(0)
+
+            # Can now requeue
+            self.env["test.queue.job"].button_that_uses_with_delay()
+            trap.assert_jobs_count(1)
+
     def test_trap_jobs_on_with_delay_assert_model_count_mismatch(self):
         recordset = self.env["test.queue.job"].create({"name": "test"})
         with trap_jobs() as trap:
@@ -219,6 +251,8 @@ def test_trap_jobs_perform(self):
             # perform the jobs
             trap.perform_enqueued_jobs()
 
+            trap.assert_jobs_count(0)
+
             logs = self.env["ir.logging"].search(
                 [
                     ("name", "=", "test_queue_job"),