mirror of
https://github.com/ggozad/oterm.git
synced 2026-10-10 01:03:21 +02:00
d077f671f1Tool identity was the bare tool name, so a name exported by two MCP
servers resolved to both toolsets and pydantic-ai rejected the agent.
MCP tools are now identified as `{server}_{tool}` and reach the model
through PrefixedToolset, making per-server selection independent.
Since the server name now reaches the provider, reject names outside
[a-zA-Z0-9_-] when the config loads and skip tools whose qualified name
is over 64 characters.
Tools load before the store so the 0.22.0 upgrade can qualify saved
selections from the connected servers' tool lists.
Point the ty pre-commit hook at the pinned version, matching CI and the
ruff hooks.
Closes #321
206 lines
7.9 KiB
Python
206 lines
7.9 KiB
Python
import pytest
|
|
from pydantic_ai import Tool as PydanticTool
|
|
from textual.app import App, ComposeResult
|
|
from textual.widgets import Checkbox
|
|
|
|
from oterm.app.widgets.tool_select import ToolSelector
|
|
|
|
|
|
def _tool_def(name: str):
|
|
def fn() -> str:
|
|
return "x"
|
|
|
|
fn.__name__ = name
|
|
return {
|
|
"name": name,
|
|
"description": f"{name} tool",
|
|
"tool": PydanticTool(fn, takes_ctx=False),
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def populated_tools(monkeypatch):
|
|
"""Inject fake builtin tools, capabilities and MCP metadata so ToolSelector has content."""
|
|
import oterm.app.widgets.tool_select as sel_mod
|
|
import oterm.tools as tools_mod
|
|
import oterm.tools.capabilities as capabilities_mod
|
|
import oterm.tools.mcp.setup as mcp_setup_mod
|
|
|
|
builtin = [_tool_def("date_time"), _tool_def("shell")]
|
|
capability_defs = [
|
|
{"name": "web_search", "description": "web search", "factory": lambda: None},
|
|
]
|
|
mcp_meta = {
|
|
"mcp_server": [{"name": "oracle", "description": "oracle tool"}],
|
|
"other_server": [{"name": "oracle", "description": "another oracle tool"}],
|
|
}
|
|
|
|
monkeypatch.setattr(tools_mod, "builtin_tools", builtin)
|
|
monkeypatch.setattr(sel_mod, "builtin_tools", builtin)
|
|
monkeypatch.setattr(sel_mod, "capability_defs", capability_defs)
|
|
monkeypatch.setattr(capabilities_mod, "capability_defs", capability_defs)
|
|
monkeypatch.setattr(sel_mod, "mcp_tool_meta", mcp_meta)
|
|
monkeypatch.setattr(mcp_setup_mod, "mcp_tool_meta", mcp_meta)
|
|
return builtin, mcp_meta
|
|
|
|
|
|
class _Host(App):
|
|
def __init__(self, selected: list[str] | None = None):
|
|
super().__init__()
|
|
self._selected = selected or []
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield ToolSelector(id="selector", selected=self._selected)
|
|
|
|
|
|
class TestToolSelector:
|
|
async def test_renders_checkbox_per_group_and_tool(self, populated_tools):
|
|
app = _Host()
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
names = {c.name for c in selector.query(Checkbox)}
|
|
assert {
|
|
"builtin",
|
|
"capabilities",
|
|
"mcp_server",
|
|
"other_server",
|
|
"date_time",
|
|
"shell",
|
|
"web_search",
|
|
"mcp_server_oracle",
|
|
"other_server_oracle",
|
|
}.issubset(names)
|
|
|
|
async def test_initial_selection_reflected_in_checkboxes(self, populated_tools):
|
|
app = _Host(selected=["date_time"])
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
date_time_cb = selector.query_one("#builtin-date_time", Checkbox)
|
|
shell_cb = selector.query_one("#builtin-shell", Checkbox)
|
|
assert date_time_cb.value is True
|
|
assert shell_cb.value is False
|
|
|
|
async def test_unavailable_selected_tools_filtered(self, populated_tools):
|
|
app = _Host(selected=["date_time", "ghost"])
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
assert "ghost" not in selector.selected
|
|
assert "date_time" in selector.selected
|
|
|
|
async def test_toggling_tool_checkbox_updates_selected(self, populated_tools):
|
|
app = _Host()
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
|
|
shell_cb = selector.query_one("#builtin-shell", Checkbox)
|
|
shell_cb.value = True
|
|
await pilot.pause()
|
|
assert "shell" in selector.selected
|
|
|
|
shell_cb.value = False
|
|
await pilot.pause()
|
|
assert "shell" not in selector.selected
|
|
|
|
async def test_group_checkbox_toggles_all_tools(self, populated_tools):
|
|
app = _Host()
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
|
|
group_cb = next(c for c in selector.query(Checkbox) if c.name == "builtin")
|
|
group_cb.value = True
|
|
await pilot.pause()
|
|
assert selector.query_one("#builtin-date_time", Checkbox).value is True
|
|
assert selector.query_one("#builtin-shell", Checkbox).value is True
|
|
|
|
async def test_capability_toggle(self, populated_tools):
|
|
app = _Host()
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
|
|
web_search_cb = selector.query_one("#capabilities-web_search", Checkbox)
|
|
web_search_cb.value = True
|
|
await pilot.pause()
|
|
assert "web_search" in selector.selected
|
|
|
|
async def test_selected_capability_survives_known_names_filter(
|
|
self, populated_tools
|
|
):
|
|
app = _Host(selected=["web_search"])
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
assert "web_search" in selector.selected
|
|
|
|
async def test_mcp_tool_toggle(self, populated_tools):
|
|
app = _Host()
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
|
|
oracle_cb = selector.query_one("#mcp_server-mcp_server_oracle", Checkbox)
|
|
oracle_cb.value = True
|
|
await pilot.pause()
|
|
assert "mcp_server_oracle" in selector.selected
|
|
|
|
async def test_same_tool_name_on_two_servers_selected_independently(
|
|
self, populated_tools
|
|
):
|
|
"""Regression for #321: selecting one server's tool leaves the other alone."""
|
|
app = _Host(selected=["mcp_server_oracle"])
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
assert (
|
|
selector.query_one("#mcp_server-mcp_server_oracle", Checkbox).value
|
|
is True
|
|
)
|
|
assert (
|
|
selector.query_one("#other_server-other_server_oracle", Checkbox).value
|
|
is False
|
|
)
|
|
|
|
async def test_group_select_all_reflects_only_its_own_server(self, populated_tools):
|
|
"""A server's select-all box ignores identically named tools on other servers."""
|
|
app = _Host(selected=["mcp_server_oracle"])
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
groups = {
|
|
c.name: c
|
|
for c in selector.query(Checkbox)
|
|
if c.has_class("tool-group-select-all")
|
|
}
|
|
assert groups["mcp_server"].value is True
|
|
assert groups["other_server"].value is False
|
|
|
|
async def test_unknown_checkbox_name_is_ignored(self, populated_tools):
|
|
"""Defensive: a Checkbox.Changed with a name we don't recognize is a no-op."""
|
|
app = _Host()
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
initial = list(selector.selected)
|
|
fake = Checkbox(label="ghost", name="ghost")
|
|
selector.on_checkbox_toggled(Checkbox.Changed(fake, True))
|
|
await pilot.pause()
|
|
assert selector.selected == initial
|
|
|
|
async def test_toggle_already_selected_tool_is_idempotent(self, populated_tools):
|
|
"""Re-checking an already-selected tool short-circuits without duplicating."""
|
|
app = _Host(selected=["shell"])
|
|
async with app.run_test() as pilot:
|
|
selector = app.query_one(ToolSelector)
|
|
await pilot.pause()
|
|
assert selector.selected == ["shell"]
|
|
|
|
shell_cb = selector.query_one("#builtin-shell", Checkbox)
|
|
# Re-fire the Changed event with value=True even though it's already True.
|
|
selector.on_checkbox_toggled(Checkbox.Changed(shell_cb, True))
|
|
await pilot.pause()
|
|
assert selector.selected == ["shell"]
|