rwm: fix storage_delete, use paging when deleting storage

rwm.py

@@ -273,10 +273,36 @@ class StorageManager:
     def storage_delete(self, bucket_name):
         """storage delete"""
 
-        bucket = self.s3.Bucket(bucket_name)
-        bucket.objects.all().delete()
-        bucket.object_versions.all().delete()
-        bucket.delete()
+        # delete all objects
+        paginator = self.s3.meta.client.get_paginator('list_objects')
+        objects = []
+        for page in paginator.paginate(Bucket=bucket_name):
+            for item in page.get("Contents", []):
+                objects.append((bucket_name, item["Key"]))
+        for item in objects:
+            self.s3.Object(*item).delete()
+
+        paginator = self.s3.meta.client.get_paginator('list_object_versions')
+
+        # delete all object versions
+        objects = []
+        for page in paginator.paginate(Bucket=bucket_name):
+            for item in page.get("Versions", []):
+                objects.append((bucket_name, item["Key"], item["VersionId"]))
+        for item in objects:
+            self.s3.ObjectVersion(*item).delete()
+
+        # delete all delete markers
+        objects = []
+        for page in paginator.paginate(Bucket=bucket_name):
+            for item in page.get("DeleteMarkers", []):
+                objects.append((bucket_name, item["Key"], item["VersionId"]))
+        for item in objects:
+            self.s3.ObjectVersion(*item).delete()
+
+        # delete bucket
+        self.s3.Bucket(bucket_name).delete()
+
         return 0
 
     @staticmethod

tests/test_storage.py

@@ -232,3 +232,20 @@ def test_storage_save_state_error_handling(tmpworkdir: str):  # pylint: disable=
     mock = Mock(side_effect=TypeError("dummy"))
     with patch.object(rwm.StorageManager, "_bucket_state", mock):
         assert rwm.StorageManager("http://localhost", "", "").storage_save_state("dummy") == 1
+
+
+@pytest.mark.skipif('PYTEST_SLOW' not in os.environ, reason='slow on devnode, runs in CI')
+def test_storage_delete(tmpworkdir: str, radosuser_admin: rwm.StorageManager):  # pylint: disable=unused-argument
+    """test storage delete"""
+
+    bucket_name = "testbuckx"
+    bucket = radosuser_admin.storage_create(bucket_name, "dummy")
+
+    bucket.upload_fileobj(BytesIO(b"dummydata0"), "dummykey")
+    for idx in range(803):
+        bucket.Object("dummykey").delete()
+        bucket.upload_fileobj(BytesIO(f"dummydata{idx}".encode()), "dummykey")
+
+    assert radosuser_admin.storage_delete(bucket.name) == 0
+
+    assert not radosuser_admin.bucket_exist(bucket.name)