dongarandClaude Opus 5
999a38a705
os.kill(pid, 0) succeeds on a zombie, and a killed orphan stays a zombie until reaped — which never happens inside a CI container with no init. The test read that as "still alive" and failed there while passing locally. It now reads the process state from /proc, with a control assertion so a broken liveness check cannot pass silently. The group kill itself was correct. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
import os
|
|
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"
|
|
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 "truncated" 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
|