tests

tests/unit/events/test_action_chunk_event.py

@@ -0,0 +1,35 @@
+import pytest
+
+from griptape.common import ActionCallDeltaMessageContent
+from griptape.events import ActionChunkEvent
+
+
+class TestCompletionChunkEvent:
+    @pytest.fixture()
+    def action_chunk_event(self):
+        return ActionChunkEvent(token="foo bar")
+
+    def test_token(self, action_chunk_event):
+        assert action_chunk_event.token == "foo bar"
+        assert action_chunk_event.index == 0
+        assert action_chunk_event.tag is None
+        assert action_chunk_event.name is None
+        assert action_chunk_event.path is None
+
+    def test_to_dict(self, action_chunk_event):
+        assert action_chunk_event.to_dict()["token"] == "foo bar"
+
+    def test_from_delta_message_content(self):
+        content = ActionCallDeltaMessageContent(
+            index=0,
+            partial_input="foo bar",
+            tag="tag",
+            name="name",
+            path="path",
+        )
+        event = ActionChunkEvent.from_delta_message_content(content)
+        assert event.token == "foo bar"
+        assert event.index == 0
+        assert event.tag == "tag"
+        assert event.name == "name"
+        assert event.path == "path"

tests/unit/events/test_completion_chunk_event.py → tests/unit/events/test_base_chunk_event.py

@@ -1,12 +1,12 @@
 import pytest
 
-from griptape.events import CompletionChunkEvent
+from griptape.events import BaseChunkEvent
 
 
 class TestCompletionChunkEvent:
     @pytest.fixture()
     def completion_chunk_event(self):
-        return CompletionChunkEvent(token="foo bar")
+        return BaseChunkEvent(token="foo bar")
 
     def test_token(self, completion_chunk_event):
         assert completion_chunk_event.token == "foo bar"

tests/unit/events/test_event_listener.py

@@ -3,7 +3,8 @@
 import pytest
 
 from griptape.events import (
-    CompletionChunkEvent,
+    ActionChunkEvent,
+    BaseChunkEvent,
     EventBus,
     EventListener,
     FinishActionsSubtaskEvent,
@@ -14,6 +15,7 @@
     StartPromptEvent,
     StartStructureRunEvent,
     StartTaskEvent,
+    TextChunkEvent,
 )
 from griptape.events.base_event import BaseEvent
 from griptape.structures import Pipeline
@@ -59,7 +61,9 @@ def test_typed_listeners(self, pipeline, mock_config):
         finish_subtask_event_handler = Mock()
         start_structure_run_event_handler = Mock()
         finish_structure_run_event_handler = Mock()
-        completion_chunk_handler = Mock()
+        base_chunk_handler = Mock()
+        text_chunk_handler = Mock()
+        action_chunk_handler = Mock()
 
         EventBus.add_event_listeners(
             [
@@ -71,7 +75,9 @@ def test_typed_listeners(self, pipeline, mock_config):
                 EventListener(finish_subtask_event_handler, event_types=[FinishActionsSubtaskEvent]),
                 EventListener(start_structure_run_event_handler, event_types=[StartStructureRunEvent]),
                 EventListener(finish_structure_run_event_handler, event_types=[FinishStructureRunEvent]),
-                EventListener(completion_chunk_handler, event_types=[CompletionChunkEvent]),
+                EventListener(base_chunk_handler, event_types=[BaseChunkEvent]),
+                EventListener(text_chunk_handler, event_types=[TextChunkEvent]),
+                EventListener(action_chunk_handler, event_types=[ActionChunkEvent]),
             ]
         )
 
@@ -88,7 +94,9 @@ def test_typed_listeners(self, pipeline, mock_config):
         finish_subtask_event_handler.assert_called_once()
         start_structure_run_event_handler.assert_called_once()
         finish_structure_run_event_handler.assert_called_once()
-        completion_chunk_handler.assert_called_once()
+        base_chunk_handler.assert_called_once()
+        text_chunk_handler.assert_called_once()
+        action_chunk_handler.assert_called_once()
 
     def test_add_remove_event_listener(self, pipeline):
         EventBus.clear_event_listeners()

tests/unit/events/test_text_chunk_event.py

@@ -0,0 +1,25 @@
+import pytest
+
+from griptape.common import TextDeltaMessageContent
+from griptape.events import TextChunkEvent
+
+
+class TestCompletionChunkEvent:
+    @pytest.fixture()
+    def text_chunk_event(self):
+        return TextChunkEvent(token="foo bar")
+
+    def test_token(self, text_chunk_event):
+        assert text_chunk_event.token == "foo bar"
+
+    def test_to_dict(self, text_chunk_event):
+        assert text_chunk_event.to_dict()["token"] == "foo bar"
+
+    def test_from_delta_message_content(self):
+        content = TextDeltaMessageContent(
+            index=0,
+            text="foo bar",
+        )
+        event = TextChunkEvent.from_delta_message_content(content)
+        assert event.token == "foo bar"
+        assert event.index == 0