Extract hook based tests to a separate file (#349)

Rather than keeping agent hook tests in the test_agent file, extract them out to a separate file for readability/better-separation-of-concerns.

Also updated the hook tests verify each hook-event individually to be more readable/easier to debug.

Author: zastrowm

tests/fixtures/mock_hook_provider.py

@@ -1,3 +1,4 @@
+from collections import deque
 from typing import Type
 
 from strands.experimental.hooks import HookEvent, HookProvider, HookRegistry
@@ -8,6 +9,9 @@ def __init__(self, event_types: list[Type]):
         self.events_received = []
         self.events_types = event_types
 
+    def get_events(self) -> deque[HookEvent]:
+        return deque(self.events_received)
+
     def register_hooks(self, registry: HookRegistry) -> None:
         for event_type in self.events_types:
             registry.add_callback(event_type, self._add_event)

tests/fixtures/mocked_model_provider.py

@@ -72,7 +72,7 @@ def map_agent_message_to_events(self, agent_message: Message) -> Iterable[dict[s
                         }
                     }
                 }
-                yield {"contentBlockDelta": {"delta": {"tool_use": {"input": json.dumps(content["toolUse"]["input"])}}}}
+                yield {"contentBlockDelta": {"delta": {"toolUse": {"input": json.dumps(content["toolUse"]["input"])}}}}
                 yield {"contentBlockStop": {}}
 
         yield {"messageStop": {"stopReason": stop_reason}}

tests/strands/agent/test_agent.py

@@ -4,7 +4,6 @@
 import os
 import textwrap
 import unittest.mock
-from unittest.mock import call
 
 import pytest
 from pydantic import BaseModel
@@ -14,12 +13,10 @@
 from strands.agent import AgentResult
 from strands.agent.conversation_manager.null_conversation_manager import NullConversationManager
 from strands.agent.conversation_manager.sliding_window_conversation_manager import SlidingWindowConversationManager
-from strands.experimental.hooks import AgentInitializedEvent, EndRequestEvent, StartRequestEvent
 from strands.handlers.callback_handler import PrintingCallbackHandler, null_callback_handler
 from strands.models.bedrock import DEFAULT_BEDROCK_MODEL_ID, BedrockModel
 from strands.types.content import Messages
 from strands.types.exceptions import ContextWindowOverflowException, EventLoopException
-from tests.fixtures.mock_hook_provider import MockHookProvider
 
 
 @pytest.fixture
@@ -40,38 +37,6 @@ def converse(*args, **kwargs):
     return mock
 
 
-@pytest.fixture
-def mock_hook_messages(mock_model, tool, agenerator):
-    """Fixture which returns a standard set of events for verifying hooks."""
-    mock_model.mock_converse.side_effect = [
-        agenerator(
-            [
-                {
-                    "contentBlockStart": {
-                        "start": {
-                            "toolUse": {
-                                "toolUseId": "t1",
-                                "name": tool.tool_spec["name"],
-                            },
-                        },
-                    },
-                },
-                {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"random_string": "abcdEfghI123"}'}}}},
-                {"contentBlockStop": {}},
-                {"messageStop": {"stopReason": "tool_use"}},
-            ],
-        ),
-        agenerator(
-            [
-                {"contentBlockDelta": {"delta": {"text": "test text"}}},
-                {"contentBlockStop": {}},
-            ],
-        ),
-    ]
-
-    return mock_model.mock_converse
-
-
 @pytest.fixture
 def system_prompt(request):
     return request.param if hasattr(request, "param") else "You are a helpful assistant."
@@ -166,11 +131,6 @@ def tools(request, tool):
     return request.param if hasattr(request, "param") else [tool_decorated]
 
 
-@pytest.fixture
-def hook_provider():
-    return MockHookProvider([AgentInitializedEvent, StartRequestEvent, EndRequestEvent])
-
-
 @pytest.fixture
 def agent(
     mock_model,
@@ -182,7 +142,6 @@ def agent(
     tool_registry,
     tool_decorated,
     request,
-    hook_provider,
 ):
     agent = Agent(
         model=mock_model,
@@ -192,9 +151,6 @@ def agent(
         tools=tools,
     )
 
-    # for now, hooks are private
-    agent._hooks.add_hook(hook_provider)
-
     # Only register the tool directly if tools wasn't parameterized
     if not hasattr(request, "param") or request.param is None:
         # Create a new function tool directly from the decorated function
@@ -789,38 +745,6 @@ async def test_agent_invoke_async(mock_model, agent, agenerator):
     assert tru_message == exp_message
 
 
-@unittest.mock.patch("strands.experimental.hooks.registry.HookRegistry.invoke_callbacks")
-def test_agent_hooks__init__(mock_invoke_callbacks):
-    """Verify that the AgentInitializedEvent is emitted on Agent construction."""
-    agent = Agent()
-
-    # Verify AgentInitialized event was invoked
-    mock_invoke_callbacks.assert_called_once()
-    assert mock_invoke_callbacks.call_args == call(AgentInitializedEvent(agent=agent))
-
-
-def test_agent_hooks__call__(agent, mock_hook_messages, hook_provider):
-    """Verify that the correct hook events are emitted as part of __call__."""
-
-    agent("test message")
-
-    assert hook_provider.events_received == [StartRequestEvent(agent=agent), EndRequestEvent(agent=agent)]
-
-
-@pytest.mark.asyncio
-async def test_agent_hooks_stream_async(agent, mock_hook_messages, hook_provider):
-    """Verify that the correct hook events are emitted as part of stream_async."""
-    iterator = agent.stream_async("test message")
-    await anext(iterator)
-    assert hook_provider.events_received == [StartRequestEvent(agent=agent)]
-
-    # iterate the rest
-    async for _ in iterator:
-        pass
-
-    assert hook_provider.events_received == [StartRequestEvent(agent=agent), EndRequestEvent(agent=agent)]
-
-
 def test_agent_tool(mock_randint, agent):
     conversation_manager_spy = unittest.mock.Mock(wraps=agent.conversation_manager)
     agent.conversation_manager = conversation_manager_spy
@@ -1051,13 +975,6 @@ async def test_agent_structured_output_async(agent, user, agenerator):
     agent.model.structured_output.assert_called_once_with(type(user), [{"role": "user", "content": [{"text": prompt}]}])
 
 
-def test_agent_hooks_structured_output(agent, user, mock_hook_messages, hook_provider, agenerator):
-    agent.model.structured_output = unittest.mock.Mock(return_value=agenerator([{"output": user}]))
-    agent.structured_output(type(user), "example prompt")
-
-    assert hook_provider.events_received == [StartRequestEvent(agent=agent), EndRequestEvent(agent=agent)]
-
-
 @pytest.mark.asyncio
 async def test_stream_async_returns_all_events(mock_event_loop_cycle, alist):
     agent = Agent()

tests/strands/agent/test_agent_hooks.py

@@ -0,0 +1,130 @@
+import unittest.mock
+from unittest.mock import call
+
+import pytest
+from pydantic import BaseModel
+
+import strands
+from strands import Agent
+from strands.experimental.hooks import AgentInitializedEvent, EndRequestEvent, StartRequestEvent
+from strands.types.content import Messages
+from tests.fixtures.mock_hook_provider import MockHookProvider
+from tests.fixtures.mocked_model_provider import MockedModelProvider
+
+
+@pytest.fixture
+def hook_provider():
+    return MockHookProvider([AgentInitializedEvent, StartRequestEvent, EndRequestEvent])
+
+
+@pytest.fixture
+def agent_tool():
+    @strands.tools.tool(name="tool_decorated")
+    def reverse(random_string: str) -> str:
+        return random_string[::-1]
+
+    return reverse
+
+
+@pytest.fixture
+def tool_use(agent_tool):
+    return {"name": agent_tool.tool_name, "toolUseId": "123", "input": {"random_string": "I invoked a tool!"}}
+
+
+@pytest.fixture
+def mock_model(tool_use):
+    agent_messages: Messages = [
+        {
+            "role": "assistant",
+            "content": [{"toolUse": tool_use}],
+        },
+        {"role": "assistant", "content": [{"text": "I invoked a tool!"}]},
+    ]
+    return MockedModelProvider(agent_messages)
+
+
+@pytest.fixture
+def agent(
+    mock_model,
+    hook_provider,
+    agent_tool,
+):
+    agent = Agent(
+        model=mock_model,
+        system_prompt="You are a helpful assistant.",
+        callback_handler=None,
+        tools=[agent_tool],
+    )
+
+    # for now, hooks are private
+    agent._hooks.add_hook(hook_provider)
+
+    return agent
+
+
+@pytest.fixture
+def user():
+    class User(BaseModel):
+        name: str
+        age: int
+
+    return User(name="Jane Doe", age=30)
+
+
+@unittest.mock.patch("strands.experimental.hooks.registry.HookRegistry.invoke_callbacks")
+def test_agent__init__hooks(mock_invoke_callbacks):
+    """Verify that the AgentInitializedEvent is emitted on Agent construction."""
+    agent = Agent()
+
+    # Verify AgentInitialized event was invoked
+    mock_invoke_callbacks.assert_called_once()
+    assert mock_invoke_callbacks.call_args == call(AgentInitializedEvent(agent=agent))
+
+
+def test_agent__call__hooks(agent, hook_provider, agent_tool, tool_use):
+    """Verify that the correct hook events are emitted as part of __call__."""
+
+    agent("test message")
+
+    events = hook_provider.get_events()
+    assert len(events) == 2
+
+    assert events.popleft() == StartRequestEvent(agent=agent)
+    assert events.popleft() == EndRequestEvent(agent=agent)
+
+
+@pytest.mark.asyncio
+async def test_agent_stream_async_hooks(agent, hook_provider, agent_tool, tool_use):
+    """Verify that the correct hook events are emitted as part of stream_async."""
+    iterator = agent.stream_async("test message")
+    await anext(iterator)
+    assert hook_provider.events_received == [StartRequestEvent(agent=agent)]
+
+    # iterate the rest
+    async for _ in iterator:
+        pass
+
+    events = hook_provider.get_events()
+    assert len(events) == 2
+
+    assert events.popleft() == StartRequestEvent(agent=agent)
+    assert events.popleft() == EndRequestEvent(agent=agent)
+
+
+def test_agent_structured_output_hooks(agent, hook_provider, user, agenerator):
+    """Verify that the correct hook events are emitted as part of structured_output."""
+
+    agent.model.structured_output = unittest.mock.Mock(return_value=agenerator([{"output": user}]))
+    agent.structured_output(type(user), "example prompt")
+
+    assert hook_provider.events_received == [StartRequestEvent(agent=agent), EndRequestEvent(agent=agent)]
+
+
+@pytest.mark.asyncio
+async def test_agent_structured_async_output_hooks(agent, hook_provider, user, agenerator):
+    """Verify that the correct hook events are emitted as part of structured_output_async."""
+
+    agent.model.structured_output = unittest.mock.Mock(return_value=agenerator([{"output": user}]))
+    await agent.structured_output_async(type(user), "example prompt")
+
+    assert hook_provider.events_received == [StartRequestEvent(agent=agent), EndRequestEvent(agent=agent)]