wehub-resource-sync
1656 lines
56 KiB
Python
1656 lines
56 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for the restricted local CLI generation backend."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from tests.litellm_stub import ensure_litellm_stub
|
|
|
|
ensure_litellm_stub()
|
|
|
|
from src.analyzer import GeminiAnalyzer # noqa: E402
|
|
from src.llm import local_cli_backend as local_cli_backend_module # noqa: E402
|
|
from src.llm.generation_backend import GenerationError, GenerationErrorCode # noqa: E402
|
|
from src.llm.local_cli_backend import ( # noqa: E402
|
|
CLAUDE_CODE_CLI_PRESET,
|
|
LocalCliGenerationBackend,
|
|
LocalCliExecutionResult,
|
|
LocalCliExtractionError,
|
|
LocalCliPreset,
|
|
OPENCODE_CLI_PRESET,
|
|
build_local_cli_env,
|
|
effective_local_cli_concurrency,
|
|
redact_diagnostic_text,
|
|
)
|
|
|
|
|
|
def _config(**overrides):
|
|
defaults = {
|
|
"generation_backend_timeout_seconds": 5,
|
|
"generation_backend_max_output_bytes": 1024 * 1024,
|
|
"generation_backend_max_concurrency": 1,
|
|
"local_cli_backend_max_concurrency": 1,
|
|
"generation_backend": "codex_cli",
|
|
"generation_fallback_backend": "",
|
|
"report_language": "zh",
|
|
}
|
|
defaults.update(overrides)
|
|
return SimpleNamespace(**defaults)
|
|
|
|
|
|
def _script(tmp_path: Path, source: str) -> str:
|
|
path = tmp_path / "mock_cli.py"
|
|
path.write_text(source, encoding="utf-8")
|
|
return str(path)
|
|
|
|
|
|
def _backend(tmp_path: Path, source: str, **config_overrides) -> LocalCliGenerationBackend:
|
|
preset = LocalCliPreset(
|
|
preset_id="codex_cli",
|
|
executable=sys.executable,
|
|
argv=(_script(tmp_path, source),),
|
|
display_name="Mock CLI",
|
|
)
|
|
return LocalCliGenerationBackend(_config(**config_overrides), preset=preset)
|
|
|
|
|
|
def test_success_uses_stdin_temp_cwd_and_usage_unavailable(tmp_path: Path) -> None:
|
|
backend = _backend(
|
|
tmp_path,
|
|
"""
|
|
import json, os, sys
|
|
prompt = sys.stdin.read()
|
|
print(json.dumps({"prompt": prompt, "cwd": os.getcwd(), "sentiment_score": 70}, ensure_ascii=False))
|
|
""",
|
|
)
|
|
|
|
result = backend.generate("hello", {}, response_validator=lambda text: json.loads(text))
|
|
payload = json.loads(result.text)
|
|
|
|
assert payload["prompt"] == "hello"
|
|
assert payload["cwd"] != os.getcwd()
|
|
assert not Path(payload["cwd"]).exists()
|
|
assert result.usage == {
|
|
"usage_available": False,
|
|
"usage_source": "unavailable",
|
|
"backend": "codex_cli",
|
|
}
|
|
assert result.diagnostics["executable"]["basename"] == Path(sys.executable).name
|
|
assert "path" not in result.diagnostics["executable"]
|
|
|
|
|
|
def test_codex_preset_reads_output_last_message_instead_of_stdout(tmp_path: Path) -> None:
|
|
final_payload = json.dumps({"prompt": "hello", "sentiment_score": 88, "source": "last_message"})
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import json, sys
|
|
args = sys.argv[1:]
|
|
output_path = args[args.index("--output-last-message") + 1]
|
|
prompt = sys.stdin.read()
|
|
with open(output_path, "w", encoding="utf-8") as handle:
|
|
handle.write(json.dumps({{"prompt": prompt, "sentiment_score": 88, "source": "last_message"}}))
|
|
print("OpenAI Codex v0.142.0")
|
|
print("23,011")
|
|
print({final_payload!r})
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="codex_cli",
|
|
executable=sys.executable,
|
|
argv=(script, "-"),
|
|
display_name="Mock Codex CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
result = backend.generate("hello", {}, response_validator=lambda text: json.loads(text))
|
|
payload = json.loads(result.text)
|
|
|
|
assert payload == {
|
|
"prompt": "hello",
|
|
"sentiment_score": 88,
|
|
"source": "last_message",
|
|
}
|
|
assert result.diagnostics["output_source"] == "output_last_message"
|
|
assert "OpenAI Codex" in result.diagnostics["stdout_preview"]
|
|
assert "final-message omitted" in result.diagnostics["stdout_preview"]
|
|
assert "last_message" not in result.diagnostics["stdout_preview"]
|
|
|
|
|
|
def test_claude_preset_runtime_argv_contains_contract_args(tmp_path: Path) -> None:
|
|
argv_path = tmp_path / "argv.json"
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import json, pathlib, sys
|
|
path = pathlib.Path({str(argv_path)!r})
|
|
path.write_text(json.dumps(sys.argv[1:]), encoding="utf-8")
|
|
print(json.dumps({{"type": "result", "subtype": "success", "result": "{{\\"sentiment_score\\": 77}}"}}))
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="claude_code_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *CLAUDE_CODE_CLI_PRESET.argv),
|
|
display_name="Mock Claude Code CLI",
|
|
extractor=CLAUDE_CODE_CLI_PRESET.extractor,
|
|
contract_args=CLAUDE_CODE_CLI_PRESET.contract_args,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="claude_code_cli"),
|
|
preset=preset,
|
|
)
|
|
|
|
result = backend.generate("prompt", {}, response_validator=lambda text: json.loads(text))
|
|
runtime_argv = json.loads(argv_path.read_text(encoding="utf-8"))
|
|
|
|
assert json.loads(result.text)["sentiment_score"] == 77
|
|
for contract_arg in CLAUDE_CODE_CLI_PRESET.contract_args:
|
|
assert contract_arg in runtime_argv
|
|
|
|
|
|
def test_missing_contract_arg_is_capability_unsupported(tmp_path: Path) -> None:
|
|
preset = LocalCliPreset(
|
|
preset_id="claude_code_cli",
|
|
executable=sys.executable,
|
|
argv=(_script(tmp_path, "print('unused')"), "--safe-mode"),
|
|
display_name="Mock Claude Code CLI",
|
|
contract_args=("--safe-mode", "--strict-mcp-config"),
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="claude_code_cli"),
|
|
preset=preset,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
assert exc_info.value.details["reason"] == "missing_runtime_contract_arg"
|
|
assert "--strict-mcp-config" in exc_info.value.details["missing_contract_args"]
|
|
|
|
|
|
def test_contract_args_must_keep_preset_order(tmp_path: Path) -> None:
|
|
preset = LocalCliPreset(
|
|
preset_id="claude_code_cli",
|
|
executable=sys.executable,
|
|
argv=(
|
|
_script(tmp_path, "print('unused')"),
|
|
"--strict-mcp-config",
|
|
"--safe-mode",
|
|
),
|
|
display_name="Mock Claude Code CLI",
|
|
contract_args=("--safe-mode", "--strict-mcp-config"),
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="claude_code_cli"),
|
|
preset=preset,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
assert exc_info.value.details["reason"] == "missing_runtime_contract_arg"
|
|
assert "--strict-mcp-config" in exc_info.value.details["missing_contract_args"]
|
|
|
|
|
|
def test_claude_extractor_uses_structured_output_in_schema_mode() -> None:
|
|
preset = LocalCliPreset(
|
|
preset_id="claude_code_cli",
|
|
executable="claude",
|
|
argv=(),
|
|
display_name="Mock Claude Code CLI",
|
|
extractor=lambda result: local_cli_backend_module._extract_claude_code_json(
|
|
result,
|
|
schema_mode=True,
|
|
),
|
|
)
|
|
|
|
text = preset.extractor(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps({
|
|
"type": "result",
|
|
"subtype": "success",
|
|
"structured_output": {"sentiment_score": "70"},
|
|
}),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
|
|
assert text == '{"sentiment_score":"70"}'
|
|
|
|
|
|
def test_claude_extractor_rejects_tool_event() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_claude_code_json(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps({"type": "tool_use", "result": "should not parse"}),
|
|
stderr="",
|
|
returncode=0,
|
|
),
|
|
schema_mode=False,
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
assert exc_info.value.reason == "unexpected_cli_event"
|
|
|
|
|
|
def test_claude_extractor_requires_result_success_envelope() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as missing_type:
|
|
local_cli_backend_module._extract_claude_code_json(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps({"subtype": "success", "result": "should not parse"}),
|
|
stderr="",
|
|
returncode=0,
|
|
),
|
|
schema_mode=False,
|
|
)
|
|
with pytest.raises(LocalCliExtractionError) as missing_subtype:
|
|
local_cli_backend_module._extract_claude_code_json(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps({"type": "result", "result": "should not parse"}),
|
|
stderr="",
|
|
returncode=0,
|
|
),
|
|
schema_mode=False,
|
|
)
|
|
|
|
assert missing_type.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
assert missing_type.value.reason == "unexpected_cli_event"
|
|
assert missing_subtype.value.error_code is GenerationErrorCode.UNKNOWN_BACKEND_ERROR
|
|
assert missing_subtype.value.reason == "cli_result_not_success"
|
|
|
|
|
|
def test_claude_schema_retry_exhaustion_maps_schema_validation_failed() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_claude_code_json(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps({
|
|
"type": "result",
|
|
"subtype": "error_max_structured_output_retries",
|
|
"is_error": True,
|
|
}),
|
|
stderr="",
|
|
returncode=0,
|
|
),
|
|
schema_mode=True,
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.SCHEMA_VALIDATION_FAILED
|
|
|
|
|
|
def test_opencode_preset_uses_prompt_file_and_safe_argv(tmp_path: Path, monkeypatch) -> None:
|
|
monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-should-not-leak")
|
|
monkeypatch.setenv("OPENAI_API_KEY", "sk-openai-should-not-leak")
|
|
monkeypatch.setenv("OPENCODE_CONFIG_CONTENT", '{"plugin":["leak"]}')
|
|
argv_path = tmp_path / "argv.json"
|
|
probe_path = tmp_path / "probe.json"
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import json, os, pathlib, stat, sys
|
|
argv = sys.argv[1:]
|
|
prompt_path = pathlib.Path(argv[argv.index("--file") + 1])
|
|
config_path = pathlib.Path.cwd() / "opencode.json"
|
|
probe = {{
|
|
"argv": argv,
|
|
"prompt": prompt_path.read_text(encoding="utf-8"),
|
|
"prompt_mode": stat.S_IMODE(prompt_path.stat().st_mode),
|
|
"cwd_mode": stat.S_IMODE(pathlib.Path.cwd().stat().st_mode),
|
|
"config": config_path.read_text(encoding="utf-8"),
|
|
"env": {{
|
|
"DEEPSEEK_API_KEY": os.environ.get("DEEPSEEK_API_KEY"),
|
|
"OPENAI_API_KEY": os.environ.get("OPENAI_API_KEY"),
|
|
"OPENCODE_CONFIG_CONTENT": os.environ.get("OPENCODE_CONFIG_CONTENT"),
|
|
"OPENCODE_CONFIG_DIR": os.environ.get("OPENCODE_CONFIG_DIR"),
|
|
}},
|
|
}}
|
|
pathlib.Path({str(argv_path)!r}).write_text(json.dumps(argv), encoding="utf-8")
|
|
pathlib.Path({str(probe_path)!r}).write_text(json.dumps(probe), encoding="utf-8")
|
|
print(json.dumps({{"type": "step_start"}}))
|
|
print(json.dumps({{"type": "text", "part": {{"text": "{{\\"sentiment_score\\": 66}}"}}}}))
|
|
print(json.dumps({{"type": "step_finish", "reason": "stop"}}))
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="opencode_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *OPENCODE_CLI_PRESET.argv),
|
|
display_name="Mock OpenCode CLI",
|
|
extractor=OPENCODE_CLI_PRESET.extractor,
|
|
contract_args=OPENCODE_CLI_PRESET.contract_args,
|
|
prompt_transport=OPENCODE_CLI_PRESET.prompt_transport,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="opencode_cli"),
|
|
preset=preset,
|
|
)
|
|
|
|
result = backend.generate("prompt from dsa", {}, response_validator=lambda text: json.loads(text))
|
|
payload = json.loads(result.text)
|
|
argv = json.loads(argv_path.read_text(encoding="utf-8"))
|
|
probe = json.loads(probe_path.read_text(encoding="utf-8"))
|
|
opencode_config = json.loads(probe["config"])
|
|
|
|
assert payload["sentiment_score"] == 66
|
|
assert argv[:4] == ["--pure", "run", "--format", "json"]
|
|
assert "--model" not in argv
|
|
assert "--file" in argv
|
|
assert argv.index("--file") > argv.index("json")
|
|
assert "--attach" not in argv
|
|
assert "--dangerously-skip-permissions" not in argv
|
|
assert probe["prompt"] == "prompt from dsa"
|
|
assert probe["prompt_mode"] == 0o600
|
|
assert probe["cwd_mode"] == 0o700
|
|
for tool_name in local_cli_backend_module._OPENCODE_DISABLED_TOOL_NAMES:
|
|
assert opencode_config["tools"][tool_name] is False
|
|
assert opencode_config["tools"]["websearch"] is False
|
|
assert opencode_config["tools"]["question"] is False
|
|
assert opencode_config["tools"]["skill"] is False
|
|
assert opencode_config["tools"]["todowrite"] is False
|
|
assert opencode_config["tools"]["lsp"] is False
|
|
assert "sk-should-not-leak" not in probe["config"]
|
|
assert "sk-openai-should-not-leak" not in probe["config"]
|
|
assert probe["env"]["DEEPSEEK_API_KEY"] is None
|
|
assert probe["env"]["OPENAI_API_KEY"] is None
|
|
assert probe["env"]["OPENCODE_CONFIG_CONTENT"] is None
|
|
assert probe["env"]["OPENCODE_CONFIG_DIR"] is None
|
|
assert result.diagnostics["opencode_project_config_written"] is True
|
|
assert "opencode_config_controlled" not in result.diagnostics
|
|
assert result.backend == "opencode_cli"
|
|
assert result.provider == "opencode_cli"
|
|
assert result.model == "opencode_cli"
|
|
assert result.usage["backend"] == "opencode_cli"
|
|
|
|
|
|
def test_opencode_static_instruction_does_not_force_stock_json_contract() -> None:
|
|
instruction = " ".join(str(arg) for arg in OPENCODE_CLI_PRESET.argv)
|
|
normalized_instruction = instruction.lower()
|
|
|
|
assert "attached prompt file" in instruction
|
|
assert "json object" not in normalized_instruction
|
|
assert "parser contract" not in normalized_instruction
|
|
for field_name in (
|
|
"sentiment_score",
|
|
"trend_prediction",
|
|
"operation_advice",
|
|
"analysis_summary",
|
|
"dashboard",
|
|
):
|
|
assert field_name not in instruction
|
|
|
|
|
|
def test_opencode_preset_accepts_free_text_without_json_validator(tmp_path: Path) -> None:
|
|
review = "## 今日复盘\n\n市场震荡，保持观察。"
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import json
|
|
print(json.dumps({{"type": "step_start"}}))
|
|
print(json.dumps({{"type": "text", "part": {{"text": {review!r}}}}}, ensure_ascii=False))
|
|
print(json.dumps({{"type": "step_finish", "reason": "stop"}}))
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="opencode_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *OPENCODE_CLI_PRESET.argv),
|
|
display_name="Mock OpenCode CLI",
|
|
extractor=OPENCODE_CLI_PRESET.extractor,
|
|
contract_args=OPENCODE_CLI_PRESET.contract_args,
|
|
prompt_transport=OPENCODE_CLI_PRESET.prompt_transport,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="opencode_cli"),
|
|
preset=preset,
|
|
)
|
|
|
|
result = backend.generate("请生成 Markdown 复盘", {}, response_validator=None)
|
|
|
|
assert result.text == review
|
|
|
|
|
|
def test_opencode_model_override_inserts_model_arg(tmp_path: Path) -> None:
|
|
argv_path = tmp_path / "argv.json"
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import json, pathlib, sys
|
|
pathlib.Path({str(argv_path)!r}).write_text(json.dumps(sys.argv[1:]), encoding="utf-8")
|
|
print(json.dumps({{"type": "step_start"}}))
|
|
print(json.dumps({{"type": "text", "part": {{"text": "{{\\"sentiment_score\\": 67}}"}}}}))
|
|
print(json.dumps({{"type": "step_finish", "reason": "stop"}}))
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="opencode_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *OPENCODE_CLI_PRESET.argv),
|
|
display_name="Mock OpenCode CLI",
|
|
extractor=OPENCODE_CLI_PRESET.extractor,
|
|
contract_args=OPENCODE_CLI_PRESET.contract_args,
|
|
prompt_transport=OPENCODE_CLI_PRESET.prompt_transport,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="opencode_cli", opencode_cli_model="provider/model"),
|
|
preset=preset,
|
|
)
|
|
|
|
result = backend.generate("prompt", {}, response_validator=lambda text: json.loads(text))
|
|
argv = json.loads(argv_path.read_text(encoding="utf-8"))
|
|
|
|
assert json.loads(result.text)["sentiment_score"] == 67
|
|
assert argv[:6] == ["--pure", "run", "--format", "json", "--model", "provider/model"]
|
|
assert argv.index("--file") > argv.index("provider/model")
|
|
|
|
|
|
def test_opencode_nonzero_json_event_error_maps_unknown_backend_error(tmp_path: Path) -> None:
|
|
script = _script(
|
|
tmp_path,
|
|
"""
|
|
import json
|
|
print(json.dumps({"type": "error", "error": {"name": "UnknownError"}}))
|
|
raise SystemExit(1)
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="opencode_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *OPENCODE_CLI_PRESET.argv),
|
|
display_name="Mock OpenCode CLI",
|
|
extractor=OPENCODE_CLI_PRESET.extractor,
|
|
contract_args=OPENCODE_CLI_PRESET.contract_args,
|
|
prompt_transport=OPENCODE_CLI_PRESET.prompt_transport,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="opencode_cli", opencode_cli_model="provider/model"),
|
|
preset=preset,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNKNOWN_BACKEND_ERROR
|
|
assert exc_info.value.details["reason"] == "cli_result_error"
|
|
|
|
|
|
def test_opencode_nonzero_pretty_json_error_maps_unknown_backend_error(tmp_path: Path) -> None:
|
|
script = _script(
|
|
tmp_path,
|
|
"""
|
|
import json
|
|
print(json.dumps({"type": "error", "error": {"name": "UnknownError"}}, indent=2))
|
|
raise SystemExit(1)
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="opencode_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *OPENCODE_CLI_PRESET.argv),
|
|
display_name="Mock OpenCode CLI",
|
|
extractor=OPENCODE_CLI_PRESET.extractor,
|
|
contract_args=OPENCODE_CLI_PRESET.contract_args,
|
|
prompt_transport=OPENCODE_CLI_PRESET.prompt_transport,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="opencode_cli", opencode_cli_model="provider/model"),
|
|
preset=preset,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNKNOWN_BACKEND_ERROR
|
|
assert exc_info.value.details["reason"] == "cli_result_error"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("event", "stream_name"),
|
|
[
|
|
({"type": "tool_use", "name": "read"}, "stdout"),
|
|
({"type": "websearch", "query": "AAPL"}, "stdout"),
|
|
({"type": "tool_result", "part": {"tool_name": "todowrite"}}, "stdout"),
|
|
({"type": "lsp", "name": "diagnostics"}, "stdout"),
|
|
({"type": "question", "text": "Continue?"}, "stdout"),
|
|
({"type": "permission", "name": "read"}, "stdout"),
|
|
({"type": "step_finish", "is_error": True}, "stdout"),
|
|
({"type": "step_finish", "error": {"name": "StepFailed"}}, "stdout"),
|
|
({"type": "error", "error": {"name": "StderrError"}}, "stderr"),
|
|
],
|
|
)
|
|
def test_opencode_nonzero_blocked_or_error_event_maps_unknown_backend_error(
|
|
tmp_path: Path,
|
|
event: dict,
|
|
stream_name: str,
|
|
) -> None:
|
|
target_stream = "sys.stderr" if stream_name == "stderr" else "sys.stdout"
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import json, sys
|
|
print(json.dumps({event!r}), file={target_stream})
|
|
raise SystemExit(1)
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="opencode_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *OPENCODE_CLI_PRESET.argv),
|
|
display_name="Mock OpenCode CLI",
|
|
extractor=OPENCODE_CLI_PRESET.extractor,
|
|
contract_args=OPENCODE_CLI_PRESET.contract_args,
|
|
prompt_transport=OPENCODE_CLI_PRESET.prompt_transport,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="opencode_cli", opencode_cli_model="provider/model"),
|
|
preset=preset,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNKNOWN_BACKEND_ERROR
|
|
assert exc_info.value.details["reason"] == "cli_result_error"
|
|
|
|
|
|
def test_opencode_runtime_rejects_unsafe_model_override(tmp_path: Path) -> None:
|
|
script = _script(
|
|
tmp_path,
|
|
"""
|
|
print("should not execute")
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="opencode_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *OPENCODE_CLI_PRESET.argv),
|
|
display_name="Mock OpenCode CLI",
|
|
extractor=OPENCODE_CLI_PRESET.extractor,
|
|
contract_args=OPENCODE_CLI_PRESET.contract_args,
|
|
prompt_transport=OPENCODE_CLI_PRESET.prompt_transport,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="opencode_cli", opencode_cli_model="provider/$MODEL"),
|
|
preset=preset,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNSAFE_CONFIG
|
|
assert exc_info.value.details["reason"] == "unsafe_opencode_cli_model"
|
|
|
|
|
|
def test_opencode_extractor_rejects_tool_event() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(
|
|
stdout="\n".join([
|
|
json.dumps({"type": "step_start"}),
|
|
json.dumps({"type": "tool_use", "name": "read"}),
|
|
json.dumps({"type": "step_finish", "reason": "stop"}),
|
|
]),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"event",
|
|
[
|
|
{"type": "websearch", "query": "AAPL"},
|
|
{"type": "question", "text": "Continue?"},
|
|
{"type": "skill", "name": "default"},
|
|
{"type": "todowrite", "items": []},
|
|
{"type": "lsp", "name": "diagnostics"},
|
|
{"type": "tool_use", "name": "websearch"},
|
|
{"type": "tool_result", "part": {"tool_name": "todowrite"}},
|
|
],
|
|
)
|
|
def test_opencode_extractor_rejects_default_tool_events(event: dict) -> None:
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(
|
|
stdout="\n".join([
|
|
json.dumps({"type": "step_start"}),
|
|
json.dumps(event),
|
|
json.dumps({"type": "step_finish", "reason": "stop"}),
|
|
]),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
|
|
|
|
def test_opencode_event_iterator_accepts_pretty_single_event_object() -> None:
|
|
events = list(local_cli_backend_module._iter_opencode_events(
|
|
json.dumps({"type": "step_start"}, indent=2)
|
|
))
|
|
|
|
assert events == [{"type": "step_start"}]
|
|
|
|
|
|
def test_opencode_extractor_accepts_json_array_event_stream() -> None:
|
|
result = local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps([
|
|
{"type": "step_start"},
|
|
{"type": "text", "text": '{"sentiment_score":'},
|
|
{"type": "text", "part": {"text": " 68}"}},
|
|
{"type": "step_finish", "reason": "stop"},
|
|
], indent=2),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
|
|
assert json.loads(result)["sentiment_score"] == 68
|
|
|
|
|
|
def test_opencode_extractor_accepts_concatenated_event_stream() -> None:
|
|
stdout = "".join([
|
|
json.dumps({"type": "step_start"}),
|
|
json.dumps({"type": "text", "text": '{"sentiment_score":'}),
|
|
json.dumps({"type": "text", "part": {"text": " 69}"}}),
|
|
json.dumps({"type": "step_finish", "reason": "end_turn"}),
|
|
])
|
|
|
|
result = local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(stdout=stdout, stderr="", returncode=0)
|
|
)
|
|
|
|
assert json.loads(result)["sentiment_score"] == 69
|
|
|
|
|
|
def test_opencode_extractor_rejects_trailing_non_json_garbage() -> None:
|
|
stdout = json.dumps({"type": "step_start"}) + " trailing"
|
|
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(stdout=stdout, stderr="", returncode=0)
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.INVALID_JSON
|
|
|
|
|
|
def test_opencode_extractor_rejects_non_event_json_shape() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(stdout=json.dumps({"message": "not an event"}), stderr="", returncode=0)
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.SCHEMA_VALIDATION_FAILED
|
|
|
|
|
|
def test_opencode_extractor_rejects_array_without_step_finish() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps([
|
|
{"type": "step_start"},
|
|
{"type": "text", "text": "hello"},
|
|
]),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.SCHEMA_VALIDATION_FAILED
|
|
assert exc_info.value.reason == "missing_step_finish"
|
|
|
|
|
|
def test_opencode_extractor_rejects_later_error_after_step_finish() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as exc_info:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps([
|
|
{"type": "step_start"},
|
|
{"type": "text", "text": "hello"},
|
|
{"type": "step_finish", "reason": "stop"},
|
|
{"type": "error", "error": {"name": "LaterError"}},
|
|
]),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
|
|
|
|
def test_opencode_extractor_requires_step_finish_and_text() -> None:
|
|
with pytest.raises(LocalCliExtractionError) as missing_finish:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(
|
|
stdout=json.dumps({"type": "text", "text": "hello"}),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
with pytest.raises(LocalCliExtractionError) as empty_text:
|
|
local_cli_backend_module._extract_opencode_json_events(
|
|
LocalCliExecutionResult(
|
|
stdout="\n".join([
|
|
json.dumps({"type": "step_start"}),
|
|
json.dumps({"type": "step_finish", "reason": "stop"}),
|
|
]),
|
|
stderr="",
|
|
returncode=0,
|
|
)
|
|
)
|
|
|
|
assert missing_finish.value.error_code is GenerationErrorCode.SCHEMA_VALIDATION_FAILED
|
|
assert empty_text.value.error_code is GenerationErrorCode.EMPTY_OUTPUT
|
|
|
|
|
|
def test_output_last_message_stdout_duplicate_is_not_double_counted(tmp_path: Path) -> None:
|
|
final_payload = json.dumps(
|
|
{
|
|
"sentiment_score": 70,
|
|
"source": "last_message",
|
|
"details": "x" * 40,
|
|
}
|
|
)
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import sys
|
|
args = sys.argv[1:]
|
|
output_path = args[args.index("--output-last-message") + 1]
|
|
with open(output_path, "w", encoding="utf-8") as handle:
|
|
handle.write({final_payload!r})
|
|
print({final_payload!r})
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(script,),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend_max_output_bytes=len(final_payload.encode("utf-8")) + 2),
|
|
preset=preset,
|
|
)
|
|
|
|
result = backend.generate("prompt", {}, response_validator=lambda text: json.loads(text))
|
|
|
|
assert json.loads(result.text)["sentiment_score"] == 70
|
|
assert result.diagnostics["stdout_final_message_omitted"] is True
|
|
assert "last_message" not in result.diagnostics["stdout_preview"]
|
|
|
|
|
|
def test_output_last_message_nonzero_exit_omits_duplicate_final_stdout_preview(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
final_payload = json.dumps(
|
|
{
|
|
"sentiment_score": 70,
|
|
"source": "secret_final_payload",
|
|
}
|
|
)
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import sys
|
|
args = sys.argv[1:]
|
|
output_path = args[args.index("--output-last-message") + 1]
|
|
with open(output_path, "w", encoding="utf-8") as handle:
|
|
handle.write({final_payload!r})
|
|
print("diagnostic: before final")
|
|
print({final_payload!r})
|
|
sys.exit(2)
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(script,),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.NON_ZERO_EXIT
|
|
assert "diagnostic: before final" in exc_info.value.details["stdout_preview"]
|
|
assert "final-message omitted" in exc_info.value.details["stdout_preview"]
|
|
assert "secret_final_payload" not in exc_info.value.details["stdout_preview"]
|
|
|
|
|
|
def test_stream_request_degrades_to_non_stream(tmp_path: Path) -> None:
|
|
progress = []
|
|
backend = _backend(tmp_path, "print('{\"sentiment_score\": 60}')")
|
|
|
|
result = backend.generate(
|
|
"prompt",
|
|
{},
|
|
stream=True,
|
|
stream_progress_callback=progress.append,
|
|
)
|
|
|
|
assert json.loads(result.text)["sentiment_score"] == 60
|
|
assert result.diagnostics["stream_degraded"] is True
|
|
assert progress
|
|
|
|
|
|
def test_stderr_does_not_affect_successful_stdout_or_json_parsing(tmp_path: Path) -> None:
|
|
analyzer = GeminiAnalyzer.__new__(GeminiAnalyzer)
|
|
analyzer._config_override = _config()
|
|
backend = _backend(
|
|
tmp_path,
|
|
"""
|
|
import sys
|
|
print('{"sentiment_score": 70, "trend_prediction": "看多"}')
|
|
print('{"bad": "stderr"}', file=sys.stderr)
|
|
""",
|
|
)
|
|
|
|
result = backend.generate(
|
|
"prompt",
|
|
{},
|
|
response_validator=analyzer._validate_json_response,
|
|
)
|
|
|
|
assert json.loads(result.text)["sentiment_score"] == 70
|
|
assert "stderr" in result.diagnostics["stderr_preview"]
|
|
|
|
|
|
def test_multiple_json_objects_fail_as_invalid_json_ambiguous(tmp_path: Path) -> None:
|
|
analyzer = GeminiAnalyzer.__new__(GeminiAnalyzer)
|
|
analyzer._config_override = _config()
|
|
backend = _backend(tmp_path, "print('{\"sentiment_score\": 70} {\"sentiment_score\": 80}')")
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {}, response_validator=analyzer._validate_json_response)
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.INVALID_JSON
|
|
assert exc_info.value.details["reason"] == "ambiguous_json"
|
|
|
|
|
|
def test_command_not_executable(monkeypatch, tmp_path: Path) -> None:
|
|
not_exec = tmp_path / "not-executable"
|
|
not_exec.write_text("#!/bin/sh\n", encoding="utf-8")
|
|
monkeypatch.setattr("src.llm.local_cli_backend.shutil.which", lambda _cmd: str(not_exec))
|
|
preset = LocalCliPreset("codex_cli", "mock", (), "Mock CLI")
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.COMMAND_NOT_EXECUTABLE
|
|
|
|
|
|
def test_command_not_found(monkeypatch) -> None:
|
|
monkeypatch.setattr("src.llm.local_cli_backend.shutil.which", lambda _cmd: None)
|
|
backend = LocalCliGenerationBackend(_config())
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.COMMAND_NOT_FOUND
|
|
|
|
|
|
def test_shell_metachar_returns_unsafe_config() -> None:
|
|
preset = LocalCliPreset("codex_cli", "mock", ("echo", "ok;rm"), "Mock CLI")
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNSAFE_CONFIG
|
|
assert exc_info.value.details["reason"] == "shell_metachar"
|
|
|
|
|
|
def test_output_last_message_arg_shell_metachar_returns_unsafe_config(tmp_path: Path) -> None:
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(_script(tmp_path, "print('ok')"),),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message;rm",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNSAFE_CONFIG
|
|
assert exc_info.value.details["reason"] == "shell_metachar"
|
|
|
|
|
|
def test_output_too_large(tmp_path: Path) -> None:
|
|
backend = _backend(
|
|
tmp_path,
|
|
"print('x' * 100)",
|
|
generation_backend_max_output_bytes=20,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.OUTPUT_TOO_LARGE
|
|
|
|
|
|
def test_output_stat_error_is_structured_and_kills_process_group(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
) -> None:
|
|
pid_file = tmp_path / "child-stat-error.pid"
|
|
backend = _backend(
|
|
tmp_path,
|
|
f"""
|
|
import subprocess, sys, time
|
|
child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
|
|
open({str(pid_file)!r}, "w", encoding="utf-8").write(str(child.pid))
|
|
sys.stdout.write("started")
|
|
sys.stdout.flush()
|
|
time.sleep(30)
|
|
""",
|
|
)
|
|
|
|
def _raise_stat_error(*_paths):
|
|
deadline = time.time() + 3
|
|
while not pid_file.exists() and time.time() < deadline:
|
|
time.sleep(0.01)
|
|
raise OSError("mock stat failure sk-secretsecretsecret")
|
|
|
|
monkeypatch.setattr(
|
|
"src.llm.local_cli_backend._combined_path_size_required",
|
|
_raise_stat_error,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNKNOWN_BACKEND_ERROR
|
|
assert exc_info.value.details["reason"] == "output_stat_failed"
|
|
assert "sk-secret" not in exc_info.value.details["error"]
|
|
child_pid = int(pid_file.read_text(encoding="utf-8"))
|
|
deadline = time.time() + 3
|
|
while time.time() < deadline:
|
|
try:
|
|
os.kill(child_pid, 0)
|
|
except OSError:
|
|
break
|
|
time.sleep(0.05)
|
|
else:
|
|
pytest.fail("child process was not terminated after output stat failure")
|
|
|
|
|
|
def test_output_read_error_is_structured_unknown_not_empty(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
) -> None:
|
|
backend = _backend(tmp_path, "print('{\"sentiment_score\": 70}')")
|
|
|
|
def _raise_read_error(_path):
|
|
raise OSError("mock read failure")
|
|
|
|
monkeypatch.setattr(
|
|
"src.llm.local_cli_backend._read_text_file_required",
|
|
_raise_read_error,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNKNOWN_BACKEND_ERROR
|
|
assert exc_info.value.details["reason"] == "output_read_failed"
|
|
|
|
|
|
def test_stdout_output_limit_is_not_double_counted(tmp_path: Path) -> None:
|
|
backend = _backend(
|
|
tmp_path,
|
|
"print('{\"sentiment_score\": 70}')",
|
|
generation_backend_max_output_bytes=30,
|
|
)
|
|
|
|
result = backend.generate("prompt", {}, response_validator=lambda text: json.loads(text))
|
|
|
|
assert json.loads(result.text)["sentiment_score"] == 70
|
|
assert result.diagnostics["output_source"] == "stdout"
|
|
|
|
|
|
def test_output_too_large_kills_process_group(tmp_path: Path) -> None:
|
|
pid_file = tmp_path / "child-output-limit.pid"
|
|
backend = _backend(
|
|
tmp_path,
|
|
f"""
|
|
import subprocess, sys, time
|
|
child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
|
|
open({str(pid_file)!r}, "w", encoding="utf-8").write(str(child.pid))
|
|
sys.stdout.write("x" * 100000)
|
|
sys.stdout.flush()
|
|
time.sleep(30)
|
|
""",
|
|
generation_backend_max_output_bytes=20,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.OUTPUT_TOO_LARGE
|
|
child_pid = int(pid_file.read_text(encoding="utf-8"))
|
|
deadline = time.time() + 3
|
|
while time.time() < deadline:
|
|
try:
|
|
os.kill(child_pid, 0)
|
|
except OSError:
|
|
break
|
|
time.sleep(0.05)
|
|
else:
|
|
pytest.fail("child process was not terminated with the process group")
|
|
|
|
|
|
def test_output_last_message_too_large(tmp_path: Path) -> None:
|
|
script = _script(
|
|
tmp_path,
|
|
"""
|
|
import sys
|
|
args = sys.argv[1:]
|
|
output_path = args[args.index("--output-last-message") + 1]
|
|
with open(output_path, "w", encoding="utf-8") as handle:
|
|
handle.write("x" * 100)
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(script,),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(generation_backend_max_output_bytes=20), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.OUTPUT_TOO_LARGE
|
|
|
|
|
|
def test_output_last_message_total_limit_includes_stdio(tmp_path: Path) -> None:
|
|
script = _script(
|
|
tmp_path,
|
|
"""
|
|
import sys
|
|
args = sys.argv[1:]
|
|
output_path = args[args.index("--output-last-message") + 1]
|
|
print("stdout bytes")
|
|
with open(output_path, "w", encoding="utf-8") as handle:
|
|
handle.write("final bytes")
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(script,),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(generation_backend_max_output_bytes=20), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.OUTPUT_TOO_LARGE
|
|
|
|
|
|
def test_empty_stdout_returns_empty_output(tmp_path: Path) -> None:
|
|
backend = _backend(tmp_path, "")
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.EMPTY_OUTPUT
|
|
assert exc_info.value.details["reason"] == "empty_stdout"
|
|
|
|
|
|
def test_missing_output_last_message_returns_empty_output(tmp_path: Path) -> None:
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(_script(tmp_path, "print('metadata only')"),),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.EMPTY_OUTPUT
|
|
assert exc_info.value.details["reason"] == "missing_last_message_output"
|
|
assert exc_info.value.details["output_source"] == "output_last_message"
|
|
|
|
|
|
def test_non_zero_exit_maps_login_required(tmp_path: Path) -> None:
|
|
backend = _backend(
|
|
tmp_path,
|
|
"""
|
|
import sys
|
|
print('not authenticated, please login', file=sys.stderr)
|
|
raise SystemExit(2)
|
|
""",
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.LOGIN_REQUIRED
|
|
assert exc_info.value.details["returncode"] == 2
|
|
|
|
|
|
def test_non_zero_exit_maps_cli_contract_unsupported(tmp_path: Path) -> None:
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(
|
|
_script(
|
|
tmp_path,
|
|
"""
|
|
import sys
|
|
print("error: unexpected argument '--output-last-message' found", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
""",
|
|
),
|
|
),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
assert exc_info.value.fallbackable is True
|
|
assert exc_info.value.details["reason"] == "cli_contract_unsupported"
|
|
assert exc_info.value.details["returncode"] == 2
|
|
assert "--output-last-message" in exc_info.value.details["stderr_preview"]
|
|
|
|
|
|
def test_claude_unknown_contract_arg_is_capability_unsupported_without_retry(tmp_path: Path) -> None:
|
|
argv_path = tmp_path / "argv.json"
|
|
count_path = tmp_path / "count.txt"
|
|
script = _script(
|
|
tmp_path,
|
|
f"""
|
|
import json, pathlib, sys
|
|
argv_path = pathlib.Path({str(argv_path)!r})
|
|
count_path = pathlib.Path({str(count_path)!r})
|
|
count = int(count_path.read_text(encoding="utf-8")) if count_path.exists() else 0
|
|
count_path.write_text(str(count + 1), encoding="utf-8")
|
|
argv = sys.argv[1:]
|
|
argv_path.write_text(json.dumps(argv), encoding="utf-8")
|
|
if "--strict-mcp-config" in argv:
|
|
print("error: unknown option '--strict-mcp-config'", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
print(json.dumps({{"type": "result", "subtype": "success", "result": "{{\\"sentiment_score\\": 90}}"}}))
|
|
""",
|
|
)
|
|
preset = LocalCliPreset(
|
|
preset_id="claude_code_cli",
|
|
executable=sys.executable,
|
|
argv=(script, *CLAUDE_CODE_CLI_PRESET.argv),
|
|
display_name="Mock Claude Code CLI",
|
|
extractor=CLAUDE_CODE_CLI_PRESET.extractor,
|
|
contract_args=CLAUDE_CODE_CLI_PRESET.contract_args,
|
|
)
|
|
backend = LocalCliGenerationBackend(
|
|
_config(generation_backend="claude_code_cli"),
|
|
preset=preset,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
runtime_argv = json.loads(argv_path.read_text(encoding="utf-8"))
|
|
assert exc_info.value.error_code is GenerationErrorCode.CAPABILITY_UNSUPPORTED
|
|
assert exc_info.value.details["reason"] == "cli_contract_unsupported"
|
|
assert "--strict-mcp-config" in runtime_argv
|
|
assert count_path.read_text(encoding="utf-8") == "1"
|
|
|
|
|
|
def test_non_zero_exit_mentions_preset_arg_without_unknown_marker_stays_generic(tmp_path: Path) -> None:
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(
|
|
_script(
|
|
tmp_path,
|
|
"""
|
|
import sys
|
|
print("failed while writing --output-last-message file", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
""",
|
|
),
|
|
),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.NON_ZERO_EXIT
|
|
assert exc_info.value.details["reason"] == "non_zero_exit"
|
|
|
|
|
|
def test_non_zero_exit_with_missing_last_message_still_maps_login_required(tmp_path: Path) -> None:
|
|
preset = LocalCliPreset(
|
|
"codex_cli",
|
|
sys.executable,
|
|
(
|
|
_script(
|
|
tmp_path,
|
|
"""
|
|
import sys
|
|
print("not authenticated, please login", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
""",
|
|
),
|
|
),
|
|
"Mock CLI",
|
|
output_last_message_arg="--output-last-message",
|
|
)
|
|
backend = LocalCliGenerationBackend(_config(), preset=preset)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.LOGIN_REQUIRED
|
|
assert exc_info.value.details["reason"] == "login_required"
|
|
|
|
|
|
def test_process_start_error_diagnostics_are_redacted(monkeypatch) -> None:
|
|
home_path = Path.home()
|
|
executable_path = str(home_path / "secret" / "bin" / "codex")
|
|
monkeypatch.setattr("src.llm.local_cli_backend.shutil.which", lambda _cmd: executable_path)
|
|
monkeypatch.setattr("src.llm.local_cli_backend.os.access", lambda _path, _mode: True)
|
|
|
|
def _raise_os_error(*_args, **_kwargs):
|
|
raise OSError(f"Exec format error: {executable_path} sk-secretsecretsecret")
|
|
|
|
monkeypatch.setattr("src.llm.local_cli_backend.subprocess.Popen", _raise_os_error)
|
|
backend = LocalCliGenerationBackend(_config())
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.UNKNOWN_BACKEND_ERROR
|
|
error = exc_info.value.details["error"]
|
|
assert str(home_path) not in error
|
|
assert "sk-secret" not in error
|
|
|
|
|
|
def test_prompt_is_passed_as_stdin_file_not_pipe(tmp_path: Path, monkeypatch) -> None:
|
|
captured = {}
|
|
|
|
def _raise_os_error(*_args, **kwargs):
|
|
stdin = kwargs.get("stdin")
|
|
captured["stdin"] = stdin
|
|
captured["stdin_closed_at_popen"] = getattr(stdin, "closed", True)
|
|
raise OSError("mock start failure")
|
|
|
|
monkeypatch.setattr("src.llm.local_cli_backend.subprocess.Popen", _raise_os_error)
|
|
backend = _backend(tmp_path, "print('unused')")
|
|
|
|
with pytest.raises(GenerationError):
|
|
backend.generate("x" * 200000, {})
|
|
|
|
stdin = captured["stdin"]
|
|
assert stdin is not subprocess.PIPE
|
|
assert hasattr(stdin, "fileno")
|
|
assert not captured["stdin_closed_at_popen"]
|
|
|
|
|
|
def test_timeout_kills_process_group(tmp_path: Path) -> None:
|
|
pid_file = tmp_path / "child.pid"
|
|
backend = _backend(
|
|
tmp_path,
|
|
f"""
|
|
import subprocess, sys, time
|
|
child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
|
|
open({str(pid_file)!r}, "w", encoding="utf-8").write(str(child.pid))
|
|
time.sleep(30)
|
|
""",
|
|
generation_backend_timeout_seconds=1,
|
|
)
|
|
|
|
with pytest.raises(GenerationError) as exc_info:
|
|
backend.generate("prompt", {})
|
|
|
|
assert exc_info.value.error_code is GenerationErrorCode.TIMEOUT
|
|
child_pid = int(pid_file.read_text(encoding="utf-8"))
|
|
deadline = time.time() + 3
|
|
while time.time() < deadline:
|
|
try:
|
|
os.kill(child_pid, 0)
|
|
except OSError:
|
|
break
|
|
time.sleep(0.05)
|
|
else:
|
|
pytest.fail("child process was not terminated with the process group")
|
|
|
|
|
|
def test_env_allowlist_and_denylist(monkeypatch) -> None:
|
|
monkeypatch.setenv("PATH", "/bin")
|
|
monkeypatch.setenv("HOME", "/tmp/home")
|
|
monkeypatch.setenv("CODEX_HOME", "/tmp/codex-home")
|
|
monkeypatch.setenv("LC_MESSAGES", "C")
|
|
monkeypatch.setenv("UNRELATED_VALUE", "leak")
|
|
monkeypatch.setenv("CODEX_CLI_TOKEN", "codex-secret")
|
|
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-secret")
|
|
monkeypatch.setenv("ANTHROPIC_MODEL", "claude")
|
|
monkeypatch.setenv("CLAUDE_CONFIG_DIR", "/tmp/claude")
|
|
monkeypatch.setenv("OPENCODE_CONFIG_CONTENT", "{}")
|
|
monkeypatch.setenv("OPENAI_API_KEY", "sk-secret")
|
|
monkeypatch.setenv("WEBHOOK_TOKEN", "token")
|
|
monkeypatch.setenv("AUTHORIZATION", "Bearer token")
|
|
|
|
child_env = build_local_cli_env()
|
|
|
|
assert child_env["PATH"] == "/bin"
|
|
assert child_env["HOME"] == "/tmp/home"
|
|
assert child_env["CODEX_HOME"] == "/tmp/codex-home"
|
|
assert child_env["LC_MESSAGES"] == "C"
|
|
assert "UNRELATED_VALUE" not in child_env
|
|
assert "CODEX_CLI_TOKEN" not in child_env
|
|
assert "ANTHROPIC_API_KEY" not in child_env
|
|
assert "ANTHROPIC_MODEL" not in child_env
|
|
assert "CLAUDE_CONFIG_DIR" not in child_env
|
|
assert "OPENCODE_CONFIG_CONTENT" not in child_env
|
|
assert "OPENAI_API_KEY" not in child_env
|
|
assert "WEBHOOK_TOKEN" not in child_env
|
|
assert "AUTHORIZATION" not in child_env
|
|
|
|
|
|
def test_env_allowlist_preserves_windows_runtime_context() -> None:
|
|
source = {
|
|
"Path": r"C:\Users\tester\AppData\Local\Microsoft\WindowsApps",
|
|
"SystemRoot": r"C:\Windows",
|
|
"WINDIR": r"C:\Windows",
|
|
"PATHEXT": ".COM;.EXE;.BAT;.CMD",
|
|
"ComSpec": r"C:\Windows\System32\cmd.exe",
|
|
"USERPROFILE": r"C:\Users\tester",
|
|
"APPDATA": r"C:\Users\tester\AppData\Roaming",
|
|
"LOCALAPPDATA": r"C:\Users\tester\AppData\Local",
|
|
"HOMEDRIVE": "C:",
|
|
"HOMEPATH": r"\Users\tester",
|
|
"OPENAI_API_KEY": "sk-secret",
|
|
"UNRELATED_VALUE": "leak",
|
|
}
|
|
|
|
child_env = build_local_cli_env(source)
|
|
|
|
for key in (
|
|
"Path",
|
|
"SystemRoot",
|
|
"WINDIR",
|
|
"PATHEXT",
|
|
"ComSpec",
|
|
"USERPROFILE",
|
|
"HOMEDRIVE",
|
|
"HOMEPATH",
|
|
):
|
|
assert child_env[key] == source[key]
|
|
assert "APPDATA" not in child_env
|
|
assert "LOCALAPPDATA" not in child_env
|
|
assert "OPENAI_API_KEY" not in child_env
|
|
assert "UNRELATED_VALUE" not in child_env
|
|
|
|
|
|
def test_generate_passes_allowlisted_windows_context_to_child_env(monkeypatch, tmp_path: Path) -> None:
|
|
windows_context = {
|
|
"SystemRoot": r"C:\Windows",
|
|
"WINDIR": r"C:\Windows",
|
|
"PATHEXT": ".COM;.EXE;.BAT;.CMD",
|
|
"ComSpec": r"C:\Windows\System32\cmd.exe",
|
|
"USERPROFILE": r"C:\Users\tester",
|
|
"HOMEDRIVE": "C:",
|
|
"HOMEPATH": r"\Users\tester",
|
|
}
|
|
for key, value in windows_context.items():
|
|
monkeypatch.setenv(key, value)
|
|
monkeypatch.setenv("APPDATA", r"C:\Users\tester\AppData\Roaming")
|
|
monkeypatch.setenv("LOCALAPPDATA", r"C:\Users\tester\AppData\Local")
|
|
monkeypatch.setenv("OPENAI_API_KEY", "sk-secret")
|
|
monkeypatch.setenv("UNRELATED_VALUE", "leak")
|
|
|
|
backend = _backend(
|
|
tmp_path,
|
|
"""
|
|
import json, os
|
|
keys = [
|
|
"SystemRoot",
|
|
"WINDIR",
|
|
"PATHEXT",
|
|
"ComSpec",
|
|
"USERPROFILE",
|
|
"APPDATA",
|
|
"LOCALAPPDATA",
|
|
"HOMEDRIVE",
|
|
"HOMEPATH",
|
|
"OPENAI_API_KEY",
|
|
"UNRELATED_VALUE",
|
|
]
|
|
print(json.dumps({key: os.environ.get(key) for key in keys}, ensure_ascii=False))
|
|
""",
|
|
)
|
|
|
|
result = backend.generate("prompt", {})
|
|
payload = json.loads(result.text)
|
|
|
|
for key, value in windows_context.items():
|
|
assert payload[key] == value
|
|
assert payload["APPDATA"] is None
|
|
assert payload["LOCALAPPDATA"] is None
|
|
assert payload["OPENAI_API_KEY"] is None
|
|
assert payload["UNRELATED_VALUE"] is None
|
|
|
|
|
|
def test_popen_session_kwargs_are_platform_specific(monkeypatch) -> None:
|
|
monkeypatch.setattr(local_cli_backend_module.os, "name", "nt")
|
|
monkeypatch.setattr(
|
|
local_cli_backend_module.subprocess,
|
|
"CREATE_NEW_PROCESS_GROUP",
|
|
0x00000200,
|
|
raising=False,
|
|
)
|
|
|
|
assert local_cli_backend_module._popen_session_kwargs() == {
|
|
"creationflags": 0x00000200,
|
|
}
|
|
|
|
monkeypatch.setattr(local_cli_backend_module.os, "name", "posix")
|
|
|
|
assert local_cli_backend_module._popen_session_kwargs() == {
|
|
"start_new_session": True,
|
|
}
|
|
|
|
|
|
def test_windows_terminate_process_group_prefers_ctrl_break(monkeypatch) -> None:
|
|
class FakeProcess:
|
|
pid = 1234
|
|
|
|
def __init__(self) -> None:
|
|
self.signals = []
|
|
self.terminated = False
|
|
self.killed = False
|
|
|
|
def poll(self):
|
|
return None
|
|
|
|
def send_signal(self, sig):
|
|
self.signals.append(sig)
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
def terminate(self):
|
|
self.terminated = True
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
|
|
monkeypatch.setattr(local_cli_backend_module.os, "name", "nt")
|
|
monkeypatch.setattr(
|
|
local_cli_backend_module.signal,
|
|
"CTRL_BREAK_EVENT",
|
|
1,
|
|
raising=False,
|
|
)
|
|
process = FakeProcess()
|
|
|
|
LocalCliGenerationBackend._terminate_process_group(process)
|
|
|
|
assert process.signals == [1]
|
|
assert process.terminated is False
|
|
assert process.killed is False
|
|
|
|
|
|
def test_windows_terminate_process_group_falls_back_to_kill(monkeypatch) -> None:
|
|
class FakeProcess:
|
|
pid = 1234
|
|
|
|
def __init__(self) -> None:
|
|
self.signals = []
|
|
self.terminated = False
|
|
self.killed = False
|
|
self._wait_calls = 0
|
|
|
|
def poll(self):
|
|
return None
|
|
|
|
def send_signal(self, sig):
|
|
self.signals.append(sig)
|
|
raise OSError("no console")
|
|
|
|
def wait(self, timeout=None):
|
|
self._wait_calls += 1
|
|
if self._wait_calls == 1:
|
|
raise subprocess.TimeoutExpired(cmd="mock", timeout=timeout)
|
|
return 0
|
|
|
|
def terminate(self):
|
|
self.terminated = True
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
|
|
monkeypatch.setattr(local_cli_backend_module.os, "name", "nt")
|
|
monkeypatch.setattr(
|
|
local_cli_backend_module.signal,
|
|
"CTRL_BREAK_EVENT",
|
|
1,
|
|
raising=False,
|
|
)
|
|
process = FakeProcess()
|
|
|
|
LocalCliGenerationBackend._terminate_process_group(process)
|
|
|
|
assert process.signals == [1]
|
|
assert process.terminated is True
|
|
assert process.killed is True
|
|
|
|
|
|
def test_diagnostics_redaction_and_truncation() -> None:
|
|
text = (
|
|
"Authorization: Bearer sk-abc123456789012345678901234567890 "
|
|
"https://user:pass@example.com/path "
|
|
+ "safe text " * 20
|
|
)
|
|
|
|
redacted = redact_diagnostic_text(text, home="/Users/example", limit=60)
|
|
|
|
assert "sk-abc" not in redacted
|
|
assert "user:pass" not in redacted
|
|
assert "<truncated>" in redacted
|
|
|
|
|
|
def test_diagnostics_redacts_webhook_urls_and_preserves_adjacent_normal_urls() -> None:
|
|
text = (
|
|
"slack=https://hooks.slack.com/services/T000/B000/super-secret "
|
|
"dingtalk=https://oapi.dingtalk.com/robot/send?access_token=abc123&foo=bar "
|
|
"docs=https://example.com/public/docs?foo=bar"
|
|
)
|
|
|
|
redacted = redact_diagnostic_text(text, limit=1000)
|
|
|
|
assert "hooks.slack.com" not in redacted
|
|
assert "oapi.dingtalk.com" not in redacted
|
|
assert "super-secret" not in redacted
|
|
assert "access_token" not in redacted
|
|
assert redacted.count("<redacted-url>") == 2
|
|
assert "https://example.com/public/docs?foo=bar" in redacted
|
|
|
|
|
|
def test_effective_local_cli_concurrency_uses_minimum() -> None:
|
|
assert effective_local_cli_concurrency(_config()) == 1
|
|
assert effective_local_cli_concurrency(
|
|
_config(generation_backend_max_concurrency=4, local_cli_backend_max_concurrency=2)
|
|
) == 2
|
|
assert effective_local_cli_concurrency(
|
|
_config(generation_backend_max_concurrency=1, local_cli_backend_max_concurrency=5)
|
|
) == 1
|
|
assert effective_local_cli_concurrency(
|
|
_config(generation_backend_max_concurrency=999, local_cli_backend_max_concurrency=999)
|
|
) == 4
|
|
|
|
|
|
def test_local_cli_concurrency_limit_serializes_subprocesses(tmp_path: Path) -> None:
|
|
events_dir = tmp_path / "events"
|
|
events_dir.mkdir()
|
|
backend = _backend(
|
|
tmp_path,
|
|
f"""
|
|
import json, os, pathlib, time
|
|
events_dir = pathlib.Path({str(events_dir)!r})
|
|
pid = os.getpid()
|
|
start = time.time()
|
|
time.sleep(0.25)
|
|
end = time.time()
|
|
(events_dir / f"{{pid}}.json").write_text(
|
|
json.dumps({{"start": start, "end": end}}),
|
|
encoding="utf-8",
|
|
)
|
|
print(json.dumps({{"sentiment_score": 60}}))
|
|
""",
|
|
generation_backend_max_concurrency=4,
|
|
local_cli_backend_max_concurrency=1,
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
results = list(executor.map(lambda _: backend.generate("prompt", {}), range(2)))
|
|
|
|
assert [json.loads(result.text)["sentiment_score"] for result in results] == [60, 60]
|
|
intervals = [
|
|
json.loads(path.read_text(encoding="utf-8"))
|
|
for path in events_dir.glob("*.json")
|
|
]
|
|
assert len(intervals) == 2
|
|
intervals.sort(key=lambda item: item["start"])
|
|
assert intervals[1]["start"] >= intervals[0]["end"]
|