andrewyng--aisuite
Rohit C Prasad
41b1591f6d
Mechanical 'black .' over root package and platform/ (109 files; the Lint workflow checks the whole repo and has been red since platform landed unformatted, with a few more violations from community merges whose branches never ran CI). No functional change: full platform suite (264) and aisuite suites verified unchanged after formatting.
118 行
4.4 KiB
Python
118 行
4.4 KiB
Python
"""Tests for the Code agent's new tools: grep (ripgrep/python), git_log, web_fetch.
|
|
|
|
No network: web_fetch is exercised via its URL guard + the HTML→text helper. grep/git_log run
|
|
against temp dirs (git_log needs a real `git`, which the dev box has).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from coworker.tools.git import git_tools
|
|
from coworker.tools.search import _py_grep, search_tools
|
|
from coworker.web.fetch import _html_to_text, make_web_fetch_tool
|
|
|
|
|
|
# -- grep ----------------------------------------------------------------------
|
|
def _seed(tmp_path):
|
|
(tmp_path / "a.py").write_text("def hello():\n return 42\n", encoding="utf-8")
|
|
(tmp_path / "b.txt").write_text("hello world\nbye\n", encoding="utf-8")
|
|
nm = tmp_path / "node_modules" / "pkg"
|
|
nm.mkdir(parents=True)
|
|
(nm / "junk.py").write_text("hello from deps\n", encoding="utf-8")
|
|
|
|
|
|
def test_grep_finds_matches_and_respects_glob(tmp_path):
|
|
_seed(tmp_path)
|
|
grep = search_tools(str(tmp_path))[0]
|
|
out = grep(pattern="hello")
|
|
files = {m["file"] for m in out["matches"]}
|
|
assert "a.py" in files and "b.txt" in files
|
|
# node_modules is ignored by both engines
|
|
assert not any("node_modules" in f for f in files)
|
|
# glob filter restricts to python files
|
|
only_py = grep(pattern="hello", glob="*.py")
|
|
assert all(m["file"].endswith(".py") for m in only_py["matches"])
|
|
assert only_py["matches"][0]["line"] == 1
|
|
|
|
|
|
def test_grep_rejects_path_escape(tmp_path):
|
|
grep = search_tools(str(tmp_path))[0]
|
|
assert "escapes" in grep(pattern="x", path="../..")["error"]
|
|
|
|
|
|
def test_py_grep_fallback_skips_ignored_dirs(tmp_path):
|
|
_seed(tmp_path)
|
|
res = _py_grep(tmp_path.resolve(), tmp_path.resolve(), "hello", None, 100)
|
|
assert res["count"] == 2 # a.py + b.txt, NOT node_modules
|
|
assert all("node_modules" not in m["file"] for m in res["matches"])
|
|
|
|
|
|
# -- git_log -------------------------------------------------------------------
|
|
def test_git_log_lists_commits(tmp_path):
|
|
ws = tmp_path / "repo"
|
|
ws.mkdir()
|
|
run = lambda *a: subprocess.run(
|
|
["git", "-C", str(ws), *a], capture_output=True, check=True
|
|
)
|
|
run("init", "-q")
|
|
run("config", "user.email", "t@t.io")
|
|
run("config", "user.name", "T")
|
|
(ws / "f.txt").write_text("1", encoding="utf-8")
|
|
run("add", "-A")
|
|
run("commit", "-qm", "first")
|
|
(ws / "f.txt").write_text("2", encoding="utf-8")
|
|
run("add", "-A")
|
|
run("commit", "-qm", "second")
|
|
|
|
git_log = git_tools(str(ws))[0]
|
|
out = git_log(max_count=10)
|
|
assert out["count"] == 2
|
|
assert out["commits"][0]["subject"] == "second" # newest first
|
|
assert set(out["commits"][0]) == {"hash", "author", "date", "subject"}
|
|
|
|
|
|
def test_git_log_errors_outside_repo(tmp_path):
|
|
git_log = git_tools(str(tmp_path))[0]
|
|
assert "error" in git_log()
|
|
|
|
|
|
# -- web_fetch -----------------------------------------------------------------
|
|
def test_web_fetch_rejects_non_http():
|
|
web_fetch = make_web_fetch_tool()
|
|
assert "http" in web_fetch("file:///etc/passwd")["error"]
|
|
assert "http" in web_fetch("javascript:alert(1)")["error"]
|
|
|
|
|
|
def test_html_to_text_strips_scripts_and_tags():
|
|
html = "<html><head><style>x{}</style></head><body><h1>Hi</h1><script>bad()</script><p>Body text</p></body></html>"
|
|
text = _html_to_text(html)
|
|
assert "Hi" in text and "Body text" in text
|
|
assert "bad()" not in text and "x{}" not in text
|
|
|
|
|
|
# -- Code agent wiring ---------------------------------------------------------
|
|
def test_code_agent_has_grep_and_git_log_not_search_files(tmp_path):
|
|
from coworker.agents.base import AgentContext
|
|
from coworker.agents.code import code_agent
|
|
|
|
ctx = AgentContext(workspace=tmp_path, executor=None, todo=None)
|
|
names = {getattr(t, "__name__", "") for t in code_agent().build_tools(ctx)}
|
|
assert "grep" in names and "git_log" in names
|
|
assert "search_files" not in names # replaced by grep
|
|
assert {"read_file", "write_file", "git_status", "git_diff"} <= names
|
|
|
|
|
|
def test_cowork_has_grep_not_search_files(tmp_path):
|
|
from coworker.agents.base import AgentContext
|
|
from coworker.agents.cowork import cowork_tool_factory
|
|
|
|
names = {
|
|
getattr(t, "__name__", "")
|
|
for t in cowork_tool_factory(AgentContext(workspace=tmp_path))
|
|
}
|
|
assert "grep" in names and "search_files" not in names
|
|
assert "git_log" not in names # git history isn't useful for Cowork
|