fix: 315764 improve _handle_deduplicated

Author: roma-valor

src/country_workspace/contrib/hope/push/orchestration.py

@@ -171,11 +171,20 @@ def _lock_and_fail_rdp(rdp_id: int, *, reason: str) -> None:
 
 
 def _handle_deduplicated(rdp: Rdp, rdp_id: int, findings_count: int) -> None:
-    origin_job = AsyncJob.objects.filter(rdp_id=rdp_id).order_by("-id").first()
-    max_findings_percent: int = origin_job.config.get("max_dedup_findings_percent", 0) if origin_job else 0
+    try:
+        origin_job = AsyncJob.objects.filter(rdp_id=rdp_id).latest("id")
+    except AsyncJob.DoesNotExist:
+        _lock_and_fail_rdp(rdp_id, reason="missing origin AsyncJob for threshold config")
+        return
+
+    max_findings_percent: int = origin_job.config.get("max_dedup_findings_percent", 0)
 
     total_individuals = qs_individuals_for_rdp(rdp=rdp).count()
-    findings_rate = findings_count / total_individuals * 100 if total_individuals > 0 else float(findings_count)
+    if total_individuals == 0:
+        _lock_and_fail_rdp(rdp_id, reason="no individuals linked to RDP; cannot compute findings rate")
+        return
+
+    findings_rate = findings_count / total_individuals * 100
 
     logger.info(
         "dedup_callback_handle: rdp_id=%s DEDUPLICATED findings_count=%s total_individuals=%s "
@@ -195,27 +204,28 @@ def _handle_deduplicated(rdp: Rdp, rdp_id: int, findings_count: int) -> None:
         return
 
     with transaction.atomic():
-        locked = lock_rdp_for_update(pk=rdp_id)
-        if locked.status != Rdp.PushStatus.DEDUP_PENDING:
+        locked_rdp = lock_rdp_for_update(pk=rdp_id)
+        if locked_rdp.status != Rdp.PushStatus.DEDUP_PENDING:
             logger.warning(
                 "dedup_callback_handle: rdp_id=%s skip push queue; status changed to %s",
                 rdp_id,
-                locked.status,
+                locked_rdp.status,
             )
             return
-        locked.status = Rdp.PushStatus.PENDING
-        locked.save(update_fields=["status"])
+        locked_rdp.status = Rdp.PushStatus.PENDING
+        locked_rdp.save(update_fields=["status"])
+
+        push_job = AsyncJob.objects.create(
+            description=f"Push RDP {rdp_id} to HOPE (post-dedup)",
+            type=AsyncJob.JobType.TASK,
+            owner=locked_rdp.pushed_by,
+            action=fqn(push_existing_rdp_core),
+            program=locked_rdp.program,
+            rdp=locked_rdp,
+            config={"rdp_id": rdp_id},
+        )
+        transaction.on_commit(push_job.queue)
 
-    push_job = AsyncJob.objects.create(
-        description=f"Push RDP {rdp_id} to HOPE (post-dedup)",
-        type=AsyncJob.JobType.TASK,
-        owner=rdp.pushed_by,
-        action=fqn(push_existing_rdp_core),
-        program=rdp.program,
-        rdp=rdp,
-        config={"rdp_id": rdp_id},
-    )
-    push_job.queue()
     logger.info(
         "dedup_callback_handle: rdp_id=%s status DEDUP_PENDING→PENDING; queued push job_id=%s",
         rdp_id,

tests/contrib/hope/push/test_dedup_callback_flow.py

@@ -206,6 +206,23 @@ def _patch_dedup_status(mocker: MockerFixture, *, state: str, findings_count: in
     return policy
 
 
+def _patch_origin_job(mocker: MockerFixture, config: dict | None = None, *, missing: bool = False):
+    qs = mocker.MagicMock()
+    if missing:
+        qs.latest.side_effect = AsyncJob.DoesNotExist
+    else:
+        qs.latest.return_value = mocker.MagicMock(config=config if config is not None else {})
+    mocker.patch.object(AsyncJob.objects, "filter", return_value=qs)
+    return qs
+
+
+def _patch_individuals_count(mocker: MockerFixture, count: int) -> None:
+    mocker.patch(
+        f"{MOD}.qs_individuals_for_rdp",
+        return_value=mocker.MagicMock(count=mocker.MagicMock(return_value=count)),
+    )
+
+
 def test_dedup_callback_handle_rdp_not_found(mocker: MockerFixture) -> None:
     _patch_rdp_for_push(mocker, not_found=True)
     set_status = mocker.patch(f"{MOD}.set_rdp_push_status")
@@ -320,25 +337,17 @@ def test_dedup_callback_handle_deduplicated_within_threshold_queues_push(mocker:
     locked_pending.status = Rdp.PushStatus.DEDUP_PENDING
     _patch_rdp_for_push(mocker, rdp)
     _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=5)
-    origin_job = mocker.MagicMock(config={"max_dedup_findings_percent": 10})
-    mocker.patch.object(
-        AsyncJob.objects,
-        "filter",
-        return_value=mocker.MagicMock(
-            order_by=mocker.MagicMock(return_value=mocker.MagicMock(first=mocker.MagicMock(return_value=origin_job)))
-        ),
-    )
-    mocker.patch(
-        f"{MOD}.qs_individuals_for_rdp", return_value=mocker.MagicMock(count=mocker.MagicMock(return_value=100))
-    )
+    _patch_origin_job(mocker, {"max_dedup_findings_percent": 10})
+    _patch_individuals_count(mocker, 100)
     mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked_pending)
     push_job = mocker.MagicMock()
     create_job = mocker.patch.object(AsyncJob.objects, "create", return_value=push_job)
+    on_commit = mocker.patch(f"{MOD}.transaction.on_commit")
 
     dedup_callback_handle(rdp_id=99)
 
     create_job.assert_called_once()
-    push_job.queue.assert_called_once()
+    on_commit.assert_called_once_with(push_job.queue)
     assert locked_pending.status == Rdp.PushStatus.PENDING
 
 
@@ -350,23 +359,16 @@ def test_dedup_callback_handle_deduplicated_skips_push_when_status_changed_under
     locked.status = Rdp.PushStatus.FAILURE
     _patch_rdp_for_push(mocker, rdp)
     _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=5)
-    origin_job = mocker.MagicMock(config={"max_dedup_findings_percent": 10})
-    mocker.patch.object(
-        AsyncJob.objects,
-        "filter",
-        return_value=mocker.MagicMock(
-            order_by=mocker.MagicMock(return_value=mocker.MagicMock(first=mocker.MagicMock(return_value=origin_job)))
-        ),
-    )
-    mocker.patch(
-        f"{MOD}.qs_individuals_for_rdp", return_value=mocker.MagicMock(count=mocker.MagicMock(return_value=100))
-    )
+    _patch_origin_job(mocker, {"max_dedup_findings_percent": 10})
+    _patch_individuals_count(mocker, 100)
     mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked)
     create_job = mocker.patch.object(AsyncJob.objects, "create")
+    on_commit = mocker.patch(f"{MOD}.transaction.on_commit")
 
     dedup_callback_handle(rdp_id=99)
 
     create_job.assert_not_called()
+    on_commit.assert_not_called()
     locked.save.assert_not_called()
 
 
@@ -376,17 +378,8 @@ def test_dedup_callback_handle_deduplicated_exceeds_threshold_marks_failure(mock
     locked.status = Rdp.PushStatus.DEDUP_PENDING
     _patch_rdp_for_push(mocker, rdp)
     _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=20)
-    origin_job = mocker.MagicMock(config={"max_dedup_findings_percent": 10})
-    mocker.patch.object(
-        AsyncJob.objects,
-        "filter",
-        return_value=mocker.MagicMock(
-            order_by=mocker.MagicMock(return_value=mocker.MagicMock(first=mocker.MagicMock(return_value=origin_job)))
-        ),
-    )
-    mocker.patch(
-        f"{MOD}.qs_individuals_for_rdp", return_value=mocker.MagicMock(count=mocker.MagicMock(return_value=100))
-    )
+    _patch_origin_job(mocker, {"max_dedup_findings_percent": 10})
+    _patch_individuals_count(mocker, 100)
     mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked)
     set_status = mocker.patch(f"{MOD}.set_rdp_push_status")
     create_job = mocker.patch.object(AsyncJob.objects, "create")
@@ -404,17 +397,8 @@ def test_dedup_callback_handle_default_threshold_is_zero(mocker: MockerFixture)
     locked.status = Rdp.PushStatus.DEDUP_PENDING
     _patch_rdp_for_push(mocker, rdp)
     _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=1)
-    origin_job = mocker.MagicMock(config={})  # no max_dedup_findings_percent
-    mocker.patch.object(
-        AsyncJob.objects,
-        "filter",
-        return_value=mocker.MagicMock(
-            order_by=mocker.MagicMock(return_value=mocker.MagicMock(first=mocker.MagicMock(return_value=origin_job)))
-        ),
-    )
-    mocker.patch(
-        f"{MOD}.qs_individuals_for_rdp", return_value=mocker.MagicMock(count=mocker.MagicMock(return_value=100))
-    )
+    _patch_origin_job(mocker, {})
+    _patch_individuals_count(mocker, 100)
     mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked)
     set_status = mocker.patch(f"{MOD}.set_rdp_push_status")
 
@@ -430,21 +414,48 @@ def test_dedup_callback_handle_zero_findings_within_threshold_queues_push(mocker
     locked_pending.status = Rdp.PushStatus.DEDUP_PENDING
     _patch_rdp_for_push(mocker, rdp)
     _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=0)
-    origin_job = mocker.MagicMock(config={})  # default threshold=0
-    mocker.patch.object(
-        AsyncJob.objects,
-        "filter",
-        return_value=mocker.MagicMock(
-            order_by=mocker.MagicMock(return_value=mocker.MagicMock(first=mocker.MagicMock(return_value=origin_job)))
-        ),
-    )
-    mocker.patch(
-        f"{MOD}.qs_individuals_for_rdp", return_value=mocker.MagicMock(count=mocker.MagicMock(return_value=100))
-    )
+    _patch_origin_job(mocker, {})
+    _patch_individuals_count(mocker, 100)
     mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked_pending)
     push_job = mocker.MagicMock()
     mocker.patch.object(AsyncJob.objects, "create", return_value=push_job)
+    on_commit = mocker.patch(f"{MOD}.transaction.on_commit")
+
+    dedup_callback_handle(rdp_id=99)
+
+    on_commit.assert_called_once_with(push_job.queue)
+
+
+def test_dedup_callback_handle_missing_origin_job_marks_failure(mocker: MockerFixture) -> None:
+    rdp = _make_rdp(mocker)
+    locked = mocker.MagicMock()
+    locked.status = Rdp.PushStatus.DEDUP_PENDING
+    _patch_rdp_for_push(mocker, rdp)
+    _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=0)
+    _patch_origin_job(mocker, missing=True)
+    mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked)
+    set_status = mocker.patch(f"{MOD}.set_rdp_push_status")
+    create_job = mocker.patch.object(AsyncJob.objects, "create")
+
+    dedup_callback_handle(rdp_id=99)
+
+    set_status.assert_called_once_with(rdp=locked, status=Rdp.PushStatus.FAILURE, hope_rdi_id="N/A")
+    create_job.assert_not_called()
+
+
+def test_dedup_callback_handle_zero_individuals_marks_failure(mocker: MockerFixture) -> None:
+    rdp = _make_rdp(mocker)
+    locked = mocker.MagicMock()
+    locked.status = Rdp.PushStatus.DEDUP_PENDING
+    _patch_rdp_for_push(mocker, rdp)
+    _patch_dedup_status(mocker, state=DeduplicationSetState.DEDUPLICATED, findings_count=0)
+    _patch_origin_job(mocker, {"max_dedup_findings_percent": 10})
+    _patch_individuals_count(mocker, 0)
+    mocker.patch(f"{MOD}.lock_rdp_for_update", return_value=locked)
+    set_status = mocker.patch(f"{MOD}.set_rdp_push_status")
+    create_job = mocker.patch.object(AsyncJob.objects, "create")
 
     dedup_callback_handle(rdp_id=99)
 
-    push_job.queue.assert_called_once()
+    set_status.assert_called_once_with(rdp=locked, status=Rdp.PushStatus.FAILURE, hope_rdi_id="N/A")
+    create_job.assert_not_called()