vitya
6bbbefaa6f
- 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>
389 lines
14 KiB
Python
389 lines
14 KiB
Python
"""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 |