Merge pull request #2132 from Projectplace/tagging-directive

Support x-amz-tagging-directive in s3 copy_object

moto/s3/responses.py

@@ -1202,6 +1202,10 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
             if mdirective is not None and mdirective == "REPLACE":
                 metadata = metadata_from_headers(request.headers)
                 new_key.set_metadata(metadata, replace=True)
+            tdirective = request.headers.get("x-amz-tagging-directive")
+            if tdirective == "REPLACE":
+                tagging = self._tagging_from_headers(request.headers)
+                new_key.set_tagging(tagging)
             template = self.response_template(S3_OBJECT_COPY_RESPONSE)
             response_headers.update(new_key.response_dict)
             return 200, response_headers, template.render(key=new_key)

tests/test_s3/test_s3.py

@@ -2143,6 +2143,34 @@ def test_boto3_copy_object_from_unversioned_to_versioned_bucket():
     obj2_version_new.should_not.equal(None)
 
 
+@mock_s3
+def test_boto3_copy_object_with_replacement_tagging():
+    client = boto3.client("s3", region_name="eu-north-1")
+    client.create_bucket(Bucket="mybucket")
+    client.put_object(
+        Bucket="mybucket", Key="original", Body=b"test", Tagging="tag=old"
+    )
+
+    client.copy_object(
+        CopySource={"Bucket": "mybucket", "Key": "original"},
+        Bucket="mybucket",
+        Key="copy1",
+        TaggingDirective="REPLACE",
+        Tagging="tag=new",
+    )
+    client.copy_object(
+        CopySource={"Bucket": "mybucket", "Key": "original"},
+        Bucket="mybucket",
+        Key="copy2",
+        TaggingDirective="COPY",
+    )
+
+    tags1 = client.get_object_tagging(Bucket="mybucket", Key="copy1")["TagSet"]
+    tags1.should.equal([{"Key": "tag", "Value": "new"}])
+    tags2 = client.get_object_tagging(Bucket="mybucket", Key="copy2")["TagSet"]
+    tags2.should.equal([{"Key": "tag", "Value": "old"}])
+
+
 @mock_s3
 def test_boto3_deleted_versionings_list():
     client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)