chore: Add unit tests for ToolboxSyncTool

Author: anubhav756

packages/toolbox-core/src/toolbox_core/sync_tool.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
 import asyncio
 from asyncio import AbstractEventLoop
 from inspect import Signature

packages/toolbox-core/tests/test_sync_tool.py

@@ -0,0 +1,248 @@
+# 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.
+
+
+import asyncio
+from inspect import Parameter, Signature
+from threading import Thread
+from typing import Any, Callable, Mapping, Union
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+
+from toolbox_core.sync_tool import ToolboxSyncTool
+from toolbox_core.tool import ToolboxTool
+
+
+@pytest.fixture
+def mock_async_tool() -> MagicMock:
+    """Fixture for a MagicMock simulating a ToolboxTool instance."""
+    tool = MagicMock(spec=ToolboxTool)
+    tool.__name__ = "mock_async_tool_name"
+    tool.__doc__ = "Mock async tool documentation."
+
+    # Create a simple signature for the mock tool
+    param_a = Parameter("a", Parameter.POSITIONAL_OR_KEYWORD, annotation=str)
+    param_b = Parameter(
+        "b", Parameter.POSITIONAL_OR_KEYWORD, annotation=int, default=10
+    )
+    tool.__signature__ = Signature(parameters=[param_a, param_b])
+
+    tool.__annotations__ = {"a": str, "b": int, "return": str}
+
+    # Mock the __call__ method to return a coroutine (MagicMock that can be awaited)
+    # We'll make the internal call return a simple value for testing the sync wrapper
+    async def mock_async_call(*args, **kwargs):
+        return f"async_called_with_{args}_{kwargs}"
+
+    tool.__call__ = MagicMock(side_effect=lambda *a, **k: mock_async_call(*a, **k))
+
+    # Mock methods that return a new async_tool
+    tool.add_auth_token_getters = MagicMock(return_value=MagicMock(spec=ToolboxTool))
+    tool.bind_params = MagicMock(return_value=MagicMock(spec=ToolboxTool))
+
+    return tool
+
+
+@pytest.fixture
+def event_loop() -> asyncio.AbstractEventLoop:
+    """Fixture for an event loop."""
+    # Using asyncio.get_event_loop() might be problematic if no loop is set.
+    # For this test setup, we'll mock `run_coroutine_threadsafe` directly.
+    return Mock(spec=asyncio.AbstractEventLoop)
+
+
+@pytest.fixture
+def mock_thread() -> MagicMock:
+    """Fixture for a mock Thread."""
+    return MagicMock(spec=Thread)
+
+
+@pytest.fixture
+def toolbox_sync_tool(
+    mock_async_tool: MagicMock,
+    event_loop: asyncio.AbstractEventLoop,
+    mock_thread: MagicMock,
+) -> ToolboxSyncTool:
+    """Fixture for a ToolboxSyncTool instance."""
+    return ToolboxSyncTool(mock_async_tool, event_loop, mock_thread)
+
+
+def test_toolbox_sync_tool_init_success(
+    mock_async_tool: MagicMock,
+    event_loop: asyncio.AbstractEventLoop,
+    mock_thread: MagicMock,
+):
+    """Tests successful initialization of ToolboxSyncTool."""
+    tool = ToolboxSyncTool(mock_async_tool, event_loop, mock_thread)
+    assert tool._ToolboxSyncTool__async_tool is mock_async_tool
+    assert tool._ToolboxSyncTool__loop is event_loop
+    assert tool._ToolboxSyncTool__thread is mock_thread
+    assert tool.__qualname__ == f"ToolboxSyncTool.{mock_async_tool.__name__}"
+
+
+def test_toolbox_sync_tool_init_type_error():
+    """Tests TypeError if async_tool is not a ToolboxTool instance."""
+    with pytest.raises(
+        TypeError, match="async_tool must be an instance of ToolboxTool"
+    ):
+        ToolboxSyncTool("not_a_toolbox_tool", Mock(), Mock())
+
+
+def test_toolbox_sync_tool_name_property(
+    toolbox_sync_tool: ToolboxSyncTool, mock_async_tool: MagicMock
+):
+    """Tests the __name__ property."""
+    assert toolbox_sync_tool.__name__ == mock_async_tool.__name__
+
+
+def test_toolbox_sync_tool_doc_property(
+    toolbox_sync_tool: ToolboxSyncTool, mock_async_tool: MagicMock
+):
+    """Tests the __doc__ property."""
+    assert toolbox_sync_tool.__doc__ == mock_async_tool.__doc__
+
+    # Test with __doc__ = None
+    mock_async_tool.__doc__ = None
+    sync_tool_no_doc = ToolboxSyncTool(mock_async_tool, Mock(), Mock())
+    assert sync_tool_no_doc.__doc__ is None
+
+
+def test_toolbox_sync_tool_signature_property(
+    toolbox_sync_tool: ToolboxSyncTool, mock_async_tool: MagicMock
+):
+    """Tests the __signature__ property."""
+    assert toolbox_sync_tool.__signature__ is mock_async_tool.__signature__
+
+
+def test_toolbox_sync_tool_annotations_property(
+    toolbox_sync_tool: ToolboxSyncTool, mock_async_tool: MagicMock
+):
+    """Tests the __annotations__ property."""
+    assert toolbox_sync_tool.__annotations__ is mock_async_tool.__annotations__
+
+
+@patch("asyncio.run_coroutine_threadsafe")
+def test_toolbox_sync_tool_call(
+    mock_run_coroutine_threadsafe: MagicMock,
+    toolbox_sync_tool: ToolboxSyncTool,
+    mock_async_tool: MagicMock,
+    event_loop: asyncio.AbstractEventLoop,
+):
+    """Tests the __call__ method."""
+    mock_future = MagicMock()
+    expected_result = "call_result"
+    mock_future.result.return_value = expected_result
+    mock_run_coroutine_threadsafe.return_value = mock_future
+
+    args_tuple = ("test_arg",)
+    kwargs_dict = {"kwarg1": "value1"}
+
+    # Create a mock coroutine to be returned by async_tool.__call__
+    mock_coro = MagicMock()  # Represents the coroutine object
+    mock_async_tool.return_value = mock_coro  # async_tool() returns mock_coro
+
+    result = toolbox_sync_tool(*args_tuple, **kwargs_dict)
+
+    mock_async_tool.assert_called_once_with(*args_tuple, **kwargs_dict)
+    mock_run_coroutine_threadsafe.assert_called_once_with(mock_coro, event_loop)
+    mock_future.result.assert_called_once_with()
+    assert result == expected_result
+
+
+def test_toolbox_sync_tool_add_auth_token_getters(
+    toolbox_sync_tool: ToolboxSyncTool,
+    mock_async_tool: MagicMock,
+    event_loop: asyncio.AbstractEventLoop,
+    mock_thread: MagicMock,
+):
+    """Tests the add_auth_token_getters method."""
+    auth_getters: Mapping[str, Callable[[], str]] = {"service1": lambda: "token1"}
+
+    # The mock_async_tool.add_auth_token_getters is already set up to return a new MagicMock
+    new_mock_async_tool = mock_async_tool.add_auth_token_getters.return_value
+    new_mock_async_tool.__name__ = "new_async_tool_with_auth"  # for __qualname__
+
+    new_sync_tool = toolbox_sync_tool.add_auth_token_getters(auth_getters)
+
+    mock_async_tool.add_auth_token_getters.assert_called_once_with(auth_getters)
+
+    assert isinstance(new_sync_tool, ToolboxSyncTool)
+    assert new_sync_tool is not toolbox_sync_tool
+    assert new_sync_tool._ToolboxSyncTool__async_tool is new_mock_async_tool
+    assert new_sync_tool._ToolboxSyncTool__loop is event_loop  # Should be the same loop
+    assert (
+        new_sync_tool._ToolboxSyncTool__thread is mock_thread
+    )  # Should be the same thread
+    assert (
+        new_sync_tool.__qualname__ == f"ToolboxSyncTool.{new_mock_async_tool.__name__}"
+    )
+
+
+def test_toolbox_sync_tool_bind_params(
+    toolbox_sync_tool: ToolboxSyncTool,
+    mock_async_tool: MagicMock,
+    event_loop: asyncio.AbstractEventLoop,
+    mock_thread: MagicMock,
+):
+    """Tests the bind_params method."""
+    bound_params: Mapping[str, Union[Callable[[], Any], Any]] = {
+        "param1": "value1",
+        "param2": lambda: "value2",
+    }
+
+    new_mock_async_tool = mock_async_tool.bind_params.return_value
+    new_mock_async_tool.__name__ = "new_async_tool_with_bound_params"
+
+    new_sync_tool = toolbox_sync_tool.bind_params(bound_params)
+
+    mock_async_tool.bind_params.assert_called_once_with(bound_params)
+
+    assert isinstance(new_sync_tool, ToolboxSyncTool)
+    assert new_sync_tool is not toolbox_sync_tool
+    assert new_sync_tool._ToolboxSyncTool__async_tool is new_mock_async_tool
+    assert new_sync_tool._ToolboxSyncTool__loop is event_loop
+    assert new_sync_tool._ToolboxSyncTool__thread is mock_thread
+    assert (
+        new_sync_tool.__qualname__ == f"ToolboxSyncTool.{new_mock_async_tool.__name__}"
+    )
+
+
+def test_toolbox_sync_tool_bind_param(
+    toolbox_sync_tool: ToolboxSyncTool,
+    mock_async_tool: MagicMock,
+    event_loop: asyncio.AbstractEventLoop,
+    mock_thread: MagicMock,
+):
+    """Tests the bind_param method."""
+    param_name = "my_param"
+    param_value = "my_value"
+
+    new_mock_async_tool = mock_async_tool.bind_params.return_value
+    new_mock_async_tool.__name__ = "new_async_tool_with_single_bound_param"
+
+    # Since bind_param calls self.bind_params, which in turn calls async_tool.bind_params,
+    # we check that async_tool.bind_params is called correctly.
+    new_sync_tool = toolbox_sync_tool.bind_param(param_name, param_value)
+
+    mock_async_tool.bind_params.assert_called_once_with({param_name: param_value})
+
+    assert isinstance(new_sync_tool, ToolboxSyncTool)
+    assert new_sync_tool is not toolbox_sync_tool
+    assert new_sync_tool._ToolboxSyncTool__async_tool is new_mock_async_tool
+    assert new_sync_tool._ToolboxSyncTool__loop is event_loop
+    assert new_sync_tool._ToolboxSyncTool__thread is mock_thread
+    assert (
+        new_sync_tool.__qualname__ == f"ToolboxSyncTool.{new_mock_async_tool.__name__}"
+    )