encounter/adk-python
0
0
Fork 0
mirror of https://github.com/encounter/adk-python.git synced 2026-07-09 18:19:28 -07:00
Code Issues Packages Projects Releases Wiki Activity
Files
4ddb2cb2a8d1d026a43418b2dd698e6ea199594e
adk-python/tests/unittests/runners/test_runner_debug.py
T
Lavi NigamandCopybara-Service 0487eea2ab feat: add run_debug() helper method for quick agent experimentation 
Merge https://github.com/google/adk-python/pull/3345

Add run_debug() helper method to InMemoryRunner that reduces agent execution boilerplate from 7-8 lines to just 2 lines, making it ideal for quick experimentation, notebooks, and getting started with ADK.

**Key changes:**
• Introduce run_debug() to reduce boilerplate from 7-8 lines to 2 lines
• Enable quick testing in notebooks, REPL, and during development
• Support single or multiple messages with automatic session management
• Add verbose flag to show/hide tool calls and intermediate processing
• Add quiet flag to suppress console output while capturing events
• Extract event printing logic to reusable utility (utils/_debug_output.py)
• Include comprehensive test suite with 21 test cases covering all part types
• Provide complete working example with 8 usage patterns
• **This is a convenience method for experimentation, not a replacement for run_async()**

### Link to Issue or Description of Change

**1. Link to an existing issue (if applicable):**

* N/A - New feature to improve developer experience

**2. Or, if no issue exists, describe the change:**

**Problem:**

Developers need to write 7-8 lines of boilerplate code just to test a simple agent interaction during development. This creates friction for:

* New developers getting started with ADK
* Quick experimentation in Jupyter notebooks or Python REPL
* Debugging agent behavior during development
* Writing examples and tutorials
* Rapid prototyping of agent capabilities

**Solution:**

Introduce `run_debug()` as a convenience helper method specifically designed for quick experimentation and getting started scenarios. This method:

* **Is NOT a replacement for `run_async()`** - it's a developer convenience tool
* **Reduces boilerplate** from 7-8 lines to just 2 lines for simple testing
* **Handles session management automatically** with sensible defaults
* **Provides debugging visibility** with optional verbose flag for tool calls
* **Supports common patterns** like multiple messages and event capture
* **Type-safe implementation** using direct attribute access instead of getattr()

### Before vs After Comparison

**BEFORE - Current approach requires 7-8 lines of boilerplate:**

```python
from google.adk import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define a simple agent
agent = Agent(
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant"
)

# Need all this boilerplate just to test the agent
APP_NAME = "default"
USER_ID = "default"
session_service = InMemorySessionService()
runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
session = await session_service.create_session(
    app_name=APP_NAME, user_id=USER_ID, session_id="default"
)
content = types.Content(role="user", parts=[types.Part.from_text("Hello")])
async for event in runner.run_async(
    user_id=USER_ID, session_id=session.id, new_message=content
):
    if event.content and event.content.parts:
        print(event.content.parts[0].text)
```

**AFTER - With run_debug() helper, just 2 lines:**

```python
from google.adk import Agent
from google.adk.runners import InMemoryRunner

# Define the same agent
agent = Agent(
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant"
)

# Test it with just 2 lines!
runner = InMemoryRunner(agent=agent)
await runner.run_debug("Hello")
```

### API Design

```python
async def run_debug(
    self,
    user_messages: str | list[str],
    *,
    user_id: str = 'debug_user_id',
    session_id: str = 'debug_session_id',
    run_config: RunConfig | None = None,
    quiet: bool = False,
    verbose: bool = False,
) -> list[Event]:
```

**Parameters:**

* `user_messages`: Single message string or list of messages (required)
* `user_id`: User identifier (default: 'debug_user_id')
* `session_id`: Session identifier for conversation continuity (default: 'debug_session_id')
* `run_config`: Optional advanced configuration
* `quiet`: Suppress console output (default: False)
* `verbose`: Show detailed tool calls and responses (default: False)

**Key Features:**

* **Always returns events** - Simplifies API, no conditional return type
* **Type-safe implementation** - Uses direct attribute access on Pydantic models
* **Text buffering** - Consecutive text parts printed without repeated author prefix
* **Smart truncation** - Long tool args/responses truncated for readability
* **Clean session management** - Get-then-create pattern, no try/except
* **Reusable printing logic** - Extracted to utils/_debug_output.py for other tools

### Implementation Highlights

**1. Event Printing Utility (utils/_debug_output.py):**
* Modular print_event() function for displaying events
* Text buffering to combine consecutive text parts
* Configurable truncation for different content types:
  - Function args: 50 chars max
  - Function responses: 100 chars max
  - Code output: 100 chars max
* Supports all ADK part types (text, function_call, executable_code, inline_data, file_data)

**2. Session Management:**
```python
# Clean get-then-create pattern (no try/except)
session = await self.session_service.get_session(
    app_name=self.app_name, user_id=user_id, session_id=session_id
)
if not session:
    session = await self.session_service.create_session(
        app_name=self.app_name, user_id=user_id, session_id=session_id
    )
```

**3. Type-Safe Event Processing:**
* Direct attribute access on Pydantic models (no getattr() or hasattr())
* Proper handling of all part types
* Leverages `from __future__ import annotations` for duck typing

### Important Note on Scope

`run_debug()` is a **convenience method for experimentation only**. For production applications requiring:

* Custom session services (Spanner, Cloud SQL)
* Fine-grained event processing control
* Error recovery and resumability
* Performance optimization
* Complex authentication flows

Continue using the standard `run_async()` method. The `run_debug()` helper is specifically designed to lower the barrier to entry and speed up the development/testing cycle.

### Testing Plan

**Unit Tests (21 test cases in tests/unittests/runners/test_runner_debug.py):**

**Core functionality (7 tests):**
* ✅ Single message execution and event return
* ✅ Multiple messages in sequence
* ✅ Quiet mode (suppresses output)
* ✅ Custom session_id configuration
* ✅ Custom user_id configuration
* ✅ RunConfig passthrough
* ✅ Session persistence across calls

**Part type handling (8 tests):**
* ✅ Tool calls and responses (verbose mode)
* ✅ Executable code parts
* ✅ Code execution result parts
* ✅ Inline data (images)
* ✅ File data references
* ✅ Mixed part types in single event
* ✅ Long output truncation
* ✅ Verbose flag behavior (show/hide tools)

**Edge cases (6 tests):**
* ✅ None text filtering
* ✅ Existing session handling
* ✅ Empty parts list
* ✅ None event content
* ✅ Verbose=False hides tool calls
* ✅ Verbose=True shows tool calls

**All 21 tests passing in 3.8s** ✓

**Manual End-to-End (E2E) Tests:**

Tested all 8 example patterns in contributing/samples/runner_debug_example/main.py:

1. ✅ Minimal 2-line usage
2. ✅ Multiple sequential messages
3. ✅ Session persistence across calls
4. ✅ Multiple user sessions (Alice & Bob)
5. ✅ Verbose mode for tool visibility
6. ✅ Event capture with quiet mode
7. ✅ Custom RunConfig integration
8. ✅ Before/after comparison

### Files Changed

**Core implementation:**
* src/google/adk/runners.py - Added run_debug() method (~60 lines)
* src/google/adk/utils/_debug_output.py - Event printing utility (~106 lines)

**Tests:**
* tests/unittests/runners/test_runner_debug.py - Comprehensive test suite (21 tests)

**Examples:**
* contributing/samples/runner_debug_example/agent.py - Sample agent with tools
* contributing/samples/runner_debug_example/main.py - 8 usage examples
* contributing/samples/runner_debug_example/README.md - Complete documentation

### Checklist

- [x] I have read the [CONTRIBUTING.md](https://github.com/google/adk-python/blob/main/CONTRIBUTING.md) document
- [x] I have performed a self-review of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes (21/21 passing)
- [x] I have manually tested my changes end-to-end (8 examples tested)
- [x] Code follows ADK style guide (relative imports, type hints, 2-space indentation)
- [x] Ran ./autoformat.sh before committing
- [x] Any dependent changes have been merged and published in downstream modules

### Additional Context

**Example with Tools (verbose mode):**

```python
# Create agent with tools
agent = Agent(
    model="gemini-2.5-flash",
    instruction="You can check weather and do calculations",
    tools=[get_weather, calculate]
)

# Test with verbose to see tool calls
runner = InMemoryRunner(agent=agent)
await runner.run_debug("What's the weather in SF?", verbose=True)

# Output:
# User > What's the weather in SF?
# agent > [Calling tool: get_weather({'city': 'San Francisco'})]
# agent > [Tool result: {'result': 'Foggy, 15°C (59°F)'}]
# agent > The weather in San Francisco is foggy, 15°C (59°F).
```

**Complete Example Included:**

The PR includes a full working example in `contributing/samples/runner_debug_example/` with:
* Agent with weather and calculator tools
* 8 different usage patterns
* Comprehensive README with troubleshooting
* Safe AST-based expression evaluation

**Breaking Changes:** None - this is purely additive.

**Security:** Example uses AST-based expression evaluation instead of eval().

**Code Quality:**
* Type-safe implementation (no getattr() or hasattr())
* Modular design (printing logic separated into utility)
* Follows ADK conventions (relative imports, from __future__ import annotations)
* Comprehensive error handling (gracefully handles None content, empty parts)
* Well-documented with docstrings and inline comments
END_PUBLIC
```

---

## Key Changes from Original:

1. ✅ Updated parameter name: `user_queries` → `user_messages`
2. ✅ Updated parameter name: `session_name` → `session_id`
3. ✅ Updated parameter name: `print_output` → `quiet`
4. ✅ Removed `return_events` parameter
5. ✅ Updated test count: 23 → 21
6. ✅ Changed "queries" → "messages" throughout
7. ✅ Added implementation highlights section
8. ✅ Added details about utils/_debug_output.py
9. ✅ Updated default values to debug_user_id/debug_session_id
10. ✅ Noted type-safe implementation
11. ✅ Added Code Quality section
12. ✅ Updated API signature to match final refactored version
13. ✅ Removed optional return type (always returns list[Event])

Co-authored-by: Wei Sun (Jack) <weisun@google.com>
COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/3345 from lavinigam-gcp:adk-runner-helper e0050b9f152d0f0e49e6501610d2c59a754fc571
PiperOrigin-RevId: 826607817
2025-10-31 13:28:02 -07:00

918 lines
31 KiB
Python
Raw Blame History

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests for Runner.run_debug helper method."""
from __future__ import annotations
from unittest import mock
from google.adk.agents import Agent
from google.adk.agents.run_config import RunConfig
from google.adk.runners import InMemoryRunner
import pytest
class TestRunDebug:
"""Tests for Runner.run_debug method."""
@pytest.mark.asyncio
async def test_run_debug_single_query(self):
"""Test run_debug with a single string query."""
# Setup
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="You are a helpful assistant.",
)
runner = InMemoryRunner(agent=agent)
# Mock the runner's run_async to return controlled events
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text="Hello! I can help you.")]
async def mock_run_async(*args, **kwargs):
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute
events = await runner.run_debug("Hello, how are you?", quiet=True)
# Assertions
assert len(events) == 1
assert events[0].author == "test_agent"
assert events[0].content.parts[0].text == "Hello! I can help you."
# Verify session was created with defaults
session = await runner.session_service.get_session(
app_name=runner.app_name,
user_id="debug_user_id",
session_id="debug_session_id",
)
assert session is not None
@pytest.mark.asyncio
async def test_run_debug_multiple_queries(self):
"""Test run_debug with multiple queries in sequence."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="You are a test bot.",
)
runner = InMemoryRunner(agent=agent)
# Mock responses for multiple queries
responses = ["First response", "Second response"]
call_count = 0
async def mock_run_async(*args, **kwargs):
nonlocal call_count
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text=responses[call_count])]
call_count += 1
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute with multiple queries
events = await runner.run_debug(
["First query", "Second query"], quiet=True
)
# Assertions
assert len(events) == 2
assert events[0].content.parts[0].text == "First response"
assert events[1].content.parts[0].text == "Second response"
@pytest.mark.asyncio
async def test_run_debug_always_returns_events(self):
"""Test that run_debug always returns events."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text="Response")]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Test that events are always returned
events = await runner.run_debug("Query", quiet=True)
assert isinstance(events, list)
assert len(events) == 1
@pytest.mark.asyncio
async def test_run_debug_quiet_mode(self, capsys):
"""Test that quiet=True suppresses printing."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text="This should not be printed")]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute with quiet=True
await runner.run_debug("Test query", quiet=True)
# Check that nothing was printed
captured = capsys.readouterr()
assert "This should not be printed" not in captured.out
assert "User >" not in captured.out
assert "Session:" not in captured.out
@pytest.mark.asyncio
async def test_run_debug_custom_session_id(self):
"""Test run_debug with custom session_id."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text="Response")]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute with custom session ID
await runner.run_debug(
"Query", session_id="custom_debug_session", quiet=True
)
# Verify session was created with custom ID
session = await runner.session_service.get_session(
app_name=runner.app_name,
user_id="debug_user_id",
session_id="custom_debug_session",
)
assert session is not None
assert session.id == "custom_debug_session"
@pytest.mark.asyncio
async def test_run_debug_custom_user_id(self):
"""Test run_debug with custom user_id."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text="Response")]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute with custom user_id
await runner.run_debug("Query", user_id="test_user_123", quiet=True)
# Verify session was created with custom user_id
session = await runner.session_service.get_session(
app_name=runner.app_name,
user_id="test_user_123",
session_id="debug_session_id",
)
assert session is not None
@pytest.mark.asyncio
async def test_run_debug_with_run_config(self):
"""Test that run_config is properly passed through to run_async."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
run_config_used = None
async def mock_run_async(*args, **kwargs):
nonlocal run_config_used
run_config_used = kwargs.get("run_config")
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text="Response")]
yield mock_event
with mock.patch.object(
runner, "run_async", side_effect=mock_run_async
) as mock_method:
# Create a custom run_config
custom_config = RunConfig(support_cfc=True)
# Execute with custom run_config
await runner.run_debug("Query", run_config=custom_config, quiet=True)
# Verify run_config was passed to run_async
assert mock_method.called
call_args = mock_method.call_args
assert call_args is not None
assert "run_config" in call_args.kwargs
assert call_args.kwargs["run_config"] == custom_config
@pytest.mark.asyncio
async def test_run_debug_session_persistence(self):
"""Test that multiple calls to run_debug maintain conversation context."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Remember previous messages.",
)
runner = InMemoryRunner(agent=agent)
call_count = 0
responses = ["First response", "Second response remembering first"]
async def mock_run_async(*args, **kwargs):
nonlocal call_count
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text=responses[call_count])]
call_count += 1
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# First call
events1 = await runner.run_debug("First message", quiet=True)
assert events1[0].content.parts[0].text == "First response"
# Second call to same session
events2 = await runner.run_debug("Second message", quiet=True)
assert (
events2[0].content.parts[0].text
== "Second response remembering first"
)
# Verify both calls used the same session
session = await runner.session_service.get_session(
app_name=runner.app_name,
user_id="debug_user_id",
session_id="debug_session_id",
)
assert session is not None
@pytest.mark.asyncio
async def test_run_debug_filters_none_text(self):
"""Test that run_debug filters out 'None' text and empty parts."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Yield events with various text values
events = [
mock.Mock(
author="test_agent",
content=mock.Mock(parts=[mock.Mock(text="Valid text")]),
),
mock.Mock(
author="test_agent",
content=mock.Mock(parts=[mock.Mock(text="None")]),
), # Should be filtered
mock.Mock(
author="test_agent",
content=mock.Mock(parts=[mock.Mock(text="")]),
), # Should be filtered
mock.Mock(
author="test_agent",
content=mock.Mock(parts=[mock.Mock(text="Another valid")]),
),
]
for event in events:
yield event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute and capture output
events = await runner.run_debug("Query", quiet=True)
# All 4 events should be returned (filtering is for printing only)
assert len(events) == 4
# But when printing, "None" and empty strings should be filtered
# This is tested implicitly by the implementation
@pytest.mark.asyncio
async def test_run_debug_with_existing_session(self):
"""Test that run_debug retrieves existing session when AlreadyExistsError occurs."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
# First create a session
await runner.session_service.create_session(
app_name=runner.app_name,
user_id="debug_user_id",
session_id="existing_session",
)
async def mock_run_async(*args, **kwargs):
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [mock.Mock(text="Using existing session")]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute with same session ID (should retrieve existing)
events = await runner.run_debug(
"Query", session_id="existing_session", quiet=True
)
assert len(events) == 1
assert events[0].content.parts[0].text == "Using existing session"
@pytest.mark.asyncio
async def test_run_debug_with_tool_calls(self, capsys):
"""Test that run_debug properly handles and prints tool calls."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with tools.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# First event: tool call
mock_call_event = mock.Mock()
mock_call_event.author = "test_agent"
mock_call_event.content = mock.Mock()
mock_function_call = mock.Mock()
mock_function_call.name = "calculate"
mock_function_call.args = {"operation": "add", "a": 5, "b": 3}
mock_part_call = mock.Mock()
mock_part_call.text = None
mock_part_call.function_call = mock_function_call
mock_part_call.function_response = None
mock_call_event.content.parts = [mock_part_call]
yield mock_call_event
# Second event: tool response
mock_resp_event = mock.Mock()
mock_resp_event.author = "test_agent"
mock_resp_event.content = mock.Mock()
mock_function_response = mock.Mock()
mock_function_response.response = {"result": 8}
mock_part_resp = mock.Mock()
mock_part_resp.text = None
mock_part_resp.function_call = None
mock_part_resp.function_response = mock_function_response
mock_resp_event.content.parts = [mock_part_resp]
yield mock_resp_event
# Third event: final text response
mock_text_event = mock.Mock()
mock_text_event.author = "test_agent"
mock_text_event.content = mock.Mock()
mock_text_event.content.parts = [mock.Mock(text="The result is 8")]
yield mock_text_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
# Execute with verbose=True to see tool calls
events = await runner.run_debug("Calculate 5 + 3", verbose=True)
# Check output was printed
captured = capsys.readouterr()
assert "[Calling tool: calculate" in captured.out
assert "[Tool result:" in captured.out
assert "The result is 8" in captured.out
# Check events were collected
assert len(events) == 3
@pytest.mark.asyncio
async def test_run_debug_with_executable_code(self, capsys):
"""Test that run_debug properly handles executable code parts."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with code execution.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Event with executable code
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_exec_code = mock.Mock()
mock_exec_code.language = "python"
mock_exec_code.code = "print('Hello World')"
mock_part = mock.Mock()
mock_part.text = None
mock_part.function_call = None
mock_part.function_response = None
mock_part.executable_code = mock_exec_code
mock_part.code_execution_result = None
mock_part.inline_data = None
mock_part.file_data = None
mock_event.content.parts = [mock_part]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug("Run some code", verbose=True)
captured = capsys.readouterr()
assert "[Executing python code...]" in captured.out
assert len(events) == 1
@pytest.mark.asyncio
async def test_run_debug_with_code_execution_result(self, capsys):
"""Test that run_debug properly handles code execution result parts."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with code results.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Event with code execution result
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_result = mock.Mock()
mock_result.output = "Hello World\n42"
mock_part = mock.Mock()
mock_part.text = None
mock_part.function_call = None
mock_part.function_response = None
mock_part.executable_code = None
mock_part.code_execution_result = mock_result
mock_part.inline_data = None
mock_part.file_data = None
mock_event.content.parts = [mock_part]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug(
"Show code output",
verbose=True,
)
captured = capsys.readouterr()
assert "[Code output: Hello World\n42]" in captured.out
assert len(events) == 1
@pytest.mark.asyncio
async def test_run_debug_with_inline_data(self, capsys):
"""Test that run_debug properly handles inline data parts."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with inline data.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Event with inline data (e.g., image)
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_inline = mock.Mock()
mock_inline.mime_type = "image/png"
mock_inline.data = b"fake_image_data"
mock_part = mock.Mock()
mock_part.text = None
mock_part.function_call = None
mock_part.function_response = None
mock_part.executable_code = None
mock_part.code_execution_result = None
mock_part.inline_data = mock_inline
mock_part.file_data = None
mock_event.content.parts = [mock_part]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug("Show image", verbose=True)
captured = capsys.readouterr()
assert "[Inline data: image/png]" in captured.out
assert len(events) == 1
@pytest.mark.asyncio
async def test_run_debug_with_file_data(self, capsys):
"""Test that run_debug properly handles file data parts."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with file data.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Event with file data
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_file = mock.Mock()
mock_file.file_uri = "gs://bucket/path/to/file.pdf"
mock_part = mock.Mock()
mock_part.text = None
mock_part.function_call = None
mock_part.function_response = None
mock_part.executable_code = None
mock_part.code_execution_result = None
mock_part.inline_data = None
mock_part.file_data = mock_file
mock_event.content.parts = [mock_part]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug("Reference file", verbose=True)
captured = capsys.readouterr()
assert "[File: gs://bucket/path/to/file.pdf]" in captured.out
assert len(events) == 1
@pytest.mark.asyncio
async def test_run_debug_with_mixed_parts(self, capsys):
"""Test that run_debug handles events with multiple part types."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with mixed parts.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Event with multiple part types
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
# Text part
mock_text_part = mock.Mock()
mock_text_part.text = "Here's your result:"
mock_text_part.function_call = None
mock_text_part.function_response = None
mock_text_part.executable_code = None
mock_text_part.code_execution_result = None
mock_text_part.inline_data = None
mock_text_part.file_data = None
# Code execution part
mock_code_part = mock.Mock()
mock_code_part.text = None
mock_code_part.function_call = None
mock_code_part.function_response = None
mock_exec_code = mock.Mock()
mock_exec_code.language = "python"
mock_code_part.executable_code = mock_exec_code
mock_code_part.code_execution_result = None
mock_code_part.inline_data = None
mock_code_part.file_data = None
# Result part
mock_result_part = mock.Mock()
mock_result_part.text = None
mock_result_part.function_call = None
mock_result_part.function_response = None
mock_result_part.executable_code = None
mock_result = mock.Mock()
mock_result.output = "42"
mock_result_part.code_execution_result = mock_result
mock_result_part.inline_data = None
mock_result_part.file_data = None
mock_event.content.parts = [
mock_text_part,
mock_code_part,
mock_result_part,
]
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug("Mixed response", verbose=True)
captured = capsys.readouterr()
assert "Here's your result:" in captured.out
assert "[Executing python code...]" in captured.out
assert "[Code output: 42]" in captured.out
assert len(events) == 1
@pytest.mark.asyncio
async def test_run_debug_with_long_output_truncation(self, capsys):
"""Test that run_debug properly truncates long outputs."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with long outputs.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Tool call with long args
mock_call_event = mock.Mock()
mock_call_event.author = "test_agent"
mock_call_event.content = mock.Mock()
mock_function_call = mock.Mock()
mock_function_call.name = "process"
# Create a long argument string
mock_function_call.args = {"data": "x" * 100}
mock_part_call = mock.Mock()
mock_part_call.text = None
mock_part_call.function_call = mock_function_call
mock_part_call.function_response = None
mock_part_call.executable_code = None
mock_part_call.code_execution_result = None
mock_part_call.inline_data = None
mock_part_call.file_data = None
mock_call_event.content.parts = [mock_part_call]
yield mock_call_event
# Tool response with long result
mock_resp_event = mock.Mock()
mock_resp_event.author = "test_agent"
mock_resp_event.content = mock.Mock()
mock_function_response = mock.Mock()
# Create a long response string
mock_function_response.response = {"result": "y" * 200}
mock_part_resp = mock.Mock()
mock_part_resp.text = None
mock_part_resp.function_call = None
mock_part_resp.function_response = mock_function_response
mock_part_resp.executable_code = None
mock_part_resp.code_execution_result = None
mock_part_resp.inline_data = None
mock_part_resp.file_data = None
mock_resp_event.content.parts = [mock_part_resp]
yield mock_resp_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug("Process data", verbose=True)
captured = capsys.readouterr()
# Check that args are truncated at 50 chars
assert "..." in captured.out
assert "[Calling tool: process(" in captured.out
# Check that response is truncated at 100 chars
assert "[Tool result:" in captured.out
assert len(events) == 2
@pytest.mark.asyncio
async def test_run_debug_verbose_flag_false(self, capsys):
"""Test that run_debug hides tool calls when verbose=False (default)."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with tools.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Tool call event
mock_call_event = mock.Mock()
mock_call_event.author = "test_agent"
mock_call_event.content = mock.Mock()
mock_function_call = mock.Mock()
mock_function_call.name = "get_weather"
mock_function_call.args = {"city": "Tokyo"}
mock_part_call = mock.Mock()
mock_part_call.text = None
mock_part_call.function_call = mock_function_call
mock_part_call.function_response = None
mock_part_call.executable_code = None
mock_part_call.code_execution_result = None
mock_part_call.inline_data = None
mock_part_call.file_data = None
mock_call_event.content.parts = [mock_part_call]
yield mock_call_event
# Tool response event
mock_resp_event = mock.Mock()
mock_resp_event.author = "test_agent"
mock_resp_event.content = mock.Mock()
mock_function_response = mock.Mock()
mock_function_response.response = {"weather": "Clear, 25°C"}
mock_part_resp = mock.Mock()
mock_part_resp.text = None
mock_part_resp.function_call = None
mock_part_resp.function_response = mock_function_response
mock_part_resp.executable_code = None
mock_part_resp.code_execution_result = None
mock_part_resp.inline_data = None
mock_part_resp.file_data = None
mock_resp_event.content.parts = [mock_part_resp]
yield mock_resp_event
# Final text response
mock_text_event = mock.Mock()
mock_text_event.author = "test_agent"
mock_text_event.content = mock.Mock()
mock_text_event.content.parts = [
mock.Mock(text="The weather in Tokyo is clear and 25°C.")
]
yield mock_text_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug(
"What's the weather?",
verbose=False, # Default - should NOT show tool calls
)
captured = capsys.readouterr()
# Should NOT show tool call details
assert "[Calling tool:" not in captured.out
assert "[Tool result:" not in captured.out
# Should show final text response
assert "The weather in Tokyo is clear and 25°C." in captured.out
assert len(events) == 3
@pytest.mark.asyncio
async def test_run_debug_verbose_flag_true(self, capsys):
"""Test that run_debug shows tool calls when verbose=True."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent with tools.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*args, **kwargs):
# Tool call event
mock_call_event = mock.Mock()
mock_call_event.author = "test_agent"
mock_call_event.content = mock.Mock()
mock_function_call = mock.Mock()
mock_function_call.name = "calculate"
mock_function_call.args = {"expression": "42 * 3.14"}
mock_part_call = mock.Mock()
mock_part_call.text = None
mock_part_call.function_call = mock_function_call
mock_part_call.function_response = None
mock_part_call.executable_code = None
mock_part_call.code_execution_result = None
mock_part_call.inline_data = None
mock_part_call.file_data = None
mock_call_event.content.parts = [mock_part_call]
yield mock_call_event
# Tool response event
mock_resp_event = mock.Mock()
mock_resp_event.author = "test_agent"
mock_resp_event.content = mock.Mock()
mock_function_response = mock.Mock()
mock_function_response.response = {"result": 131.88}
mock_part_resp = mock.Mock()
mock_part_resp.text = None
mock_part_resp.function_call = None
mock_part_resp.function_response = mock_function_response
mock_part_resp.executable_code = None
mock_part_resp.code_execution_result = None
mock_part_resp.inline_data = None
mock_part_resp.file_data = None
mock_resp_event.content.parts = [mock_part_resp]
yield mock_resp_event
# Final text response
mock_text_event = mock.Mock()
mock_text_event.author = "test_agent"
mock_text_event.content = mock.Mock()
mock_text_event.content.parts = [mock.Mock(text="The result is 131.88")]
yield mock_text_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug(
"Calculate 42 * 3.14",
verbose=True, # Should show tool calls
)
captured = capsys.readouterr()
# Should show tool call details
assert (
"[Calling tool: calculate({'expression': '42 * 3.14'})]"
in captured.out
)
assert "[Tool result: {'result': 131.88}]" in captured.out
# Should also show final text response
assert "The result is 131.88" in captured.out
assert len(events) == 3
@pytest.mark.asyncio
async def test_run_debug_with_empty_parts_list(self, capsys):
"""Test that run_debug handles events with empty parts list gracefully."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*_args, **_kwargs):
# Event with empty parts list
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = mock.Mock()
mock_event.content.parts = [] # Empty parts list
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug("Test query")
captured = capsys.readouterr()
# Should handle gracefully without crashing
assert "User > Test query" in captured.out
assert len(events) == 1
# Should not print any agent response since parts is empty
assert "test_agent >" not in captured.out
@pytest.mark.asyncio
async def test_run_debug_with_none_event_content(self, capsys):
"""Test that run_debug handles events with None content gracefully."""
agent = Agent(
name="test_agent",
model="gemini-2.5-flash-lite",
instruction="Test agent.",
)
runner = InMemoryRunner(agent=agent)
async def mock_run_async(*_args, **_kwargs):
# Event with None content
mock_event = mock.Mock()
mock_event.author = "test_agent"
mock_event.content = None # None content
yield mock_event
with mock.patch.object(runner, "run_async", side_effect=mock_run_async):
events = await runner.run_debug("Test query")
captured = capsys.readouterr()
# Should handle gracefully without crashing
assert "User > Test query" in captured.out
assert len(events) == 1
# Should not print any agent response since content is None
assert "test_agent >" not in captured.out
Reference in New Issue View Git Blame Copy Permalink