mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Improved AzureAI Package Test Coverage (#3452)
* improved azureai test coverage * small fix * fix failing tests
This commit is contained in:
Giles Odigwe
committed by
GitHub
Unverified
GitHub
parent
474b31393d
commit
f6c4e078ec
@@ -19,6 +19,7 @@ from azure.ai.agents.models import (
|
||||
Agent,
|
||||
CodeInterpreterToolDefinition,
|
||||
)
|
||||
from azure.identity.aio import AzureCliCredential
|
||||
from pydantic import BaseModel
|
||||
|
||||
from agent_framework_azure_ai import (
|
||||
@@ -464,7 +465,77 @@ def test_as_agent_with_hosted_tools(
|
||||
|
||||
assert isinstance(agent, ChatAgent)
|
||||
# Should have HostedCodeInterpreterTool in the default_options tools
|
||||
assert any(isinstance(t, HostedCodeInterpreterTool) for t in (agent.default_options.get("tools") or []))
|
||||
assert any(isinstance(t, HostedCodeInterpreterTool) for t in (agent.default_options.get("tools") or [])) # type: ignore
|
||||
|
||||
|
||||
def test_as_agent_with_dict_function_tools_validates(
|
||||
azure_ai_unit_test_env: dict[str, str],
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test as_agent validates dict-format function tools require implementations."""
|
||||
# Dict-based function tool (as returned by some Azure AI SDK operations)
|
||||
dict_function_tool = { # type: ignore
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "dict_based_function",
|
||||
"description": "A function defined as dict",
|
||||
"parameters": {"type": "object", "properties": {}},
|
||||
},
|
||||
}
|
||||
|
||||
mock_agent = MagicMock(spec=Agent)
|
||||
mock_agent.id = "agent-id"
|
||||
mock_agent.name = "Agent"
|
||||
mock_agent.description = None
|
||||
mock_agent.instructions = None
|
||||
mock_agent.model = "gpt-4"
|
||||
mock_agent.temperature = None
|
||||
mock_agent.top_p = None
|
||||
mock_agent.tools = [dict_function_tool]
|
||||
|
||||
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
|
||||
|
||||
with pytest.raises(ServiceInitializationError) as exc_info:
|
||||
provider.as_agent(mock_agent)
|
||||
|
||||
assert "dict_based_function" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_as_agent_with_dict_function_tools_provided(
|
||||
azure_ai_unit_test_env: dict[str, str],
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test as_agent succeeds when dict-format function tools have implementations provided."""
|
||||
dict_function_tool = { # type: ignore
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "dict_based_function",
|
||||
"description": "A function defined as dict",
|
||||
"parameters": {"type": "object", "properties": {}},
|
||||
},
|
||||
}
|
||||
|
||||
mock_agent = MagicMock(spec=Agent)
|
||||
mock_agent.id = "agent-id"
|
||||
mock_agent.name = "Agent"
|
||||
mock_agent.description = None
|
||||
mock_agent.instructions = None
|
||||
mock_agent.model = "gpt-4"
|
||||
mock_agent.temperature = None
|
||||
mock_agent.top_p = None
|
||||
mock_agent.tools = [dict_function_tool]
|
||||
|
||||
@tool
|
||||
def dict_based_function() -> str:
|
||||
"""A function implementation."""
|
||||
return "result"
|
||||
|
||||
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
|
||||
|
||||
agent = provider.as_agent(mock_agent, tools=dict_based_function)
|
||||
|
||||
assert isinstance(agent, ChatAgent)
|
||||
assert agent.id == "agent-id"
|
||||
|
||||
|
||||
# endregion
|
||||
@@ -729,8 +800,6 @@ def test_from_azure_ai_agent_tools_unknown_dict() -> None:
|
||||
@skip_if_azure_ai_integration_tests_disabled
|
||||
async def test_integration_create_agent() -> None:
|
||||
"""Integration test: Create an agent using the provider."""
|
||||
from azure.identity.aio import AzureCliCredential
|
||||
|
||||
async with (
|
||||
AzureCliCredential() as credential,
|
||||
AzureAIAgentsProvider(credential=credential) as provider,
|
||||
@@ -753,8 +822,6 @@ async def test_integration_create_agent() -> None:
|
||||
@skip_if_azure_ai_integration_tests_disabled
|
||||
async def test_integration_get_agent() -> None:
|
||||
"""Integration test: Get an existing agent using the provider."""
|
||||
from azure.identity.aio import AzureCliCredential
|
||||
|
||||
async with (
|
||||
AzureCliCredential() as credential,
|
||||
AzureAIAgentsProvider(credential=credential) as provider,
|
||||
@@ -779,8 +846,6 @@ async def test_integration_get_agent() -> None:
|
||||
@skip_if_azure_ai_integration_tests_disabled
|
||||
async def test_integration_create_and_run() -> None:
|
||||
"""Integration test: Create an agent and run a conversation."""
|
||||
from azure.identity.aio import AzureCliCredential
|
||||
|
||||
async with (
|
||||
AzureCliCredential() as credential,
|
||||
AzureAIAgentsProvider(credential=credential) as provider,
|
||||
|
||||
@@ -26,16 +26,19 @@ from agent_framework import (
|
||||
tool,
|
||||
)
|
||||
from agent_framework._serialization import SerializationMixin
|
||||
from agent_framework.exceptions import ServiceInitializationError
|
||||
from agent_framework.exceptions import ServiceInitializationError, ServiceInvalidRequestError
|
||||
from azure.ai.agents.models import (
|
||||
AgentsNamedToolChoice,
|
||||
AgentsNamedToolChoiceType,
|
||||
AgentsToolChoiceOptionMode,
|
||||
CodeInterpreterToolDefinition,
|
||||
FileInfo,
|
||||
MessageDeltaChunk,
|
||||
MessageDeltaTextContent,
|
||||
MessageDeltaTextFileCitationAnnotation,
|
||||
MessageDeltaTextFilePathAnnotation,
|
||||
MessageDeltaTextUrlCitationAnnotation,
|
||||
MessageInputTextBlock,
|
||||
RequiredFunctionToolCall,
|
||||
RequiredMcpToolCall,
|
||||
RunStatus,
|
||||
@@ -593,8 +596,6 @@ async def test_azure_ai_chat_client_prepare_options_with_none_tool_choice(
|
||||
|
||||
run_options, _ = await chat_client._prepare_options([], chat_options) # type: ignore
|
||||
|
||||
from azure.ai.agents.models import AgentsToolChoiceOptionMode
|
||||
|
||||
assert run_options["tool_choice"] == AgentsToolChoiceOptionMode.NONE
|
||||
|
||||
|
||||
@@ -608,8 +609,6 @@ async def test_azure_ai_chat_client_prepare_options_with_auto_tool_choice(
|
||||
|
||||
run_options, _ = await chat_client._prepare_options([], chat_options) # type: ignore
|
||||
|
||||
from azure.ai.agents.models import AgentsToolChoiceOptionMode
|
||||
|
||||
assert run_options["tool_choice"] == AgentsToolChoiceOptionMode.AUTO
|
||||
|
||||
|
||||
@@ -1942,3 +1941,205 @@ def test_azure_ai_chat_client_init_with_auto_created_agents_client(
|
||||
assert client.agent_id == "test-agent"
|
||||
assert client.credential is mock_azure_credential
|
||||
assert client._should_close_client is True # Should close since we created it # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_azure_ai_chat_client_prepare_options_with_mapping_response_format(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_options with Mapping-based response_format (runtime JSON schema)."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client)
|
||||
|
||||
# Runtime JSON schema dict
|
||||
response_format_dict = {
|
||||
"type": "json_schema",
|
||||
"json_schema": {
|
||||
"name": "TestSchema",
|
||||
"schema": {"type": "object", "properties": {"name": {"type": "string"}}},
|
||||
},
|
||||
}
|
||||
|
||||
chat_options: ChatOptions = {"response_format": response_format_dict} # type: ignore[typeddict-item]
|
||||
|
||||
run_options, _ = await chat_client._prepare_options([], chat_options) # type: ignore
|
||||
|
||||
assert "response_format" in run_options
|
||||
# Should pass through as-is for Mapping types
|
||||
assert run_options["response_format"] == response_format_dict
|
||||
|
||||
|
||||
async def test_azure_ai_chat_client_prepare_options_with_invalid_response_format(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_options with invalid response_format raises error."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client)
|
||||
|
||||
# Invalid response_format (not BaseModel or Mapping)
|
||||
chat_options: ChatOptions = {"response_format": "invalid_format"} # type: ignore[typeddict-item]
|
||||
|
||||
with pytest.raises(ServiceInvalidRequestError, match="response_format must be a Pydantic BaseModel"):
|
||||
await chat_client._prepare_options([], chat_options) # type: ignore
|
||||
|
||||
|
||||
async def test_azure_ai_chat_client_prepare_tool_definitions_with_agent_tool_resources(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_tool_definitions_and_resources copies tool_resources from agent definition."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
|
||||
|
||||
# Create mock agent definition with tool_resources
|
||||
mock_agent_definition = MagicMock()
|
||||
mock_agent_definition.tools = []
|
||||
mock_agent_definition.tool_resources = {"code_interpreter": {"file_ids": ["file-123"]}}
|
||||
|
||||
run_options: dict[str, Any] = {}
|
||||
options: dict[str, Any] = {}
|
||||
|
||||
await chat_client._prepare_tool_definitions_and_resources(options, mock_agent_definition, run_options) # type: ignore
|
||||
|
||||
# Verify tool_resources was copied to run_options
|
||||
assert "tool_resources" in run_options
|
||||
assert run_options["tool_resources"] == {"code_interpreter": {"file_ids": ["file-123"]}}
|
||||
|
||||
|
||||
def test_azure_ai_chat_client_prepare_mcp_resources_with_dict_approval_mode(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_mcp_resources with dict-based approval mode (always_require_approval)."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client)
|
||||
|
||||
# MCP tool with dict-based approval mode
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
approval_mode={"always_require_approval": {"tool1", "tool2"}},
|
||||
)
|
||||
|
||||
result = chat_client._prepare_mcp_resources([mcp_tool]) # type: ignore
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["server_label"] == "Test_MCP"
|
||||
assert "require_approval" in result[0]
|
||||
assert result[0]["require_approval"] == {"always": {"tool1", "tool2"}}
|
||||
|
||||
|
||||
def test_azure_ai_chat_client_prepare_mcp_resources_with_never_require_dict(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_mcp_resources with dict-based approval mode (never_require_approval)."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client)
|
||||
|
||||
# MCP tool with never_require_approval dict
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
approval_mode={"never_require_approval": {"safe_tool"}},
|
||||
)
|
||||
|
||||
result = chat_client._prepare_mcp_resources([mcp_tool]) # type: ignore
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["require_approval"] == {"never": {"safe_tool"}}
|
||||
|
||||
|
||||
def test_azure_ai_chat_client_prepare_messages_with_function_result(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_messages extracts function_result content."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client)
|
||||
|
||||
function_result = Content.from_function_result(call_id='["run_123", "call_456"]', result="test result")
|
||||
messages = [ChatMessage(role=Role.USER, contents=[function_result])]
|
||||
|
||||
additional_messages, instructions, required_action_results = chat_client._prepare_messages(messages) # type: ignore
|
||||
|
||||
# function_result should be extracted, not added to additional_messages
|
||||
assert additional_messages is None
|
||||
assert required_action_results is not None
|
||||
assert len(required_action_results) == 1
|
||||
assert required_action_results[0].type == "function_result"
|
||||
|
||||
|
||||
def test_azure_ai_chat_client_prepare_messages_with_raw_content_block(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_messages handles raw MessageInputContentBlock in content."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client)
|
||||
|
||||
# Create content with raw_representation that is a MessageInputContentBlock
|
||||
raw_block = MessageInputTextBlock(text="Raw block text")
|
||||
custom_content = Content(type="custom", raw_representation=raw_block)
|
||||
messages = [ChatMessage(role=Role.USER, contents=[custom_content])]
|
||||
|
||||
additional_messages, instructions, required_action_results = chat_client._prepare_messages(messages) # type: ignore
|
||||
|
||||
assert additional_messages is not None
|
||||
assert len(additional_messages) == 1
|
||||
assert len(additional_messages[0].content) == 1
|
||||
assert additional_messages[0].content[0] == raw_block
|
||||
|
||||
|
||||
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_mcp_tool(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_tools_for_azure_ai with HostedMCPTool."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
|
||||
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP Server",
|
||||
url="https://example.com/mcp",
|
||||
allowed_tools=["tool1", "tool2"],
|
||||
)
|
||||
|
||||
tool_definitions = await chat_client._prepare_tools_for_azure_ai([mcp_tool]) # type: ignore
|
||||
|
||||
assert len(tool_definitions) >= 1
|
||||
# The McpTool.definitions property returns the tool definitions
|
||||
# Verify the MCP tool was converted correctly by checking the definition type
|
||||
mcp_def = tool_definitions[0]
|
||||
assert mcp_def.get("type") == "mcp"
|
||||
|
||||
|
||||
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_tool_definition(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_tools_for_azure_ai with ToolDefinition passthrough."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
|
||||
|
||||
# Pass a ToolDefinition directly - should be passed through as-is
|
||||
tool_def = CodeInterpreterToolDefinition()
|
||||
|
||||
tool_definitions = await chat_client._prepare_tools_for_azure_ai([tool_def]) # type: ignore
|
||||
|
||||
assert len(tool_definitions) == 1
|
||||
assert tool_definitions[0] is tool_def
|
||||
|
||||
|
||||
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_dict_passthrough(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_tools_for_azure_ai with dict passthrough."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
|
||||
|
||||
# Pass a dict tool definition - should be passed through as-is
|
||||
dict_tool = {"type": "function", "function": {"name": "test_func", "parameters": {}}}
|
||||
|
||||
tool_definitions = await chat_client._prepare_tools_for_azure_ai([dict_tool]) # type: ignore
|
||||
|
||||
assert len(tool_definitions) == 1
|
||||
assert tool_definitions[0] is dict_tool
|
||||
|
||||
|
||||
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_unsupported_type(
|
||||
mock_agents_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _prepare_tools_for_azure_ai raises error for unsupported tool type."""
|
||||
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
|
||||
|
||||
# Pass an unsupported tool type
|
||||
class UnsupportedTool:
|
||||
pass
|
||||
|
||||
unsupported_tool = UnsupportedTool()
|
||||
|
||||
with pytest.raises(ServiceInitializationError, match="Unsupported tool type"):
|
||||
await chat_client._prepare_tools_for_azure_ai([unsupported_tool]) # type: ignore
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from collections.abc import AsyncGenerator, AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Annotated, Any
|
||||
@@ -35,6 +36,7 @@ from azure.ai.projects.models import (
|
||||
ResponseTextFormatConfigurationJsonSchema,
|
||||
WebSearchPreviewTool,
|
||||
)
|
||||
from azure.core.exceptions import ResourceNotFoundError
|
||||
from azure.identity.aio import AzureCliCredential
|
||||
from openai.types.responses.parsed_response import ParsedResponse
|
||||
from openai.types.responses.response import Response as OpenAIResponse
|
||||
@@ -580,6 +582,107 @@ async def test_close_client_when_should_close_false(mock_project_client: MagicMo
|
||||
mock_project_client.close.assert_not_called()
|
||||
|
||||
|
||||
async def test_configure_azure_monitor_success(mock_project_client: MagicMock) -> None:
|
||||
"""Test configure_azure_monitor successfully configures Azure Monitor."""
|
||||
client = create_test_azure_ai_client(mock_project_client)
|
||||
|
||||
# Mock the telemetry connection string retrieval
|
||||
mock_project_client.telemetry.get_application_insights_connection_string = AsyncMock(
|
||||
return_value="InstrumentationKey=test-key;IngestionEndpoint=https://test.endpoint"
|
||||
)
|
||||
|
||||
mock_configure = MagicMock()
|
||||
mock_views = MagicMock(return_value=[])
|
||||
mock_resource = MagicMock()
|
||||
mock_enable = MagicMock()
|
||||
|
||||
with (
|
||||
patch.dict(
|
||||
"sys.modules",
|
||||
{"azure.monitor.opentelemetry": MagicMock(configure_azure_monitor=mock_configure)},
|
||||
),
|
||||
patch("agent_framework.observability.create_metric_views", mock_views),
|
||||
patch("agent_framework.observability.create_resource", return_value=mock_resource),
|
||||
patch("agent_framework.observability.enable_instrumentation", mock_enable),
|
||||
):
|
||||
await client.configure_azure_monitor(enable_sensitive_data=True)
|
||||
|
||||
# Verify connection string was retrieved
|
||||
mock_project_client.telemetry.get_application_insights_connection_string.assert_called_once()
|
||||
|
||||
# Verify Azure Monitor was configured
|
||||
mock_configure.assert_called_once()
|
||||
call_kwargs = mock_configure.call_args[1]
|
||||
assert call_kwargs["connection_string"] == "InstrumentationKey=test-key;IngestionEndpoint=https://test.endpoint"
|
||||
|
||||
# Verify instrumentation was enabled with sensitive data flag
|
||||
mock_enable.assert_called_once_with(enable_sensitive_data=True)
|
||||
|
||||
|
||||
async def test_configure_azure_monitor_resource_not_found(mock_project_client: MagicMock) -> None:
|
||||
"""Test configure_azure_monitor handles ResourceNotFoundError gracefully."""
|
||||
client = create_test_azure_ai_client(mock_project_client)
|
||||
|
||||
# Mock the telemetry to raise ResourceNotFoundError
|
||||
mock_project_client.telemetry.get_application_insights_connection_string = AsyncMock(
|
||||
side_effect=ResourceNotFoundError("No Application Insights found")
|
||||
)
|
||||
|
||||
# Should not raise, just log warning and return
|
||||
await client.configure_azure_monitor()
|
||||
|
||||
# Verify connection string retrieval was attempted
|
||||
mock_project_client.telemetry.get_application_insights_connection_string.assert_called_once()
|
||||
|
||||
|
||||
async def test_configure_azure_monitor_import_error(mock_project_client: MagicMock) -> None:
|
||||
"""Test configure_azure_monitor raises ImportError when azure-monitor-opentelemetry is not installed."""
|
||||
client = create_test_azure_ai_client(mock_project_client)
|
||||
|
||||
# Mock the telemetry connection string retrieval
|
||||
mock_project_client.telemetry.get_application_insights_connection_string = AsyncMock(
|
||||
return_value="InstrumentationKey=test-key"
|
||||
)
|
||||
|
||||
# Mock the import to fail
|
||||
with (
|
||||
patch.dict(sys.modules, {"azure.monitor.opentelemetry": None}),
|
||||
patch("builtins.__import__", side_effect=ImportError("No module named 'azure.monitor.opentelemetry'")),
|
||||
pytest.raises(ImportError, match="azure-monitor-opentelemetry is required"),
|
||||
):
|
||||
await client.configure_azure_monitor()
|
||||
|
||||
|
||||
async def test_configure_azure_monitor_with_custom_resource(mock_project_client: MagicMock) -> None:
|
||||
"""Test configure_azure_monitor uses custom resource when provided."""
|
||||
client = create_test_azure_ai_client(mock_project_client)
|
||||
|
||||
mock_project_client.telemetry.get_application_insights_connection_string = AsyncMock(
|
||||
return_value="InstrumentationKey=test-key"
|
||||
)
|
||||
|
||||
custom_resource = MagicMock()
|
||||
mock_configure = MagicMock()
|
||||
|
||||
with (
|
||||
patch.dict(
|
||||
"sys.modules",
|
||||
{"azure.monitor.opentelemetry": MagicMock(configure_azure_monitor=mock_configure)},
|
||||
),
|
||||
patch("agent_framework.observability.create_metric_views") as mock_views,
|
||||
patch("agent_framework.observability.create_resource") as mock_create_resource,
|
||||
patch("agent_framework.observability.enable_instrumentation"),
|
||||
):
|
||||
mock_views.return_value = []
|
||||
|
||||
await client.configure_azure_monitor(resource=custom_resource)
|
||||
|
||||
# Verify custom resource was used, not create_resource
|
||||
mock_create_resource.assert_not_called()
|
||||
call_kwargs = mock_configure.call_args[1]
|
||||
assert call_kwargs["resource"] is custom_resource
|
||||
|
||||
|
||||
async def test_agent_creation_with_instructions(
|
||||
mock_project_client: MagicMock,
|
||||
) -> None:
|
||||
@@ -676,8 +779,6 @@ async def test_use_latest_version_agent_not_found(
|
||||
mock_project_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test _get_agent_reference_or_create when use_latest_version=True but agent doesn't exist."""
|
||||
from azure.core.exceptions import ResourceNotFoundError
|
||||
|
||||
client = create_test_azure_ai_client(mock_project_client, agent_name="non-existing-agent", use_latest_version=True)
|
||||
|
||||
# Mock ResourceNotFoundError when trying to retrieve agent
|
||||
@@ -971,6 +1072,128 @@ def test_get_conversation_id_with_parsed_response_no_conversation() -> None:
|
||||
assert result == "resp_parsed_12345"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_basic() -> None:
|
||||
"""Test _prepare_mcp_tool with basic HostedMCPTool."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP Server",
|
||||
url="https://example.com/mcp",
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert result["server_label"] == "Test_MCP_Server"
|
||||
assert result["server_url"] == "https://example.com/mcp"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_description() -> None:
|
||||
"""Test _prepare_mcp_tool with description."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
description="A test MCP server",
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert result["server_description"] == "A test MCP server"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_project_connection_id() -> None:
|
||||
"""Test _prepare_mcp_tool with project_connection_id in additional_properties."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
additional_properties={"project_connection_id": "conn-123"},
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert result["project_connection_id"] == "conn-123"
|
||||
assert "headers" not in result # headers should not be set when project_connection_id is present
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_headers() -> None:
|
||||
"""Test _prepare_mcp_tool with headers (no project_connection_id)."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
headers={"Authorization": "Bearer token123"},
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert result["headers"] == {"Authorization": "Bearer token123"}
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_allowed_tools() -> None:
|
||||
"""Test _prepare_mcp_tool with allowed_tools."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
allowed_tools=["tool1", "tool2"],
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert set(result["allowed_tools"]) == {"tool1", "tool2"}
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_approval_mode_always_require() -> None:
|
||||
"""Test _prepare_mcp_tool with string approval_mode 'always_require'."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
approval_mode="always_require",
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert result["require_approval"] == "always"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_approval_mode_never_require() -> None:
|
||||
"""Test _prepare_mcp_tool with string approval_mode 'never_require'."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
approval_mode="never_require",
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert result["require_approval"] == "never"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_dict_approval_mode_always() -> None:
|
||||
"""Test _prepare_mcp_tool with dict approval_mode containing always_require_approval."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
approval_mode={"always_require_approval": {"dangerous_tool", "risky_tool"}},
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert "require_approval" in result
|
||||
assert "always" in result["require_approval"]
|
||||
assert set(result["require_approval"]["always"]["tool_names"]) == {"dangerous_tool", "risky_tool"}
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_dict_approval_mode_never() -> None:
|
||||
"""Test _prepare_mcp_tool with dict approval_mode containing never_require_approval."""
|
||||
mcp_tool = HostedMCPTool(
|
||||
name="Test MCP",
|
||||
url="https://example.com/mcp",
|
||||
approval_mode={"never_require_approval": {"safe_tool"}},
|
||||
)
|
||||
|
||||
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
|
||||
|
||||
assert "require_approval" in result
|
||||
assert "never" in result["require_approval"]
|
||||
assert set(result["require_approval"]["never"]["tool_names"]) == {"safe_tool"}
|
||||
|
||||
|
||||
def test_from_azure_ai_tools() -> None:
|
||||
"""Test from_azure_ai_tools."""
|
||||
# Test MCP tool
|
||||
|
||||
@@ -424,6 +424,42 @@ def test_provider_as_agent(mock_project_client: MagicMock) -> None:
|
||||
assert call_kwargs["model_deployment_name"] == "gpt-4"
|
||||
|
||||
|
||||
def test_provider_merge_tools_skips_function_tool_dicts(mock_project_client: MagicMock) -> None:
|
||||
"""Test that _merge_tools skips function tool dicts but keeps other hosted tools."""
|
||||
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
|
||||
|
||||
# Create a mock FunctionTool to provide as implementation
|
||||
mock_ai_function = create_mock_ai_function("my_function", "My function description")
|
||||
|
||||
# Definition tools include a function tool (dict) and an MCP tool
|
||||
definition_tools = [
|
||||
{"type": "function", "name": "my_function", "parameters": {}}, # Should be skipped
|
||||
{"type": "mcp", "server_label": "my_mcp", "server_url": "http://localhost:8080"}, # Should be converted
|
||||
]
|
||||
|
||||
# Call _merge_tools with user-provided function implementation
|
||||
merged = provider._merge_tools(definition_tools, [mock_ai_function]) # type: ignore
|
||||
|
||||
# Should have 2 items: the converted HostedMCPTool and the user-provided FunctionTool
|
||||
assert len(merged) == 2
|
||||
|
||||
# Check that the function tool dict was NOT included (it was skipped)
|
||||
function_dicts = [t for t in merged if isinstance(t, dict) and t.get("type") == "function"]
|
||||
assert len(function_dicts) == 0
|
||||
|
||||
# Check that the MCP tool was converted to HostedMCPTool
|
||||
from agent_framework import HostedMCPTool
|
||||
|
||||
mcp_tools = [t for t in merged if isinstance(t, HostedMCPTool)]
|
||||
assert len(mcp_tools) == 1
|
||||
assert mcp_tools[0].name == "my mcp" # server_label with _ replaced by space
|
||||
|
||||
# Check that the user-provided FunctionTool was included
|
||||
ai_functions = [t for t in merged if isinstance(t, FunctionTool)]
|
||||
assert len(ai_functions) == 1
|
||||
assert ai_functions[0].name == "my_function"
|
||||
|
||||
|
||||
async def test_provider_context_manager(mock_project_client: MagicMock) -> None:
|
||||
"""Test AzureAIProjectAgentProvider async context manager."""
|
||||
with patch("agent_framework_azure_ai._project_provider.AIProjectClient") as mock_ai_project_client:
|
||||
|
||||
@@ -0,0 +1,431 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from agent_framework import (
|
||||
Content,
|
||||
FunctionTool,
|
||||
HostedCodeInterpreterTool,
|
||||
HostedFileSearchTool,
|
||||
HostedImageGenerationTool,
|
||||
HostedMCPTool,
|
||||
HostedWebSearchTool,
|
||||
)
|
||||
from agent_framework.exceptions import ServiceInitializationError, ServiceInvalidRequestError
|
||||
from azure.ai.agents.models import CodeInterpreterToolDefinition
|
||||
from pydantic import BaseModel
|
||||
|
||||
from agent_framework_azure_ai._shared import (
|
||||
_convert_response_format, # type: ignore
|
||||
_convert_sdk_tool, # type: ignore
|
||||
_extract_project_connection_id, # type: ignore
|
||||
_prepare_mcp_tool_for_azure_ai, # type: ignore
|
||||
create_text_format_config,
|
||||
from_azure_ai_agent_tools,
|
||||
from_azure_ai_tools,
|
||||
to_azure_ai_agent_tools,
|
||||
to_azure_ai_tools,
|
||||
)
|
||||
|
||||
|
||||
def test_extract_project_connection_id_direct() -> None:
|
||||
"""Test extracting project_connection_id from direct key."""
|
||||
result = _extract_project_connection_id({"project_connection_id": "my-connection"})
|
||||
assert result == "my-connection"
|
||||
|
||||
|
||||
def test_extract_project_connection_id_from_connection_name() -> None:
|
||||
"""Test extracting project_connection_id from connection.name structure."""
|
||||
result = _extract_project_connection_id({"connection": {"name": "my-connection"}})
|
||||
assert result == "my-connection"
|
||||
|
||||
|
||||
def test_extract_project_connection_id_none() -> None:
|
||||
"""Test returns None when no connection info."""
|
||||
assert _extract_project_connection_id(None) is None
|
||||
assert _extract_project_connection_id({}) is None
|
||||
|
||||
|
||||
def test_to_azure_ai_agent_tools_empty() -> None:
|
||||
"""Test converting empty/None tools list."""
|
||||
assert to_azure_ai_agent_tools(None) == []
|
||||
assert to_azure_ai_agent_tools([]) == []
|
||||
|
||||
|
||||
def test_to_azure_ai_agent_tools_function_tool() -> None:
|
||||
"""Test converting FunctionTool to tool definition."""
|
||||
|
||||
def my_func(arg: str) -> str:
|
||||
"""My function."""
|
||||
return arg
|
||||
|
||||
func_tool = FunctionTool(func=my_func, name="my_func", description="My function.") # type: ignore
|
||||
result = to_azure_ai_agent_tools([func_tool]) # type: ignore
|
||||
assert len(result) == 1
|
||||
assert result[0]["type"] == "function"
|
||||
assert result[0]["function"]["name"] == "my_func"
|
||||
|
||||
|
||||
def test_to_azure_ai_agent_tools_code_interpreter() -> None:
|
||||
"""Test converting HostedCodeInterpreterTool."""
|
||||
tool = HostedCodeInterpreterTool()
|
||||
result = to_azure_ai_agent_tools([tool])
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], CodeInterpreterToolDefinition)
|
||||
|
||||
|
||||
def test_to_azure_ai_agent_tools_web_search_missing_connection() -> None:
|
||||
"""Test HostedWebSearchTool raises without connection info."""
|
||||
tool = HostedWebSearchTool()
|
||||
with pytest.raises(ServiceInitializationError, match="Bing search tool requires"):
|
||||
to_azure_ai_agent_tools([tool])
|
||||
|
||||
|
||||
def test_to_azure_ai_agent_tools_dict_passthrough() -> None:
|
||||
"""Test dict tools pass through unchanged."""
|
||||
tool_dict = {"type": "custom", "config": "value"}
|
||||
result = to_azure_ai_agent_tools([tool_dict])
|
||||
assert result[0] == tool_dict
|
||||
|
||||
|
||||
def test_to_azure_ai_agent_tools_unsupported_type() -> None:
|
||||
"""Test unsupported tool type raises error."""
|
||||
|
||||
class UnsupportedTool:
|
||||
pass
|
||||
|
||||
with pytest.raises(ServiceInitializationError, match="Unsupported tool type"):
|
||||
to_azure_ai_agent_tools([UnsupportedTool()]) # type: ignore
|
||||
|
||||
|
||||
def test_from_azure_ai_agent_tools_empty() -> None:
|
||||
"""Test converting empty/None tools list."""
|
||||
assert from_azure_ai_agent_tools(None) == []
|
||||
assert from_azure_ai_agent_tools([]) == []
|
||||
|
||||
|
||||
def test_from_azure_ai_agent_tools_code_interpreter() -> None:
|
||||
"""Test converting CodeInterpreterToolDefinition."""
|
||||
tool = CodeInterpreterToolDefinition()
|
||||
result = from_azure_ai_agent_tools([tool])
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], HostedCodeInterpreterTool)
|
||||
|
||||
|
||||
def test_convert_sdk_tool_code_interpreter() -> None:
|
||||
"""Test _convert_sdk_tool with code_interpreter type."""
|
||||
tool = MagicMock()
|
||||
tool.type = "code_interpreter"
|
||||
result = _convert_sdk_tool(tool)
|
||||
assert isinstance(result, HostedCodeInterpreterTool)
|
||||
|
||||
|
||||
def test_convert_sdk_tool_function_returns_none() -> None:
|
||||
"""Test _convert_sdk_tool with function type returns None."""
|
||||
tool = MagicMock()
|
||||
tool.type = "function"
|
||||
result = _convert_sdk_tool(tool)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_convert_sdk_tool_mcp_returns_none() -> None:
|
||||
"""Test _convert_sdk_tool with mcp type returns None."""
|
||||
tool = MagicMock()
|
||||
tool.type = "mcp"
|
||||
result = _convert_sdk_tool(tool)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_convert_sdk_tool_file_search() -> None:
|
||||
"""Test _convert_sdk_tool with file_search type."""
|
||||
tool = MagicMock()
|
||||
tool.type = "file_search"
|
||||
tool.file_search = MagicMock()
|
||||
tool.file_search.vector_store_ids = ["vs-1", "vs-2"]
|
||||
result = _convert_sdk_tool(tool)
|
||||
assert isinstance(result, HostedFileSearchTool)
|
||||
assert len(result.inputs) == 2 # type: ignore
|
||||
|
||||
|
||||
def test_convert_sdk_tool_bing_grounding() -> None:
|
||||
"""Test _convert_sdk_tool with bing_grounding type."""
|
||||
tool = MagicMock()
|
||||
tool.type = "bing_grounding"
|
||||
tool.bing_grounding = MagicMock()
|
||||
tool.bing_grounding.connection_id = "conn-123"
|
||||
result = _convert_sdk_tool(tool)
|
||||
assert isinstance(result, HostedWebSearchTool)
|
||||
assert result.additional_properties["connection_id"] == "conn-123" # type: ignore
|
||||
|
||||
|
||||
def test_convert_sdk_tool_bing_custom_search() -> None:
|
||||
"""Test _convert_sdk_tool with bing_custom_search type."""
|
||||
tool = MagicMock()
|
||||
tool.type = "bing_custom_search"
|
||||
tool.bing_custom_search = MagicMock()
|
||||
tool.bing_custom_search.connection_id = "conn-123"
|
||||
tool.bing_custom_search.instance_name = "my-instance"
|
||||
result = _convert_sdk_tool(tool)
|
||||
assert isinstance(result, HostedWebSearchTool)
|
||||
assert result.additional_properties["custom_connection_id"] == "conn-123" # type: ignore
|
||||
assert result.additional_properties["custom_instance_name"] == "my-instance" # type: ignore
|
||||
|
||||
|
||||
def test_to_azure_ai_tools_empty() -> None:
|
||||
"""Test converting empty/None tools list."""
|
||||
assert to_azure_ai_tools(None) == []
|
||||
assert to_azure_ai_tools([]) == []
|
||||
|
||||
|
||||
def test_to_azure_ai_tools_code_interpreter_with_file_ids() -> None:
|
||||
"""Test converting HostedCodeInterpreterTool with file inputs."""
|
||||
tool = HostedCodeInterpreterTool(
|
||||
inputs=[Content.from_hosted_file(file_id="file-123")] # type: ignore
|
||||
)
|
||||
result = to_azure_ai_tools([tool])
|
||||
assert len(result) == 1
|
||||
assert result[0]["type"] == "code_interpreter"
|
||||
assert result[0]["container"]["file_ids"] == ["file-123"]
|
||||
|
||||
|
||||
def test_to_azure_ai_tools_function_tool() -> None:
|
||||
"""Test converting FunctionTool."""
|
||||
|
||||
def my_func(arg: str) -> str:
|
||||
"""My function."""
|
||||
return arg
|
||||
|
||||
func_tool = FunctionTool(func=my_func, name="my_func", description="My function.") # type: ignore
|
||||
result = to_azure_ai_tools([func_tool]) # type: ignore
|
||||
assert len(result) == 1
|
||||
assert result[0]["type"] == "function"
|
||||
assert result[0]["name"] == "my_func"
|
||||
|
||||
|
||||
def test_to_azure_ai_tools_file_search() -> None:
|
||||
"""Test converting HostedFileSearchTool."""
|
||||
tool = HostedFileSearchTool(
|
||||
inputs=[Content.from_hosted_vector_store(vector_store_id="vs-123")], # type: ignore
|
||||
max_results=10,
|
||||
)
|
||||
result = to_azure_ai_tools([tool])
|
||||
assert len(result) == 1
|
||||
assert result[0]["type"] == "file_search"
|
||||
assert result[0]["vector_store_ids"] == ["vs-123"]
|
||||
assert result[0]["max_num_results"] == 10
|
||||
|
||||
|
||||
def test_to_azure_ai_tools_web_search_with_location() -> None:
|
||||
"""Test converting HostedWebSearchTool with user location."""
|
||||
tool = HostedWebSearchTool(
|
||||
additional_properties={
|
||||
"user_location": {
|
||||
"city": "Seattle",
|
||||
"country": "US",
|
||||
"region": "WA",
|
||||
"timezone": "PST",
|
||||
}
|
||||
}
|
||||
)
|
||||
result = to_azure_ai_tools([tool])
|
||||
assert len(result) == 1
|
||||
assert result[0]["type"] == "web_search_preview"
|
||||
|
||||
|
||||
def test_to_azure_ai_tools_image_generation() -> None:
|
||||
"""Test converting HostedImageGenerationTool."""
|
||||
tool = HostedImageGenerationTool(
|
||||
options={"model_id": "gpt-image-1", "image_size": "1024x1024"},
|
||||
additional_properties={"quality": "high"},
|
||||
)
|
||||
result = to_azure_ai_tools([tool])
|
||||
assert len(result) == 1
|
||||
assert result[0]["type"] == "image_generation"
|
||||
assert result[0]["model"] == "gpt-image-1"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_basic() -> None:
|
||||
"""Test basic MCP tool conversion."""
|
||||
tool = HostedMCPTool(name="my tool", url="http://localhost:8080")
|
||||
result = _prepare_mcp_tool_for_azure_ai(tool)
|
||||
assert result["server_label"] == "my_tool"
|
||||
assert "http://localhost:8080" in result["server_url"]
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_description() -> None:
|
||||
"""Test MCP tool with description."""
|
||||
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", description="My MCP server")
|
||||
result = _prepare_mcp_tool_for_azure_ai(tool)
|
||||
assert result["server_description"] == "My MCP server"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_with_headers() -> None:
|
||||
"""Test MCP tool with headers (no project_connection_id)."""
|
||||
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", headers={"X-Api-Key": "secret"})
|
||||
result = _prepare_mcp_tool_for_azure_ai(tool)
|
||||
assert result["headers"] == {"X-Api-Key": "secret"}
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_project_connection_takes_precedence() -> None:
|
||||
"""Test project_connection_id takes precedence over headers."""
|
||||
tool = HostedMCPTool(
|
||||
name="my tool",
|
||||
url="http://localhost:8080",
|
||||
headers={"X-Api-Key": "secret"},
|
||||
additional_properties={"project_connection_id": "my-conn"},
|
||||
)
|
||||
result = _prepare_mcp_tool_for_azure_ai(tool)
|
||||
assert result["project_connection_id"] == "my-conn"
|
||||
assert "headers" not in result
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_approval_mode_always() -> None:
|
||||
"""Test MCP tool with always_require approval mode."""
|
||||
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", approval_mode="always_require")
|
||||
result = _prepare_mcp_tool_for_azure_ai(tool)
|
||||
assert result["require_approval"] == "always"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_approval_mode_never() -> None:
|
||||
"""Test MCP tool with never_require approval mode."""
|
||||
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", approval_mode="never_require")
|
||||
result = _prepare_mcp_tool_for_azure_ai(tool)
|
||||
assert result["require_approval"] == "never"
|
||||
|
||||
|
||||
def test_prepare_mcp_tool_approval_mode_dict() -> None:
|
||||
"""Test MCP tool with dict approval mode."""
|
||||
tool = HostedMCPTool(
|
||||
name="my tool",
|
||||
url="http://localhost:8080",
|
||||
approval_mode={
|
||||
"always_require_approval": {"sensitive_tool"},
|
||||
"never_require_approval": {"safe_tool"},
|
||||
},
|
||||
)
|
||||
result = _prepare_mcp_tool_for_azure_ai(tool)
|
||||
# The last assignment wins in the current implementation
|
||||
assert "require_approval" in result
|
||||
|
||||
|
||||
def test_create_text_format_config_pydantic_model() -> None:
|
||||
"""Test creating text format config from Pydantic model."""
|
||||
|
||||
class MySchema(BaseModel):
|
||||
name: str
|
||||
value: int
|
||||
|
||||
result = create_text_format_config(MySchema)
|
||||
assert result["type"] == "json_schema"
|
||||
assert result["name"] == "MySchema"
|
||||
assert result["strict"] is True
|
||||
|
||||
|
||||
def test_create_text_format_config_json_schema_mapping() -> None:
|
||||
"""Test creating text format config from json_schema mapping."""
|
||||
config = {
|
||||
"type": "json_schema",
|
||||
"json_schema": {
|
||||
"name": "MyResponse",
|
||||
"schema": {"type": "object", "properties": {"name": {"type": "string"}}},
|
||||
},
|
||||
}
|
||||
result = create_text_format_config(config)
|
||||
assert result["type"] == "json_schema"
|
||||
assert result["name"] == "MyResponse"
|
||||
|
||||
|
||||
def test_create_text_format_config_json_object() -> None:
|
||||
"""Test creating text format config for json_object type."""
|
||||
result = create_text_format_config({"type": "json_object"})
|
||||
assert result["type"] == "json_object"
|
||||
|
||||
|
||||
def test_create_text_format_config_text() -> None:
|
||||
"""Test creating text format config for text type."""
|
||||
result = create_text_format_config({"type": "text"})
|
||||
assert result["type"] == "text"
|
||||
|
||||
|
||||
def test_create_text_format_config_invalid_raises() -> None:
|
||||
"""Test invalid response_format raises error."""
|
||||
with pytest.raises(ServiceInvalidRequestError):
|
||||
create_text_format_config({"type": "invalid"})
|
||||
|
||||
|
||||
def test_convert_response_format_with_format_key() -> None:
|
||||
"""Test _convert_response_format with nested format key."""
|
||||
config = {"format": {"type": "json_object"}}
|
||||
result = _convert_response_format(config)
|
||||
assert result["type"] == "json_object"
|
||||
|
||||
|
||||
def test_convert_response_format_json_schema_missing_schema_raises() -> None:
|
||||
"""Test json_schema without schema raises error."""
|
||||
with pytest.raises(ServiceInvalidRequestError, match="requires a schema"):
|
||||
_convert_response_format({"type": "json_schema", "json_schema": {}})
|
||||
|
||||
|
||||
def test_from_azure_ai_tools_mcp_approval_mode_always() -> None:
|
||||
"""Test from_azure_ai_tools converts MCP require_approval='always' to approval_mode."""
|
||||
tools = [
|
||||
{
|
||||
"type": "mcp",
|
||||
"server_label": "my_mcp",
|
||||
"server_url": "http://localhost:8080",
|
||||
"require_approval": "always",
|
||||
}
|
||||
]
|
||||
result = from_azure_ai_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], HostedMCPTool)
|
||||
assert result[0].approval_mode == "always_require"
|
||||
|
||||
|
||||
def test_from_azure_ai_tools_mcp_approval_mode_never() -> None:
|
||||
"""Test from_azure_ai_tools converts MCP require_approval='never' to approval_mode."""
|
||||
tools = [
|
||||
{
|
||||
"type": "mcp",
|
||||
"server_label": "my_mcp",
|
||||
"server_url": "http://localhost:8080",
|
||||
"require_approval": "never",
|
||||
}
|
||||
]
|
||||
result = from_azure_ai_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], HostedMCPTool)
|
||||
assert result[0].approval_mode == "never_require"
|
||||
|
||||
|
||||
def test_from_azure_ai_tools_mcp_approval_mode_dict_always() -> None:
|
||||
"""Test from_azure_ai_tools converts MCP dict require_approval with 'always' key."""
|
||||
tools = [
|
||||
{
|
||||
"type": "mcp",
|
||||
"server_label": "my_mcp",
|
||||
"server_url": "http://localhost:8080",
|
||||
"require_approval": {"always": {"tool_names": ["sensitive_tool", "dangerous_tool"]}},
|
||||
}
|
||||
]
|
||||
result = from_azure_ai_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], HostedMCPTool)
|
||||
assert result[0].approval_mode == {"always_require_approval": {"sensitive_tool", "dangerous_tool"}}
|
||||
|
||||
|
||||
def test_from_azure_ai_tools_mcp_approval_mode_dict_never() -> None:
|
||||
"""Test from_azure_ai_tools converts MCP dict require_approval with 'never' key."""
|
||||
tools = [
|
||||
{
|
||||
"type": "mcp",
|
||||
"server_label": "my_mcp",
|
||||
"server_url": "http://localhost:8080",
|
||||
"require_approval": {"never": {"tool_names": ["safe_tool"]}},
|
||||
}
|
||||
]
|
||||
result = from_azure_ai_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], HostedMCPTool)
|
||||
assert result[0].approval_mode == {"never_require_approval": {"safe_tool"}}
|
||||
Reference in New Issue
Block a user