fix(drivers-event-listener-griptape-cloud): add type/timestamp fallbacks for custom events

griptape/drivers/event_listener/griptape_cloud_event_listener_driver.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import time
 from typing import Optional
 from urllib.parse import urljoin
 
@@ -72,8 +73,8 @@ def try_publish_event_payload_batch(self, event_payload_batch: list[dict]) -> No
     def _get_event_request(self, event_payload: dict) -> dict:
         return {
             "payload": event_payload,
-            "timestamp": event_payload["timestamp"],
-            "type": event_payload["type"],
+            "timestamp": event_payload.get("timestamp", time.time()),
+            "type": event_payload.get("type", "UserEvent"),
         }
 
     def _post_event(self, json: list[dict] | dict) -> None:

tests/unit/drivers/event_listener/test_griptape_cloud_event_listener_driver.py

@@ -1,6 +1,6 @@
 import os
 import time
-from unittest.mock import MagicMock, Mock
+from unittest.mock import ANY, MagicMock, Mock
 
 import pytest
 
@@ -104,3 +104,27 @@ def test_try_publish_event_payload_batch(self, mock_post, driver):
                 json=driver._get_event_request(event.to_dict()),
                 headers={"Authorization": "Bearer foo bar"},
             )
+
+    @pytest.mark.parametrize(
+        ("event", "expected"),
+        [
+            (
+                MockEvent().to_dict(),
+                {
+                    "payload": {
+                        "id": ANY,
+                        "meta": {},
+                        "timestamp": ANY,
+                        "type": "MockEvent",
+                    },
+                    "timestamp": ANY,
+                    "type": "MockEvent",
+                },
+            ),
+            ({}, {"payload": {}, "timestamp": ANY, "type": "UserEvent"}),
+        ],
+    )
+    def test__get_event_request(self, driver, event, expected):
+        result = driver._get_event_request(event)
+
+        assert result == expected