Fix S3 delete_objects errors and ensure test coverage

Author: rajvi0106

wp1/logic/selection.py

@@ -144,7 +144,7 @@ def delete_keys_from_storage(keys):
     )
 
     fully_successful = True
-    errors = []
+
     # TODO: Check for errors in the response. For now, just pretend
     # it's always successful.
     # for e in (resp if resp is not None else {}).get('Errors', []):
@@ -154,13 +154,15 @@ def delete_keys_from_storage(keys):
 
     if isinstance(resp, dict):
         errors = resp.get("Errors", [])
-    if errors:
-        fully_successful = False
-        for e in errors:
-            logger.warning(
-                "Error deleting %r:[code=%r, msg=%r]",
-                (e["Key"], e["Code"], e["Message"]),
-            )
+        if errors:
+            fully_successful = False
+            for e in errors:
+                logger.warning(
+                    "Error deleting %r: [code=%r, msg=%r]",
+                    e["Key"],
+                    e["Code"],
+                    e["Message"],
+                )
 
     return fully_successful
 

wp1/logic/selection_test.py

@@ -383,6 +383,33 @@ def test_delete_objects_non_dict_response(self, patched_connect_storage):
         result = logic_selection.delete_keys_from_storage([b"key1"])
         self.assertTrue(result)
 
+    @patch("wp1.logic.selection.connect_storage")
+    def test_delete_keys_from_storage_with_errors(self, patched_connect_storage):
+        s3 = MagicMock()
+        bucket = MagicMock()
+        patched_connect_storage.return_value = s3
+        s3.bucket = bucket
+        bucket.delete_objects.return_value = {
+            "Errors": [
+                {
+                    "Key": "object/key/1",
+                    "Code": "AccessDenied",
+                    "Message": "Access Denied",
+                }
+            ]
+        }
+        actual = logic_selection.delete_keys_from_storage(
+            [b"object/key/1", b"object/key/2"]
+        )
+
+        bucket.delete_objects.assert_called_once_with(
+            Delete={
+                "Objects": [{"Key": "object/key/1"}, {"Key": "object/key/2"}],
+                "Quiet": True,
+            }
+        )
+        self.assertFalse(actual)
+
     def test_zim_file_requested_at_for(self):
         self._insert_selections()
         actual = logic_selection.zim_file_requested_at_for(self.wp10db, "xyz1")