wehub-resource-sync
656 lines
22 KiB
Python
656 lines
22 KiB
Python
"""Test Gemini provider functionality.
|
||
|
||
Verifies:
|
||
- Provider initialization and attributes
|
||
- wrap_tool returns Python callables with correct metadata
|
||
- wrap_tools returns list of callables
|
||
- Callables are AFC-compatible (work with google-genai's function map)
|
||
- handle_response dispatches function calls via stored executors (backward compat)
|
||
- _process_execution_result helper
|
||
"""
|
||
|
||
import inspect
|
||
from unittest.mock import MagicMock, Mock
|
||
|
||
import pytest
|
||
|
||
from composio.client.types import Tool, tool_list_response
|
||
from composio.core.models.base import allow_tracking
|
||
from composio.core.provider import AgenticProvider
|
||
|
||
try:
|
||
import composio_gemini as _composio_gemini # noqa: F401
|
||
|
||
HAS_COMPOSIO_GEMINI = True
|
||
except ImportError:
|
||
HAS_COMPOSIO_GEMINI = False
|
||
|
||
try:
|
||
from google.genai import types as genai_types
|
||
|
||
HAS_GENAI = True
|
||
except ImportError:
|
||
genai_types = None # type: ignore[assignment]
|
||
HAS_GENAI = False
|
||
|
||
pytestmark = [
|
||
pytest.mark.gemini,
|
||
pytest.mark.skipif(
|
||
not HAS_COMPOSIO_GEMINI, reason="composio_gemini package not installed"
|
||
),
|
||
]
|
||
|
||
requires_genai = pytest.mark.skipif(
|
||
not HAS_GENAI, reason="google-genai package not installed"
|
||
)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def disable_telemetry():
|
||
"""Disable telemetry for all tests to prevent thread issues."""
|
||
token = allow_tracking.set(False)
|
||
yield
|
||
allow_tracking.reset(token)
|
||
|
||
|
||
def create_mock_tool(
|
||
slug: str,
|
||
toolkit_slug: str,
|
||
version: str = "12012025_00",
|
||
input_parameters: dict | None = None,
|
||
description: str = "Test tool for provider testing",
|
||
) -> Tool:
|
||
"""Create a mock tool for testing."""
|
||
return Tool(
|
||
name=f"Test {slug}",
|
||
slug=slug,
|
||
description=description,
|
||
input_parameters=input_parameters
|
||
or {"type": "object", "properties": {}, "required": []},
|
||
output_parameters={},
|
||
available_versions=[version],
|
||
version=version,
|
||
scopes=[],
|
||
toolkit=tool_list_response.ItemToolkit(
|
||
name=toolkit_slug.title(), slug=toolkit_slug, logo=""
|
||
),
|
||
deprecated=tool_list_response.ItemDeprecated(
|
||
available_versions=[version],
|
||
displayName=f"Test {slug}",
|
||
version=version,
|
||
toolkit=tool_list_response.ItemDeprecatedToolkit(logo=""),
|
||
is_deprecated=False,
|
||
),
|
||
is_deprecated=False,
|
||
no_auth=False,
|
||
tags=[],
|
||
)
|
||
|
||
|
||
def create_mock_execute_tool():
|
||
"""Create a mock execute_tool function matching AgenticProviderExecuteFn."""
|
||
mock_fn = Mock()
|
||
mock_fn.return_value = {
|
||
"data": {"result": "success"},
|
||
"error": None,
|
||
"successful": True,
|
||
}
|
||
return mock_fn
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Initialization
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGeminiProviderInitialization:
|
||
def test_initialization(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
assert provider is not None
|
||
assert provider.name == "gemini"
|
||
assert isinstance(provider, AgenticProvider)
|
||
|
||
def test_has_empty_executors_on_init(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
assert provider._executors == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# wrap_tool – callable creation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestWrapTool:
|
||
def test_returns_callable(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool("GITHUB_STAR_REPO", "github")
|
||
result = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
assert callable(result)
|
||
assert inspect.isfunction(result)
|
||
|
||
def test_callable_has_correct_name(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool("GITHUB_STAR_REPO", "github")
|
||
result = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
assert result.__name__ == "GITHUB_STAR_REPO"
|
||
|
||
def test_callable_has_correct_doc(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"GITHUB_STAR_REPO", "github", description="Star a GitHub repository"
|
||
)
|
||
result = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
assert result.__doc__ == "Star a GitHub repository"
|
||
|
||
def test_callable_has_typed_signature(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"GITHUB_CREATE_ISSUE",
|
||
"github",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"title": {"type": "string"},
|
||
"body": {"type": "string"},
|
||
},
|
||
"required": ["title"],
|
||
},
|
||
)
|
||
result = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
|
||
sig = inspect.signature(result)
|
||
assert "title" in sig.parameters
|
||
assert "body" in sig.parameters
|
||
assert sig.parameters["title"].annotation is str
|
||
assert sig.parameters["body"].annotation is str
|
||
|
||
def test_callable_has_annotations(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"TEST_TOOL",
|
||
"test",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {"name": {"type": "string"}},
|
||
"required": ["name"],
|
||
},
|
||
)
|
||
result = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
|
||
assert "name" in result.__annotations__
|
||
assert result.__annotations__["name"] is str
|
||
assert result.__annotations__["return"] is dict
|
||
|
||
def test_stores_executor(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool("GITHUB_STAR_REPO", "github")
|
||
execute_tool = create_mock_execute_tool()
|
||
provider.wrap_tool(tool, execute_tool)
|
||
|
||
assert "GITHUB_STAR_REPO" in provider._executors
|
||
stored_execute_tool, _aliases = provider._executors["GITHUB_STAR_REPO"]
|
||
assert stored_execute_tool is execute_tool
|
||
|
||
def test_callable_executes_correctly(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"MY_TOOL",
|
||
"test",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {"query": {"type": "string"}},
|
||
"required": ["query"],
|
||
},
|
||
)
|
||
execute_tool = create_mock_execute_tool()
|
||
func = provider.wrap_tool(tool, execute_tool)
|
||
|
||
result = func(query="hello")
|
||
execute_tool.assert_called_once_with("MY_TOOL", {"query": "hello"})
|
||
# _process_execution_result extracts data when successful
|
||
assert result == {"result": "success"}
|
||
|
||
def test_callable_processes_error_result(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"MY_TOOL",
|
||
"test",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {"q": {"type": "string"}},
|
||
"required": [],
|
||
},
|
||
)
|
||
execute_tool = Mock()
|
||
execute_tool.return_value = {
|
||
"data": {},
|
||
"error": "Auth failed",
|
||
"successful": False,
|
||
}
|
||
func = provider.wrap_tool(tool, execute_tool)
|
||
result = func(q="test")
|
||
assert result["error"] == "Auth failed"
|
||
|
||
def test_callable_converts_pydantic_args_to_dicts(self):
|
||
"""AFC may pass Pydantic GeneratedModel instances as kwargs; they must be
|
||
converted to plain dicts before reaching execute_tool so the Composio
|
||
API can JSON-serialize them.
|
||
"""
|
||
from pydantic import BaseModel
|
||
|
||
from composio_gemini import GeminiProvider
|
||
|
||
class FakeQuery(BaseModel):
|
||
use_case: str = ""
|
||
known_fields: str = ""
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"COMPOSIO_SEARCH_TOOLS",
|
||
"composio",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"queries": {
|
||
"type": "array",
|
||
"items": {
|
||
"type": "object",
|
||
"title": "Query",
|
||
"properties": {
|
||
"use_case": {"type": "string"},
|
||
"known_fields": {"type": "string"},
|
||
},
|
||
},
|
||
},
|
||
},
|
||
"required": [],
|
||
},
|
||
)
|
||
execute_tool = create_mock_execute_tool()
|
||
func = provider.wrap_tool(tool, execute_tool)
|
||
|
||
# Simulate what the SDK does: pass Pydantic model instances
|
||
func(queries=[FakeQuery(use_case="summarize email")])
|
||
|
||
call_args = execute_tool.call_args[0][1]
|
||
# The queries value must be a plain list of dicts, not Pydantic models
|
||
assert isinstance(call_args["queries"], list)
|
||
assert isinstance(call_args["queries"][0], dict)
|
||
assert call_args["queries"][0]["use_case"] == "summarize email"
|
||
|
||
def test_array_param_has_parameterized_type(self):
|
||
"""Array parameters must produce List[X], not bare List.
|
||
|
||
The google-genai SDK rejects bare List because it generates
|
||
{"type": "ARRAY"} without "items".
|
||
"""
|
||
import typing
|
||
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"TOOL_WITH_ARRAY",
|
||
"test",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"tags": {"type": "array", "items": {"type": "string"}},
|
||
"name": {"type": "string"},
|
||
},
|
||
"required": ["name"],
|
||
},
|
||
)
|
||
func = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
|
||
sig = inspect.signature(func)
|
||
tags_annotation = sig.parameters["tags"].annotation
|
||
# Must be a parameterized generic like List[str], not bare list
|
||
assert typing.get_origin(tags_annotation) is list
|
||
assert typing.get_args(tags_annotation) != ()
|
||
|
||
def test_reserved_keyword_handling(self):
|
||
"""Parameters named 'for' or 'async' are substituted and reinstated."""
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"TOOL_WITH_RESERVED",
|
||
"test",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"for": {"type": "string"},
|
||
"name": {"type": "string"},
|
||
},
|
||
"required": [],
|
||
},
|
||
)
|
||
execute_tool = create_mock_execute_tool()
|
||
func = provider.wrap_tool(tool, execute_tool)
|
||
|
||
# The callable's signature should use the cleaned name
|
||
sig = inspect.signature(func)
|
||
assert "for_rs" in sig.parameters
|
||
assert "for" not in sig.parameters
|
||
|
||
# When called with the cleaned name, it reinstates the original
|
||
func(for_rs="test_value", name="hello")
|
||
call_args = execute_tool.call_args
|
||
assert call_args[0][1]["for"] == "test_value"
|
||
assert call_args[0][1]["name"] == "hello"
|
||
|
||
def test_wrap_same_tool_twice_does_not_corrupt_schema(self):
|
||
"""Wrapping the same Tool object twice must not corrupt the schema."""
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"TOOL_RESERVED",
|
||
"test",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"for": {"type": "string"},
|
||
"name": {"type": "string"},
|
||
},
|
||
"required": ["for"],
|
||
},
|
||
)
|
||
execute_tool = create_mock_execute_tool()
|
||
|
||
func1 = provider.wrap_tool(tool, execute_tool)
|
||
func2 = provider.wrap_tool(tool, execute_tool)
|
||
|
||
# Both callables must have the cleaned parameter
|
||
sig1 = inspect.signature(func1)
|
||
sig2 = inspect.signature(func2)
|
||
assert "for_rs" in sig1.parameters
|
||
assert "for_rs" in sig2.parameters
|
||
|
||
# Original tool schema must be unchanged
|
||
assert "for" in tool.input_parameters["properties"]
|
||
assert "for_rs" not in tool.input_parameters["properties"]
|
||
|
||
def test_reserved_keyword_stays_required(self):
|
||
"""A reserved keyword listed in 'required' must remain required after renaming."""
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"TOOL_REQUIRED_RESERVED",
|
||
"test",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"for": {"type": "string"},
|
||
"name": {"type": "string"},
|
||
},
|
||
"required": ["for", "name"],
|
||
},
|
||
)
|
||
func = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
|
||
sig = inspect.signature(func)
|
||
# Both params should be required (no default value)
|
||
assert sig.parameters["for_rs"].default is inspect.Parameter.empty
|
||
assert sig.parameters["name"].default is inspect.Parameter.empty
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# wrap_tools
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestWrapTools:
|
||
def test_returns_list_of_callables(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tools = [
|
||
create_mock_tool("GITHUB_STAR_REPO", "github"),
|
||
create_mock_tool("GMAIL_SEND_EMAIL", "gmail"),
|
||
]
|
||
result = provider.wrap_tools(tools, create_mock_execute_tool())
|
||
|
||
assert len(result) == 2
|
||
assert all(inspect.isfunction(f) for f in result)
|
||
|
||
def test_stores_all_executors(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tools = [
|
||
create_mock_tool("GITHUB_STAR_REPO", "github"),
|
||
create_mock_tool("GMAIL_SEND_EMAIL", "gmail"),
|
||
]
|
||
provider.wrap_tools(tools, create_mock_execute_tool())
|
||
|
||
assert "GITHUB_STAR_REPO" in provider._executors
|
||
assert "GMAIL_SEND_EMAIL" in provider._executors
|
||
|
||
def test_empty_list(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
result = provider.wrap_tools([], create_mock_execute_tool())
|
||
assert result == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# AFC compatibility (requires google-genai)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@requires_genai
|
||
class TestAFCCompatibility:
|
||
"""Verify callables work with google-genai's AFC pipeline."""
|
||
|
||
def _wrap_one_tool(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool(
|
||
"GITHUB_STAR_REPO",
|
||
"github",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"owner": {"type": "string"},
|
||
"repo": {"type": "string"},
|
||
},
|
||
"required": ["owner", "repo"],
|
||
},
|
||
description="Star a GitHub repository",
|
||
)
|
||
func = provider.wrap_tool(tool, create_mock_execute_tool())
|
||
return func
|
||
|
||
def test_callable_is_function(self):
|
||
"""inspect.isfunction must be True for t_tool() to call from_callable."""
|
||
func = self._wrap_one_tool()
|
||
assert inspect.isfunction(func)
|
||
|
||
def test_callable_not_afc_incompatible(self):
|
||
"""Callables should not be flagged as AFC-incompatible."""
|
||
func = self._wrap_one_tool()
|
||
# The SDK checks: isinstance(tool, types.Tool) and tool.function_declarations
|
||
# A callable is not a types.Tool, so it should not be flagged.
|
||
assert not isinstance(func, genai_types.Tool)
|
||
|
||
def test_callable_enters_function_map(self):
|
||
"""The callable should appear in the function map built by the SDK."""
|
||
func = self._wrap_one_tool()
|
||
# Simulate what get_function_map does: callable(tool) -> function_map[tool.__name__]
|
||
assert callable(func)
|
||
function_map = {}
|
||
if callable(func):
|
||
function_map[func.__name__] = func
|
||
assert "GITHUB_STAR_REPO" in function_map
|
||
|
||
def test_callables_in_generate_content_config(self):
|
||
"""Wrapped callables can be passed to GenerateContentConfig without error."""
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tools = [
|
||
create_mock_tool(
|
||
"GITHUB_STAR_REPO",
|
||
"github",
|
||
input_parameters={
|
||
"type": "object",
|
||
"properties": {
|
||
"owner": {"type": "string"},
|
||
"repo": {"type": "string"},
|
||
},
|
||
"required": ["owner", "repo"],
|
||
},
|
||
),
|
||
]
|
||
wrapped = provider.wrap_tools(tools, create_mock_execute_tool())
|
||
|
||
# This should not raise
|
||
config = genai_types.GenerateContentConfig(tools=wrapped)
|
||
assert config.tools is not None
|
||
assert len(config.tools) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# handle_response (backward compat)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@requires_genai
|
||
class TestHandleResponse:
|
||
def _create_mock_response(self, function_calls: list[tuple[str, dict]]):
|
||
parts = []
|
||
for name, args in function_calls:
|
||
part = MagicMock()
|
||
part.function_call = MagicMock()
|
||
part.function_call.name = name
|
||
part.function_call.args = args
|
||
parts.append(part)
|
||
|
||
response = MagicMock()
|
||
response.candidates = [MagicMock()]
|
||
response.candidates[0].content = MagicMock()
|
||
response.candidates[0].content.parts = parts
|
||
return response
|
||
|
||
def test_executes_function_call(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
tool = create_mock_tool("GITHUB_STAR_REPO", "github")
|
||
execute_tool = create_mock_execute_tool()
|
||
provider.wrap_tools([tool], execute_tool)
|
||
|
||
response = self._create_mock_response(
|
||
[("GITHUB_STAR_REPO", {"repo": "composio/composio"})]
|
||
)
|
||
function_responses, executed = provider.handle_response(response)
|
||
|
||
assert executed is True
|
||
assert len(function_responses) == 1
|
||
execute_tool.assert_called_once_with(
|
||
slug="GITHUB_STAR_REPO",
|
||
arguments={"repo": "composio/composio"},
|
||
)
|
||
|
||
def test_no_function_calls(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
response = MagicMock()
|
||
response.candidates = [MagicMock()]
|
||
text_part = MagicMock()
|
||
text_part.function_call = None
|
||
response.candidates[0].content = MagicMock()
|
||
response.candidates[0].content.parts = [text_part]
|
||
|
||
function_responses, executed = provider.handle_response(response)
|
||
assert executed is False
|
||
assert function_responses == []
|
||
|
||
def test_unknown_function_name(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
response = self._create_mock_response([("UNKNOWN_TOOL", {"param": "value"})])
|
||
function_responses, executed = provider.handle_response(response)
|
||
assert executed is False
|
||
assert function_responses == []
|
||
|
||
def test_no_candidates(self):
|
||
from composio_gemini import GeminiProvider
|
||
|
||
provider = GeminiProvider()
|
||
response = MagicMock()
|
||
response.candidates = []
|
||
function_responses, executed = provider.handle_response(response)
|
||
assert executed is False
|
||
assert function_responses == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _process_execution_result
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProcessExecutionResult:
|
||
def test_non_dict_result(self):
|
||
from composio_gemini.provider import _process_execution_result
|
||
|
||
assert _process_execution_result("hello") == {"result": "hello"}
|
||
assert _process_execution_result(42) == {"result": 42}
|
||
|
||
def test_successful_with_dict_data(self):
|
||
from composio_gemini.provider import _process_execution_result
|
||
|
||
result = _process_execution_result(
|
||
{"data": {"key": "value"}, "error": None, "successful": True}
|
||
)
|
||
assert result == {"key": "value"}
|
||
|
||
def test_successful_with_non_dict_data(self):
|
||
from composio_gemini.provider import _process_execution_result
|
||
|
||
result = _process_execution_result(
|
||
{"data": "plain text", "error": None, "successful": True}
|
||
)
|
||
assert result == {"result": "plain text"}
|
||
|
||
def test_failed_result(self):
|
||
from composio_gemini.provider import _process_execution_result
|
||
|
||
result = _process_execution_result(
|
||
{"data": {}, "error": "Something went wrong", "successful": False}
|
||
)
|
||
assert result["error"] == "Something went wrong"
|
||
|
||
def test_passthrough_dict(self):
|
||
from composio_gemini.provider import _process_execution_result
|
||
|
||
result = _process_execution_result({"custom": "response"})
|
||
assert result == {"custom": "response"}
|