KOKO\Mimi
282 lines
9.7 KiB
Python
282 lines
9.7 KiB
Python
import json
|
||
import subprocess
|
||
|
||
import pytest
|
||
|
||
from scripts import execute
|
||
|
||
|
||
def write_implementation_agent_profile(
|
||
root, *, model="gpt-5.6-terra", reasoning="xhigh"
|
||
):
|
||
profile = root / ".codex" / "agents" / "implementation-agent.toml"
|
||
profile.parent.mkdir(parents=True)
|
||
profile.write_text(
|
||
f'model = "{model}"\nmodel_reasoning_effort = "{reasoning}"\n',
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def test_invoke_codex_uses_utf8_for_unicode_prompt(tmp_path, monkeypatch):
|
||
phase_dir = tmp_path / "phases" / "unicode-phase"
|
||
phase_dir.mkdir(parents=True)
|
||
(phase_dir / "index.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"project": "FESA Structural Solver",
|
||
"phase": "unicode-phase",
|
||
"steps": [{"step": 0, "name": "unicode", "status": "pending"}],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
(phase_dir / "step0.md").write_text("Euler–Bernoulli 보", encoding="utf-8")
|
||
write_implementation_agent_profile(tmp_path)
|
||
|
||
invocation = {}
|
||
|
||
def capture_run(command, **kwargs):
|
||
invocation["command"] = command
|
||
invocation["kwargs"] = kwargs
|
||
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
|
||
|
||
monkeypatch.setattr(execute, "ROOT", tmp_path)
|
||
monkeypatch.setattr(execute.subprocess, "run", capture_run)
|
||
|
||
executor = execute.StepExecutor("unicode-phase")
|
||
executor._invoke_codex({"step": 0, "name": "unicode"}, "승인된 설계 – ")
|
||
|
||
assert "Euler–Bernoulli 보" in invocation["kwargs"]["input"]
|
||
assert invocation["kwargs"]["encoding"] == "utf-8"
|
||
sandbox_index = invocation["command"].index("--sandbox")
|
||
assert invocation["command"][sandbox_index + 1] == "workspace-write"
|
||
|
||
|
||
def test_invoke_codex_uses_implementation_agent_command_contract(
|
||
tmp_path, monkeypatch
|
||
):
|
||
phase_dir = tmp_path / "phases" / "agent-phase"
|
||
phase_dir.mkdir(parents=True)
|
||
(phase_dir / "index.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"project": "FESA Structural Solver",
|
||
"phase": "agent-phase",
|
||
"steps": [{"step": 0, "name": "agent", "status": "pending"}],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
(phase_dir / "step0.md").write_text("agent routing", encoding="utf-8")
|
||
write_implementation_agent_profile(tmp_path)
|
||
|
||
invocation = {}
|
||
|
||
def capture_run(command, **kwargs):
|
||
invocation["command"] = command
|
||
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
|
||
|
||
monkeypatch.setattr(execute, "ROOT", tmp_path)
|
||
monkeypatch.setattr(execute.subprocess, "run", capture_run)
|
||
|
||
executor = execute.StepExecutor("agent-phase")
|
||
executor._invoke_codex({"step": 0, "name": "agent"}, "")
|
||
|
||
model_index = invocation["command"].index("--model")
|
||
config_index = invocation["command"].index("--config")
|
||
assert invocation["command"][model_index + 1] == "gpt-5.6-terra"
|
||
assert invocation["command"][config_index + 1] == 'model_reasoning_effort="xhigh"'
|
||
adjacent_arguments = list(zip(invocation["command"], invocation["command"][1:]))
|
||
assert ("--disable", "unified_exec") in adjacent_arguments
|
||
|
||
|
||
def test_invoke_codex_accepts_explicit_danger_full_access_override(
|
||
tmp_path, monkeypatch
|
||
):
|
||
phase_dir = tmp_path / "phases" / "sandbox-phase"
|
||
phase_dir.mkdir(parents=True)
|
||
(phase_dir / "index.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"project": "FESA Structural Solver",
|
||
"phase": "sandbox-phase",
|
||
"steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
(phase_dir / "step0.md").write_text("sandbox override", encoding="utf-8")
|
||
write_implementation_agent_profile(tmp_path)
|
||
|
||
invocation = {}
|
||
|
||
def capture_run(command, **kwargs):
|
||
invocation["command"] = command
|
||
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
|
||
|
||
monkeypatch.setattr(execute, "ROOT", tmp_path)
|
||
monkeypatch.setattr(execute.subprocess, "run", capture_run)
|
||
monkeypatch.setenv("FESA_HARNESS_CODEX_SANDBOX", "danger-full-access")
|
||
|
||
executor = execute.StepExecutor("sandbox-phase")
|
||
executor._invoke_codex({"step": 0, "name": "sandbox"}, "")
|
||
|
||
sandbox_index = invocation["command"].index("--sandbox")
|
||
assert invocation["command"][sandbox_index + 1] == "danger-full-access"
|
||
|
||
|
||
def test_invoke_codex_timeout_writes_structured_failure(tmp_path, monkeypatch):
|
||
phase_dir = tmp_path / "phases" / "timeout-phase"
|
||
phase_dir.mkdir(parents=True)
|
||
(phase_dir / "index.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"project": "FESA Structural Solver",
|
||
"phase": "timeout-phase",
|
||
"steps": [{"step": 0, "name": "timeout", "status": "pending"}],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
(phase_dir / "step0.md").write_text("timeout", encoding="utf-8")
|
||
write_implementation_agent_profile(tmp_path)
|
||
|
||
def timeout_run(command, **kwargs):
|
||
raise subprocess.TimeoutExpired(
|
||
command, kwargs["timeout"], output="partial stdout", stderr="partial stderr"
|
||
)
|
||
|
||
monkeypatch.setattr(execute, "ROOT", tmp_path)
|
||
monkeypatch.setattr(execute.subprocess, "run", timeout_run)
|
||
|
||
executor = execute.StepExecutor("timeout-phase")
|
||
output = executor._invoke_codex({"step": 0, "name": "timeout"}, "")
|
||
|
||
assert output["exitCode"] == 124
|
||
assert output["stdout"] == "partial stdout"
|
||
assert "Codex timed out after 1800 seconds." in output["stderr"]
|
||
stored = json.loads((phase_dir / "step0-output.json").read_text(encoding="utf-8"))
|
||
assert stored == output
|
||
|
||
|
||
def test_execute_single_step_retries_structured_codex_timeout(tmp_path, monkeypatch):
|
||
phase_dir = tmp_path / "phases" / "retry-phase"
|
||
phase_dir.mkdir(parents=True)
|
||
(phase_dir / "index.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"project": "FESA Structural Solver",
|
||
"phase": "retry-phase",
|
||
"steps": [{"step": 0, "name": "retry", "status": "pending"}],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
(phase_dir / "step0.md").write_text("retry", encoding="utf-8")
|
||
write_implementation_agent_profile(tmp_path)
|
||
attempts = []
|
||
|
||
def timeout_run(command, **kwargs):
|
||
attempts.append(command)
|
||
raise subprocess.TimeoutExpired(command, kwargs["timeout"])
|
||
|
||
monkeypatch.setattr(execute, "ROOT", tmp_path)
|
||
monkeypatch.setattr(execute.subprocess, "run", timeout_run)
|
||
|
||
executor = execute.StepExecutor("retry-phase")
|
||
executor.MAX_RETRIES = 2
|
||
monkeypatch.setattr(executor, "_commit_step", lambda *args: None)
|
||
monkeypatch.setattr(executor, "_update_top_index", lambda *args: None)
|
||
|
||
with pytest.raises(SystemExit, match="1"):
|
||
executor._execute_single_step(
|
||
{"step": 0, "name": "retry", "status": "pending"}, ""
|
||
)
|
||
|
||
assert len(attempts) == 2
|
||
index = json.loads((phase_dir / "index.json").read_text(encoding="utf-8"))
|
||
assert index["steps"][0]["status"] == "error"
|
||
assert "Codex timed out after 1800 seconds." in index["steps"][0]["error_message"]
|
||
|
||
|
||
def test_invoke_codex_rejects_unknown_sandbox_override(tmp_path, monkeypatch):
|
||
phase_dir = tmp_path / "phases" / "sandbox-phase"
|
||
phase_dir.mkdir(parents=True)
|
||
(phase_dir / "index.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"project": "FESA Structural Solver",
|
||
"phase": "sandbox-phase",
|
||
"steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
(phase_dir / "step0.md").write_text("sandbox override", encoding="utf-8")
|
||
|
||
monkeypatch.setattr(execute, "ROOT", tmp_path)
|
||
monkeypatch.setenv("FESA_HARNESS_CODEX_SANDBOX", "read-only")
|
||
|
||
executor = execute.StepExecutor("sandbox-phase")
|
||
with pytest.raises(execute.CodexEnvironmentError, match="FESA_HARNESS_CODEX_SANDBOX"):
|
||
executor._invoke_codex({"step": 0, "name": "sandbox"}, "")
|
||
|
||
|
||
def test_codex_environment_failure_detects_missing_sandbox_helper_on_zero_exit():
|
||
output = {
|
||
"exitCode": 0,
|
||
"stdout": "",
|
||
"stderr": (
|
||
"windows sandbox: orchestrator_helper_launch_failed: "
|
||
"helper=codex-windows-sandbox-setup.exe, error=program not found"
|
||
),
|
||
}
|
||
|
||
assert execute.StepExecutor._codex_environment_failure(output) == (
|
||
"Codex Windows sandbox helper를 실행할 수 없습니다."
|
||
)
|
||
|
||
|
||
def test_run_git_decodes_output_as_utf8(tmp_path, monkeypatch):
|
||
invocation = {}
|
||
|
||
def capture_run(command, **kwargs):
|
||
invocation["command"] = command
|
||
invocation["kwargs"] = kwargs
|
||
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
|
||
|
||
monkeypatch.setattr(execute.subprocess, "run", capture_run)
|
||
executor = object.__new__(execute.StepExecutor)
|
||
executor._root = str(tmp_path)
|
||
|
||
executor._run_git("status", "--short")
|
||
|
||
assert invocation["kwargs"]["encoding"] == "utf-8"
|
||
|
||
|
||
def test_feature_commit_message_is_windows_console_safe():
|
||
message = execute.StepExecutor.FEAT_MSG.format(
|
||
phase="linear-static-3d-euler-beam", num=0, name="requirements-baseline"
|
||
)
|
||
|
||
message.encode("cp949")
|
||
|
||
|
||
def test_configure_utf8_stdio_reconfigures_both_streams(monkeypatch):
|
||
class ReconfigurableStream:
|
||
def __init__(self):
|
||
self.encodings = []
|
||
|
||
def reconfigure(self, *, encoding):
|
||
self.encodings.append(encoding)
|
||
|
||
stdout = ReconfigurableStream()
|
||
stderr = ReconfigurableStream()
|
||
monkeypatch.setattr(execute.sys, "stdout", stdout)
|
||
monkeypatch.setattr(execute.sys, "stderr", stderr)
|
||
|
||
execute._configure_utf8_stdio()
|
||
|
||
assert stdout.encodings == ["utf-8"]
|
||
assert stderr.encodings == ["utf-8"]
|