rein-openweights/tests/test_tools.py

import subprocess

import pytest

from rein_openweights.tools import (
    ToolPolicyError,
    call_tool,
    edit_file,
    git_add_commit,
    git_diff,
    git_log,
    git_status,
    glob_files,
    grep_files,
    openai_tool_schemas,
    read_file,
    write_file,
)


def _init_repo(tmp_path):
    subprocess.run(["git", "init", "-q"], cwd=tmp_path, check=True)
    subprocess.run(["git", "config", "user.email", "t@example.com"], cwd=tmp_path, check=True)
    subprocess.run(["git", "config", "user.name", "t"], cwd=tmp_path, check=True)
    subprocess.run(["git", "commit", "-q", "--allow-empty", "-m", "init"], cwd=tmp_path, check=True)
    return tmp_path


def test_write_then_read_file(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "a.txt", "hello")
    assert read_file(repo, "a.txt") == "hello"


def test_edit_file_replaces_first_occurrence(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "a.txt", "foo foo")
    edit_file(repo, "a.txt", "foo", "bar")
    assert read_file(repo, "a.txt") == "bar foo"


def test_edit_file_missing_old_raises(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "a.txt", "hello")
    with pytest.raises(ToolPolicyError, match="old string not found"):
        edit_file(repo, "a.txt", "nope", "x")


def test_path_traversal_is_blocked(tmp_path):
    repo = _init_repo(tmp_path)
    with pytest.raises(ToolPolicyError, match="escapes repo root"):
        read_file(repo, "../outside.txt")


def test_glob_files(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "src/a.py", "1")
    write_file(repo, "src/b.py", "2")
    write_file(repo, "src/c.txt", "3")
    result = glob_files(repo, "**/*.py")
    assert "src/a.py" in result
    assert "src/b.py" in result
    assert "src/c.txt" not in result


def test_grep_files(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "a.txt", "hello world\nneedle here\n")
    result = grep_files(repo, "needle")
    assert "a.txt:2:needle here" in result


def test_grep_files_no_matches(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "a.txt", "hello")
    assert grep_files(repo, "zzz") == "(no matches)"


def test_git_status_diff_log_and_commit(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "a.txt", "hello")
    assert "a.txt" in git_status(repo)
    git_add_commit(repo, "add a.txt")
    assert git_status(repo) == "(clean)"
    assert "add a.txt" in git_log(repo)


def test_git_diff_after_edit(tmp_path):
    repo = _init_repo(tmp_path)
    write_file(repo, "a.txt", "hello")
    git_add_commit(repo, "add a.txt")
    edit_file(repo, "a.txt", "hello", "goodbye")
    assert "goodbye" in git_diff(repo)


def test_call_tool_dispatches_by_name(tmp_path):
    repo = _init_repo(tmp_path)
    call_tool(repo, "write_file", {"path": "a.txt", "content": "x"})
    assert read_file(repo, "a.txt") == "x"


def test_call_tool_unknown_name_raises(tmp_path):
    repo = _init_repo(tmp_path)
    with pytest.raises(ToolPolicyError, match="unknown tool"):
        call_tool(repo, "delete_everything", {})


def test_openai_tool_schemas_cover_all_tools():
    schemas = openai_tool_schemas()
    names = {s["function"]["name"] for s in schemas}
    assert names == {
        "read_file",
        "write_file",
        "edit_file",
        "glob_files",
        "grep_files",
        "git_status",
        "git_diff",
        "git_log",
        "git_add_commit",
    }