feat: refactor APIClient and ToolRegistry to classes
- api_client.py: APIClient class with ProviderConfig types, chat_stream() returns str instead of printing, module-level wrappers for backward compat - tools.py: ToolRegistry class with instance methods, pure Python list_files (os.walk) and search_in_files (re.search) replacing find/grep for Windows compat, English tool descriptions and error messages, module-level wrappers for backward compat - 74 new tests (21 api_client + 53 tools) Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
359
tests/test_api_client.py
Normal file
359
tests/test_api_client.py
Normal file
@@ -0,0 +1,359 @@
|
||||
"""Tests for meeting_room.api_client — APIClient class and module-level wrappers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from meeting_room.api_client import APIClient, chat, chat_stream, init_providers
|
||||
from meeting_room.models import ProviderConfig
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures & helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
PROVIDERS = {
|
||||
"testprov": ProviderConfig(
|
||||
base_url="https://api.test.example.com/v1",
|
||||
api_key="sk-test-key-123",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _make_chat_response(
|
||||
content: str = "Hello!",
|
||||
tool_calls: list[dict] | None = None,
|
||||
) -> dict:
|
||||
"""Build a minimal OpenAI-style chat completion response."""
|
||||
message: dict = {"content": content}
|
||||
if tool_calls:
|
||||
message["tool_calls"] = tool_calls
|
||||
return {"choices": [{"message": message}]}
|
||||
|
||||
|
||||
def _make_stream_lines(chunks: list[str]) -> list[str]:
|
||||
"""Build SSE lines that look like what OpenAI streaming returns.
|
||||
|
||||
Each chunk becomes a ``data: {json}`` line. A ``data: [DONE]``
|
||||
sentinel is appended automatically.
|
||||
"""
|
||||
lines: list[str] = []
|
||||
for text in chunks:
|
||||
delta = {"choices": [{"delta": {"content": text}}]}
|
||||
lines.append(f"data: {json.dumps(delta)}")
|
||||
lines.append("data: [DONE]")
|
||||
return lines
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# APIClient — construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAPIClientInit:
|
||||
def test_stores_providers(self) -> None:
|
||||
client = APIClient(PROVIDERS)
|
||||
assert "testprov" in client.providers
|
||||
assert client.providers["testprov"].base_url == "https://api.test.example.com/v1"
|
||||
|
||||
def test_empty_providers_ok(self) -> None:
|
||||
client = APIClient({})
|
||||
assert client.providers == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# APIClient.chat
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAPIClientChat:
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_basic_text_response(self, mock_client_cls: MagicMock) -> None:
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = _make_chat_response("Hi there")
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_client_instance = mock_client_cls.return_value
|
||||
mock_client_instance.post.return_value = mock_resp
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
result = client.chat("testprov", "gpt-4o-mini", [{"role": "user", "content": "Hi"}])
|
||||
|
||||
assert result == {"content": "Hi there", "tool_calls": None}
|
||||
mock_client_instance.post.assert_called_once()
|
||||
call_args = mock_client_instance.post.call_args
|
||||
assert call_args[0][0] == "/chat/completions"
|
||||
payload = call_args[1]["json"]
|
||||
assert payload["model"] == "gpt-4o-mini"
|
||||
assert payload["messages"] == [{"role": "user", "content": "Hi"}]
|
||||
assert payload["temperature"] == 0.7
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_custom_temperature(self, mock_client_cls: MagicMock) -> None:
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = _make_chat_response("cold")
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_client_cls.return_value.post.return_value = mock_resp
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
client.chat("testprov", "m", [], temperature=0.2)
|
||||
|
||||
payload = mock_client_cls.return_value.post.call_args[1]["json"]
|
||||
assert payload["temperature"] == 0.2
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_tool_calls_in_response(self, mock_client_cls: MagicMock) -> None:
|
||||
tool_calls = [
|
||||
{
|
||||
"id": "call_abc",
|
||||
"function": {"name": "read_file", "arguments": '{"path": "/tmp/x"}'},
|
||||
"type": "function",
|
||||
}
|
||||
]
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = _make_chat_response("", tool_calls=tool_calls)
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_client_cls.return_value.post.return_value = mock_resp
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
result = client.chat("testprov", "m", [])
|
||||
|
||||
assert result["tool_calls"] == [
|
||||
{"id": "call_abc", "name": "read_file", "arguments": '{"path": "/tmp/x"}'}
|
||||
]
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_tools_sent_in_payload(self, mock_client_cls: MagicMock) -> None:
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = _make_chat_response("ok")
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_client_cls.return_value.post.return_value = mock_resp
|
||||
|
||||
tools = [{"type": "function", "function": {"name": "search"}}]
|
||||
client = APIClient(PROVIDERS)
|
||||
client.chat("testprov", "m", [], tools=tools)
|
||||
|
||||
payload = mock_client_cls.return_value.post.call_args[1]["json"]
|
||||
assert "tools" in payload
|
||||
assert payload["tool_choice"] == "auto"
|
||||
|
||||
def test_unknown_provider_raises(self) -> None:
|
||||
client = APIClient(PROVIDERS)
|
||||
with pytest.raises(ValueError, match="Unknown provider: nonexistent"):
|
||||
client.chat("nonexistent", "m", [])
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_empty_content_yields_empty_string(self, mock_client_cls: MagicMock) -> None:
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"choices": [{"message": {"content": None}}]}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_client_cls.return_value.post.return_value = mock_resp
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
result = client.chat("testprov", "m", [])
|
||||
assert result["content"] == ""
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_http_error_propagates(self, mock_client_cls: MagicMock) -> None:
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
|
||||
"Bad Request",
|
||||
request=MagicMock(),
|
||||
response=httpx.Response(400),
|
||||
)
|
||||
mock_client_cls.return_value.post.return_value = mock_resp
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
with pytest.raises(httpx.HTTPStatusError):
|
||||
client.chat("testprov", "m", [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# APIClient.chat_stream
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAPIClientChatStream:
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_accumulates_text(self, mock_client_cls: MagicMock) -> None:
|
||||
"""chat_stream must return the concatenated text, NOT print."""
|
||||
lines = _make_stream_lines(["Hello", " world", "!"])
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_lines.return_value = lines
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
mock_client_instance = mock_client_cls.return_value
|
||||
mock_client_instance.stream.return_value = mock_response
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
result = client.chat_stream("testprov", "gpt-4o-mini", [{"role": "user", "content": "Hi"}])
|
||||
|
||||
assert result == "Hello world!"
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_stream_payload_includes_stream_flag(self, mock_client_cls: MagicMock) -> None:
|
||||
lines = _make_stream_lines(["x"])
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_lines.return_value = lines
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
mock_client_cls.return_value.stream.return_value = mock_response
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
client.chat_stream("testprov", "m", [])
|
||||
|
||||
call_args = mock_client_cls.return_value.stream.call_args
|
||||
payload = call_args[1]["json"]
|
||||
assert payload["stream"] is True
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_empty_stream_returns_empty_string(self, mock_client_cls: MagicMock) -> None:
|
||||
lines = _make_stream_lines([])
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_lines.return_value = lines
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
mock_client_cls.return_value.stream.return_value = mock_response
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
result = client.chat_stream("testprov", "m", [])
|
||||
assert result == ""
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_malformed_lines_skipped(self, mock_client_cls: MagicMock) -> None:
|
||||
"""Lines that are not valid SSE data should be silently skipped."""
|
||||
raw_lines = [
|
||||
"", # empty line — skipped
|
||||
"not SSE", # no "data: " prefix — skipped
|
||||
"data: {bad json", # malformed JSON — skipped
|
||||
'data: {"choices":[{"delta":{"content":"ok"}}]}', # valid
|
||||
"data: [DONE]",
|
||||
]
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_lines.return_value = raw_lines
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
mock_client_cls.return_value.stream.return_value = mock_response
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
result = client.chat_stream("testprov", "m", [])
|
||||
assert result == "ok"
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_does_not_print(self, mock_client_cls: MagicMock, capsys: pytest.CaptureFixture) -> None:
|
||||
"""chat_stream must NOT write to stdout."""
|
||||
lines = _make_stream_lines(["Hello", " world"])
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_lines.return_value = lines
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
mock_client_cls.return_value.stream.return_value = mock_response
|
||||
|
||||
client = APIClient(PROVIDERS)
|
||||
result = client.chat_stream("testprov", "m", [])
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == ""
|
||||
assert result == "Hello world"
|
||||
|
||||
def test_unknown_provider_raises(self) -> None:
|
||||
client = APIClient(PROVIDERS)
|
||||
with pytest.raises(ValueError, match="Unknown provider: nope"):
|
||||
client.chat_stream("nope", "m", [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module-level wrappers — backward compatibility
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestModuleLevelWrappers:
|
||||
def test_init_providers_creates_default_client(self) -> None:
|
||||
import meeting_room.api_client as mod
|
||||
|
||||
config = {
|
||||
"providers": {
|
||||
"myprov": {"base_url": "https://api.example.com/v1", "api_key": "sk-abc"},
|
||||
}
|
||||
}
|
||||
init_providers(config)
|
||||
assert mod._default_client is not None
|
||||
assert "myprov" in mod._default_client.providers
|
||||
assert mod._default_client.providers["myprov"].api_key == "sk-abc"
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_module_chat_delegates(self, mock_client_cls: MagicMock) -> None:
|
||||
import meeting_room.api_client as mod
|
||||
|
||||
init_providers({"providers": {"prov": {"base_url": "https://x/v1", "api_key": "k"}}})
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = _make_chat_response("delegated")
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_client_cls.return_value.post.return_value = mock_resp
|
||||
|
||||
result = mod.chat("prov", "m", [])
|
||||
assert result["content"] == "delegated"
|
||||
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_module_chat_stream_delegates(self, mock_client_cls: MagicMock) -> None:
|
||||
import meeting_room.api_client as mod
|
||||
|
||||
init_providers({"providers": {"prov": {"base_url": "https://x/v1", "api_key": "k"}}})
|
||||
lines = _make_stream_lines(["abc"])
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_lines.return_value = lines
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
mock_client_cls.return_value.stream.return_value = mock_response
|
||||
|
||||
result = mod.chat_stream("prov", "m", [])
|
||||
assert result == "abc"
|
||||
|
||||
def test_module_chat_without_init_raises(self) -> None:
|
||||
import meeting_room.api_client as mod
|
||||
|
||||
mod._default_client = None # ensure clean state
|
||||
with pytest.raises(RuntimeError, match="init_providers"):
|
||||
mod.chat("prov", "m", [])
|
||||
|
||||
def test_module_chat_stream_without_init_raises(self) -> None:
|
||||
import meeting_room.api_client as mod
|
||||
|
||||
mod._default_client = None
|
||||
with pytest.raises(RuntimeError, match="init_providers"):
|
||||
mod.chat_stream("prov", "m", [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# APIClient._get_client — httpx.Client construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetClient:
|
||||
@patch("meeting_room.api_client.httpx.Client")
|
||||
def test_client_created_with_correct_params(self, mock_client_cls: MagicMock) -> None:
|
||||
client = APIClient(PROVIDERS)
|
||||
# Trigger _get_client by calling chat with a mocked response
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = _make_chat_response("ok")
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_client_cls.return_value.post.return_value = mock_resp
|
||||
|
||||
client.chat("testprov", "m", [])
|
||||
|
||||
mock_client_cls.assert_called_once_with(
|
||||
base_url="https://api.test.example.com/v1",
|
||||
headers={"Authorization": "Bearer sk-test-key-123"},
|
||||
timeout=120.0,
|
||||
)
|
||||
389
tests/test_tools.py
Normal file
389
tests/test_tools.py
Normal file
@@ -0,0 +1,389 @@
|
||||
"""Tests for meeting_room.tools — ToolRegistry class and module-level wrappers."""
|
||||
|
||||
import os
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
from meeting_room.tools import (
|
||||
TOOL_SCHEMAS,
|
||||
TOOL_SETS,
|
||||
ToolRegistry,
|
||||
execute_tool,
|
||||
get_tool_schemas,
|
||||
set_workdir,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_workdir(tmp_path):
|
||||
"""Create a temporary directory with some files for testing."""
|
||||
# Create directories
|
||||
(tmp_path / "src").mkdir()
|
||||
(tmp_path / "src" / "sub").mkdir()
|
||||
(tmp_path / "docs").mkdir()
|
||||
(tmp_path / ".git" / "objects").mkdir(parents=True)
|
||||
(tmp_path / "node_modules" / "pkg").mkdir(parents=True)
|
||||
(tmp_path / "__pycache__").mkdir()
|
||||
|
||||
# Create files
|
||||
(tmp_path / "README.md").write_text("# Hello\nWorld", encoding="utf-8")
|
||||
(tmp_path / "src" / "main.py").write_text("def main():\n pass\n", encoding="utf-8")
|
||||
(tmp_path / "src" / "utils.py").write_text("def add(a, b):\n return a + b\n", encoding="utf-8")
|
||||
(tmp_path / "src" / "sub" / "deep.py").write_text("# deep file\nx = 42\n", encoding="utf-8")
|
||||
(tmp_path / "docs" / "notes.txt").write_text("Meeting notes\n", encoding="utf-8")
|
||||
(tmp_path / "config.yaml").write_text("key: value\n", encoding="utf-8")
|
||||
# Files in skip dirs
|
||||
(tmp_path / ".git" / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
|
||||
(tmp_path / "node_modules" / "pkg" / "index.js").write_text("module.exports = {};\n", encoding="utf-8")
|
||||
(tmp_path / "__pycache__" / "cache.pyc").write_text("bytecode", encoding="utf-8")
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def reg(tmp_workdir):
|
||||
"""ToolRegistry pointed at the temp workdir."""
|
||||
return ToolRegistry(workdir=str(tmp_workdir))
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# TOOL_SCHEMAS — structure checks
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolSchemas:
|
||||
def test_schema_names_match_expected(self):
|
||||
names = {s["function"]["name"] for s in TOOL_SCHEMAS}
|
||||
expected = {"read_file", "list_files", "search_in_files", "web_search", "web_fetch", "write_file", "run_command"}
|
||||
assert names == expected
|
||||
|
||||
def test_each_schema_has_required_fields(self):
|
||||
for s in TOOL_SCHEMAS:
|
||||
func = s["function"]
|
||||
assert "name" in func
|
||||
assert "description" in func
|
||||
assert "parameters" in func
|
||||
params = func["parameters"]
|
||||
assert params["type"] == "object"
|
||||
assert "properties" in params
|
||||
|
||||
def test_descriptions_are_in_english(self):
|
||||
"""All descriptions must be in English (no Cyrillic)."""
|
||||
for s in TOOL_SCHEMAS:
|
||||
desc = s["function"]["description"]
|
||||
# Check no Cyrillic characters
|
||||
assert not re.search(r"[а-яА-ЯёЁ]", desc), f"Cyrillic in description: {desc}"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# TOOL_SETS — grouping checks
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolSets:
|
||||
def test_all_sets_exist(self):
|
||||
for name in ("all", "full", "files", "readonly", "web", "none"):
|
||||
assert name in TOOL_SETS
|
||||
|
||||
def test_all_equals_full(self):
|
||||
assert TOOL_SETS["all"] == TOOL_SETS["full"]
|
||||
|
||||
def test_none_is_empty(self):
|
||||
assert TOOL_SETS["none"] == []
|
||||
|
||||
def test_readonly_subset_of_files(self):
|
||||
assert set(TOOL_SETS["readonly"]).issubset(set(TOOL_SETS["files"]))
|
||||
|
||||
def test_web_only_web_tools(self):
|
||||
assert set(TOOL_SETS["web"]) == {"web_search", "web_fetch"}
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# ToolRegistry — construction and path resolution
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolRegistryInit:
|
||||
def test_default_workdir(self):
|
||||
r = ToolRegistry()
|
||||
assert r._workdir == os.path.abspath(".")
|
||||
|
||||
def test_custom_workdir(self, tmp_path):
|
||||
r = ToolRegistry(workdir=str(tmp_path))
|
||||
assert r._workdir == str(tmp_path)
|
||||
|
||||
def test_resolve_absolute_path(self, reg):
|
||||
abs_path = os.path.abspath(os.path.join(os.sep, "tmp", "some", "file.txt"))
|
||||
assert reg._resolve_path(abs_path) == abs_path
|
||||
|
||||
def test_resolve_relative_path(self, reg, tmp_workdir):
|
||||
result = reg._resolve_path("src/main.py")
|
||||
assert result == os.path.normpath(os.path.join(str(tmp_workdir), "src/main.py"))
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# get_tool_schemas
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestGetToolSchemas:
|
||||
def test_all_set(self, reg):
|
||||
schemas = reg.get_tool_schemas("all")
|
||||
assert len(schemas) == len(TOOL_SCHEMAS)
|
||||
|
||||
def test_none_set(self, reg):
|
||||
schemas = reg.get_tool_schemas("none")
|
||||
assert schemas == []
|
||||
|
||||
def test_readonly_set(self, reg):
|
||||
schemas = reg.get_tool_schemas("readonly")
|
||||
names = {s["function"]["name"] for s in schemas}
|
||||
assert names == {"read_file", "list_files", "search_in_files"}
|
||||
|
||||
def test_explicit_list(self, reg):
|
||||
schemas = reg.get_tool_schemas(["read_file", "web_search"])
|
||||
names = {s["function"]["name"] for s in schemas}
|
||||
assert names == {"read_file", "web_search"}
|
||||
|
||||
def test_unknown_set_falls_back_to_all(self, reg):
|
||||
schemas = reg.get_tool_schemas("nonexistent_set")
|
||||
assert len(schemas) == len(TOOL_SCHEMAS)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# execute_tool dispatch
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestExecuteTool:
|
||||
def test_unknown_tool(self, reg):
|
||||
result = reg.execute_tool("nonexistent_tool", {})
|
||||
assert "Unknown tool" in result
|
||||
|
||||
def test_execute_read_file(self, reg):
|
||||
result = reg.execute_tool("read_file", {"path": "README.md"})
|
||||
assert "Hello" in result
|
||||
|
||||
def test_execute_write_then_read(self, reg, tmp_workdir):
|
||||
reg.execute_tool("write_file", {"path": "new_file.txt", "content": "test content"})
|
||||
result = reg.execute_tool("read_file", {"path": "new_file.txt"})
|
||||
assert "test content" in result
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# tool_read_file
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolReadFile:
|
||||
def test_read_existing_file(self, reg):
|
||||
result = reg.tool_read_file("README.md")
|
||||
assert "Hello" in result
|
||||
|
||||
def test_read_with_offset(self, reg):
|
||||
result = reg.tool_read_file("README.md", offset=2)
|
||||
assert "World" in result
|
||||
# Line numbers start at offset
|
||||
assert "2|" in result
|
||||
|
||||
def test_read_with_limit(self, reg):
|
||||
result = reg.tool_read_file("src/main.py", limit=1)
|
||||
# Should have only 1 content line
|
||||
lines = [l for l in result.split("\n") if "|" in l]
|
||||
assert len(lines) == 1
|
||||
|
||||
def test_read_nonexistent_file(self, reg):
|
||||
result = reg.tool_read_file("nonexistent.txt")
|
||||
assert "not found" in result
|
||||
|
||||
def test_read_shows_total_lines(self, reg):
|
||||
result = reg.tool_read_file("README.md")
|
||||
assert "lines" in result
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# tool_list_files — pure Python, Windows-compatible
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolListFiles:
|
||||
def test_list_root(self, reg, tmp_workdir):
|
||||
result = reg.tool_list_files()
|
||||
# Must include README.md and config.yaml at root
|
||||
assert "README.md" in result
|
||||
assert "config.yaml" in result
|
||||
|
||||
def test_list_skips_git(self, reg):
|
||||
result = reg.tool_list_files()
|
||||
# .git contents should NOT appear
|
||||
assert ".git" not in result or "HEAD" not in result
|
||||
|
||||
def test_list_skips_node_modules(self, reg):
|
||||
result = reg.tool_list_files()
|
||||
assert "index.js" not in result
|
||||
|
||||
def test_list_skips_pycache(self, reg):
|
||||
result = reg.tool_list_files()
|
||||
assert "cache.pyc" not in result
|
||||
|
||||
def test_list_with_pattern(self, reg):
|
||||
result = reg.tool_list_files(pattern="*.py")
|
||||
lines = result.strip().split("\n") if result.strip() else []
|
||||
# Only .py files should appear
|
||||
for line in lines:
|
||||
if line.strip():
|
||||
assert line.endswith(".py") or os.path.isdir(line), f"Non-Python file in filtered results: {line}"
|
||||
|
||||
def test_list_with_path(self, reg, tmp_workdir):
|
||||
result = reg.tool_list_files(path="src")
|
||||
assert "main.py" in result
|
||||
assert "utils.py" in result
|
||||
|
||||
def test_list_depth_limit(self, reg):
|
||||
result = reg.tool_list_files(max_depth=1)
|
||||
# With depth 1, should include root dir and its immediate files
|
||||
# but NOT deeply nested files like src/sub/deep.py
|
||||
# (it may list the src directory itself)
|
||||
assert "config.yaml" in result
|
||||
|
||||
def test_list_nonexistent_dir(self, reg):
|
||||
result = reg.tool_list_files(path="nonexistent_dir")
|
||||
assert "not found" in result
|
||||
|
||||
def test_no_find_command_used(self):
|
||||
"""Verify list_files implementation does not shell out to find."""
|
||||
import inspect
|
||||
source = inspect.getsource(ToolRegistry.tool_list_files)
|
||||
assert "subprocess" not in source
|
||||
assert '"find"' not in source
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# tool_search_in_files — pure Python, Windows-compatible
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolSearchInFiles:
|
||||
def test_search_basic(self, reg):
|
||||
result = reg.tool_search_in_files(query="def main")
|
||||
assert "main.py" in result
|
||||
|
||||
def test_search_regex(self, reg):
|
||||
result = reg.tool_search_in_files(query=r"def \w+\(")
|
||||
# Should find function definitions
|
||||
assert "def " in result
|
||||
|
||||
def test_search_with_file_pattern(self, reg):
|
||||
result = reg.tool_search_in_files(query="value", file_pattern="*.yaml")
|
||||
assert "config.yaml" in result
|
||||
|
||||
def test_search_no_matches(self, reg):
|
||||
result = reg.tool_search_in_files(query="zzz_nonexistent_pattern_xyz")
|
||||
assert "No matches" in result
|
||||
|
||||
def test_search_invalid_regex(self, reg):
|
||||
result = reg.tool_search_in_files(query="[invalid")
|
||||
assert "invalid regex" in result
|
||||
|
||||
def test_search_nonexistent_dir(self, reg):
|
||||
result = reg.tool_search_in_files(query="test", path="nonexistent_dir")
|
||||
assert "not found" in result
|
||||
|
||||
def test_search_skips_git(self, reg):
|
||||
result = reg.tool_search_in_files(query="refs")
|
||||
# Should not find content inside .git/HEAD
|
||||
# (unless something else matches, but the file .git/HEAD itself should be skipped)
|
||||
lines = result.strip().split("\n") if result.strip() else []
|
||||
for line in lines:
|
||||
assert ".git" not in line
|
||||
|
||||
def test_search_shows_line_numbers(self, reg):
|
||||
result = reg.tool_search_in_files(query="def add")
|
||||
# Format is filepath:linenum:content
|
||||
assert ":" in result
|
||||
|
||||
def test_no_grep_command_used(self):
|
||||
"""Verify search_in_files implementation does not shell out to grep."""
|
||||
import inspect
|
||||
source = inspect.getsource(ToolRegistry.tool_search_in_files)
|
||||
assert "subprocess" not in source
|
||||
assert '"grep"' not in source
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# tool_write_file
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolWriteFile:
|
||||
def test_write_new_file(self, reg, tmp_workdir):
|
||||
result = reg.tool_write_file("output.txt", "hello world")
|
||||
assert "Wrote" in result
|
||||
assert os.path.isfile(os.path.join(str(tmp_workdir), "output.txt"))
|
||||
|
||||
def test_write_creates_subdirs(self, reg, tmp_workdir):
|
||||
result = reg.tool_write_file("sub/dir/test.txt", "nested")
|
||||
assert "Wrote" in result
|
||||
assert os.path.isfile(os.path.join(str(tmp_workdir), "sub", "dir", "test.txt"))
|
||||
|
||||
def test_write_overwrites(self, reg, tmp_workdir):
|
||||
reg.tool_write_file("overwrite.txt", "first")
|
||||
reg.tool_write_file("overwrite.txt", "second")
|
||||
result = reg.tool_read_file("overwrite.txt")
|
||||
assert "second" in result
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# tool_run_command
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolRunCommand:
|
||||
def test_run_simple_command(self, reg):
|
||||
result = reg.tool_run_command("echo hello")
|
||||
assert "hello" in result.lower()
|
||||
|
||||
def test_run_command_timeout(self, reg):
|
||||
# Use a command that sleeps longer than the timeout
|
||||
result = reg.tool_run_command("ping -n 10 127.0.0.1" if os.name == "nt" else "sleep 30", timeout=1)
|
||||
assert "Timeout" in result or "timeout" in result.lower()
|
||||
|
||||
def test_run_command_nonzero_exit(self, reg):
|
||||
result = reg.tool_run_command("exit 1" if os.name != "nt" else "cmd /c exit 1")
|
||||
# Should show exit code
|
||||
assert "exit code" in result or result.strip() != ""
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Module-level backward compatibility wrappers
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestModuleLevelWrappers:
|
||||
def test_set_workdir_and_execute(self, tmp_workdir):
|
||||
set_workdir(str(tmp_workdir))
|
||||
result = execute_tool("read_file", {"path": "README.md"})
|
||||
assert "Hello" in result
|
||||
|
||||
def test_get_tool_schemas_default(self):
|
||||
# Without set_workdir, should still work (defaults to ".")
|
||||
schemas = get_tool_schemas("all")
|
||||
assert len(schemas) == len(TOOL_SCHEMAS)
|
||||
|
||||
def test_get_tool_schemas_module_level(self):
|
||||
schemas = get_tool_schemas("readonly")
|
||||
names = {s["function"]["name"] for s in schemas}
|
||||
assert "read_file" in names
|
||||
assert "web_search" not in names
|
||||
|
||||
def test_execute_unknown_tool(self):
|
||||
result = execute_tool("does_not_exist", {})
|
||||
assert "Unknown tool" in result
|
||||
|
||||
|
||||
# Need re import for Cyrillic check
|
||||
import re
|
||||
Reference in New Issue
Block a user