nisse/nisse-dev/tests/test_tools.py

import os
import time
from pathlib import Path

import pytest

from nisse_dev.tools import EXECUTOR_TOOLS, MAX_OUTPUT_CHARS, Workspace


@pytest.fixture
def ws(tmp_path: Path) -> Workspace:
    return Workspace(tmp_path / "work", timeout_s=10)


def test_manifest_is_small_and_flat() -> None:
    # A15: fewer tools, flatter arguments for the local executor.
    names = [t["function"]["name"] for t in EXECUTOR_TOOLS]
    assert names == ["read_file", "write_file", "list_dir", "run_command"]
    for tool in EXECUTOR_TOOLS:
        props = tool["function"]["parameters"]["properties"]
        assert all(p["type"] == "string" for p in props.values())


def test_write_then_read_round_trip(ws: Workspace) -> None:
    assert ws.write_file("src/app.py", "print('hi')\n").success
    outcome = ws.read_file("src/app.py")
    assert outcome.success
    assert outcome.result == "print('hi')\n"


def test_relative_paths_land_in_the_workspace(ws: Workspace) -> None:
    ws.write_file("notes.txt", "x")
    assert (ws.root / "notes.txt").exists()


def test_read_missing_file_is_a_failed_result_not_an_exception(
    ws: Workspace,
) -> None:
    outcome = ws.read_file("nope.txt")
    assert not outcome.success
    assert "could not read" in outcome.result


def test_list_dir(ws: Workspace) -> None:
    ws.write_file("a.txt", "a")
    ws.write_file("sub/b.txt", "b")
    outcome = ws.list_dir(".")
    assert outcome.success
    assert outcome.result.splitlines() == ["a.txt", "sub/"]


def test_list_empty_dir(ws: Workspace) -> None:
    assert ws.list_dir(".").result == "(empty)"


def test_run_command_success_reports_exit_code_and_stdout(ws: Workspace) -> None:
    outcome = ws.run_command("echo hello")
    assert outcome.success
    assert "exit code: 0" in outcome.result
    assert "hello" in outcome.result


def test_nonzero_exit_is_a_result_not_an_error(ws: Workspace) -> None:
    outcome = ws.run_command("exit 3")
    assert not outcome.success  # feeds the A12 failure count
    assert "exit code: 3" in outcome.result


def test_command_runs_in_the_workspace(ws: Workspace) -> None:
    ws.write_file("marker.txt", "here")
    assert "marker.txt" in ws.run_command("ls").result


def test_command_timeout_is_reported(tmp_path: Path) -> None:
    ws = Workspace(tmp_path / "work", timeout_s=1)
    outcome = ws.run_command("sleep 30")
    assert not outcome.success
    assert "timed out" in outcome.result


def _is_running(pid: int) -> bool:
    """os.kill(pid, 0) is not enough: a killed orphan stays a zombie until
    something reaps it, and no init does that inside a CI container. Read the
    process state instead."""
    try:
        stat = Path(f"/proc/{pid}/stat").read_text()
    except FileNotFoundError:
        return False
    # State is the first field after the parenthesised comm.
    return stat.rsplit(")", 1)[1].split()[0] != "Z"


def test_timeout_kills_the_whole_process_group(tmp_path: Path) -> None:
    ws = Workspace(tmp_path / "work", timeout_s=1)
    pidfile = ws.root / "child.pid"
    # A backgrounded grandchild must not outlive the timeout.
    ws.run_command(f"(sleep 30 & echo $! > {pidfile}); sleep 30")
    assert pidfile.exists()
    pid = int(pidfile.read_text().strip())
    assert _is_running(os.getpid()), "liveness check is broken, not the kill"
    # SIGKILL to a process group is asynchronous, and the grandchild is an
    # orphan by now, so its exit is not synchronised with our return. Wait
    # for death rather than assuming it is instantaneous.
    deadline = time.monotonic() + 5
    while _is_running(pid) and time.monotonic() < deadline:
        time.sleep(0.05)
    assert not _is_running(pid)


def test_agent_credentials_are_not_inherited(
    ws: Workspace, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("NISSE_LLM_TOKEN", "sk-secret")
    monkeypatch.setenv("PATH", os.environ["PATH"])
    outcome = ws.run_command("echo [${NISSE_LLM_TOKEN}]")
    assert "sk-secret" not in outcome.result
    assert "[]" in outcome.result


def test_stdin_is_closed_so_prompts_fail_fast(ws: Workspace) -> None:
    outcome = ws.run_command("read line && echo got")
    assert not outcome.success
    assert "got" not in outcome.result


def test_long_output_is_truncated(ws: Workspace) -> None:
    outcome = ws.run_command("printf 'x%.0s' $(seq 1 20000)")
    assert len(outcome.result) < MAX_OUTPUT_CHARS + 200
    assert "characters omitted" in outcome.result


def test_dispatch_unknown_tool_and_missing_argument(ws: Workspace) -> None:
    assert not ws.dispatch("fly", {}).success
    missing = ws.dispatch("read_file", {})
    assert not missing.success
    assert "missing argument" in missing.result


def test_write_outside_the_workspace_is_refused(ws: Workspace, tmp_path: Path) -> None:
    # Not sandboxing: a file written outside the repo is invisible to git, so
    # the work would vanish while the model truthfully reported success.
    # Models really do invent absolute paths like /workspace.
    stray = tmp_path / "elsewhere" / "slugify.py"
    outcome = ws.write_file(str(stray), "x")
    assert not outcome.success
    assert "outside the workspace" in outcome.result
    assert str(ws.root) in outcome.result  # tells it where to write instead
    assert not stray.exists()


def test_escaping_via_parent_traversal_is_refused(ws: Workspace) -> None:
    outcome = ws.write_file("../escaped.txt", "x")
    assert not outcome.success
    assert not (ws.root.parent / "escaped.txt").exists()


def test_describe_names_the_root(ws: Workspace) -> None:
    assert str(ws.root) in ws.describe()