[S3]Add Eventbridge notification for object storage class change

moto/s3/models.py

@@ -230,9 +230,25 @@ def set_metadata(self, metadata: Any, replace: bool = False) -> None:
             self._metadata = {}  # type: ignore
         self._metadata.update(metadata)
 
-    def set_storage_class(self, storage: Optional[str]) -> None:
+
+    @property
+    def storage_class_status(self) -> Optional[str]:
+        return self._storage_class
+
+
+    @storage_class_status.setter
+    def storage_class_status(self, storage: Optional[str]) -> None:
         if storage is not None and storage not in STORAGE_CLASS:
             raise InvalidStorageClass(storage=storage)
+        if self._storage_class != storage:
+            s3_backend = s3_backends[self.account_id][self.partition]
+            bucket = s3_backend.get_bucket(self.bucket_name)  # type: ignore
+            notifications.send_event(
+                self.account_id,
+                notifications.S3NotificationEvent.LIFECYCLE_TRANSITION_EVENT,
+                bucket,
+                key=self,
+            )
         self._storage_class = storage
 
     def set_expiry(self, expiry: Optional[datetime.datetime]) -> None:

tests/test_s3/test_s3_eventbridge_integration.py

@@ -340,3 +340,32 @@ def test_delete_object_tagging_notification():
     assert event_message["region"] == REGION_NAME
     assert event_message["detail"]["bucket"]["name"] == bucket_name
     assert event_message["detail"]["reason"] == "ObjectTagging"
+
+
+@mock_aws
+def test_storage_class_change_notifications():
+    resource_names = _seteup_bucket_notification_eventbridge()
+    bucket_name = resource_names["bucket_name"]
+    s3_client = boto3.client("s3", region_name=REGION_NAME)
+
+    s3_client.put_object(Bucket=bucket_name, Key="keyname", Body="bodyofnewobject")
+
+    # Change the storage class
+    new_class = "GLACIER"
+    copy_source = {
+        "Bucket": bucket_name,
+        "Key": "keyname"
+    }
+
+    s3_client.copy_object(copy_source, bucket_name, "keyname_copy", ExtraArgs={'StorageClass': new_class})
+
+    events = _get_send_events()
+    assert len(events) == 3
+    event_message = json.loads(events[2]["message"])
+    print(event_message)
+    assert event_message["detail-type"] == "Object Created"
+    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"] == "ObjectCreated"