thaloco/pyro-mcp
1
Fork 0
Code Pull requests Releases Packages Activity Actions

Add model-native workspace file operations

Browse source 
Remove shell-escaped file mutation from the stable workspace flow by adding explicit file and patch tools across the CLI, SDK, and MCP surfaces.

This adds workspace file list/read/write plus unified text patch application, backed by new guest and manager file primitives that stay scoped to started workspaces and /workspace only. Patch application is preflighted on the host, file writes stay text-only and bounded, and the existing diff/export/reset semantics remain intact.

The milestone also updates the 3.2.0 roadmap, public contract, docs, examples, and versioning, and includes focused coverage for the new helper module and dispatch paths.

Validation:
- uv lock
- UV_CACHE_DIR=.uv-cache make check
- UV_CACHE_DIR=.uv-cache make dist-check
- real guest-backed smoke for workspace file read, patch apply, exec, export, and delete
This commit is contained in:
Thales Maciel 2026-03-12 22:03:25 -03:00
parent dbb71a3174
commit ab02ae46c7
27 changed files with 3068 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

327
tests/test_api.py
View file

@ -55,6 +55,10 @@ def test_pyro_create_server_registers_vm_run(tmp_path: Path) -> None:
assert "workspace_diff" in tool_names
assert "workspace_sync_push" in tool_names
assert "workspace_export" in tool_names
assert "workspace_file_list" in tool_names
assert "workspace_file_read" in tool_names
assert "workspace_file_write" in tool_names
assert "workspace_patch_apply" in tool_names
assert "workspace_disk_export" in tool_names
assert "workspace_disk_list" in tool_names
assert "workspace_disk_read" in tool_names
@ -230,6 +234,23 @@ def test_pyro_workspace_methods_delegate_to_manager(tmp_path: Path) -> None:
command='sh -lc \'printf "%s\\n" "$API_TOKEN"\'',
secret_env={"API_TOKEN": "API_TOKEN"},
)
listed_files = pyro.list_workspace_files(workspace_id, path="/workspace", recursive=True)
file_read = pyro.read_workspace_file(workspace_id, "note.txt")
file_write = pyro.write_workspace_file(
workspace_id,
"src/app.py",
text="print('hello from file op')\n",
)
patch_result = pyro.apply_workspace_patch(
workspace_id,
patch=(
"--- a/note.txt\n"
"+++ b/note.txt\n"
"@@ -1 +1 @@\n"
"-ok\n"
"+patched\n"
),
)
diff_payload = pyro.diff_workspace(workspace_id)
snapshot = pyro.create_snapshot(workspace_id, "checkpoint")
snapshots = pyro.list_snapshots(workspace_id)
@ -273,13 +294,19 @@ def test_pyro_workspace_methods_delegate_to_manager(tmp_path: Path) -> None:
{"name": "FILE_TOKEN", "source_kind": "file"},
]
assert executed["stdout"] == "[REDACTED]\n"
assert any(entry["path"] == "/workspace/note.txt" for entry in listed_files["entries"])
assert file_read["content"] == "ok\n"
assert file_write["path"] == "/workspace/src/app.py"
assert file_write["bytes_written"] == len("print('hello from file op')\n".encode("utf-8"))
assert patch_result["changed"] is True
assert patch_result["entries"] == [{"path": "/workspace/note.txt", "status": "modified"}]
assert created["workspace_seed"]["mode"] == "directory"
assert synced["workspace_sync"]["destination"] == "/workspace/subdir"
assert diff_payload["changed"] is True
assert snapshot["snapshot"]["snapshot_name"] == "checkpoint"
assert snapshots["count"] == 2
assert exported["output_path"] == str(export_path)
assert export_path.read_text(encoding="utf-8") == "ok\n"
assert export_path.read_text(encoding="utf-8") == "patched\n"
assert shell_output["output"].count("[REDACTED]") >= 1
assert shell_closed["closed"] is True
assert service["state"] == "running"
@ -540,6 +567,304 @@ def test_pyro_create_server_workspace_disk_tools_delegate() -> None:
]
def test_pyro_workspace_file_methods_delegate_to_manager() -> None:
calls: list[tuple[str, dict[str, Any]]] = []
class StubManager:
def list_workspace_files(
self,
workspace_id: str,
*,
path: str = "/workspace",
recursive: bool = False,
) -> dict[str, Any]:
calls.append(
(
"list_workspace_files",
{
"workspace_id": workspace_id,
"path": path,
"recursive": recursive,
},
)
)
return {"workspace_id": workspace_id, "entries": []}
def read_workspace_file(
self,
workspace_id: str,
path: str,
*,
max_bytes: int = 65536,
) -> dict[str, Any]:
calls.append(
(
"read_workspace_file",
{
"workspace_id": workspace_id,
"path": path,
"max_bytes": max_bytes,
},
)
)
return {"workspace_id": workspace_id, "content": "hello\n"}
def write_workspace_file(
self,
workspace_id: str,
path: str,
*,
text: str,
) -> dict[str, Any]:
calls.append(
(
"write_workspace_file",
{
"workspace_id": workspace_id,
"path": path,
"text": text,
},
)
)
return {"workspace_id": workspace_id, "bytes_written": len(text.encode("utf-8"))}
def apply_workspace_patch(
self,
workspace_id: str,
*,
patch: str,
) -> dict[str, Any]:
calls.append(
(
"apply_workspace_patch",
{
"workspace_id": workspace_id,
"patch": patch,
},
)
)
return {"workspace_id": workspace_id, "changed": True}
pyro = Pyro(manager=cast(Any, StubManager()))
listed = pyro.list_workspace_files("workspace-123", path="/workspace/src", recursive=True)
read = pyro.read_workspace_file("workspace-123", "note.txt", max_bytes=4096)
written = pyro.write_workspace_file("workspace-123", "src/app.py", text="print('hi')\n")
patched = pyro.apply_workspace_patch(
"workspace-123",
patch="--- a/note.txt\n+++ b/note.txt\n@@ -1 +1 @@\n-old\n+new\n",
)
assert listed["entries"] == []
assert read["content"] == "hello\n"
assert written["bytes_written"] == len("print('hi')\n".encode("utf-8"))
assert patched["changed"] is True
assert calls == [
(
"list_workspace_files",
{
"workspace_id": "workspace-123",
"path": "/workspace/src",
"recursive": True,
},
),
(
"read_workspace_file",
{
"workspace_id": "workspace-123",
"path": "note.txt",
"max_bytes": 4096,
},
),
(
"write_workspace_file",
{
"workspace_id": "workspace-123",
"path": "src/app.py",
"text": "print('hi')\n",
},
),
(
"apply_workspace_patch",
{
"workspace_id": "workspace-123",
"patch": "--- a/note.txt\n+++ b/note.txt\n@@ -1 +1 @@\n-old\n+new\n",
},
),
]
def test_pyro_create_server_workspace_file_tools_delegate() -> None:
calls: list[tuple[str, dict[str, Any]]] = []
class StubManager:
def list_workspace_files(
self,
workspace_id: str,
*,
path: str = "/workspace",
recursive: bool = False,
) -> dict[str, Any]:
calls.append(
(
"list_workspace_files",
{
"workspace_id": workspace_id,
"path": path,
"recursive": recursive,
},
)
)
return {"workspace_id": workspace_id, "entries": []}
def read_workspace_file(
self,
workspace_id: str,
path: str,
*,
max_bytes: int = 65536,
) -> dict[str, Any]:
calls.append(
(
"read_workspace_file",
{
"workspace_id": workspace_id,
"path": path,
"max_bytes": max_bytes,
},
)
)
return {"workspace_id": workspace_id, "content": "hello\n"}
def write_workspace_file(
self,
workspace_id: str,
path: str,
*,
text: str,
) -> dict[str, Any]:
calls.append(
(
"write_workspace_file",
{
"workspace_id": workspace_id,
"path": path,
"text": text,
},
)
)
return {"workspace_id": workspace_id, "bytes_written": len(text.encode("utf-8"))}
def apply_workspace_patch(
self,
workspace_id: str,
*,
patch: str,
) -> dict[str, Any]:
calls.append(
(
"apply_workspace_patch",
{
"workspace_id": workspace_id,
"patch": patch,
},
)
)
return {"workspace_id": workspace_id, "changed": True}
pyro = Pyro(manager=cast(Any, StubManager()))
def _extract_structured(raw_result: object) -> dict[str, Any]:
if not isinstance(raw_result, tuple) or len(raw_result) != 2:
raise TypeError("unexpected call_tool result shape")
_, structured = raw_result
if not isinstance(structured, dict):
raise TypeError("expected structured dictionary result")
return cast(dict[str, Any], structured)
async def _run() -> tuple[dict[str, Any], ...]:
server = pyro.create_server()
listed = _extract_structured(
await server.call_tool(
"workspace_file_list",
{
"workspace_id": "workspace-123",
"path": "/workspace/src",
"recursive": True,
},
)
)
read = _extract_structured(
await server.call_tool(
"workspace_file_read",
{
"workspace_id": "workspace-123",
"path": "note.txt",
"max_bytes": 4096,
},
)
)
written = _extract_structured(
await server.call_tool(
"workspace_file_write",
{
"workspace_id": "workspace-123",
"path": "src/app.py",
"text": "print('hi')\n",
},
)
)
patched = _extract_structured(
await server.call_tool(
"workspace_patch_apply",
{
"workspace_id": "workspace-123",
"patch": "--- a/note.txt\n+++ b/note.txt\n@@ -1 +1 @@\n-old\n+new\n",
},
)
)
return listed, read, written, patched
listed, read, written, patched = asyncio.run(_run())
assert listed["entries"] == []
assert read["content"] == "hello\n"
assert written["bytes_written"] == len("print('hi')\n".encode("utf-8"))
assert patched["changed"] is True
assert calls == [
(
"list_workspace_files",
{
"workspace_id": "workspace-123",
"path": "/workspace/src",
"recursive": True,
},
),
(
"read_workspace_file",
{
"workspace_id": "workspace-123",
"path": "note.txt",
"max_bytes": 4096,
},
),
(
"write_workspace_file",
{
"workspace_id": "workspace-123",
"path": "src/app.py",
"text": "print('hi')\n",
},
),
(
"apply_workspace_patch",
{
"workspace_id": "workspace-123",
"patch": "--- a/note.txt\n+++ b/note.txt\n@@ -1 +1 @@\n-old\n+new\n",
},
),
]
def test_pyro_create_server_workspace_status_shell_and_service_delegate() -> None:
calls: list[tuple[str, dict[str, Any]]] = []

194
tests/test_cli.py
View file

@ -116,6 +116,37 @@ def test_cli_subcommand_help_includes_examples_and_guidance() -> None:
assert "--output" in workspace_export_help
assert "Export one file or directory from `/workspace`" in workspace_export_help
workspace_file_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "file"
).format_help()
assert "model-native tree inspection and text edits" in workspace_file_help
assert "pyro workspace file read WORKSPACE_ID src/app.py" in workspace_file_help
workspace_file_list_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "list"
).format_help()
assert "--recursive" in workspace_file_list_help
workspace_file_read_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "read"
).format_help()
assert "--max-bytes" in workspace_file_read_help
workspace_file_write_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "write"
).format_help()
assert "--text" in workspace_file_write_help
workspace_patch_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "patch"
).format_help()
assert "Apply add/modify/delete unified text patches" in workspace_patch_help
workspace_patch_apply_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "patch"), "apply"
).format_help()
assert "--patch" in workspace_patch_apply_help
workspace_stop_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "stop"
).format_help()
@ -682,6 +713,169 @@ def test_cli_workspace_export_prints_human_output(
assert "artifact_type=file" in output
def test_cli_workspace_file_commands_print_human_and_json(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def list_workspace_files(
self,
workspace_id: str,
*,
path: str,
recursive: bool,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "/workspace/src"
assert recursive is True
return {
"workspace_id": workspace_id,
"path": path,
"recursive": recursive,
"entries": [
{
"path": "/workspace/src/app.py",
"artifact_type": "file",
"size_bytes": 14,
"link_target": None,
}
],
"execution_mode": "guest_vsock",
}
def read_workspace_file(
self,
workspace_id: str,
path: str,
*,
max_bytes: int,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "src/app.py"
assert max_bytes == 4096
return {
"workspace_id": workspace_id,
"path": "/workspace/src/app.py",
"size_bytes": 14,
"max_bytes": max_bytes,
"content": "print('hi')\n",
"truncated": False,
"execution_mode": "guest_vsock",
}
def write_workspace_file(
self,
workspace_id: str,
path: str,
*,
text: str,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "src/app.py"
assert text == "print('hello')\n"
return {
"workspace_id": workspace_id,
"path": "/workspace/src/app.py",
"size_bytes": len(text.encode("utf-8")),
"bytes_written": len(text.encode("utf-8")),
"execution_mode": "guest_vsock",
}
def apply_workspace_patch(
self,
workspace_id: str,
*,
patch: str,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert patch.startswith("--- a/src/app.py")
return {
"workspace_id": workspace_id,
"changed": True,
"summary": {"total": 1, "added": 0, "modified": 1, "deleted": 0},
"entries": [{"path": "/workspace/src/app.py", "status": "modified"}],
"patch": patch,
"execution_mode": "guest_vsock",
}
class ListParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="list",
workspace_id="workspace-123",
path="/workspace/src",
recursive=True,
json=False,
)
class ReadParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="read",
workspace_id="workspace-123",
path="src/app.py",
max_bytes=4096,
json=True,
)
class WriteParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="write",
workspace_id="workspace-123",
path="src/app.py",
text="print('hello')\n",
json=False,
)
class PatchParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="patch",
workspace_patch_command="apply",
workspace_id="workspace-123",
patch=(
"--- a/src/app.py\n"
"+++ b/src/app.py\n"
"@@ -1 +1 @@\n"
"-print('hi')\n"
"+print('hello')\n"
),
json=False,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: ListParser())
cli.main()
list_output = capsys.readouterr().out
assert "Workspace path: /workspace/src (recursive=yes)" in list_output
assert "/workspace/src/app.py [file]" in list_output
monkeypatch.setattr(cli, "_build_parser", lambda: ReadParser())
cli.main()
read_payload = json.loads(capsys.readouterr().out)
assert read_payload["path"] == "/workspace/src/app.py"
assert read_payload["content"] == "print('hi')\n"
monkeypatch.setattr(cli, "_build_parser", lambda: WriteParser())
cli.main()
write_output = capsys.readouterr().out
assert "[workspace-file-write] workspace_id=workspace-123" in write_output
monkeypatch.setattr(cli, "_build_parser", lambda: PatchParser())
cli.main()
patch_output = capsys.readouterr().out
assert "[workspace-patch] workspace_id=workspace-123 total=1" in patch_output
def test_cli_workspace_stop_and_start_print_human_output(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],

36
tests/test_public_contract.py
View file

@ -24,6 +24,12 @@ from pyro_mcp.contract import (
PUBLIC_CLI_WORKSPACE_DISK_READ_FLAGS,
PUBLIC_CLI_WORKSPACE_EXEC_FLAGS,
PUBLIC_CLI_WORKSPACE_EXPORT_FLAGS,
PUBLIC_CLI_WORKSPACE_FILE_LIST_FLAGS,
PUBLIC_CLI_WORKSPACE_FILE_READ_FLAGS,
PUBLIC_CLI_WORKSPACE_FILE_SUBCOMMANDS,
PUBLIC_CLI_WORKSPACE_FILE_WRITE_FLAGS,
PUBLIC_CLI_WORKSPACE_PATCH_APPLY_FLAGS,
PUBLIC_CLI_WORKSPACE_PATCH_SUBCOMMANDS,
PUBLIC_CLI_WORKSPACE_RESET_FLAGS,
PUBLIC_CLI_WORKSPACE_SERVICE_LIST_FLAGS,
PUBLIC_CLI_WORKSPACE_SERVICE_LOGS_FLAGS,
@ -121,6 +127,36 @@ def test_public_cli_help_lists_commands_and_run_flags() -> None:
).format_help()
for flag in PUBLIC_CLI_WORKSPACE_EXPORT_FLAGS:
assert flag in workspace_export_help_text
workspace_file_help_text = _subparser_choice(
_subparser_choice(parser, "workspace"), "file"
).format_help()
for subcommand_name in PUBLIC_CLI_WORKSPACE_FILE_SUBCOMMANDS:
assert subcommand_name in workspace_file_help_text
workspace_file_list_help_text = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "list"
).format_help()
for flag in PUBLIC_CLI_WORKSPACE_FILE_LIST_FLAGS:
assert flag in workspace_file_list_help_text
workspace_file_read_help_text = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "read"
).format_help()
for flag in PUBLIC_CLI_WORKSPACE_FILE_READ_FLAGS:
assert flag in workspace_file_read_help_text
workspace_file_write_help_text = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "write"
).format_help()
for flag in PUBLIC_CLI_WORKSPACE_FILE_WRITE_FLAGS:
assert flag in workspace_file_write_help_text
workspace_patch_help_text = _subparser_choice(
_subparser_choice(parser, "workspace"), "patch"
).format_help()
for subcommand_name in PUBLIC_CLI_WORKSPACE_PATCH_SUBCOMMANDS:
assert subcommand_name in workspace_patch_help_text
workspace_patch_apply_help_text = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "patch"), "apply"
).format_help()
for flag in PUBLIC_CLI_WORKSPACE_PATCH_APPLY_FLAGS:
assert flag in workspace_patch_apply_help_text
workspace_disk_help_text = _subparser_choice(
_subparser_choice(parser, "workspace"), "disk"
).format_help()

61
tests/test_server.py
View file

@ -36,6 +36,10 @@ def test_create_server_registers_vm_tools(tmp_path: Path) -> None:
assert "workspace_stop" in tool_names
assert "workspace_diff" in tool_names
assert "workspace_export" in tool_names
assert "workspace_file_list" in tool_names
assert "workspace_file_read" in tool_names
assert "workspace_file_write" in tool_names
assert "workspace_patch_apply" in tool_names
assert "workspace_disk_export" in tool_names
assert "workspace_disk_list" in tool_names
assert "workspace_disk_read" in tool_names
@ -247,6 +251,51 @@ def test_workspace_tools_round_trip(tmp_path: Path) -> None:
},
)
)
listed_files = _extract_structured(
await server.call_tool(
"workspace_file_list",
{
"workspace_id": workspace_id,
"path": "/workspace",
"recursive": True,
},
)
)
file_read = _extract_structured(
await server.call_tool(
"workspace_file_read",
{
"workspace_id": workspace_id,
"path": "note.txt",
"max_bytes": 4096,
},
)
)
file_written = _extract_structured(
await server.call_tool(
"workspace_file_write",
{
"workspace_id": workspace_id,
"path": "src/app.py",
"text": "print('hello from file op')\n",
},
)
)
patched = _extract_structured(
await server.call_tool(
"workspace_patch_apply",
{
"workspace_id": workspace_id,
"patch": (
"--- a/note.txt\n"
"+++ b/note.txt\n"
"@@ -1 +1 @@\n"
"-ok\n"
"+patched\n"
),
},
)
)
diffed = _extract_structured(
await server.call_tool("workspace_diff", {"workspace_id": workspace_id})
)
@ -338,6 +387,10 @@ def test_workspace_tools_round_trip(tmp_path: Path) -> None:
created,
synced,
executed,
listed_files,
file_read,
file_written,
patched,
diffed,
snapshot,
snapshots,
@ -357,6 +410,10 @@ def test_workspace_tools_round_trip(tmp_path: Path) -> None:
created,
synced,
executed,
listed_files,
file_read,
file_written,
patched,
diffed,
snapshot,
snapshots,
@ -379,6 +436,10 @@ def test_workspace_tools_round_trip(tmp_path: Path) -> None:
]
assert synced["workspace_sync"]["destination"] == "/workspace/subdir"
assert executed["stdout"] == "[REDACTED]\n"
assert any(entry["path"] == "/workspace/note.txt" for entry in listed_files["entries"])
assert file_read["content"] == "ok\n"
assert file_written["path"] == "/workspace/src/app.py"
assert patched["changed"] is True
assert diffed["changed"] is True
assert snapshot["snapshot"]["snapshot_name"] == "checkpoint"
assert [entry["snapshot_name"] for entry in snapshots["snapshots"]] == [

85
tests/test_vm_manager.py
View file

@ -552,6 +552,91 @@ def test_workspace_diff_and_export_round_trip(tmp_path: Path) -> None:
assert logs["count"] == 0
def test_workspace_file_ops_and_patch_round_trip(tmp_path: Path) -> None:
seed_dir = tmp_path / "seed"
seed_dir.mkdir()
src_dir = seed_dir / "src"
src_dir.mkdir()
(src_dir / "app.py").write_text('print("bug")\n', encoding="utf-8")
manager = VmManager(
backend_name="mock",
base_dir=tmp_path / "vms",
network_manager=TapNetworkManager(enabled=False),
)
workspace_id = str(
manager.create_workspace(
environment="debian:12-base",
allow_host_compat=True,
seed_path=seed_dir,
)["workspace_id"]
)
listing = manager.list_workspace_files(workspace_id, path="src", recursive=True)
assert listing["entries"] == [
{
"path": "/workspace/src/app.py",
"artifact_type": "file",
"size_bytes": 13,
"link_target": None,
}
]
read_payload = manager.read_workspace_file(workspace_id, "src/app.py")
assert read_payload["content"] == 'print("bug")\n'
written = manager.write_workspace_file(
workspace_id,
"src/generated/out.txt",
text="generated\n",
)
assert written["bytes_written"] == 10
patch_payload = manager.apply_workspace_patch(
workspace_id,
patch=(
"--- a/src/app.py\n"
"+++ b/src/app.py\n"
"@@ -1 +1 @@\n"
'-print("bug")\n'
'+print("fixed")\n'
"--- /dev/null\n"
"+++ b/src/new.py\n"
"@@ -0,0 +1 @@\n"
'+print("new")\n'
),
)
assert patch_payload["changed"] is True
assert patch_payload["summary"] == {
"total": 2,
"added": 1,
"modified": 1,
"deleted": 0,
}
executed = manager.exec_workspace(
workspace_id,
command="python3 src/app.py && cat src/new.py && cat src/generated/out.txt",
timeout_seconds=30,
)
assert executed["stdout"] == 'fixed\nprint("new")\ngenerated\n'
diff_payload = manager.diff_workspace(workspace_id)
assert diff_payload["changed"] is True
assert diff_payload["summary"]["added"] == 2
assert diff_payload["summary"]["modified"] == 1
output_path = tmp_path / "exported-app.py"
export_payload = manager.export_workspace(
workspace_id,
path="src/app.py",
output_path=output_path,
)
assert export_payload["artifact_type"] == "file"
assert output_path.read_text(encoding="utf-8") == 'print("fixed")\n'
def test_workspace_export_directory_uses_exact_output_path(tmp_path: Path) -> None:
seed_dir = tmp_path / "seed"
nested_dir = seed_dir / "src"

427
tests/test_workspace_files.py Normal file
View file

@ -0,0 +1,427 @@
from __future__ import annotations
import os
from pathlib import Path
import pytest
from pyro_mcp.workspace_files import (
WORKSPACE_FILE_MAX_BYTES,
WORKSPACE_PATCH_MAX_BYTES,
WorkspacePatchHunk,
WorkspaceTextPatch,
apply_unified_text_patch,
delete_workspace_path,
list_workspace_files,
normalize_workspace_path,
parse_unified_text_patch,
read_workspace_file,
workspace_host_path,
write_workspace_file,
)
def test_workspace_files_list_read_write_and_delete(tmp_path: Path) -> None:
workspace_dir = tmp_path / "workspace"
workspace_dir.mkdir()
(workspace_dir / "src").mkdir()
(workspace_dir / "src" / "note.txt").write_text("hello\n", encoding="utf-8")
os.symlink("note.txt", workspace_dir / "src" / "note-link")
listing = list_workspace_files(
workspace_dir,
workspace_path="/workspace/src",
recursive=True,
)
assert listing.path == "/workspace/src"
assert listing.artifact_type == "directory"
assert [entry.to_payload() for entry in listing.entries] == [
{
"path": "/workspace/src/note-link",
"artifact_type": "symlink",
"size_bytes": 8,
"link_target": "note.txt",
},
{
"path": "/workspace/src/note.txt",
"artifact_type": "file",
"size_bytes": 6,
"link_target": None,
},
]
read_payload = read_workspace_file(
workspace_dir,
workspace_path="/workspace/src/note.txt",
)
assert read_payload.content_bytes == b"hello\n"
written = write_workspace_file(
workspace_dir,
workspace_path="/workspace/generated/out.txt",
text="generated\n",
)
assert written.bytes_written == 10
assert (workspace_dir / "generated" / "out.txt").read_text(encoding="utf-8") == "generated\n"
deleted = delete_workspace_path(
workspace_dir,
workspace_path="/workspace/generated/out.txt",
)
assert deleted.deleted is True
assert not (workspace_dir / "generated" / "out.txt").exists()
def test_workspace_file_read_and_delete_reject_unsupported_paths(tmp_path: Path) -> None:
workspace_dir = tmp_path / "workspace"
workspace_dir.mkdir()
(workspace_dir / "dir").mkdir()
(workspace_dir / "file.txt").write_text("ok\n", encoding="utf-8")
os.symlink("file.txt", workspace_dir / "link.txt")
with pytest.raises(RuntimeError, match="regular files"):
read_workspace_file(workspace_dir, workspace_path="/workspace/dir")
with pytest.raises(RuntimeError, match="regular files"):
read_workspace_file(workspace_dir, workspace_path="/workspace/link.txt")
with pytest.raises(RuntimeError, match="does not support directories"):
delete_workspace_path(workspace_dir, workspace_path="/workspace/dir")
def test_workspace_file_helpers_cover_single_paths_and_path_validation(tmp_path: Path) -> None:
workspace_dir = tmp_path / "workspace"
workspace_dir.mkdir()
(workspace_dir / "note.txt").write_text("hello\n", encoding="utf-8")
listing = list_workspace_files(
workspace_dir,
workspace_path="/workspace/note.txt",
recursive=False,
)
assert listing.path == "/workspace/note.txt"
assert listing.artifact_type == "file"
assert [entry.path for entry in listing.entries] == ["/workspace/note.txt"]
assert normalize_workspace_path("src/app.py") == "/workspace/src/app.py"
assert workspace_host_path(workspace_dir, "src/app.py") == workspace_dir / "src" / "app.py"
with pytest.raises(ValueError, match="must not be empty"):
normalize_workspace_path(" ")
with pytest.raises(ValueError, match="must stay inside /workspace"):
normalize_workspace_path("..")
with pytest.raises(ValueError, match="must stay inside /workspace"):
normalize_workspace_path("/tmp/outside")
with pytest.raises(ValueError, match="must stay inside /workspace"):
normalize_workspace_path("/")
def test_workspace_file_read_limits_and_write_validation(tmp_path: Path) -> None:
workspace_dir = tmp_path / "workspace"
workspace_dir.mkdir()
(workspace_dir / "big.txt").write_text("hello\n", encoding="utf-8")
(workspace_dir / "dir").mkdir()
real_dir = workspace_dir / "real"
real_dir.mkdir()
os.symlink("real", workspace_dir / "linked")
with pytest.raises(ValueError, match="max_bytes must be positive"):
read_workspace_file(workspace_dir, workspace_path="/workspace/big.txt", max_bytes=0)
with pytest.raises(ValueError, match="at most"):
read_workspace_file(
workspace_dir,
workspace_path="/workspace/big.txt",
max_bytes=WORKSPACE_FILE_MAX_BYTES + 1,
)
with pytest.raises(RuntimeError, match="exceeds the maximum supported size"):
read_workspace_file(workspace_dir, workspace_path="/workspace/big.txt", max_bytes=4)
with pytest.raises(RuntimeError, match="regular file targets"):
write_workspace_file(workspace_dir, workspace_path="/workspace/dir", text="nope\n")
with pytest.raises(RuntimeError, match="symlinked parent"):
write_workspace_file(
workspace_dir,
workspace_path="/workspace/linked/out.txt",
text="nope\n",
)
with pytest.raises(ValueError, match="at most"):
write_workspace_file(
workspace_dir,
workspace_path="/workspace/huge.txt",
text="x" * (WORKSPACE_FILE_MAX_BYTES + 1),
)
def test_workspace_file_list_rejects_unsupported_filesystem_types(tmp_path: Path) -> None:
workspace_dir = tmp_path / "workspace"
workspace_dir.mkdir()
fifo_path = workspace_dir / "pipe"
os.mkfifo(fifo_path)
with pytest.raises(RuntimeError, match="unsupported workspace path type"):
list_workspace_files(workspace_dir, workspace_path="/workspace", recursive=True)
def test_parse_and_apply_unified_text_patch_round_trip() -> None:
patch_text = """--- a/src/app.py
+++ b/src/app.py
@@ -1,2 +1,3 @@
print("old")
-print("bug")
+print("fixed")
+print("done")
--- /dev/null
+++ b/src/new.py
@@ -0,0 +1 @@
+print("new")
--- a/src/remove.py
+++ /dev/null
@@ -1 +0,0 @@
-print("remove")
"""
patches = parse_unified_text_patch(patch_text)
assert [(item.path, item.status) for item in patches] == [
("/workspace/src/app.py", "modified"),
("/workspace/src/new.py", "added"),
("/workspace/src/remove.py", "deleted"),
]
modified = apply_unified_text_patch(
path="/workspace/src/app.py",
patch=patches[0],
before_text='print("old")\nprint("bug")\n',
)
added = apply_unified_text_patch(
path="/workspace/src/new.py",
patch=patches[1],
before_text=None,
)
deleted = apply_unified_text_patch(
path="/workspace/src/remove.py",
patch=patches[2],
before_text='print("remove")\n',
)
assert modified == 'print("old")\nprint("fixed")\nprint("done")\n'
assert added == 'print("new")\n'
assert deleted is None
def test_parse_unified_text_patch_rejects_unsupported_features() -> None:
with pytest.raises(ValueError, match="unsupported patch feature"):
parse_unified_text_patch(
"""diff --git a/file.txt b/file.txt
old mode 100644
--- a/file.txt
+++ b/file.txt
@@ -1 +1 @@
-old
+new
"""
)
with pytest.raises(ValueError, match="rename and copy patches are not supported"):
parse_unified_text_patch(
"""--- a/old.txt
+++ b/new.txt
@@ -1 +1 @@
-old
+new
"""
)
def test_parse_unified_text_patch_handles_git_headers_and_validation_errors() -> None:
parsed = parse_unified_text_patch(
"""
diff --git a/file.txt b/file.txt
index 1234567..89abcde 100644
--- /workspace/file.txt
+++ /workspace/file.txt
@@ -1 +1 @@
-old
+new
\\ No newline at end of file
"""
)
assert parsed[0].path == "/workspace/file.txt"
with pytest.raises(ValueError, match="must not be empty"):
parse_unified_text_patch("")
with pytest.raises(ValueError, match="invalid patch header"):
parse_unified_text_patch("oops\n")
with pytest.raises(ValueError, match="missing '\\+\\+\\+' header"):
parse_unified_text_patch("--- a/file.txt\n")
with pytest.raises(ValueError, match="has no hunks"):
parse_unified_text_patch("--- a/file.txt\n+++ b/file.txt\n")
with pytest.raises(ValueError, match="line counts do not match"):
parse_unified_text_patch(
"""--- a/file.txt
+++ b/file.txt
@@ -1,2 +1,1 @@
-old
+new
"""
)
with pytest.raises(ValueError, match="invalid patch hunk line"):
parse_unified_text_patch(
"""--- a/file.txt
+++ b/file.txt
@@ -1 +1 @@
?bad
"""
)
with pytest.raises(ValueError, match="at most"):
parse_unified_text_patch("x" * (WORKSPACE_PATCH_MAX_BYTES + 1))
with pytest.raises(ValueError, match="patch must target a workspace path"):
parse_unified_text_patch("--- /dev/null\n+++ /dev/null\n")
with pytest.raises(ValueError, match="patch must contain at least one file change"):
parse_unified_text_patch(
"""diff --git a/file.txt b/file.txt
index 1234567..89abcde 100644
"""
)
with pytest.raises(ValueError, match="unsupported patch feature"):
parse_unified_text_patch(
"""--- a/file.txt
+++ b/file.txt
new mode 100644
"""
)
with pytest.raises(ValueError, match="invalid patch hunk header"):
parse_unified_text_patch(
"""--- a/file.txt
+++ b/file.txt
@@ -1 +1 @@
old
@@bogus
"""
)
parsed = parse_unified_text_patch(
"""--- a/file.txt
+++ b/file.txt
index 1234567..89abcde 100644
@@ -1 +1 @@
-old
+new
@@ -3 +3 @@
-before
+after
"""
)
assert len(parsed[0].hunks) == 2
def test_apply_unified_text_patch_rejects_context_mismatches() -> None:
patch = parse_unified_text_patch(
"""--- a/file.txt
+++ b/file.txt
@@ -1 +1 @@
-before
+after
"""
)[0]
with pytest.raises(RuntimeError, match="patch context does not match"):
apply_unified_text_patch(
path="/workspace/file.txt",
patch=patch,
before_text="different\n",
)
with pytest.raises(RuntimeError, match="patch context does not match"):
apply_unified_text_patch(
path="/workspace/file.txt",
patch=WorkspaceTextPatch(
path="/workspace/file.txt",
status="modified",
hunks=[
WorkspacePatchHunk(
old_start=1,
old_count=1,
new_start=1,
new_count=1,
lines=[" same\n"],
)
],
),
before_text="",
)
def test_apply_unified_text_patch_rejects_range_prefix_delete_and_size_errors() -> None:
with pytest.raises(RuntimeError, match="out of range"):
apply_unified_text_patch(
path="/workspace/file.txt",
patch=WorkspaceTextPatch(
path="/workspace/file.txt",
status="modified",
hunks=[
WorkspacePatchHunk(
old_start=3,
old_count=1,
new_start=3,
new_count=1,
lines=["-old\n", "+new\n"],
)
],
),
before_text="old\n",
)
with pytest.raises(RuntimeError, match="invalid patch line prefix"):
apply_unified_text_patch(
path="/workspace/file.txt",
patch=WorkspaceTextPatch(
path="/workspace/file.txt",
status="modified",
hunks=[
WorkspacePatchHunk(
old_start=1,
old_count=0,
new_start=1,
new_count=0,
lines=["?bad\n"],
)
],
),
before_text="",
)
with pytest.raises(RuntimeError, match="delete patch did not remove all content"):
apply_unified_text_patch(
path="/workspace/file.txt",
patch=WorkspaceTextPatch(
path="/workspace/file.txt",
status="deleted",
hunks=[
WorkspacePatchHunk(
old_start=1,
old_count=1,
new_start=1,
new_count=0,
lines=["-first\n"],
)
],
),
before_text="first\nsecond\n",
)
huge_payload = "x" * (WORKSPACE_FILE_MAX_BYTES + 1)
with pytest.raises(RuntimeError, match="exceeds the maximum supported size"):
apply_unified_text_patch(
path="/workspace/file.txt",
patch=WorkspaceTextPatch(
path="/workspace/file.txt",
status="added",
hunks=[
WorkspacePatchHunk(
old_start=0,
old_count=0,
new_start=1,
new_count=1,
lines=[f"+{huge_payload}"],
)
],
),
before_text=None,
)