KOKO\Mimi
349 lines
15 KiB
Python
349 lines
15 KiB
Python
import os
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from scripts import execute
|
|
from test_execute import run_step, step_runtime
|
|
|
|
|
|
def set_owned_paths(runtime, paths):
|
|
index = runtime.executor._read_phase_index()
|
|
index["steps"][0]["owned_paths"] = paths
|
|
runtime.executor._write_json(runtime.index_file, index)
|
|
|
|
|
|
def link_directory(runtime, link, target):
|
|
if os.name == "nt":
|
|
# Junctions exercise Windows link resolution without symlink privileges.
|
|
runtime.real_run(
|
|
["cmd", "/c", "mklink", "/J", str(link), str(target)],
|
|
cwd=runtime.root, capture_output=True, check=True,
|
|
)
|
|
else:
|
|
link.symlink_to(target, target_is_directory=True)
|
|
|
|
|
|
@pytest.mark.parametrize("paths", [
|
|
None, "src/a.py", [1], [""], ["."], ["src/"], ["src/**"],
|
|
["../a.py"], ["src/../a.py"], ["/a.py"], ["C:/a.py"], ["C:a.py"],
|
|
["\\\\server\\share\\a.py"], ["src/a.py:stream"], [".git/config"],
|
|
["phases/runtime/index.json"], ["phases/other/step0.md"],
|
|
["src/a.py", "src\\a.py"], ["src/a.py\x00"], ["src/a.py."],
|
|
])
|
|
def test_owned_paths_rejects_ambiguous_or_unsafe_entries(step_runtime, paths):
|
|
runtime = step_runtime
|
|
set_owned_paths(runtime, paths)
|
|
with pytest.raises(execute.HarnessError, match="owned_paths"):
|
|
runtime.executor._read_phase_index()
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "nt", reason="Windows path identity")
|
|
def test_owned_paths_rejects_case_alias_duplicates(step_runtime):
|
|
set_owned_paths(step_runtime, ["src/a.py", "SRC/A.PY"])
|
|
with pytest.raises(execute.HarnessError, match="owned_paths"):
|
|
step_runtime.executor._read_phase_index()
|
|
|
|
|
|
def test_missing_pending_ownership_stops_before_branch_or_state_changes(step_runtime):
|
|
runtime = step_runtime
|
|
index = runtime.executor._read_phase_index()
|
|
del index["steps"][0]["owned_paths"]
|
|
runtime.executor._write_json(runtime.index_file, index)
|
|
original = runtime.index_file.read_bytes()
|
|
branch = runtime.git("symbolic-ref", "HEAD")
|
|
with pytest.raises(execute.HarnessError, match="owned_paths"):
|
|
runtime.executor.run()
|
|
assert runtime.calls == []
|
|
assert runtime.index_file.read_bytes() == original
|
|
assert runtime.git("symbolic-ref", "HEAD") == branch
|
|
|
|
|
|
def test_completed_history_without_ownership_remains_readable(step_runtime):
|
|
runtime = step_runtime
|
|
index = runtime.executor._read_phase_index()
|
|
index["steps"][0] = {"step": 0, "name": "runtime", "status": "completed", "summary": "historical"}
|
|
runtime.executor._write_json(runtime.index_file, index)
|
|
assert execute.StepExecutor("runtime")._read_phase_index()["steps"][0]["summary"] == "historical"
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["src", "escape/file.py"])
|
|
def test_pending_ownership_rejects_directory_or_link_escape(step_runtime, path):
|
|
runtime = step_runtime
|
|
if path == "src":
|
|
(runtime.root / "src").mkdir()
|
|
else:
|
|
outside = runtime.root.parent / (runtime.root.name + "-outside")
|
|
outside.mkdir()
|
|
link_directory(runtime, runtime.root / "escape", outside)
|
|
set_owned_paths(runtime, [path])
|
|
with pytest.raises(execute.HarnessError, match="owned_paths"):
|
|
runtime.executor._preflight_repository()
|
|
assert runtime.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("history", [None, {}, ["bad"], [{"attempt": 1}]])
|
|
def test_malformed_attempt_history_is_rejected_without_erasure(step_runtime, history):
|
|
runtime = step_runtime
|
|
index = runtime.executor._read_phase_index()
|
|
index["steps"][0]["attempt_history"] = history
|
|
runtime.executor._write_json(runtime.index_file, index)
|
|
original = runtime.index_file.read_bytes()
|
|
with pytest.raises(execute.HarnessError, match="attempt_history"):
|
|
runtime.executor._read_phase_index()
|
|
assert runtime.index_file.read_bytes() == original
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("owned_paths", ["unassigned.md"]), ("attempt_history", [])])
|
|
def test_agent_cannot_change_executor_or_planning_metadata(step_runtime, field, value):
|
|
runtime = step_runtime
|
|
|
|
def tamper(runtime):
|
|
index = runtime.executor._read_phase_index()
|
|
index["steps"][0][field] = value
|
|
runtime.executor._write_json(runtime.index_file, index)
|
|
return {"status": "completed"}
|
|
|
|
runtime.responses = [tamper]
|
|
with pytest.raises(execute.HarnessError, match="immutable metadata"):
|
|
run_step(runtime)
|
|
current = runtime.executor._read_phase_index()["steps"][0]
|
|
assert current["owned_paths"] == []
|
|
assert "attempt_history" not in current
|
|
assert current["status"] == "error"
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("reported_status", []), ("classification", []), ("exit_code", True), ("attempt", True)])
|
|
def test_history_record_types_fail_closed(step_runtime, field, value):
|
|
runtime = step_runtime
|
|
runtime.responses = [{"status": "error"}]
|
|
with pytest.raises(SystemExit):
|
|
run_step(runtime)
|
|
index = runtime.executor._read_phase_index()
|
|
index["steps"][0]["attempt_history"][0][field] = value
|
|
runtime.executor._write_json(runtime.index_file, index)
|
|
with pytest.raises(execute.HarnessError, match="attempt_history"):
|
|
runtime.executor._read_phase_index()
|
|
|
|
|
|
def test_all_pending_steps_need_ownership_before_execution(step_runtime):
|
|
runtime = step_runtime
|
|
index = runtime.executor._read_phase_index()
|
|
index["steps"].append({"step": 1, "name": "later", "status": "pending"})
|
|
(runtime.index_file.parent / "step1.md").write_text("Later Step", encoding="utf-8")
|
|
runtime.executor._write_json(runtime.index_file, index)
|
|
executor = execute.StepExecutor("runtime")
|
|
with pytest.raises(execute.HarnessError, match="Step 1 requires an owned_paths"):
|
|
executor.run()
|
|
assert runtime.calls == []
|
|
assert "created_at" not in executor._read_phase_index()
|
|
|
|
|
|
def test_metadata_commit_failure_preserves_implementation_commit_and_reason(step_runtime, monkeypatch):
|
|
runtime = step_runtime
|
|
set_owned_paths(runtime, ["approved.py"])
|
|
|
|
def change(runtime):
|
|
(runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
|
|
return {"status": "completed"}
|
|
|
|
original_run_git = runtime.executor._run_git
|
|
|
|
def fail_metadata_commit(*args):
|
|
if args[:2] == ("commit", "-m") and args[-1].startswith("chore("):
|
|
return subprocess.CompletedProcess(args, 1, stdout="", stderr="metadata commit failed")
|
|
return original_run_git(*args)
|
|
|
|
monkeypatch.setattr(runtime.executor, "_run_git", fail_metadata_commit)
|
|
runtime.responses = [change]
|
|
with pytest.raises(SystemExit) as stopped:
|
|
run_step(runtime)
|
|
assert stopped.value.code == 1
|
|
current = runtime.executor._read_phase_index()["steps"][0]
|
|
assert current["status"] == "error"
|
|
assert current["attempt_history"][-1]["reason"] == current["error_message"]
|
|
assert current["attempt_history"][-1]["classification"] == "commit-error"
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "2"
|
|
assert runtime.git("show", "HEAD:approved.py") == "approved\n"
|
|
assert runtime.executor._read_top_index()["phases"][0]["status"] == "error"
|
|
|
|
|
|
@pytest.mark.parametrize("staged", [False, True])
|
|
@pytest.mark.parametrize("result", ["completed", "timeout"])
|
|
def test_unassigned_changes_stop_commit_and_retry_without_reverting_files(step_runtime, staged, result):
|
|
runtime = step_runtime
|
|
set_owned_paths(runtime, ["approved.py"])
|
|
|
|
def change(runtime):
|
|
(runtime.root / "approved.py").write_text("approved change\n", encoding="utf-8")
|
|
(runtime.root / "unassigned.md").write_text("user change\n", encoding="utf-8")
|
|
if staged:
|
|
runtime.git("add", "--", "unassigned.md")
|
|
if result == "timeout":
|
|
return subprocess.TimeoutExpired("codex", 1800)
|
|
return {"status": "completed"}
|
|
|
|
runtime.responses = [change, {"status": "completed"}]
|
|
with pytest.raises(SystemExit) as stopped:
|
|
run_step(runtime)
|
|
assert stopped.value.code == 1
|
|
assert len(runtime.calls) == 1
|
|
current = runtime.executor._read_phase_index()["steps"][0]
|
|
assert current["status"] == "error"
|
|
assert "unassigned.md" in current["error_message"]
|
|
assert current["attempt_history"][0]["classification"] == "ownership-violation"
|
|
assert "unassigned.md" in current["attempt_history"][0]["reason"]
|
|
assert (runtime.root / "unassigned.md").read_text(encoding="utf-8") == "user change\n"
|
|
assert (runtime.root / "approved.py").read_text(encoding="utf-8") == "approved change\n"
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
|
|
assert runtime.git("diff", "--cached", "--name-only").strip() == ("unassigned.md" if staged else "")
|
|
|
|
|
|
def test_empty_ownership_allows_only_metadata_commit(step_runtime):
|
|
runtime = step_runtime
|
|
runtime.responses = [{"status": "completed"}]
|
|
assert run_step(runtime)
|
|
assert runtime.git("log", "-1", "--format=%s").strip() == "chore(runtime): step 0 output"
|
|
assert runtime.git("diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD").strip() == "phases/runtime/index.json"
|
|
|
|
|
|
def test_owned_creation_modification_deletion_and_rename_commit_separately(step_runtime):
|
|
runtime = step_runtime
|
|
for name in ("modify.py", "delete.py", "old.py"):
|
|
(runtime.root / name).write_text("original content\n", encoding="utf-8")
|
|
runtime.git("add", "--", "modify.py", "delete.py", "old.py")
|
|
runtime.git("commit", "--quiet", "-m", "test: seed tracked files")
|
|
set_owned_paths(runtime, ["created.py", "modify.py", "delete.py", "old.py", "new.py"])
|
|
|
|
def change(runtime):
|
|
(runtime.root / "created.py").write_text("created\n", encoding="utf-8")
|
|
(runtime.root / "modify.py").write_text("modified\n", encoding="utf-8")
|
|
(runtime.root / "delete.py").unlink()
|
|
runtime.git("mv", "--", "old.py", "new.py")
|
|
return {"status": "completed"}
|
|
|
|
runtime.responses = [change]
|
|
assert run_step(runtime)
|
|
assert set(runtime.git("diff-tree", "--no-commit-id", "--no-renames", "--name-only", "-r", "HEAD~1").splitlines()) == {"created.py", "modify.py", "delete.py", "old.py", "new.py"}
|
|
assert runtime.git("diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD").strip() == "phases/runtime/index.json"
|
|
assert runtime.git("status", "--porcelain").strip() == ""
|
|
|
|
|
|
@pytest.mark.parametrize("owned", [["old.py"], ["new.py"]])
|
|
def test_rename_requires_both_endpoints(step_runtime, owned):
|
|
runtime = step_runtime
|
|
(runtime.root / "old.py").write_text("content\n", encoding="utf-8")
|
|
runtime.git("add", "--", "old.py")
|
|
runtime.git("commit", "--quiet", "-m", "test: seed rename")
|
|
set_owned_paths(runtime, owned)
|
|
|
|
def rename(runtime):
|
|
runtime.git("mv", "--", "old.py", "new.py")
|
|
return {"status": "completed"}
|
|
|
|
runtime.responses = [rename]
|
|
with pytest.raises(SystemExit) as stopped:
|
|
run_step(runtime)
|
|
assert stopped.value.code == 1
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "2"
|
|
assert (runtime.root / "new.py").is_file()
|
|
|
|
|
|
@pytest.mark.parametrize("injected_path", ["unassigned.md", "phases/runtime/index.json"])
|
|
def test_new_unassigned_staged_path_is_rejected_before_commit(step_runtime, monkeypatch, injected_path):
|
|
runtime = step_runtime
|
|
set_owned_paths(runtime, ["approved.py"])
|
|
|
|
def change(runtime):
|
|
(runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
|
|
return {"status": "completed"}
|
|
|
|
original_run_git = runtime.executor._run_git
|
|
|
|
def inject_during_add(*args):
|
|
result = original_run_git(*args)
|
|
if "add" in args:
|
|
if injected_path == "unassigned.md":
|
|
(runtime.root / injected_path).write_text("late change\n", encoding="utf-8")
|
|
runtime.git("add", "--", injected_path)
|
|
return result
|
|
|
|
monkeypatch.setattr(runtime.executor, "_run_git", inject_during_add)
|
|
runtime.responses = [change]
|
|
with pytest.raises(SystemExit) as stopped:
|
|
run_step(runtime)
|
|
assert stopped.value.code == 1
|
|
assert injected_path in runtime.executor._read_phase_index()["steps"][0]["error_message"]
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
|
|
assert (runtime.root / injected_path).is_file()
|
|
|
|
|
|
def test_owned_path_replaced_by_external_junction_is_rejected(step_runtime):
|
|
runtime = step_runtime
|
|
set_owned_paths(runtime, ["src/approved.py"])
|
|
outside = runtime.root.parent / (runtime.root.name + "-outside")
|
|
outside.mkdir()
|
|
(outside / "approved.py").write_text("outside data\n", encoding="utf-8")
|
|
|
|
def substitute(runtime):
|
|
link_directory(runtime, runtime.root / "src", outside)
|
|
return {"status": "completed"}
|
|
|
|
runtime.responses = [substitute]
|
|
with pytest.raises(SystemExit) as stopped:
|
|
run_step(runtime)
|
|
assert stopped.value.code == 1
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
|
|
assert (outside / "approved.py").read_text(encoding="utf-8") == "outside data\n"
|
|
|
|
|
|
def test_late_unassigned_staging_survives_index_reset(step_runtime, monkeypatch):
|
|
runtime = step_runtime
|
|
set_owned_paths(runtime, ["approved.py"])
|
|
|
|
def change(runtime):
|
|
(runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
|
|
return {"status": "completed"}
|
|
|
|
original_run_git = runtime.executor._run_git
|
|
|
|
def inject_before_reset(*args):
|
|
if "reset" in args:
|
|
(runtime.root / "unassigned.md").write_text("staged user change\n", encoding="utf-8")
|
|
runtime.git("add", "--", "unassigned.md")
|
|
return original_run_git(*args)
|
|
|
|
monkeypatch.setattr(runtime.executor, "_run_git", inject_before_reset)
|
|
runtime.responses = [change]
|
|
with pytest.raises(SystemExit) as stopped:
|
|
run_step(runtime)
|
|
assert stopped.value.code == 1
|
|
assert "unassigned.md" in runtime.git("diff", "--cached", "--name-only").splitlines()
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
|
|
|
|
|
|
@pytest.mark.parametrize("when", ["preflight", "after-attempt"])
|
|
def test_owned_hard_link_is_rejected(step_runtime, when):
|
|
runtime = step_runtime
|
|
outside = runtime.root.parent / (runtime.root.name + "-outside.py")
|
|
outside.write_text("external data\n", encoding="utf-8")
|
|
set_owned_paths(runtime, ["approved.py"])
|
|
|
|
def substitute(runtime):
|
|
(runtime.root / "approved.py").hardlink_to(outside)
|
|
return {"status": "completed"}
|
|
|
|
if when == "preflight":
|
|
substitute(runtime)
|
|
with pytest.raises(execute.HarnessError, match="owned_paths"):
|
|
runtime.executor._preflight_repository()
|
|
assert runtime.calls == []
|
|
else:
|
|
runtime.responses = [substitute]
|
|
with pytest.raises(SystemExit) as stopped:
|
|
run_step(runtime)
|
|
assert stopped.value.code == 1
|
|
assert outside.read_text(encoding="utf-8") == "external data\n"
|
|
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
|