wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
410 lines
15 KiB
Python
410 lines
15 KiB
Python
"""Tests for built-in tools and unified tool registry behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import types
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from deeptutor.core.tool_protocol import BaseTool, ToolDefinition, ToolParameter, ToolResult
|
|
from deeptutor.runtime.registry.tool_registry import ToolRegistry
|
|
from deeptutor.services.path_service import PathService
|
|
from deeptutor.services.sandbox.spec import ExecResult
|
|
from deeptutor.tools.builtin import (
|
|
BrainstormTool,
|
|
CodeExecutionTool,
|
|
ExecTool,
|
|
GeoGebraAnalysisTool,
|
|
PaperSearchToolWrapper,
|
|
RAGTool,
|
|
ReasonTool,
|
|
WebSearchTool,
|
|
)
|
|
|
|
|
|
def _install_module(
|
|
monkeypatch: pytest.MonkeyPatch, fullname: str, **attrs: Any
|
|
) -> types.ModuleType:
|
|
"""Install a fake module (and missing parent packages) into sys.modules."""
|
|
parts = fullname.split(".")
|
|
for idx in range(1, len(parts)):
|
|
pkg_name = ".".join(parts[:idx])
|
|
if pkg_name not in sys.modules:
|
|
pkg = types.ModuleType(pkg_name)
|
|
pkg.__path__ = [] # type: ignore[attr-defined]
|
|
monkeypatch.setitem(sys.modules, pkg_name, pkg)
|
|
if idx > 1:
|
|
parent = sys.modules[".".join(parts[: idx - 1])]
|
|
# Use monkeypatch (not raw setattr) so the parent package's
|
|
# attribute is restored on teardown — otherwise a real
|
|
# submodule the parent exposes (e.g. ``llm.config``) stays
|
|
# shadowed by the fake and leaks into later tests.
|
|
monkeypatch.setattr(parent, parts[idx - 1], pkg, raising=False)
|
|
|
|
module = types.ModuleType(fullname)
|
|
for key, value in attrs.items():
|
|
setattr(module, key, value)
|
|
monkeypatch.setitem(sys.modules, fullname, module)
|
|
if len(parts) > 1:
|
|
parent = sys.modules[".".join(parts[:-1])]
|
|
monkeypatch.setattr(parent, parts[-1], module, raising=False)
|
|
return module
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_tool_reports_generated_public_artifacts(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
path_service = PathService(workspace_root=tmp_path / "data")
|
|
workdir = path_service.get_task_workspace("chat", "turn_1") / "exec"
|
|
|
|
class FakeSandboxService:
|
|
async def run(self, request, *, user_id: str):
|
|
assert user_id == "user-1"
|
|
assert request.command == "python build_pdf.py"
|
|
output_dir = PathService(workspace_root=tmp_path / "data").get_task_workspace(
|
|
"chat", "turn_1"
|
|
)
|
|
assert request.workdir == str(output_dir / "exec")
|
|
work = output_dir / "exec"
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
(work / "report.pdf").write_bytes(b"%PDF-1.4\n")
|
|
(work / "build_pdf.py").write_text("print('internal')", encoding="utf-8")
|
|
return ExecResult(stdout="created report.pdf\n", exit_code=0)
|
|
|
|
import deeptutor.services.sandbox as sandbox_pkg
|
|
import deeptutor.services.sandbox.artifacts as sandbox_artifacts
|
|
|
|
monkeypatch.setattr(sandbox_pkg, "get_sandbox_service", lambda: FakeSandboxService())
|
|
monkeypatch.setattr(sandbox_artifacts, "get_path_service", lambda: path_service)
|
|
|
|
result = await ExecTool().execute(
|
|
command="python build_pdf.py",
|
|
_sandbox_user_id="user-1",
|
|
_sandbox_workdir=str(workdir),
|
|
)
|
|
|
|
assert result.success is True
|
|
assert "Generated artifacts" in result.content
|
|
assert "report.pdf" in result.content
|
|
# The model is told to mention the exact filename (the UI linkifies it); the
|
|
# raw download URL is delivered out-of-band (sources/metadata), never in the
|
|
# model-facing text, so the model can't paste it.
|
|
assert "clickable link" in result.content
|
|
assert "/api/outputs/" not in result.content
|
|
assert "build_pdf.py" not in result.content
|
|
assert result.metadata["artifacts"][0]["filename"] == "report.pdf"
|
|
assert (
|
|
result.metadata["artifacts"][0]["url"]
|
|
== "/api/outputs/workspace/chat/chat/turn_1/exec/report.pdf"
|
|
)
|
|
assert result.sources[0]["url"].endswith("/report.pdf")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_brainstorm_tool_passes_llm_arguments(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_brainstorm(**kwargs: Any) -> dict[str, Any]:
|
|
captured.update(kwargs)
|
|
return {"answer": "## 1. Test idea\n- Rationale: worth exploring"}
|
|
|
|
_install_module(monkeypatch, "deeptutor.tools.brainstorm", brainstorm=fake_brainstorm)
|
|
|
|
result = await BrainstormTool().execute(
|
|
topic="agent-native tutoring",
|
|
context="Focus on fast ideation",
|
|
model="gpt-test",
|
|
)
|
|
|
|
assert "Test idea" in result.content
|
|
assert captured["topic"] == "agent-native tutoring"
|
|
assert captured["context"] == "Focus on fast ideation"
|
|
assert captured["model"] == "gpt-test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rag_tool_forwards_query_and_extra_kwargs(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_rag_search(**kwargs: Any) -> dict[str, Any]:
|
|
captured.update(kwargs)
|
|
return {"answer": "grounded answer", "provider": "fake"}
|
|
|
|
_install_module(monkeypatch, "deeptutor.tools.rag_tool", rag_search=fake_rag_search)
|
|
|
|
result = await RAGTool().execute(
|
|
query="what is a tensor",
|
|
kb_name="demo-kb",
|
|
mode="hybrid",
|
|
only_need_context=True,
|
|
)
|
|
|
|
assert result.content == "grounded answer"
|
|
assert captured["query"] == "what is a tensor"
|
|
assert captured["kb_name"] == "demo-kb"
|
|
assert captured["mode"] == "hybrid"
|
|
assert captured["only_need_context"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rag_tool_rejects_empty_query(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
called = False
|
|
|
|
async def fake_rag_search(**_kwargs: Any) -> dict[str, Any]:
|
|
nonlocal called
|
|
called = True
|
|
return {"answer": "should not run"}
|
|
|
|
_install_module(monkeypatch, "deeptutor.tools.rag_tool", rag_search=fake_rag_search)
|
|
|
|
with pytest.raises(ValueError, match="RAG query must be a non-empty string"):
|
|
await RAGTool().execute(query=" ", kb_name="demo-kb")
|
|
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_tool_wraps_sync_function(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_web_search(**kwargs: Any) -> dict[str, Any]:
|
|
captured.update(kwargs)
|
|
return {
|
|
"answer": "web summary",
|
|
"citations": [{"url": "https://example.com", "title": "Example"}],
|
|
}
|
|
|
|
_install_module(monkeypatch, "deeptutor.tools.web_search", web_search=fake_web_search)
|
|
|
|
result = await WebSearchTool().execute(query="latest benchmark", output_dir="/tmp/out")
|
|
|
|
assert result.content == "web summary"
|
|
assert captured["query"] == "latest benchmark"
|
|
assert captured["output_dir"] == "/tmp/out"
|
|
assert result.sources[0]["url"] == "https://example.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_code_execution_tool_runs_python_via_sandbox(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from pathlib import Path
|
|
|
|
path_service = PathService(workspace_root=tmp_path / "data")
|
|
workdir = path_service.get_task_workspace("chat", "turn_1") / "code_runs"
|
|
|
|
class FakeSandboxService:
|
|
async def run(self, request, *, user_id: str):
|
|
assert user_id == "user-1"
|
|
assert request.command == "python3 main.py"
|
|
run_dir = Path(request.workdir)
|
|
assert run_dir.parent == workdir
|
|
assert run_dir.name.startswith("python_")
|
|
# The tool wrote the source file into the run dir before invoking us.
|
|
assert (run_dir / "main.py").read_text(encoding="utf-8") == "print(2 + 2)"
|
|
(run_dir / "result.txt").write_text("ok", encoding="utf-8")
|
|
return ExecResult(stdout="4\n", exit_code=0)
|
|
|
|
import deeptutor.services.sandbox as sandbox_pkg
|
|
import deeptutor.services.sandbox.artifacts as sandbox_artifacts
|
|
|
|
monkeypatch.setattr(sandbox_pkg, "get_sandbox_service", lambda: FakeSandboxService())
|
|
monkeypatch.setattr(sandbox_artifacts, "get_path_service", lambda: path_service)
|
|
|
|
result = await CodeExecutionTool().execute(
|
|
language="python",
|
|
code="print(2 + 2)",
|
|
_sandbox_user_id="user-1",
|
|
_sandbox_workdir=str(workdir),
|
|
)
|
|
|
|
assert result.success is True
|
|
assert "4" in result.content
|
|
assert result.metadata["language"] == "python"
|
|
assert result.metadata["command"] == "python3 main.py"
|
|
# The program-generated file surfaces; the source file we wrote is filtered.
|
|
artifact_names = [row["filename"] for row in result.metadata["artifacts"]]
|
|
assert "result.txt" in artifact_names
|
|
assert "main.py" not in artifact_names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_code_execution_tool_compiles_cpp(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from pathlib import Path
|
|
|
|
path_service = PathService(workspace_root=tmp_path / "data")
|
|
workdir = path_service.get_task_workspace("chat", "turn_1") / "code_runs"
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeSandboxService:
|
|
async def run(self, request, *, user_id: str):
|
|
captured["command"] = request.command
|
|
run_dir = Path(request.workdir)
|
|
assert run_dir.name.startswith("cpp_")
|
|
assert (run_dir / "main.cpp").exists()
|
|
return ExecResult(stdout="hi\n", exit_code=0)
|
|
|
|
import deeptutor.services.sandbox as sandbox_pkg
|
|
import deeptutor.services.sandbox.artifacts as sandbox_artifacts
|
|
|
|
monkeypatch.setattr(sandbox_pkg, "get_sandbox_service", lambda: FakeSandboxService())
|
|
monkeypatch.setattr(sandbox_artifacts, "get_path_service", lambda: path_service)
|
|
|
|
result = await CodeExecutionTool().execute(
|
|
language="cpp",
|
|
code="int main(){}",
|
|
_sandbox_workdir=str(workdir),
|
|
)
|
|
|
|
assert result.success is True
|
|
assert captured["command"] == "c++ -std=c++17 -O2 main.cpp -o prog && ./prog"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_code_execution_tool_rejects_bad_input() -> None:
|
|
tool = CodeExecutionTool()
|
|
with pytest.raises(ValueError, match="Unsupported language"):
|
|
await tool.execute(language="ruby", code="puts 1")
|
|
with pytest.raises(ValueError, match="non-empty 'code'"):
|
|
await tool.execute(language="python", code=" ")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reason_tool_passes_llm_arguments(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_reason(**kwargs: Any) -> dict[str, Any]:
|
|
captured.update(kwargs)
|
|
return {"answer": "reasoned"}
|
|
|
|
_install_module(monkeypatch, "deeptutor.tools.reason", reason=fake_reason)
|
|
|
|
result = await ReasonTool().execute(
|
|
query="derive the formula",
|
|
context="prior work",
|
|
api_key="key",
|
|
base_url="url",
|
|
model="gpt-test",
|
|
)
|
|
|
|
assert result.content == "reasoned"
|
|
assert captured["model"] == "gpt-test"
|
|
assert captured["context"] == "prior work"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_paper_search_tool_formats_papers(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class FakeArxivSearchTool:
|
|
async def search_papers(self, **kwargs: Any) -> list[dict[str, Any]]:
|
|
assert kwargs["query"] == "graph learning"
|
|
return [
|
|
{
|
|
"title": "Graph Learning 101",
|
|
"year": 2024,
|
|
"authors": ["Ada", "Grace"],
|
|
"arxiv_id": "1234.5678",
|
|
"url": "https://arxiv.org/abs/1234.5678",
|
|
"abstract": "A compact abstract.",
|
|
}
|
|
]
|
|
|
|
_install_module(
|
|
monkeypatch,
|
|
"deeptutor.tools.paper_search_tool",
|
|
ArxivSearchTool=FakeArxivSearchTool,
|
|
)
|
|
|
|
result = await PaperSearchToolWrapper().execute(query="graph learning")
|
|
|
|
assert "Graph Learning 101" in result.content
|
|
assert result.sources[0]["provider"] == "arxiv"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_geogebra_analysis_tool_handles_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class FakeVisionSolverAgent:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
self.kwargs = kwargs
|
|
|
|
async def process(self, **kwargs: Any) -> dict[str, Any]:
|
|
assert kwargs["question_text"] == "analyze this"
|
|
return {
|
|
"has_image": True,
|
|
"final_ggb_commands": ["A=(0,0)", "B=(1,0)"],
|
|
"analysis_output": {
|
|
"constraints": ["AB = 1"],
|
|
"geometric_relations": [{"description": "A and B are on x-axis"}],
|
|
},
|
|
"image_is_reference": False,
|
|
}
|
|
|
|
def format_ggb_block(self, commands: list[str]) -> str:
|
|
return "\n".join(commands)
|
|
|
|
_install_module(
|
|
monkeypatch,
|
|
"deeptutor.agents.vision_solver.vision_solver_agent",
|
|
VisionSolverAgent=FakeVisionSolverAgent,
|
|
)
|
|
_install_module(
|
|
monkeypatch,
|
|
"deeptutor.services.llm.config",
|
|
get_llm_config=lambda: SimpleNamespace(api_key="k", base_url="u"),
|
|
)
|
|
|
|
result = await GeoGebraAnalysisTool().execute(
|
|
question="analyze this",
|
|
image_base64="ZmFrZQ==",
|
|
language="en",
|
|
)
|
|
|
|
assert result.success is True
|
|
assert "A=(0,0)" in result.content
|
|
assert result.metadata["commands_count"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_registry_resolves_aliases_and_argument_mapping() -> None:
|
|
class DummyTool(BaseTool):
|
|
def __init__(self, tool_name: str) -> None:
|
|
self._tool_name = tool_name
|
|
self.calls: list[dict[str, Any]] = []
|
|
|
|
def get_definition(self) -> ToolDefinition:
|
|
param_name = {
|
|
"rag": "query",
|
|
"code_execution": "code",
|
|
}[self._tool_name]
|
|
return ToolDefinition(
|
|
name=self._tool_name,
|
|
description="dummy",
|
|
parameters=[ToolParameter(name=param_name, type="string")],
|
|
)
|
|
|
|
async def execute(self, **kwargs: Any) -> ToolResult:
|
|
self.calls.append(kwargs)
|
|
return ToolResult(content=self._tool_name)
|
|
|
|
rag = DummyTool("rag")
|
|
code = DummyTool("code_execution")
|
|
|
|
registry = ToolRegistry()
|
|
registry.register(rag)
|
|
registry.register(code)
|
|
|
|
rag_result = await registry.execute("rag_hybrid", query="find this")
|
|
code_result = await registry.execute("run_code", language="python", code="print(1)")
|
|
|
|
assert rag_result.content == "rag"
|
|
assert rag.calls[0]["mode"] == "hybrid"
|
|
assert rag.calls[0]["query"] == "find this"
|
|
# The `run_code` alias resolves to code_execution and forwards kwargs
|
|
# verbatim — no natural-language `query`→`intent` remapping any more.
|
|
assert code_result.content == "code_execution"
|
|
assert code.calls[0]["code"] == "print(1)"
|
|
assert code.calls[0]["language"] == "python"
|