FESA/tests/harness/test_execute.py

import os
import subprocess
import sys
from pathlib import Path

import pytest

import scripts.execute as execute_module
from scripts.execute import (
    CodexEnvironmentError,
    StepExecutor,
    build_codex_command,
    build_codex_environment,
    configure_standard_streams,
    normalize_codex_add_dirs,
    run_utf8_process,
)


def test_normalize_codex_add_dirs_accepts_existing_absolute_directories(tmp_path):
    first = tmp_path / "tool"
    second = tmp_path / "runtime"
    first.mkdir()
    second.mkdir()

    result = normalize_codex_add_dirs([str(first), str(second)])

    assert result == (first.resolve(), second.resolve())


@pytest.mark.parametrize("raw", ["relative/tool", "missing"])
def test_normalize_codex_add_dirs_rejects_unsafe_paths(tmp_path, raw):
    candidate = raw if raw.startswith("relative") else str(tmp_path / raw)

    with pytest.raises(CodexEnvironmentError):
        normalize_codex_add_dirs([candidate])


def test_build_codex_command_grants_only_explicit_directories(tmp_path):
    root = tmp_path / "repo"
    tool = tmp_path / "tool"
    runtime = tmp_path / "runtime"
    root.mkdir()
    tool.mkdir()
    runtime.mkdir()
    codex = tmp_path / "codex.exe"

    command = build_codex_command(
        codex,
        root,
        (tool.resolve(), runtime.resolve()),
    )

    assert command == [
        str(codex),
        "exec",
        "--json",
        "--sandbox",
        "workspace-write",
        "--dangerously-bypass-hook-trust",
        "--add-dir",
        str(tool.resolve()),
        "--add-dir",
        str(runtime.resolve()),
        "--cd",
        str(root.resolve()),
        "-",
    ]


def test_build_codex_environment_appends_tool_directories(tmp_path):
    tool = (tmp_path / "tool").resolve()
    runtime = (tmp_path / "runtime").resolve()
    base = {"PATH": os.pathsep.join(("existing-one", "existing-two")), "KEEP": "value"}

    result = build_codex_environment(base, (tool, runtime))

    assert result["PATH"].split(os.pathsep) == [
        "existing-one",
        "existing-two",
        str(tool),
        str(runtime),
    ]
    assert result["KEEP"] == "value"
    assert result is not base


def test_run_utf8_process_sends_prompt_as_utf8(tmp_path):
    verifier = tmp_path / "verify_utf8.py"
    verifier.write_text(
        "import sys\n"
        "payload = sys.stdin.buffer.read()\n"
        "text = payload.decode('utf-8')\n"
        "sys.stdout.buffer.write(text.encode('utf-8'))\n",
        encoding="utf-8",
    )
    prompt = "FESA 한글 prompt"

    result = run_utf8_process(
        [sys.executable, str(verifier)],
        cwd=tmp_path,
        prompt=prompt,
        env=os.environ.copy(),
        timeout=10,
    )

    assert result.returncode == 0
    assert result.stdout == prompt


def test_invoke_codex_records_timeout_with_partial_output(tmp_path, monkeypatch):
    phase_dir = tmp_path / "phases" / "test-phase"
    phase_dir.mkdir(parents=True)
    (phase_dir / "step0.md").write_text("# Timed step\n", encoding="utf-8")
    codex = tmp_path / "codex.exe"
    codex.touch()

    executor = StepExecutor.__new__(StepExecutor)
    executor._root = str(tmp_path)
    executor._phase_dir = phase_dir
    executor._codex_add_dirs = ()

    monkeypatch.setattr(
        execute_module.shutil, "which", lambda executable: str(codex)
    )

    def raise_timeout(*args, **kwargs):
        raise subprocess.TimeoutExpired(
            cmd=["codex", "exec"],
            timeout=1800,
            output=b"partial stdout",
            stderr=b"partial stderr",
        )

    monkeypatch.setattr(execute_module, "run_utf8_process", raise_timeout)

    output = executor._invoke_codex(
        {"step": 0, "name": "timed-step"}, "preamble\n"
    )

    assert output["exitCode"] == 124
    assert output["stdout"] == "partial stdout"
    assert "partial stderr" in output["stderr"]
    assert "timed out after 1800 seconds" in output["stderr"]
    assert (
        execute_module.json.loads(
            (phase_dir / "step0-output.json").read_text(encoding="utf-8")
        )
        == output
    )


def test_execute_single_step_retries_timeout_three_times(tmp_path, monkeypatch):
    index_file = tmp_path / "index.json"
    index_file.write_text(
        execute_module.json.dumps(
            {
                "steps": [
                    {"step": 0, "name": "timed-step", "status": "pending"}
                ]
            }
        ),
        encoding="utf-8",
    )

    executor = StepExecutor.__new__(StepExecutor)
    executor._index_file = index_file
    executor._total = 1
    executor._project = "FESA"
    executor._phase_dir_name = "test-phase"
    attempts = 0

    def timeout_output(step, preamble):
        nonlocal attempts
        attempts += 1
        return {
            "step": 0,
            "name": "timed-step",
            "exitCode": 124,
            "stdout": "",
            "stderr": "Codex timed out after 1800 seconds.",
        }

    monkeypatch.setattr(executor, "_invoke_codex", timeout_output)
    monkeypatch.setattr(executor, "_commit_step", lambda *args: None)
    monkeypatch.setattr(executor, "_update_top_index", lambda *args: None)

    with pytest.raises(SystemExit) as exit_info:
        executor._execute_single_step(
            {"step": 0, "name": "timed-step", "status": "pending"}, ""
        )

    assert exit_info.value.code == 1
    assert attempts == 3
    index = execute_module.json.loads(index_file.read_text(encoding="utf-8"))
    assert index["steps"][0]["status"] == "error"
    assert "timed out after 1800 seconds" in index["steps"][0]["error_message"]


def test_configure_standard_streams_replaces_unencodable_status_characters():
    class RecordingStream:
        def __init__(self):
            self.calls = []

        def reconfigure(self, **kwargs):
            self.calls.append(kwargs)

    stdout = RecordingStream()
    stderr = RecordingStream()

    configure_standard_streams(stdout, stderr)

    assert stdout.calls == [{"errors": "replace"}]
    assert stderr.calls == [{"errors": "replace"}]