Added testcases

Author: Rakshil Modi

tests/functional/s3/__init__.py

@@ -257,6 +257,74 @@ def mp_copy_responses(self):
             self.complete_mpu_response(),
         ]
 
+    def list_object_versions_request(
+        self, bucket, prefix=None, **override_kwargs
+    ):
+        params = {
+            'Bucket': bucket,
+        }
+        if prefix is not None:
+            params['Prefix'] = prefix
+        params.update(override_kwargs)
+        return 'ListObjectVersions', params
+
+    def list_object_versions_response(self, keys, **override_kwargs):
+        versions = []
+        delete_markers = []
+        for key in keys:
+            version = {
+                'Key': key,
+                'VersionId': 'version1',
+                'LastModified': '00:00:00Z',
+                'Size': 100,
+                'ETag': '"foo-1"',
+                'IsLatest': False,
+            }
+            delete_marker = {
+                'Key': key,
+                'VersionId': 'deletemarker1',
+                'LastModified': '00:00:00Z',
+                'IsLatest': True,
+            }
+            if override_kwargs:
+                version.update(override_kwargs)
+                delete_marker.update(override_kwargs)
+
+            versions.append(version)
+            delete_markers.append(delete_marker)
+        return {'Versions': versions, 'DeleteMarkers': delete_markers}
+
+    def delete_objects_request(
+        self, bucket, objects_to_delete, **override_kwargs
+    ):
+        params = {
+            'Bucket': bucket,
+            'Delete': {
+                'Objects': objects_to_delete,
+                'Quiet': False,
+            },
+        }
+        params.update(override_kwargs)
+        return 'DeleteObjects', params
+
+    def delete_objects_response(self, keys, **override_kwargs):
+        deleted = []
+
+        for key in keys:
+            # Create deleted entries for each version that was deleted
+            deleted_objects = [
+                {'Key': key, 'VersionId': 'version1'},
+                {'Key': key, 'VersionId': 'deletemarker1'},
+            ]
+
+            if override_kwargs:
+                for obj in deleted_objects:
+                    obj.update(override_kwargs)
+
+            deleted.extend(deleted_objects)
+
+        return {'Deleted': deleted}
+
 
 class BaseS3CLIRunnerTest(unittest.TestCase):
     def setUp(self):

tests/functional/s3/test_rm_command.py

@@ -71,6 +71,49 @@ def test_recursive_delete_with_requests(self):
             ]
         )
 
+    def test_delete_all_versions_of_single_object(self):
+        self.parsed_responses = [
+            self.list_object_versions_response(['mykey']),
+            self.delete_objects_response(['mykey']),
+        ]
+        cmdline = f'{self.prefix} s3://mybucket/mykey --all-versions'
+        self.run_cmd(cmdline, expected_rc=0)
+
+        self.assert_operations_called(
+            [
+                self.list_object_versions_request('mybucket', prefix='mykey'),
+                self.delete_objects_request(
+                    'mybucket',
+                    [
+                        {'Key': 'mykey', 'VersionId': 'version1'},
+                        {'Key': 'mykey', 'VersionId': 'deletemarker1'},
+                    ],
+                ),
+            ]
+        )
+
+    def test_recursive_delete_all_versions_of_all_objects(self):
+        self.parsed_responses = [
+            self.list_object_versions_response(['mykey1', 'mykey2']),
+            self.delete_objects_response(['mykey1', 'mykey2']),
+        ]
+        cmdline = f'{self.prefix} s3://mybucket/ --all-versions --recursive'
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assert_operations_called(
+            [
+                self.list_object_versions_request('mybucket', prefix=''),
+                self.delete_objects_request(
+                    'mybucket',
+                    [
+                        {'Key': 'mykey1', 'VersionId': 'version1'},
+                        {'Key': 'mykey2', 'VersionId': 'version1'},
+                        {'Key': 'mykey1', 'VersionId': 'deletemarker1'},
+                        {'Key': 'mykey2', 'VersionId': 'deletemarker1'},
+                    ],
+                ),
+            ]
+        )
+
 
 class TestRmWithCRTClient(BaseCRTTransferClientTest):
     def test_delete_using_crt_client(self):