NRL-1798 Build result in separate function

Author: sandyforresternhs

scripts/delete_pointers_by_id.py

@@ -59,6 +59,44 @@ def _load_pointers_from_file(pointers_file: str) -> list[str]:
     return [line.strip() for line in content.splitlines() if line.strip()]
 
 
+def _build_and_write_result(
+    pointers_to_delete,
+    ods_code,
+    matched_pointers,
+    mismatched_pointers,
+    not_found_pointers,
+    pointers_deleted,
+    failed_deletes,
+    start_time,
+    end_time,
+    output_file_path,
+):
+    result = {
+        "pointers_to_delete": len(pointers_to_delete),
+        "ods_code": ods_code,
+        "ods_code_matched": {"count": len(matched_pointers), "ids": matched_pointers},
+        "ods_code_mismatched": {
+            "count": len(mismatched_pointers),
+            "ids": mismatched_pointers,
+        },
+        "pointer_not_found": {
+            "count": len(not_found_pointers),
+            "ids": not_found_pointers,
+        },
+        "deleted_pointers": {"count": len(pointers_deleted), "ids": pointers_deleted},
+        "failed_deletes": {"count": len(failed_deletes), "ids": failed_deletes},
+        "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
+    }
+    try:
+        _write_result_file(result, output_file_path)
+    except Exception as exc:
+        result["_output_error"] = (
+            f"Failed to write result file {output_file_path}: {exc}"
+        )
+    _print_summary(result)
+    return result
+
+
 def _write_result_file(result: Dict[str, Any], output_file: str) -> None:
     """
     Atomically write result dict to output_file as JSON.
@@ -237,25 +275,20 @@ def _delete_pointers_by_id(
         script_dir, f"delete_results_{ods_code}_{stamp}.json"
     )
 
-    if len(pointers_to_delete) == 0:
-        result = {
-            "pointers_to_delete": 0,
-            "ods_code": ods_code,
-            "ods_code_matched": {"count": 0, "ids": []},
-            "ods_code_mismatched": {"count": 0, "ids": []},
-            "pointer_not_found": {"count": 0, "ids": []},
-            "deleted_pointers": {"count": 0, "ids": []},
-            "failed_deletes": {"count": 0, "ids": []},
-            "deletes-took-secs": 0,
-        }
-        try:
-            _write_result_file(result, output_file_path)
-        except Exception as exc:
-            result["_output_error"] = (
-                f"Failed to write result file {output_file_path}: {exc}"
-            )
-        _print_summary(result)
-        return result
+    if not pointers_to_delete:
+        end_time = datetime.now(tz=timezone.utc)
+        return _build_and_write_result(
+            pointers_to_delete,
+            ods_code,
+            [],
+            [],
+            [],
+            [],
+            [],
+            start_time,
+            end_time,
+            output_file_path,
+        )
 
     print(
         f"Validating {len(pointers_to_delete)} pointers against ODS code {ods_code}..."
@@ -268,33 +301,21 @@ def _delete_pointers_by_id(
         f"Validate pointer's ODS code: {len(matched_pointers)} matched, {len(mismatched_pointers)} mismatched"
     )
 
-    if len(matched_pointers) == 0:
+    if not matched_pointers:
         print(f"None of the pointer IDs are a match for ODS code {ods_code}. Exiting.")
         end_time = datetime.now(tz=timezone.utc)
-        result = {
-            "pointers_to_delete": len(pointers_to_delete),
-            "ods_code": ods_code,
-            "ods_code_matched": {
-                "count": len(matched_pointers),
-                "ids": matched_pointers,
-            },
-            "ods_code_mismatched": {
-                "count": len(mismatched_pointers),
-                "ids": mismatched_pointers,
-            },
-            "pointer_not_found": {"count": 0, "ids": []},
-            "deleted_pointers": {"count": 0, "ids": []},
-            "failed_deletes": {"count": 0, "ids": []},
-            "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
-        }
-        try:
-            _write_result_file(result, output_file_path)
-        except Exception as exc:
-            result["_output_error"] = (
-                f"Failed to write result file {output_file_path}: {exc}"
-            )
-        _print_summary(result)
-        return result
+        return _build_and_write_result(
+            pointers_to_delete,
+            ods_code,
+            matched_pointers,
+            mismatched_pointers,
+            [],
+            [],
+            [],
+            start_time,
+            end_time,
+            output_file_path,
+        )
 
     print(f"Checking existence of {len(matched_pointers)} pointers in {table_name}...")
     existing_pointers, not_found_pointers = _batch_get_existing_pointers(
@@ -305,69 +326,40 @@ def _delete_pointers_by_id(
         f"Found {len(existing_pointers)} existing pointers to delete, {len(not_found_pointers)} not found."
     )
 
-    if len(existing_pointers) == 0:
+    if not existing_pointers:
         print("No pointers found to delete. Exiting.")
         end_time = datetime.now(tz=timezone.utc)
-        result = {
-            "pointers_to_delete": len(pointers_to_delete),
-            "ods_code": ods_code,
-            "ods_code_matched": {
-                "count": len(matched_pointers),
-                "ids": matched_pointers,
-            },
-            "ods_code_mismatched": {
-                "count": len(mismatched_pointers),
-                "ids": mismatched_pointers,
-            },
-            "pointer_not_found": {
-                "count": len(not_found_pointers),
-                "ids": not_found_pointers,
-            },
-            "deleted_pointers": {"count": 0, "ids": []},
-            "failed_deletes": {"count": 0, "ids": []},
-            "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
-        }
-        try:
-            _write_result_file(result, output_file_path)
-        except Exception as exc:
-            result["_output_error"] = (
-                f"Failed to write result file {output_file_path}: {exc}"
-            )
-        _print_summary(result)
-        return result
+        return _build_and_write_result(
+            pointers_to_delete,
+            ods_code,
+            matched_pointers,
+            mismatched_pointers,
+            not_found_pointers,
+            [],
+            [],
+            start_time,
+            end_time,
+            output_file_path,
+        )
 
     # Proceed with deletion using BatchWriteItem
     pointers_deleted, failed_deletes = _batch_delete_pointers(
         table_name, existing_pointers
     )
 
     end_time = datetime.now(tz=timezone.utc)
-
-    result = {
-        "pointers_to_delete": len(pointers_to_delete),
-        "ods_code": ods_code,
-        "ods_code_matched": {"count": len(matched_pointers), "ids": matched_pointers},
-        "ods_code_mismatched": {
-            "count": len(mismatched_pointers),
-            "ids": mismatched_pointers,
-        },
-        "pointer_not_found": {
-            "count": len(not_found_pointers),
-            "ids": not_found_pointers,
-        },
-        "deleted_pointers": {"count": len(pointers_deleted), "ids": pointers_deleted},
-        "failed_deletes": {"count": len(failed_deletes), "ids": failed_deletes},
-        "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
-    }
-
-    try:
-        _write_result_file(result, output_file_path)
-    except Exception as exc:
-        result["_output_error"] = (
-            f"Failed to write result file {output_file_path}: {exc}"
-        )
-
-    _print_summary(result)
+    result = _build_and_write_result(
+        pointers_to_delete,
+        ods_code,
+        matched_pointers,
+        mismatched_pointers,
+        not_found_pointers,
+        pointers_deleted,
+        failed_deletes,
+        start_time,
+        end_time,
+        output_file_path,
+    )
     print(" Done")
     return result
 

scripts/tests/test_delete_pointers_by_id.py

@@ -142,65 +142,83 @@ def test_both_params_provided(self):
                 "t", "G3H9E", pointers_to_delete=["a"], pointers_file="./f"
             )
 
-    @patch("delete_pointers_by_id._write_result_file")
+    @patch("delete_pointers_by_id._build_and_write_result")
     @patch("delete_pointers_by_id._check_pointers_match_ods_code")
     @patch("delete_pointers_by_id._batch_get_existing_pointers")
     @patch("delete_pointers_by_id._batch_delete_pointers")
-    def test_no_matched_ods_codes(self, mock_delete, mock_get, mock_check, mock_write):
+    def test_no_matched_ods_codes(self, mock_delete, mock_get, mock_check, mock_build):
         mock_check.return_value = ([], ["RAT-1", "RAT-2"])
+        mock_build.return_value = {
+            "ods_code_matched": {"count": 0},
+            "ods_code_mismatched": {"count": 2},
+        }
         result = _delete_pointers_by_id(
             "t", "G3H9E", pointers_to_delete=["RAT-1", "RAT-2"]
         )
         assert result["ods_code_matched"]["count"] == 0
         assert result["ods_code_mismatched"]["count"] == 2
-        mock_write.assert_called_once()
+        mock_build.assert_called_once()
 
-    @patch("delete_pointers_by_id._write_result_file")
+    @patch("delete_pointers_by_id._build_and_write_result")
     @patch("delete_pointers_by_id._check_pointers_match_ods_code")
     @patch("delete_pointers_by_id._batch_get_existing_pointers")
     @patch("delete_pointers_by_id._batch_delete_pointers")
-    def test_successful_flow(self, mock_delete, mock_get, mock_check, mock_write):
+    def test_successful_flow(self, mock_delete, mock_get, mock_check, mock_build):
         ids = ["G3H9E-1", "G3H9E-2"]
         mock_check.return_value = (ids, [])
         mock_get.return_value = (ids, [])
         mock_delete.return_value = (ids, [])
+        mock_build.return_value = {
+            "pointers_to_delete": 2,
+            "deleted_pointers": {"count": 2},
+            "pointer_not_found": {"count": 0},
+        }
         result = _delete_pointers_by_id("t", "G3H9E", pointers_to_delete=ids)
         assert result["pointers_to_delete"] == 2
         assert result["deleted_pointers"]["count"] == 2
         assert result["pointer_not_found"]["count"] == 0
-        mock_write.assert_called_once()
+        mock_build.assert_called_once()
 
-    @patch("delete_pointers_by_id._write_result_file")
+    @patch("delete_pointers_by_id._build_and_write_result")
     @patch("delete_pointers_by_id._check_pointers_match_ods_code")
     @patch("delete_pointers_by_id._batch_get_existing_pointers")
     @patch("delete_pointers_by_id._batch_delete_pointers")
-    def test_partial_with_failures(self, mock_delete, mock_get, mock_check, mock_write):
+    def test_partial_with_failures(self, mock_delete, mock_get, mock_check, mock_build):
         matched = ["G3H9E-1", "G3H9E-2", "G3H9E-3"]
         mock_check.return_value = (matched, ["RAT-1"])
         mock_get.return_value = (matched, [])
         mock_delete.return_value = (["G3H9E-1", "G3H9E-2"], ["G3H9E-3"])
+        mock_build.return_value = {
+            "ods_code_mismatched": {"count": 1},
+            "deleted_pointers": {"count": 2},
+            "failed_deletes": {"count": 1, "ids": ["G3H9E-3"]},
+        }
         result = _delete_pointers_by_id(
             "t", "G3H9E", pointers_to_delete=matched + ["RAT-1"]
         )
         assert result["ods_code_mismatched"]["count"] == 1
         assert result["deleted_pointers"]["count"] == 2
         assert result["failed_deletes"]["count"] == 1
         assert "G3H9E-3" in result["failed_deletes"]["ids"]
-        mock_write.assert_called_once()
+        mock_build.assert_called_once()
 
-    @patch("delete_pointers_by_id._write_result_file")
+    @patch("delete_pointers_by_id._build_and_write_result")
     @patch("delete_pointers_by_id._check_pointers_match_ods_code")
     @patch("delete_pointers_by_id._batch_get_existing_pointers")
     @patch("delete_pointers_by_id._batch_delete_pointers")
     def test_some_pointers_not_found(
-        self, mock_delete, mock_get, mock_check, mock_write
+        self, mock_delete, mock_get, mock_check, mock_build
     ):
         matched = ["G3H9E-1", "G3H9E-2", "G3H9E-3"]
         mock_check.return_value = (matched, [])
         mock_get.return_value = (["G3H9E-1", "G3H9E-2"], ["G3H9E-3"])
         mock_delete.return_value = (["G3H9E-1", "G3H9E-2"], [])
+        mock_build.return_value = {
+            "pointer_not_found": {"count": 1, "ids": ["G3H9E-3"]},
+            "deleted_pointers": {"count": 2},
+        }
         result = _delete_pointers_by_id("t", "G3H9E", pointers_to_delete=matched)
         assert result["pointer_not_found"]["count"] == 1
         assert "G3H9E-3" in result["pointer_not_found"]["ids"]
         assert result["deleted_pointers"]["count"] == 2
-        mock_write.assert_called_once()
+        mock_build.assert_called_once()