from types import SimpleNamespace import pytest from pydantic_ai import Tool as PydanticTool from pydantic_ai.messages import ToolReturnPart from oterm.app.widgets.chat import ChatContainer, _resolve_tools from oterm.types import ChatModel class _FakeServer: def __init__(self) -> None: self.filtered_args: list = [] def filtered(self, predicate): self.filtered_args.append(predicate) return f"toolset-{id(predicate)}" class TestResolveTools: def test_empty_returns_empty(self): tools, toolsets, capabilities = _resolve_tools([]) assert tools == [] assert toolsets == [] assert capabilities == [] def test_unknown_tool_logged_and_dropped(self, monkeypatch): import oterm.log monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", []) monkeypatch.setattr("oterm.app.widgets.chat.mcp_tool_meta", {}) before = len(oterm.log.log_lines) tools, toolsets, capabilities = _resolve_tools(["ghost"]) assert tools == [] assert toolsets == [] assert capabilities == [] messages = [msg for _, msg in oterm.log.log_lines[before:]] assert any("unavailable tools" in m and "ghost" in m for m in messages) def test_known_tool_returned(self, monkeypatch): def sample() -> str: return "x" tool = PydanticTool(sample, takes_ctx=False) monkeypatch.setattr( "oterm.app.widgets.chat.builtin_tools", [{"name": "sample", "description": "", "tool": tool}], ) monkeypatch.setattr("oterm.app.widgets.chat.mcp_tool_meta", {}) tools, toolsets, capabilities = _resolve_tools(["sample"]) assert tools == [tool] assert toolsets == [] assert capabilities == [] def test_capability_resolved(self, monkeypatch): from pydantic_ai.capabilities import WebSearch import oterm.log monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", []) monkeypatch.setattr("oterm.app.widgets.chat.mcp_tool_meta", {}) before = len(oterm.log.log_lines) tools, toolsets, capabilities = _resolve_tools(["web_search"]) assert tools == [] assert toolsets == [] assert len(capabilities) == 1 assert isinstance(capabilities[0], WebSearch) messages = [msg for _, msg in oterm.log.log_lines[before:]] assert not any("unavailable tools" in m for m in messages) def test_mcp_tools_become_prefixed_filtered_toolset(self, monkeypatch): from pydantic_ai.toolsets import PrefixedToolset server = _FakeServer() monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", []) monkeypatch.setattr( "oterm.app.widgets.chat.mcp_tool_meta", {"oracle": [{"name": "ask_oracle", "description": ""}]}, ) monkeypatch.setattr("oterm.app.widgets.chat.mcp_servers", {"oracle": server}) tools, toolsets, capabilities = _resolve_tools(["oracle_ask_oracle"]) assert tools == [] assert capabilities == [] assert len(toolsets) == 1 assert isinstance(toolsets[0], PrefixedToolset) assert toolsets[0].prefix == "oracle" # The filter predicate sees unprefixed names, as it wraps the raw toolset. predicate = server.filtered_args[0] assert predicate(None, SimpleNamespace(name="ask_oracle")) is True assert predicate(None, SimpleNamespace(name="other")) is False def test_same_tool_name_on_two_servers_resolves_both_independently( self, monkeypatch ): """Regression for #321: a name shared by two servers must not conflict.""" k8s, grafana = _FakeServer(), _FakeServer() monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", []) monkeypatch.setattr( "oterm.app.widgets.chat.mcp_tool_meta", { "k8s": [{"name": "query_prometheus", "description": ""}], "grafana": [{"name": "query_prometheus", "description": ""}], }, ) monkeypatch.setattr( "oterm.app.widgets.chat.mcp_servers", {"k8s": k8s, "grafana": grafana} ) _, toolsets, _ = _resolve_tools(["k8s_query_prometheus"]) assert [ts.prefix for ts in toolsets] == ["k8s"] assert grafana.filtered_args == [] _, toolsets, _ = _resolve_tools( ["k8s_query_prometheus", "grafana_query_prometheus"] ) assert sorted(ts.prefix for ts in toolsets) == ["grafana", "k8s"] def test_unqualified_mcp_tool_name_is_unavailable(self, monkeypatch): """A bare MCP tool name resolves to no server.""" import oterm.log server = _FakeServer() monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", []) monkeypatch.setattr( "oterm.app.widgets.chat.mcp_tool_meta", {"oracle": [{"name": "ask_oracle", "description": ""}]}, ) monkeypatch.setattr("oterm.app.widgets.chat.mcp_servers", {"oracle": server}) before = len(oterm.log.log_lines) _, toolsets, _ = _resolve_tools(["ask_oracle"]) assert toolsets == [] messages = [msg for _, msg in oterm.log.log_lines[before:]] assert any("unavailable tools" in m and "ask_oracle" in m for m in messages) class TestMcpToolNameCollisions: """Regression for #321, driving real toolsets through a real agent.""" @staticmethod def _two_servers(monkeypatch): from pydantic_ai.toolsets import FunctionToolset def k8s_query_prometheus() -> str: return "k8s" def grafana_query_prometheus() -> str: return "grafana" k8s = FunctionToolset(id="k8s") k8s.add_function(k8s_query_prometheus, name="query_prometheus") grafana = FunctionToolset(id="grafana") grafana.add_function(grafana_query_prometheus, name="query_prometheus") monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", []) monkeypatch.setattr( "oterm.app.widgets.chat.mcp_tool_meta", { "k8s": [{"name": "query_prometheus", "description": ""}], "grafana": [{"name": "query_prometheus", "description": ""}], }, ) monkeypatch.setattr( "oterm.app.widgets.chat.mcp_servers", {"k8s": k8s, "grafana": grafana} ) async def test_both_servers_expose_the_shared_name_without_conflict( self, monkeypatch ): from pydantic_ai.models.test import TestModel from oterm.agent import get_agent self._two_servers(monkeypatch) _, toolsets, _ = _resolve_tools( ["k8s_query_prometheus", "grafana_query_prometheus"] ) agent = get_agent(model="llama3", toolsets=toolsets) result = await agent.run("go", model=TestModel()) called = { part.tool_name for message in result.all_messages() for part in message.parts if hasattr(part, "tool_name") } assert {"k8s_query_prometheus", "grafana_query_prometheus"} <= called async def test_selecting_one_server_calls_only_that_server(self, monkeypatch): from pydantic_ai.models.test import TestModel from oterm.agent import get_agent self._two_servers(monkeypatch) _, toolsets, _ = _resolve_tools(["grafana_query_prometheus"]) agent = get_agent(model="llama3", toolsets=toolsets) result = await agent.run("go", model=TestModel()) returns = [ (part.tool_name, part.content) for message in result.all_messages() for part in message.parts if isinstance(part, ToolReturnPart) ] assert returns == [("grafana_query_prometheus", "grafana")] class TestRebuildAgent: def test_success_populates_agent(self): cm = ChatModel(model="llama3", provider="ollama") container = ChatContainer(chat_model=cm, messages=[]) assert container.agent is not None assert container._agent_error is None def test_selected_capability_reaches_agent(self): from pydantic_ai.capabilities import WebSearch cm = ChatModel(model="llama3", provider="ollama", tools=["web_search"]) container = ChatContainer(chat_model=cm, messages=[]) assert container.agent is not None capabilities = [] container.agent._root_capability.apply(capabilities.append) assert any(isinstance(c, WebSearch) for c in capabilities) def test_failure_captures_error_message(self, app_config): cm = ChatModel(model="m", provider="openai-compat/does-not-exist") container = ChatContainer(chat_model=cm, messages=[]) assert container.agent is None assert container._agent_error is not None assert "not configured" in container._agent_error class TestStreamAgentGuards: async def test_stream_agent_raises_when_agent_is_none(self, app_config): cm = ChatModel(model="m", provider="openai-compat/missing") container = ChatContainer(chat_model=cm, messages=[]) assert container.agent is None with pytest.raises(RuntimeError, match="not configured"): async for _ in container.stream_agent("hello"): pass class TestResponseTaskErrors: async def test_model_http_error_notifies(self, store, chat_model): from collections.abc import AsyncIterator from pydantic_ai import Agent from pydantic_ai.exceptions import ModelHTTPError from pydantic_ai.messages import ModelMessage from pydantic_ai.models.function import AgentInfo, FunctionModel from textual.app import App, ComposeResult chat_id = await store.save_chat(chat_model) chat_model.id = chat_id async def stream_fn( messages: list[ModelMessage], info: AgentInfo ) -> AsyncIterator[str]: raise ModelHTTPError(status_code=500, model_name="x", body="boom") yield # pragma: no cover class _Host(App): def compose(self) -> ComposeResult: yield ChatContainer(chat_model=chat_model, messages=[]) app = _Host() async with app.run_test() as pilot: container = app.query_one(ChatContainer) container.agent = Agent(FunctionModel(stream_function=stream_fn)) await container.response_task("hello") await pilot.pause() assert any( "error running your request" in n.message for n in list(app._notifications) ) stored = await store.get_messages(chat_id) assert stored == [] assert container.messages == []