test: restore missing rejections test in request confirmation
Co-authored-by: Xuan Yang <xygoogle@google.com> PiperOrigin-RevId: 953575535
This commit is contained in:
committed by
Copybara-Service
parent
4b002c4b56
commit
65d8ea7d82
@@ -587,3 +587,108 @@ async def test_request_confirmation_processor_dynamic_success():
|
||||
assert (
|
||||
args[4][MOCK_FUNCTION_CALL_ID] == user_confirmation
|
||||
) # tool_confirmation_dict
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"tools, original_args, confirmation_args, expected_exception_match",
|
||||
[
|
||||
(
|
||||
[],
|
||||
{"param1": "test"},
|
||||
{"param1": "test"},
|
||||
"is not registered",
|
||||
),
|
||||
(
|
||||
[FunctionTool(mock_tool, require_confirmation=False)],
|
||||
{"param1": "test"},
|
||||
{"param1": "test"},
|
||||
"does not require confirmation",
|
||||
),
|
||||
(
|
||||
[FunctionTool(mock_tool, require_confirmation=True)],
|
||||
{"param1": "test"},
|
||||
{"param1": "tampered"},
|
||||
"arguments mismatch",
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_confirmation_processor_rejections(
|
||||
tools, original_args, confirmation_args, expected_exception_match
|
||||
):
|
||||
"""Test various validation rejections in request confirmation processor."""
|
||||
agent = LlmAgent(name="test_agent", tools=tools)
|
||||
invocation_context = await testing_utils.create_invocation_context(
|
||||
agent=agent
|
||||
)
|
||||
llm_request = LlmRequest()
|
||||
|
||||
original_function_call = types.FunctionCall(
|
||||
name=MOCK_TOOL_NAME, args=original_args, id=MOCK_FUNCTION_CALL_ID
|
||||
)
|
||||
|
||||
# 1. Event with the original tool call
|
||||
invocation_context.session.events.append(
|
||||
Event(
|
||||
author=agent.name,
|
||||
content=types.Content(
|
||||
parts=[types.Part(function_call=original_function_call)]
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# 2. Confirmation request event from the agent to the client.
|
||||
confirmation_function_call = types.FunctionCall(
|
||||
name=MOCK_TOOL_NAME, args=confirmation_args, id=MOCK_FUNCTION_CALL_ID
|
||||
)
|
||||
tool_confirmation = ToolConfirmation(confirmed=False, hint="test hint")
|
||||
tool_confirmation_args = {
|
||||
"originalFunctionCall": confirmation_function_call.model_dump(
|
||||
exclude_none=True, by_alias=True
|
||||
),
|
||||
"toolConfirmation": tool_confirmation.model_dump(
|
||||
by_alias=True, exclude_none=True
|
||||
),
|
||||
}
|
||||
|
||||
invocation_context.session.events.append(
|
||||
Event(
|
||||
author=agent.name,
|
||||
content=types.Content(
|
||||
parts=[
|
||||
types.Part(
|
||||
function_call=types.FunctionCall(
|
||||
name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
|
||||
args=tool_confirmation_args,
|
||||
id=MOCK_CONFIRMATION_FUNCTION_CALL_ID,
|
||||
)
|
||||
)
|
||||
]
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# 3. Event with the user's confirmation response.
|
||||
user_confirmation = ToolConfirmation(confirmed=True)
|
||||
invocation_context.session.events.append(
|
||||
Event(
|
||||
author="user",
|
||||
content=types.Content(
|
||||
parts=[
|
||||
types.Part(
|
||||
function_response=types.FunctionResponse(
|
||||
name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
|
||||
id=MOCK_CONFIRMATION_FUNCTION_CALL_ID,
|
||||
response={
|
||||
"response": user_confirmation.model_dump_json()
|
||||
},
|
||||
)
|
||||
)
|
||||
]
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match=expected_exception_match):
|
||||
async for _ in request_processor.run_async(invocation_context, llm_request):
|
||||
pass
|
||||
|
||||
Reference in New Issue
Block a user