S3: add EventBridge notification for Object Tagging (#7673)

moto/s3/models.py

@@ -2291,6 +2291,11 @@ def put_object_tagging(
     ) -> FakeKey:
         if key is None:
             raise MissingKey(key=key_name)
+
+        # get bucket for eventbridge notification
+        # we can assume that the key has its bucket
+        bucket = self.get_bucket(key.bucket_name)  # type: ignore
+
         tags_input = self.tagger.convert_dict_to_tags_input(tags)
         # Validation custom to S3
         if tags:
@@ -2304,6 +2309,12 @@ def put_object_tagging(
             raise InvalidTagError(errmsg)
         self.tagger.delete_all_tags_for_resource(key.arn)
         self.tagger.tag_resource(key.arn, tags_input)
+        notifications.send_event(
+            self.account_id,
+            notifications.S3NotificationEvent.OBJECT_TAGGING_PUT_EVENT,
+            bucket,
+            key,
+        )
         return key
 
     def get_bucket_tagging(self, bucket_name: str) -> Dict[str, List[Dict[str, str]]]:

tests/test_s3/test_s3_eventbridge_integration.py

@@ -81,7 +81,7 @@ def test_put_object_notification_ObjectCreated_PUT():
     s3_client.put_object(Bucket=bucket_name, Key="keyname", Body="bodyofnewobject")
 
     events = _get_send_events()
-    assert len(events) == 1
+    assert len(events) == 2
     event_message = json.loads(events[0]["message"])
     assert event_message["detail-type"] == "Object Created"
     assert event_message["source"] == "aws.s3"
@@ -138,7 +138,7 @@ def test_copy_object_notification():
     )
 
     events = _get_send_events()
-    assert len(events) == 2  # [PutObject event, CopyObject event]
+    assert len(events) == 3  # [PutObject, ObjectTagging, CopyObject]
     event_message = json.loads(events[-1]["message"])
     assert event_message["detail-type"] == "Object Created"
     assert event_message["source"] == "aws.s3"
@@ -188,7 +188,7 @@ def test_complete_multipart_upload_notification():
     )
 
     events = _get_send_events()
-    assert len(events) == 2  # [PutObject event, CompleteMultipartUpload event]
+    assert len(events) == 3  # [PutObject, ObjectTagging, CompleteMultipartUpload]
     event_message = json.loads(events[-1]["message"])
     assert event_message["detail-type"] == "Object Created"
     assert event_message["source"] == "aws.s3"
@@ -212,8 +212,8 @@ def test_delete_object_notification():
     s3_client.delete_object(Bucket=bucket_name, Key="keyname")
 
     events = _get_send_events()
-    assert len(events) == 2
-    event_message = json.loads(events[1]["message"])
+    assert len(events) == 3  # [PutObject, ObjectTagging, DeleteObect]
+    event_message = json.loads(events[-1]["message"])
     assert event_message["detail-type"] == "Object Deleted"
     assert event_message["source"] == "aws.s3"
     assert event_message["account"] == ACCOUNT_ID
@@ -235,18 +235,60 @@ def test_restore_key_notifications():
 
     events = _get_send_events()
     event_names = [json.loads(e["message"])["detail"]["reason"] for e in events]
-    assert event_names == ["ObjectCreated", "ObjectRestore"]
+    assert event_names == ["ObjectCreated", "ObjectTagging", "ObjectRestore"]
 
     # Finish the Object Restoration - restore Completes immediately by default
     key.load()
 
     events = _get_send_events()
     event_names = [json.loads(e["message"])["detail"]["reason"] for e in events]
-    assert event_names == ["ObjectCreated", "ObjectRestore", "ObjectRestore"]
+    assert event_names == [
+        "ObjectCreated",
+        "ObjectTagging",
+        "ObjectRestore",
+        "ObjectRestore",
+    ]
 
     # Sanity check - loading the Key does not mean the Restore-event is fired every time
     key.load()
 
     events = _get_send_events()
     event_names = [json.loads(e["message"])["detail"]["reason"] for e in events]
-    assert event_names == ["ObjectCreated", "ObjectRestore", "ObjectRestore"]
+    assert event_names == [
+        "ObjectCreated",
+        "ObjectTagging",
+        "ObjectRestore",
+        "ObjectRestore",
+    ]
+
+
+@mock_aws
+def test_put_object_tagging_notification():
+    resource_names = _seteup_bucket_notification_eventbridge()
+    bucket_name = resource_names["bucket_name"]
+    s3_client = boto3.client("s3", region_name=REGION_NAME)
+
+    # Put Object
+    s3_client.put_object(Bucket=bucket_name, Key="keyname", Body="bodyofnewobject")
+
+    # Put Object Tagging
+    s3_client.put_object_tagging(
+        Bucket=bucket_name,
+        Key="keyname",
+        Tagging={
+            "TagSet": [
+                {"Key": "item1", "Value": "foo"},
+                {"Key": "item2", "Value": "bar"},
+            ]
+        },
+    )
+
+    events = _get_send_events()
+    assert len(events) == 3
+    event_message = json.loads(events[2]["message"])
+    assert event_message["detail-type"] == "Object Tags Added"
+    assert event_message["source"] == "aws.s3"
+    assert event_message["account"] == ACCOUNT_ID
+    assert event_message["region"] == REGION_NAME
+    assert event_message["detail"]["bucket"]["name"] == bucket_name
+    assert event_message["detail"]["reason"] == "ObjectTagging"