김경종
325 lines
13 KiB
Python
325 lines
13 KiB
Python
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
def load_execute():
|
|
module_path = Path(__file__).resolve().parent / "execute.py"
|
|
spec = importlib.util.spec_from_file_location("execute", module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def write_phase(root: Path, phase_dir: str = "0-mvp", phase_name: str = "0-mvp", steps=None):
|
|
phase_path = root / "phases" / phase_dir
|
|
phase_path.mkdir(parents=True)
|
|
if steps is None:
|
|
steps = [
|
|
{
|
|
"step": 1,
|
|
"name": "Docs",
|
|
"status": "pending",
|
|
"summary": "",
|
|
"allowed_paths": ["docs/*.md"],
|
|
}
|
|
]
|
|
(phase_path / "index.json").write_text(
|
|
json.dumps({"project": "FESA", "phase": phase_name, "steps": steps}, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
(phase_path / "step1.md").write_text("# Step 1\n", encoding="utf-8")
|
|
return phase_path
|
|
|
|
|
|
def make_executor(execute, root: Path, phase_dir: str = "0-mvp"):
|
|
with patch.object(execute, "ROOT", root):
|
|
return execute.StepExecutor(phase_dir)
|
|
|
|
|
|
class ExecuteRunnerSafetyTests(unittest.TestCase):
|
|
def test_scaffold_loads_execute_module(self):
|
|
execute = load_execute()
|
|
|
|
self.assertTrue(hasattr(execute, "StepExecutor"))
|
|
|
|
def test_branch_name_uses_codex_prefix_and_sanitized_phase(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root, phase_name="linear truss/1d")
|
|
executor = make_executor(execute, root)
|
|
|
|
self.assertEqual(executor._branch_name(), "codex/linear-truss-1d")
|
|
|
|
def test_finalize_push_uses_codex_branch_name(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root, phase_name="0-mvp")
|
|
executor = make_executor(execute, root)
|
|
executor._auto_push = True
|
|
calls = []
|
|
|
|
def fake_git(*args):
|
|
calls.append(args)
|
|
if args == ("diff", "--cached", "--quiet"):
|
|
return subprocess.CompletedProcess(args, 0, "", "")
|
|
return subprocess.CompletedProcess(args, 0, "", "")
|
|
|
|
with patch.object(executor, "_run_git", side_effect=fake_git):
|
|
with patch.object(executor, "_validate_before_commit", create=True):
|
|
with patch("builtins.print"):
|
|
executor._finalize()
|
|
|
|
self.assertIn(("push", "-u", "origin", "codex/0-mvp"), calls)
|
|
|
|
def test_finalize_stages_only_phase_indexes(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
(root / "phases" / "index.json").write_text('{"phases":[]}', encoding="utf-8")
|
|
executor = make_executor(execute, root)
|
|
calls = []
|
|
|
|
def fake_git(*args):
|
|
calls.append(args)
|
|
if args == ("diff", "--cached", "--quiet"):
|
|
return subprocess.CompletedProcess(args, 1, "", "")
|
|
return subprocess.CompletedProcess(args, 0, "", "")
|
|
|
|
with patch.object(executor, "_run_git", side_effect=fake_git):
|
|
with patch.object(executor, "_validate_before_commit"):
|
|
with patch("builtins.print"):
|
|
executor._finalize()
|
|
|
|
self.assertNotIn(("add", "-A"), calls)
|
|
self.assertIn(("add", "--", "phases/0-mvp/index.json", "phases/index.json"), calls)
|
|
|
|
def test_assert_clean_worktree_exits_when_git_status_has_changes(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
|
|
with patch.object(
|
|
executor,
|
|
"_run_git",
|
|
return_value=subprocess.CompletedProcess([], 0, " M AGENTS.md\n?? scratch.txt\n", ""),
|
|
):
|
|
with patch("builtins.print"):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
executor._assert_clean_worktree("before checkout")
|
|
|
|
self.assertEqual(cm.exception.code, 1)
|
|
|
|
def test_run_checks_clean_worktree_before_checkout(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
calls = []
|
|
|
|
def record(name):
|
|
def inner(*args, **kwargs):
|
|
calls.append(name)
|
|
return inner
|
|
|
|
with patch.object(executor, "_assert_clean_worktree", side_effect=record("clean")):
|
|
with patch.object(executor, "_checkout_branch", side_effect=record("checkout")):
|
|
with patch.object(executor, "_print_header"):
|
|
with patch.object(executor, "_check_blockers"):
|
|
with patch.object(executor, "_load_guardrails", return_value=""):
|
|
with patch.object(executor, "_ensure_created_at"):
|
|
with patch.object(executor, "_execute_all_steps"):
|
|
with patch.object(executor, "_finalize"):
|
|
executor.run()
|
|
|
|
self.assertLess(calls.index("clean"), calls.index("checkout"))
|
|
|
|
def test_step_allowlist_accepts_exact_prefix_and_glob_paths(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
patterns = ["AGENTS.md", "docs/", "scripts/*.py"]
|
|
|
|
self.assertTrue(executor._path_allowed("AGENTS.md", patterns))
|
|
self.assertTrue(executor._path_allowed("docs/PRD.md", patterns))
|
|
self.assertTrue(executor._path_allowed("scripts/execute.py", patterns))
|
|
self.assertFalse(executor._path_allowed(".codex/hooks.json", patterns))
|
|
|
|
def test_step_without_allowed_paths_is_rejected_before_codex_invocation(self):
|
|
execute = load_execute()
|
|
steps = [{"step": 1, "name": "Unsafe", "status": "pending", "summary": ""}]
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root, steps=steps)
|
|
executor = make_executor(execute, root)
|
|
|
|
with patch("builtins.print"):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
executor._validate_step_allowlist(steps[0])
|
|
|
|
self.assertEqual(cm.exception.code, 1)
|
|
|
|
def test_classify_step_changes_splits_allowed_housekeeping_and_disallowed_paths(self):
|
|
execute = load_execute()
|
|
step = {
|
|
"step": 1,
|
|
"name": "Docs",
|
|
"status": "completed",
|
|
"summary": "",
|
|
"allowed_paths": ["docs/*.md"],
|
|
}
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
changed = [
|
|
"docs/PRD.md",
|
|
"phases/0-mvp/index.json",
|
|
"phases/0-mvp/step1-output.json",
|
|
"scripts/execute.py",
|
|
]
|
|
|
|
allowed, housekeeping, disallowed = executor._classify_step_changes(1, step, changed)
|
|
|
|
self.assertEqual(allowed, ["docs/PRD.md"])
|
|
self.assertEqual(housekeeping, ["phases/0-mvp/index.json", "phases/0-mvp/step1-output.json"])
|
|
self.assertEqual(disallowed, ["scripts/execute.py"])
|
|
|
|
def test_commit_step_stages_only_explicit_allowed_and_housekeeping_paths(self):
|
|
execute = load_execute()
|
|
step = {
|
|
"step": 1,
|
|
"name": "Docs",
|
|
"status": "completed",
|
|
"summary": "",
|
|
"allowed_paths": ["docs/*.md"],
|
|
}
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
calls = []
|
|
|
|
def fake_git(*args):
|
|
calls.append(args)
|
|
if args in {
|
|
("diff", "--quiet", "--cached", "--"),
|
|
("diff", "--cached", "--quiet"),
|
|
}:
|
|
return subprocess.CompletedProcess(args, 1, "", "")
|
|
return subprocess.CompletedProcess(args, 0, "", "")
|
|
|
|
with patch.object(
|
|
executor,
|
|
"_changed_paths",
|
|
return_value=[
|
|
"docs/PRD.md",
|
|
"phases/0-mvp/index.json",
|
|
"phases/0-mvp/step1-output.json",
|
|
],
|
|
):
|
|
with patch.object(executor, "_run_git", side_effect=fake_git):
|
|
with patch.object(executor, "_validate_before_commit", create=True):
|
|
with patch("builtins.print"):
|
|
executor._commit_step(step, "Docs")
|
|
|
|
self.assertNotIn(("add", "-A"), calls)
|
|
self.assertIn(("add", "--", "docs/PRD.md"), calls)
|
|
self.assertIn(("add", "--", "phases/0-mvp/index.json", "phases/0-mvp/step1-output.json"), calls)
|
|
|
|
def test_validate_before_commit_runs_python_selftest_then_workspace_validation(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
commands = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
commands.append(cmd)
|
|
return subprocess.CompletedProcess(cmd, 0, "ok", "")
|
|
|
|
with patch.object(execute.subprocess, "run", side_effect=fake_run):
|
|
with patch("builtins.print"):
|
|
executor._validate_before_commit("feat(0-mvp): step 1")
|
|
|
|
self.assertEqual(
|
|
commands,
|
|
[
|
|
[sys.executable, "-m", "unittest", "discover", "-s", "scripts", "-p", "test_*.py"],
|
|
[sys.executable, "scripts/validate_workspace.py"],
|
|
],
|
|
)
|
|
|
|
def test_validate_before_commit_exits_before_commit_when_validation_fails(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
return subprocess.CompletedProcess(cmd, 1, "bad", "failed")
|
|
|
|
with patch.object(execute.subprocess, "run", side_effect=fake_run):
|
|
with patch("builtins.print"):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
executor._validate_before_commit("feat(0-mvp): step 1")
|
|
|
|
self.assertEqual(cm.exception.code, 1)
|
|
|
|
def test_invoke_codex_passes_prompt_through_stdin(self):
|
|
execute = load_execute()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_phase(root)
|
|
executor = make_executor(execute, root)
|
|
step = {"step": 1, "name": "Docs"}
|
|
long_preamble = "x" * 40000
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
return subprocess.CompletedProcess(cmd, 0, '{"event":"done"}\n', "")
|
|
|
|
with patch.object(execute.subprocess, "run", side_effect=fake_run) as run_mock:
|
|
with patch.object(executor, "_codex_command", return_value="codex.cmd"):
|
|
executor._invoke_codex(step, long_preamble)
|
|
|
|
cmd = run_mock.call_args.args[0]
|
|
kwargs = run_mock.call_args.kwargs
|
|
self.assertEqual(
|
|
cmd,
|
|
["codex.cmd", "exec", "--dangerously-bypass-approvals-and-sandbox", "--json", "-"],
|
|
)
|
|
self.assertEqual(kwargs["input"], long_preamble + "# Step 1\n")
|
|
self.assertEqual(kwargs["cwd"], str(root))
|
|
|
|
def test_codex_command_prefers_windows_cmd_or_exe_shim(self):
|
|
execute = load_execute()
|
|
|
|
def fake_which(name):
|
|
return {
|
|
"codex.cmd": "C:/tools/codex.cmd",
|
|
"codex.exe": "C:/tools/codex.exe",
|
|
"codex": "C:/tools/codex",
|
|
}.get(name)
|
|
|
|
with patch.object(execute.shutil, "which", side_effect=fake_which):
|
|
self.assertEqual(execute.StepExecutor._codex_command(), "C:/tools/codex.cmd")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|