135 lines
4.2 KiB
Python
135 lines
4.2 KiB
Python
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from glas_harness.contract import SandboxHandle
|
|
from glas_harness.transport import ExecutionTransport, TransportError, transport_from_sandbox
|
|
|
|
|
|
def test_local_namespace_transport_requires_pid_and_workspace(tmp_path: Path) -> None:
|
|
sandbox = SandboxHandle(
|
|
sandbox_id="sbx",
|
|
host="localhost",
|
|
reachability={"pid": "4321", "workspace_dir": str(tmp_path)},
|
|
)
|
|
transport = transport_from_sandbox(sandbox)
|
|
|
|
assert transport.kind == "local_namespace"
|
|
assert transport.workspace == str(tmp_path)
|
|
assert transport.command(["git", "status"]) == [
|
|
"nsenter",
|
|
"--target",
|
|
"4321",
|
|
"--mount",
|
|
"--pid",
|
|
"--net",
|
|
"--uts",
|
|
"--ipc",
|
|
"--",
|
|
"sh",
|
|
"-c",
|
|
'cd "$1" && shift && exec "$@"',
|
|
"sh",
|
|
str(tmp_path),
|
|
"git",
|
|
"status",
|
|
]
|
|
|
|
|
|
def test_remote_transport_wraps_command_without_local_shell() -> None:
|
|
sandbox = SandboxHandle(
|
|
sandbox_id="sbx",
|
|
host="sandboxer01",
|
|
reachability={"ssh": "agent@sandboxer01", "remote_dir": "/tmp/sbx"},
|
|
)
|
|
transport = transport_from_sandbox(sandbox)
|
|
|
|
assert transport.kind == "ssh"
|
|
assert transport.command(["git", "-C", "/tmp/sbx", "status"]) == [
|
|
"ssh",
|
|
"agent@sandboxer01",
|
|
"sh -c 'cd \"$1\" && shift && exec \"$@\"' sh /tmp/sbx git -C /tmp/sbx status",
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"reachability",
|
|
[
|
|
{},
|
|
{"workspace_dir": "/tmp/ws"},
|
|
{"pid": "12"},
|
|
{"ssh": "agent@host"},
|
|
{"remote_dir": "/tmp/ws"},
|
|
{
|
|
"pid": "12",
|
|
"workspace_dir": "/tmp/ws",
|
|
"ssh": "agent@host",
|
|
"remote_dir": "/tmp/ws",
|
|
},
|
|
],
|
|
)
|
|
def test_incomplete_or_ambiguous_reachability_fails_closed(reachability) -> None:
|
|
with pytest.raises(TransportError):
|
|
transport_from_sandbox(
|
|
SandboxHandle(sandbox_id="sbx", host="localhost", reachability=reachability)
|
|
)
|
|
|
|
|
|
def test_local_task_file_is_private_and_removable(tmp_path: Path) -> None:
|
|
(tmp_path / ".git").mkdir()
|
|
transport = ExecutionTransport(
|
|
kind="local_namespace", workspace=str(tmp_path), pid=123
|
|
)
|
|
task_path = transport.write_task_file({"title": "bounded", "description": "safe"})
|
|
path = Path(task_path)
|
|
|
|
assert path.parent == tmp_path / ".git"
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
assert json.loads(path.read_text())["title"] == "bounded"
|
|
|
|
transport.remove_file(task_path)
|
|
assert not path.exists()
|
|
|
|
|
|
def test_remote_task_file_uses_ssh_stdin_and_cleanup() -> None:
|
|
transport = ExecutionTransport(
|
|
kind="ssh", workspace="/tmp/sbx", ssh_target="agent@sandboxer01"
|
|
)
|
|
completed = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("glas_harness.transport.subprocess.run", return_value=completed) as run:
|
|
task_path = transport.write_task_file({"title": "bounded"})
|
|
transport.remove_file(task_path)
|
|
|
|
create = run.call_args_list[0]
|
|
assert create.args[0][:2] == ["ssh", "agent@sandboxer01"]
|
|
assert create.kwargs["input"] == '{"title": "bounded"}'
|
|
assert "cat >" in create.args[0][2]
|
|
assert "rm -f" in run.call_args_list[1].args[0][2]
|
|
|
|
|
|
def test_resolve_executable_fails_inside_transport() -> None:
|
|
transport = ExecutionTransport(
|
|
kind="ssh", workspace="/tmp/sbx", ssh_target="agent@sandboxer01"
|
|
)
|
|
with patch.object(
|
|
ExecutionTransport,
|
|
"run",
|
|
return_value=MagicMock(returncode=127, stdout="", stderr="not found"),
|
|
):
|
|
with pytest.raises(TransportError, match="not installed inside"):
|
|
transport.resolve_executable("rein-aharness")
|
|
|
|
|
|
def test_git_head_fails_closed_on_transport_error() -> None:
|
|
transport = ExecutionTransport(
|
|
kind="ssh", workspace="/tmp/sbx", ssh_target="agent@sandboxer01"
|
|
)
|
|
with patch.object(
|
|
ExecutionTransport,
|
|
"run",
|
|
return_value=MagicMock(returncode=1, stdout="", stderr="permission denied"),
|
|
):
|
|
with pytest.raises(TransportError, match="inside the selected sandbox"):
|
|
transport.git_head()
|