Merge remote-tracking branch 'origin/main' into feat/search-tools

Made-with: Cursor

tests/tools/test_filesystem_tools.py

@@ -321,6 +321,22 @@ class TestWorkspaceRestriction:
         assert "Test Skill" in result
         assert "Error" not in result
 
+    @pytest.mark.asyncio
+    async def test_read_allowed_in_media_dir(self, tmp_path, monkeypatch):
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+        media_dir = tmp_path / "media"
+        media_dir.mkdir()
+        media_file = media_dir / "photo.txt"
+        media_file.write_text("shared media", encoding="utf-8")
+
+        monkeypatch.setattr("nanobot.agent.tools.filesystem.get_media_dir", lambda: media_dir)
+
+        tool = ReadFileTool(workspace=workspace, allowed_dir=workspace)
+        result = await tool.execute(path=str(media_file))
+        assert "shared media" in result
+        assert "Error" not in result
+
     @pytest.mark.asyncio
     async def test_extra_dirs_does_not_widen_write(self, tmp_path):
         from nanobot.agent.tools.filesystem import WriteFileTool

tests/tools/test_tool_registry.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+from nanobot.agent.tools.base import Tool
+from nanobot.agent.tools.registry import ToolRegistry
+
+
+class _FakeTool(Tool):
+    def __init__(self, name: str):
+        self._name = name
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def description(self) -> str:
+        return f"{self._name} tool"
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {"type": "object", "properties": {}}
+
+    async def execute(self, **kwargs: Any) -> Any:
+        return kwargs
+
+
+def _tool_names(definitions: list[dict[str, Any]]) -> list[str]:
+    names: list[str] = []
+    for definition in definitions:
+        fn = definition.get("function", {})
+        names.append(fn.get("name", ""))
+    return names
+
+
+def test_get_definitions_orders_builtins_then_mcp_tools() -> None:
+    registry = ToolRegistry()
+    registry.register(_FakeTool("mcp_git_status"))
+    registry.register(_FakeTool("write_file"))
+    registry.register(_FakeTool("mcp_fs_list"))
+    registry.register(_FakeTool("read_file"))
+
+    assert _tool_names(registry.get_definitions()) == [
+        "read_file",
+        "write_file",
+        "mcp_fs_list",
+        "mcp_git_status",
+    ]

tests/tools/test_tool_validation.py

@@ -1,5 +1,14 @@
 from typing import Any
 
+from nanobot.agent.tools import (
+    ArraySchema,
+    IntegerSchema,
+    ObjectSchema,
+    Schema,
+    StringSchema,
+    tool_parameters,
+    tool_parameters_schema,
+)
 from nanobot.agent.tools.base import Tool
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.agent.tools.shell import ExecTool
@@ -41,6 +50,103 @@ class SampleTool(Tool):
         return "ok"
 
 
+@tool_parameters(
+    tool_parameters_schema(
+        query=StringSchema(min_length=2),
+        count=IntegerSchema(2, minimum=1, maximum=10),
+        required=["query", "count"],
+    )
+)
+class DecoratedSampleTool(Tool):
+    @property
+    def name(self) -> str:
+        return "decorated_sample"
+
+    @property
+    def description(self) -> str:
+        return "decorated sample tool"
+
+    async def execute(self, **kwargs: Any) -> str:
+        return f"ok:{kwargs['count']}"
+
+
+def test_schema_validate_value_matches_tool_validate_params() -> None:
+    """ObjectSchema.validate_value 与 validate_json_schema_value、Tool.validate_params 一致。"""
+    root = tool_parameters_schema(
+        query=StringSchema(min_length=2),
+        count=IntegerSchema(2, minimum=1, maximum=10),
+        required=["query", "count"],
+    )
+    obj = ObjectSchema(
+        query=StringSchema(min_length=2),
+        count=IntegerSchema(2, minimum=1, maximum=10),
+        required=["query", "count"],
+    )
+    params = {"query": "h", "count": 2}
+
+    class _Mini(Tool):
+        @property
+        def name(self) -> str:
+            return "m"
+
+        @property
+        def description(self) -> str:
+            return ""
+
+        @property
+        def parameters(self) -> dict[str, Any]:
+            return root
+
+        async def execute(self, **kwargs: Any) -> str:
+            return ""
+
+    expected = _Mini().validate_params(params)
+    assert Schema.validate_json_schema_value(params, root, "") == expected
+    assert obj.validate_value(params, "") == expected
+    assert IntegerSchema(0, minimum=1).validate_value(0, "n") == ["n must be >= 1"]
+
+
+def test_schema_classes_equivalent_to_sample_tool_parameters() -> None:
+    """Schema 类生成的 JSON Schema 应与手写 dict 一致，便于校验行为一致。"""
+    built = tool_parameters_schema(
+        query=StringSchema(min_length=2),
+        count=IntegerSchema(2, minimum=1, maximum=10),
+        mode=StringSchema("", enum=["fast", "full"]),
+        meta=ObjectSchema(
+            tag=StringSchema(""),
+            flags=ArraySchema(StringSchema("")),
+            required=["tag"],
+        ),
+        required=["query", "count"],
+    )
+    assert built == SampleTool().parameters
+
+
+def test_tool_parameters_returns_fresh_copy_per_access() -> None:
+    tool = DecoratedSampleTool()
+
+    first = tool.parameters
+    second = tool.parameters
+
+    assert first == second
+    assert first is not second
+    assert first["properties"] is not second["properties"]
+
+    first["properties"]["query"]["minLength"] = 99
+    assert tool.parameters["properties"]["query"]["minLength"] == 2
+
+
+async def test_registry_executes_decorated_tool_end_to_end() -> None:
+    reg = ToolRegistry()
+    reg.register(DecoratedSampleTool())
+
+    ok = await reg.execute("decorated_sample", {"query": "hello", "count": "3"})
+    assert ok == "ok:3"
+
+    err = await reg.execute("decorated_sample", {"query": "h", "count": 3})
+    assert "Invalid parameters" in err
+
+
 def test_validate_params_missing_required() -> None:
     tool = SampleTool()
     errors = tool.validate_params({"query": "hi"})
@@ -142,6 +248,19 @@ def test_exec_guard_blocks_quoted_home_path_outside_workspace(tmp_path) -> None:
     assert error == "Error: Command blocked by safety guard (path outside working dir)"
 
 
+def test_exec_guard_allows_media_path_outside_workspace(tmp_path, monkeypatch) -> None:
+    media_dir = tmp_path / "media"
+    media_dir.mkdir()
+    media_file = media_dir / "photo.jpg"
+    media_file.write_text("ok", encoding="utf-8")
+
+    monkeypatch.setattr("nanobot.agent.tools.shell.get_media_dir", lambda: media_dir)
+
+    tool = ExecTool(restrict_to_workspace=True)
+    error = tool._guard_command(f'cat "{media_file}"', str(tmp_path / "workspace"))
+    assert error is None
+
+
 def test_exec_guard_blocks_windows_drive_root_outside_workspace(monkeypatch) -> None:
     import nanobot.agent.tools.shell as shell_mod