From f45af76c9b595be87fe511378435f098ef45c284 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 14 Jul 2026 18:20:32 +0800 Subject: [PATCH] test/tools: cover workspace, file, run_command and PTY --- tests/test_tools/__init__.py | 0 tests/test_tools/conftest.py | 19 ++++++++ tests/test_tools/helpers.py | 20 ++++++++ tests/test_tools/test_file.py | 40 +++++++++++++++ tests/test_tools/test_process.py | 78 ++++++++++++++++++++++++++++++ tests/test_tools/test_workspace.py | 55 +++++++++++++++++++++ 6 files changed, 212 insertions(+) create mode 100644 tests/test_tools/__init__.py create mode 100644 tests/test_tools/conftest.py create mode 100644 tests/test_tools/helpers.py create mode 100644 tests/test_tools/test_file.py create mode 100644 tests/test_tools/test_process.py create mode 100644 tests/test_tools/test_workspace.py diff --git a/tests/test_tools/__init__.py b/tests/test_tools/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_tools/conftest.py b/tests/test_tools/conftest.py new file mode 100644 index 0000000..b3ba328 --- /dev/null +++ b/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() diff --git a/tests/test_tools/helpers.py b/tests/test_tools/helpers.py new file mode 100644 index 0000000..f94d81f --- /dev/null +++ b/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) diff --git a/tests/test_tools/test_file.py b/tests/test_tools/test_file.py new file mode 100644 index 0000000..bd2d68d --- /dev/null +++ b/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") diff --git a/tests/test_tools/test_process.py b/tests/test_tools/test_process.py new file mode 100644 index 0000000..d7d80d8 --- /dev/null +++ b/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() diff --git a/tests/test_tools/test_workspace.py b/tests/test_tools/test_workspace.py new file mode 100644 index 0000000..b66bd1e --- /dev/null +++ b/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()