glas-harness/tests/test_transport.py

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)},
    )
    sandbox._owner_execute = MagicMock(return_value=MagicMock(
        timed_out=False, output_truncated=False, exit_code=0, stdout="ok", stderr=""
    ))
    transport = transport_from_sandbox(sandbox)
    assert transport.kind == "local_namespace"
    assert transport.workspace == str(tmp_path)
    with patch("glas_harness.transport.subprocess.run") as run:
        assert transport.run(["git", "status"], timeout=30).stdout == "ok"
    run.assert_not_called()
    sandbox._owner_execute.assert_called_once_with(["git", "status"], None, 30)
    with pytest.raises(TransportError, match="owner execution"):
        transport.command(["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(
    "ssh_target",
    ["-v", "-oProxyCommand", "agent@-host", "-agent@host", "agent@host extra"],
)
def test_remote_transport_rejects_ssh_option_and_invalid_targets(ssh_target) -> None:
    sandbox = SandboxHandle(
        sandbox_id="sbx",
        host="sandboxer01",
        reachability={"ssh": ssh_target, "remote_dir": "/tmp/sbx"},
    )

    with pytest.raises(TransportError, match="single non-option host"):
        transport_from_sandbox(sandbox)


def test_direct_remote_transport_cannot_bypass_ssh_target_validation() -> None:
    transport = ExecutionTransport(kind="ssh", workspace="/tmp/sbx", ssh_target="-v")

    with pytest.raises(TransportError, match="single non-option host"):
        transport.command(["true"])


@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_uses_owner_stdin_and_cleanup(tmp_path: Path) -> None:
    owner = MagicMock(return_value=MagicMock(
        timed_out=False, output_truncated=False, exit_code=0, stdout="", stderr=""
    ))
    transport = ExecutionTransport(
        kind="local_namespace", workspace=str(tmp_path), pid=123, owner_execute=owner,
        timeout_seconds=7,
    )
    task_path = transport.write_task_file({"title": "private prompt"})
    command, stdin, timeout = owner.call_args.args
    assert "umask 077; set -C; cat" in command[2]
    assert "private prompt" not in str(command)
    assert json.loads(stdin) == {"title": "private prompt"}
    assert timeout == 7
    assert Path(task_path).parent == tmp_path / ".git"
    assert not Path(task_path).exists()  # No host write, even with a fake owner.
    transport.remove_file(task_path)
    owner.assert_called_with(["rm", "-f", "--", task_path], None, 7)


def test_local_descriptor_without_owner_fails_closed():
    with pytest.raises(TransportError, match="owner execution binding"):
        transport_from_sandbox(SandboxHandle(
            sandbox_id="sbx", host="localhost",
            reachability={"pid": "123", "workspace_dir": "/tmp/ws"},
        ))


@pytest.mark.parametrize("timed_out,truncated", [(True, False), (False, True)])
def test_owner_timeout_and_truncation_fail_closed(timed_out, truncated):
    import subprocess
    owner = MagicMock(return_value=MagicMock(
        timed_out=timed_out, output_truncated=truncated
    ))
    transport = ExecutionTransport(
        kind="local_namespace", workspace="/tmp/ws", owner_execute=owner,
        timeout_seconds=3,
    )
    with pytest.raises(subprocess.TimeoutExpired if timed_out else TransportError):
        transport.run(["true"], timeout=30)
    owner.assert_called_once_with(["true"], None, 3)


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][:3] == ["ssh", "--", "agent@sandboxer01"]
    assert create.kwargs["input"] == '{"title": "bounded"}'
    assert "cat >" in create.args[0][3]
    assert "rm -f" in run.call_args_list[1].args[0][3]


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()