Merge remote-tracking branch 'origin/main' into pr-2722

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_mcp_tool.py

@@ -196,7 +196,7 @@ async def test_execute_re_raises_external_cancellation() -> None:
 
     wrapper = _make_wrapper(SimpleNamespace(call_tool=call_tool), timeout=10)
     task = asyncio.create_task(wrapper.execute())
-    await started.wait()
+    await asyncio.wait_for(started.wait(), timeout=1.0)
 
     task.cancel()
 

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"})
@@ -95,6 +201,14 @@ def test_exec_extract_absolute_paths_keeps_full_windows_path() -> None:
     assert paths == [r"C:\user\workspace\txt"]
 
 
+def test_exec_extract_absolute_paths_captures_windows_drive_root_path() -> None:
+    """Windows drive root paths like `E:\\` must be extracted for workspace guarding."""
+    # Note: raw strings cannot end with a single backslash.
+    cmd = "dir E:\\"
+    paths = ExecTool._extract_absolute_paths(cmd)
+    assert paths == ["E:\\"]
+
+
 def test_exec_extract_absolute_paths_ignores_relative_posix_segments() -> None:
     cmd = ".venv/bin/python script.py"
     paths = ExecTool._extract_absolute_paths(cmd)
@@ -134,6 +248,58 @@ 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
+
+    class FakeWindowsPath:
+        def __init__(self, raw: str) -> None:
+            self.raw = raw.rstrip("\\") + ("\\" if raw.endswith("\\") else "")
+
+        def resolve(self) -> "FakeWindowsPath":
+            return self
+
+        def expanduser(self) -> "FakeWindowsPath":
+            return self
+
+        def is_absolute(self) -> bool:
+            return len(self.raw) >= 3 and self.raw[1:3] == ":\\"
+
+        @property
+        def parents(self) -> list["FakeWindowsPath"]:
+            if not self.is_absolute():
+                return []
+            trimmed = self.raw.rstrip("\\")
+            if len(trimmed) <= 2:
+                return []
+            idx = trimmed.rfind("\\")
+            if idx <= 2:
+                return [FakeWindowsPath(trimmed[:2] + "\\")]
+            parent = FakeWindowsPath(trimmed[:idx])
+            return [parent, *parent.parents]
+
+        def __eq__(self, other: object) -> bool:
+            return isinstance(other, FakeWindowsPath) and self.raw.lower() == other.raw.lower()
+
+    monkeypatch.setattr(shell_mod, "Path", FakeWindowsPath)
+
+    tool = ExecTool(restrict_to_workspace=True)
+    error = tool._guard_command("dir E:\\", "E:\\workspace")
+    assert error == "Error: Command blocked by safety guard (path outside working dir)"
+
+
 # --- cast_params tests ---