modify docu
This commit is contained in:
김경종
@@ -0,0 +1,286 @@
|
||||
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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user