S3: Adding a validation method for s3 notification event name (#7253)

moto/s3/models.py

@@ -41,6 +41,7 @@ from moto.s3.exceptions import (
     HeadOnDeleteMarker,
     InvalidBucketName,
     InvalidNotificationDestination,
+    InvalidNotificationEvent,
     InvalidPart,
     InvalidPublicAccessBlockConfiguration,
     InvalidRequest,
@@ -821,6 +822,9 @@ class Notification(BaseModel):
             random.choice(string.ascii_letters + string.digits) for _ in range(50)
         )
         self.arn = arn
+        for event_name in events:
+            if not notifications.S3NotificationEvent.is_event_valid(event_name):
+                raise InvalidNotificationEvent(event_name)
         self.events = events
         self.filters = filters if filters else {}
 

moto/s3/notifications.py

@@ -49,6 +49,16 @@ class S3NotificationEvent(str, Enum):
     def events(self) -> List[str]:
         return sorted([item.value for item in S3NotificationEvent])
 
+    @classmethod
+    def is_event_valid(self, event_name: str) -> bool:
+        # Ex) s3:ObjectCreated:Put
+        if event_name in self.events():
+            return True
+        # Ex) event name without `s3:` like ObjectCreated:Put
+        if event_name in [e[:3] for e in self.events()]:
+            return True
+        return False
+
 
 def _get_s3_event(
     event_name: str, bucket: Any, key: Any, notification_id: str

moto/s3/responses.py

@@ -35,7 +35,6 @@ from .exceptions import (
     InvalidMaxPartArgument,
     InvalidMaxPartNumberArgument,
     InvalidNotificationARN,
-    InvalidNotificationEvent,
     InvalidObjectState,
     InvalidPartOrder,
     InvalidRange,
@@ -62,7 +61,6 @@ from .models import (
     get_canned_acl,
     s3_backends,
 )
-from .notifications import S3NotificationEvent
 from .select_object_content import serialize_select
 from .utils import (
     ARCHIVE_STORAGE_CLASSES,
@@ -2121,10 +2119,6 @@ class S3Response(BaseResponse):
                     if not isinstance(n["Event"], list):
                         n["Event"] = [n["Event"]]
 
-                    for event in n["Event"]:
-                        if event not in S3NotificationEvent.events():
-                            raise InvalidNotificationEvent(event)
-
                     # Parse out the filters:
                     if n.get("Filter"):
                         # Error if S3Key is blank: