smart-org/apps/api/tests/test_engine_tools.py

"""The toolkit's guards: path safety, the git refusal and bounded commands.

These are the checks that run *before* anything reaches the sandbox, so they
are tested as pure functions wherever possible — a path bug must be
reproducible without a container.
"""

from __future__ import annotations

import pytest

from app.application.engine.protocol import Tag, TagName, edit_blocks_of
from app.application.engine.tools import (
    RUN_DEFAULT_TIMEOUT,
    RUN_MAX_TIMEOUT,
    ToolExecutor,
    UnsafePathError,
    safe_join,
    tool_key_for,
)
from tests.engine_fakes import FakeSandbox

ROOT = "/workspace/.worktrees/task-1"


# ---------------------------------------------------------------------------
# Path safety
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "relative",
    ["app/main.py", "./app/main.py", "app/deep/nested/file.py", "a.txt"],
)
def test_a_normal_relative_path_is_allowed(relative):
    assert safe_join(ROOT, relative) == f"{ROOT}/{relative.lstrip('./')}".replace(
        "/./", "/"
    )


@pytest.mark.parametrize(
    "relative",
    ["/etc/passwd", "~/.ssh/id_rsa", "../../etc/passwd", "app/../../secrets", ".git/config"],
)
def test_an_escaping_path_is_refused(relative):
    with pytest.raises(UnsafePathError):
        safe_join(ROOT, relative)


def test_an_empty_path_is_refused():
    with pytest.raises(UnsafePathError):
        safe_join(ROOT, "   ")


def test_the_git_directory_is_never_readable():
    with pytest.raises(UnsafePathError):
        safe_join(ROOT, ".git/objects/ab/cdef")


# ---------------------------------------------------------------------------
# The git refusal
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "command",
    ["git push origin dev", "git remote add origin x", "git config user.email a@b.c"],
)
async def test_git_commands_are_refused_before_the_sandbox(command):
    sandbox = FakeSandbox()
    executor = ToolExecutor(
        sandbox=sandbox, session_external_id="sess-1", root=ROOT
    )
    outcome = await executor.execute(
        Tag(name=TagName.RUN, attrs={"cmd": command}, body="")
    )
    assert outcome.ok is False
    assert outcome.refused is True
    assert "پلتفرم" in outcome.output
    assert sandbox.executed == []


async def test_other_git_commands_are_allowed_through():
    """Only push/remote/config are refused: the platform owns the rest."""
    sandbox = FakeSandbox(commands={"git": []})
    executor = ToolExecutor(
        sandbox=sandbox, session_external_id="sess-1", root=ROOT
    )
    outcome = await executor.execute(
        Tag(name=TagName.RUN, attrs={"cmd": "git status --porcelain -uall"}, body="")
    )
    assert outcome.ok is True
    assert len(sandbox.executed) == 1


# ---------------------------------------------------------------------------
# Bounds
# ---------------------------------------------------------------------------


async def test_run_clamps_the_timeout_to_the_ceiling():
    sandbox = FakeSandbox()
    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
    await executor.execute(
        Tag(name=TagName.RUN, attrs={"cmd": "sleep 1", "timeout": "99999"}, body="")
    )
    assert sandbox.executed[0]["timeout_seconds"] == RUN_MAX_TIMEOUT


async def test_run_defaults_to_a_sane_timeout():
    sandbox = FakeSandbox()
    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
    await executor.execute(Tag(name=TagName.RUN, attrs={"cmd": "ls"}, body=""))
    assert sandbox.executed[0]["timeout_seconds"] == RUN_DEFAULT_TIMEOUT


async def test_a_huge_output_is_truncated_to_its_tail():
    sandbox = FakeSandbox(
        commands={"noisy": [type("R", (), {"exit_code": 0, "stdout": "x" * 30000, "stderr": ""})()]}
    )
    executor = ToolExecutor(
        sandbox=sandbox, session_external_id="sess-1", root=ROOT, max_output_chars=1000
    )
    outcome = await executor.execute(
        Tag(name=TagName.RUN, attrs={"cmd": "noisy"}, body="")
    )
    assert outcome.ok is True
    assert len(outcome.output) <= 1100
    assert outcome.output.endswith("x" * 100)


# ---------------------------------------------------------------------------
# The tools themselves
# ---------------------------------------------------------------------------


async def test_write_then_read_round_trips_inside_the_root():
    sandbox = FakeSandbox()
    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
    written = await executor.execute(
        Tag(name=TagName.WRITE_FILE, attrs={"path": "app/a.py"}, body="print(1)")
    )
    read = await executor.execute(
        Tag(name=TagName.READ_FILE, attrs={"path": "app/a.py"}, body="")
    )
    assert written.ok and read.ok
    assert "print(1)" in read.output
    assert sandbox.writes[0]["path"] == f"{ROOT}/app/a.py"


async def test_list_dir_and_search_run_quoted_commands():
    sandbox = FakeSandbox()
    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
    await executor.execute(Tag(name=TagName.LIST_DIR, attrs={"path": "app"}, body=""))
    await executor.execute(
        Tag(name=TagName.SEARCH, attrs={"pattern": "TODO", "path": "app"}, body="")
    )
    assert sandbox.executed[0]["command"].startswith("ls -la -- ")
    assert "grep -RIn -e TODO" in sandbox.executed[1]["command"]


async def test_a_missing_file_reports_not_found_without_raising():
    sandbox = FakeSandbox()
    executor = ToolExecutor(sandbox=sandbox, session_external_id="sess-1", root=ROOT)
    outcome = await executor.execute(
        Tag(name=TagName.READ_FILE, attrs={"path": "app/none.py"}, body="")
    )
    assert outcome.ok is False
    assert "none.py" in outcome.output


def test_tool_keys_map_onto_the_existing_catalog():
    assert tool_key_for(Tag(name=TagName.READ_FILE, attrs={}, body="")) == (
        "sandbox.read_file"
    )
    assert tool_key_for(Tag(name=TagName.RUN, attrs={}, body="")) == (
        "sandbox.run_command"
    )


def test_edit_blocks_refuse_a_malformed_body():
    tag = Tag(name=TagName.EDIT_FILE, attrs={"path": "a"}, body="just text")
    with pytest.raises(ValueError):
        edit_blocks_of(tag)