KOKO\Mimi
123 lines
3.1 KiB
Python
123 lines
3.1 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts.execute import (
|
|
CodexEnvironmentError,
|
|
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_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"}]
|