diff --git a/moto/s3/responses.py b/moto/s3/responses.py index fb8c504b4..e1ab93860 100644 --- a/moto/s3/responses.py +++ b/moto/s3/responses.py @@ -838,27 +838,35 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin): def _bucket_response_delete_keys(self, request, body, bucket_name): template = self.response_template(S3_DELETE_KEYS_RESPONSE) + body_dict = xmltodict.parse(body) - keys = minidom.parseString(body).getElementsByTagName("Key") - deleted_names = [] - error_names = [] - if len(keys) == 0: + objects = body_dict["Delete"].get("Object", []) + if not isinstance(objects, list): + # We expect a list of objects, but when there is a single node xmltodict does not + # return a list. + objects = [objects] + if len(objects) == 0: raise MalformedXML() - for k in keys: - key_name = k.firstChild.nodeValue + deleted_objects = [] + error_names = [] + + for object_ in objects: + key_name = object_["Key"] + version_id = object_.get("VersionId", None) + success = self.backend.delete_key( - bucket_name, undo_clean_key_name(key_name) + bucket_name, undo_clean_key_name(key_name), version_id=version_id ) if success: - deleted_names.append(key_name) + deleted_objects.append((key_name, version_id)) else: error_names.append(key_name) return ( 200, {}, - template.render(deleted=deleted_names, delete_errors=error_names), + template.render(deleted=deleted_objects, delete_errors=error_names), ) def _handle_range_header(self, request, headers, response_content): @@ -1852,9 +1860,10 @@ S3_BUCKET_GET_VERSIONS = """ S3_DELETE_KEYS_RESPONSE = """ -{% for k in deleted %} +{% for k, v in deleted %} {{k}} +{% if v %}{{v}}{% endif %} {% endfor %} {% for k in delete_errors %} diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py index c52b614e0..86b892315 100644 --- a/tests/test_s3/test_s3.py +++ b/tests/test_s3/test_s3.py @@ -2218,6 +2218,29 @@ def test_boto3_deleted_versionings_list(): assert len(listed["Contents"]) == 1 +@mock_s3 +def test_boto3_delete_objects_for_specific_version_id(): + client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + client.create_bucket(Bucket="blah") + client.put_bucket_versioning( + Bucket="blah", VersioningConfiguration={"Status": "Enabled"} + ) + + client.put_object(Bucket="blah", Key="test1", Body=b"test1a") + client.put_object(Bucket="blah", Key="test1", Body=b"test1b") + + response = client.list_object_versions(Bucket="blah", Prefix="test1") + id_to_delete = [v["VersionId"] for v in response["Versions"] if v["IsLatest"]][0] + + response = client.delete_objects( + Bucket="blah", Delete={"Objects": [{"Key": "test1", "VersionId": id_to_delete}]} + ) + assert response["Deleted"] == [{"Key": "test1", "VersionId": id_to_delete}] + + listed = client.list_objects_v2(Bucket="blah") + assert len(listed["Contents"]) == 1 + + @mock_s3 def test_boto3_delete_versioned_bucket(): client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)