chore: extract hook based tests to a separate file

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

@@ -67,7 +67,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
@@ -162,11 +159,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,
@@ -178,7 +170,6 @@ def agent(
     tool_registry,
     tool_decorated,
     request,
-    hook_provider,
 ):
     agent = Agent(
         model=mock_model,
@@ -188,9 +179,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
@@ -720,48 +708,6 @@ def test_agent__call__callback(mock_model, agent, callback_handler):
     )
 
 
-@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_hooks_structured_output(agent, mock_hook_messages, hook_provider):
-    """Verify that the correct hook events are emitted as part of structured_output."""
-
-    expected_user = User(name="Jane Doe", age=30, email="jane@doe.com")
-    agent.model.structured_output = unittest.mock.Mock(return_value=[{"output": expected_user}])
-    agent.structured_output(User, "example prompt")
-
-    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

tests/strands/agent/test_agent_hooks.py

@@ -0,0 +1,120 @@
+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
+
+
+# mock the User(name='Jane Doe', age=30)
+class User(BaseModel):
+    """A user of the system."""
+
+    name: str
+    age: int
+
+
+@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, 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_hooks_stream_async(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_hooks_structured_output(agent, hook_provider):
+    """Verify that the correct hook events are emitted as part of structured_output."""
+
+    expected_user = User(name="Jane Doe", age=30)
+    agent.model.structured_output = unittest.mock.Mock(return_value=[{"output": expected_user}])
+    agent.structured_output(User, "example prompt")
+
+    assert hook_provider.events_received == [StartRequestEvent(agent=agent), EndRequestEvent(agent=agent)]