Strengthen regression coverage for #25 and #26

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## Unreleased
 
+**Added:**
+
+- Deterministic regression coverage for job arrivals during claiming (#25) and
+  propagation of thread-pool logging failures in the missing-process test (#26).
+
 **Fixed:**
 
 - Avoid duplicate recurring-task enqueues when multiple schedulers race on the

tests/test_executions.py

@@ -1,6 +1,10 @@
+from concurrent.futures import ThreadPoolExecutor
 from datetime import timedelta
+from unittest import skipUnless
+from unittest.mock import patch
 
-from django.test import TestCase
+from django.db import connections
+from django.test import TestCase, TransactionTestCase
 from django.utils import timezone
 
 import steady_queue
@@ -14,6 +18,7 @@
     ScheduledExecution,
     Semaphore,
 )
+from steady_queue.models.claimed_execution import ClaimedExecutionQuerySet
 from tests.dummy.tasks import dummy_task, limited_task, limited_task_with_lambda_key
 
 
@@ -106,6 +111,57 @@ def test_claim_with_zero_limit_returns_empty(self):
         self.assertEqual(ReadyExecution.objects.count(), 1)
 
 
+class ConcurrentClaimTestCase(TestHelperMixin, TransactionTestCase):
+    databases = {"default", "queue"}
+
+    @skipUnless(
+        connections["queue"].features.has_select_for_update_skip_locked,
+        "The queue database must support SELECT FOR UPDATE SKIP LOCKED",
+    )
+    def test_higher_priority_arrival_does_not_change_claimed_candidates(self):
+        """#25: keep the selected jobs stable across insertion and retrieval."""
+        process = self.create_test_process()
+        original = self.create_job_in_queue("default", priority=0)
+        bulk_create = ClaimedExecutionQuerySet.bulk_create
+
+        def enqueue_higher_priority_job():
+            try:
+                return self.create_job_in_queue("default", priority=10).pk
+            finally:
+                connections.close_all()
+
+        def insert_claims_then_enqueue(queryset, *args, **kwargs):
+            result = bulk_create(queryset, *args, **kwargs)
+            # Commit an arrival on another connection after claims are inserted,
+            # but before claiming() retrieves them. No sleeps or timing races.
+            with ThreadPoolExecutor(max_workers=1) as executor:
+                arrival = executor.submit(enqueue_higher_priority_job)
+                arrivals.append(arrival.result(timeout=10))
+            return result
+
+        arrivals = []
+        with patch.object(
+            ClaimedExecutionQuerySet, "bulk_create", insert_claims_then_enqueue
+        ):
+            claimed = ReadyExecution.objects.claim(["*"], 1, process.pk)
+
+        self.assertEqual([execution.job_id for execution in claimed], [original.pk])
+        self.assertFalse(
+            ReadyExecution.objects.filter(
+                job_id__in=ClaimedExecution.objects.values("job_id")
+            ).exists()
+        )
+        self.assertEqual(
+            list(ReadyExecution.objects.values_list("job_id", flat=True)), arrivals
+        )
+
+        # Draining the remaining queue must not try to claim the original again.
+        next_claimed = ReadyExecution.objects.claim(["*"], 1, process.pk)
+        self.assertEqual([execution.job_id for execution in next_claimed], arrivals)
+        self.assertEqual(ReadyExecution.objects.claim(["*"], 1, process.pk), [])
+        self.assertEqual(ClaimedExecution.objects.count(), 2)
+
+
 class ClaimedExecutionTestCase(TestHelperMixin, TestCase):
     """Tests for ClaimedExecution behavior."""
 

tests/test_process_lifecycle.py

@@ -1,5 +1,5 @@
 from datetime import timedelta
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 from django.test import TestCase
 from django.utils import timezone
@@ -272,23 +272,37 @@ class PoolPostNilProcessTest(TestCase):
 
     def test_pool_post_does_not_access_execution_process(self):
         """Pool.post should not crash when execution.process is None."""
-        import time
-
         from steady_queue.processes.pool import Pool
 
-        pool = Pool(size=1, on_idle=lambda: None)
+        on_idle = MagicMock()
+        pool = Pool(size=1, on_idle=on_idle, worker_name="test-worker")
+        self.addCleanup(pool.executor.shutdown, wait=True)
 
         execution = MagicMock()
         execution.process = None
         execution.job_id = 1
         execution.job.class_name = "test_task"
         execution.pk = 1
 
-        # Should not raise AttributeError
-        pool.post(execution)
-        # Give the thread pool time to execute
-        time.sleep(0.2)
-        pool.shutdown()
+        submit = pool.executor.submit
+        futures = []
+
+        def capture_future(*args, **kwargs):
+            future = submit(*args, **kwargs)
+            futures.append(future)
+            return future
+
+        with self.assertLogs("steady_queue", level="INFO") as logs:
+            with patch.object(pool.executor, "submit", capture_future):
+                pool.post(execution)
+            # Exceptions after perform() live in the Future. Checking only that
+            # perform() ran allowed the original #26 logging crash to pass.
+            futures[0].result(timeout=5)
 
-        # Verify perform was called successfully
         execution.perform.assert_called_once()
+        self.assertEqual(
+            logs.output,
+            ["INFO:steady_queue:test-worker completed job 1 test_task"],
+        )
+        self.assertEqual(pool.idle_threads, 1)
+        on_idle.assert_called_once_with()