test/tools: cover workspace, file, run_command and PTY

tests/test_tools/conftest.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from plyngent.tools.workspace import clear_workspace_root, set_workspace_root
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+
+@pytest.fixture
+def workspace(tmp_path: Path) -> Iterator[Path]:
+    clear_workspace_root()
+    root = set_workspace_root(tmp_path)
+    yield root
+    clear_workspace_root()

tests/test_tools/helpers.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import inspect
+from typing import TYPE_CHECKING, Any, cast
+
+if TYPE_CHECKING:
+    from plyngent.agent import ToolDefinition
+
+
+def call_sync(definition: ToolDefinition, *args: object, **kwargs: object) -> str:
+    result: Any = definition.handler(*args, **kwargs)
+    assert not inspect.isawaitable(result)
+    return cast("str", result)
+
+
+async def call_async(definition: ToolDefinition, *args: object, **kwargs: object) -> str:
+    result: Any = definition.handler(*args, **kwargs)
+    if inspect.isawaitable(result):
+        result = await result
+    return cast("str", result)

tests/test_tools/test_file.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from plyngent.tools.file import edit_replace, listdir, read_file, write_file
+from tests.test_tools.helpers import call_sync
+
+
+def test_write_read_listdir_edit(workspace: object) -> None:
+    del workspace
+    assert "wrote" in call_sync(write_file, "notes/a.txt", "hello world")
+    assert call_sync(read_file, "notes/a.txt") == "hello world"
+    listing = call_sync(listdir, "notes")
+    assert "a.txt" in listing
+    assert "file" in listing
+    result = call_sync(edit_replace, "notes/a.txt", "world", "there")
+    assert "replaced" in result
+    assert call_sync(read_file, "notes/a.txt") == "hello there"
+
+
+def test_edit_replace_first_only(workspace: object) -> None:
+    del workspace
+    _ = call_sync(write_file, "t.txt", "aa aa")
+    _ = call_sync(edit_replace, "t.txt", "aa", "bb")
+    assert call_sync(read_file, "t.txt") == "bb aa"
+
+
+def test_edit_missing_old_string(workspace: object) -> None:
+    del workspace
+    _ = call_sync(write_file, "t.txt", "x")
+    assert "not found" in call_sync(edit_replace, "t.txt", "missing", "y")
+
+
+def test_read_offset_limit(workspace: object) -> None:
+    del workspace
+    _ = call_sync(write_file, "lines.txt", "a\nb\nc\nd\n")
+    assert call_sync(read_file, "lines.txt", offset=1, limit=2) == "b\nc\n"
+
+
+def test_listdir_missing(workspace: object) -> None:
+    del workspace
+    assert "error" in call_sync(listdir, "nope")

tests/test_tools/test_process.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+from plyngent.tools.process import close_pty, open_pty, read_pty, run_command
+from plyngent.tools.process.pty_session import PtyManager
+from plyngent.tools.workspace import set_command_denylist
+from tests.test_tools.helpers import call_async, call_sync
+
+
+async def test_run_command_echo(workspace: object) -> None:
+    del workspace
+    out = await call_async(run_command, ["echo", "hi"])
+    assert "exit_code=0" in out
+    assert "hi" in out
+
+
+async def test_run_command_denied(workspace: object) -> None:
+    del workspace
+    out = await call_async(run_command, ["rm", "-rf", "."])
+    assert "denied" in out
+
+
+async def test_run_command_cwd(workspace: object) -> None:
+    assert isinstance(workspace, Path)
+    sub = workspace / "sub"
+    sub.mkdir()
+    _ = (sub / "f.txt").write_text("z", encoding="utf-8")
+    out = await call_async(run_command, ["ls"], cwd="sub")
+    assert "f.txt" in out
+
+
+async def test_run_command_timeout(workspace: object) -> None:
+    del workspace
+    out = await call_async(run_command, ["sleep", "5"], timeout_seconds=0.2)
+    assert "timed out" in out
+
+
+def test_pty_open_read_close(workspace: object) -> None:
+    del workspace
+    try:
+        opened = call_sync(open_pty, ["sleep", "30"])
+        assert opened.startswith("session_id=")
+        session_id = int(opened.split("=", 1)[1])
+        data = call_sync(read_pty, session_id, timeout=0.05)
+        assert isinstance(data, str)
+        closed = call_sync(close_pty, session_id)
+        assert "closed" in closed
+        assert "error" in call_sync(read_pty, session_id)
+    finally:
+        PtyManager.close_all()
+        set_command_denylist(None)
+
+
+def test_pty_denied(workspace: object) -> None:
+    del workspace
+    assert "denied" in call_sync(open_pty, ["sudo", "ls"])
+
+
+def test_pty_echo_output(workspace: object) -> None:
+    del workspace
+    try:
+        opened = call_sync(open_pty, ["/bin/echo", "hello-pty"])
+        session_id = int(opened.split("=", 1)[1])
+        chunks: list[str] = []
+        for _ in range(20):
+            chunk = call_sync(read_pty, session_id, timeout=0.1)
+            if chunk:
+                chunks.append(chunk)
+                if "hello-pty" in "".join(chunks):
+                    break
+            time.sleep(0.05)
+        text = "".join(chunks)
+        _ = call_sync(close_pty, session_id)
+        assert "hello-pty" in text
+    finally:
+        PtyManager.close_all()

tests/test_tools/test_workspace.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import pytest
+
+from plyngent.tools import (
+    WorkspaceError,
+    check_command_allowed,
+    clear_workspace_root,
+    get_workspace_root,
+    resolve_path,
+    set_command_denylist,
+    set_path_denylist,
+)
+
+
+def test_resolve_relative_and_absolute(workspace: object) -> None:
+    from pathlib import Path
+
+    assert isinstance(workspace, Path)
+    _ = (workspace / "a.txt").write_text("x", encoding="utf-8")
+    assert resolve_path("a.txt") == workspace / "a.txt"
+    assert resolve_path(workspace / "a.txt") == workspace / "a.txt"
+
+
+def test_escape_rejected(workspace: object) -> None:
+    del workspace
+    with pytest.raises(WorkspaceError, match="escapes"):
+        _ = resolve_path("../outside")
+
+
+def test_path_denylist(workspace: object) -> None:
+    from pathlib import Path
+
+    assert isinstance(workspace, Path)
+    secrets = workspace / "secrets"
+    secrets.mkdir()
+    _ = (secrets / "key").write_text("k", encoding="utf-8")
+    set_path_denylist(["/secrets/"])
+    with pytest.raises(WorkspaceError, match="denied"):
+        _ = resolve_path("secrets/key")
+    set_path_denylist(None)
+
+
+def test_command_denylist(workspace: object) -> None:
+    del workspace
+    with pytest.raises(WorkspaceError, match="denied"):
+        check_command_allowed(["rm", "-rf", "/"])
+    check_command_allowed(["echo", "ok"])
+    set_command_denylist(None)
+
+
+def test_root_required() -> None:
+    clear_workspace_root()
+    with pytest.raises(WorkspaceError, match="not set"):
+        _ = get_workspace_root()