131 lines
4.2 KiB
Python
131 lines
4.2 KiB
Python
import asyncio
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
from agentci.adapters.codex import CodexClient, _session_id
|
|
from agentci.domain.models import AgentResult
|
|
|
|
|
|
def test_enables_codegraph_in_shared_codex_config() -> None:
|
|
root = Path(__file__).parents[1]
|
|
config = tomllib.loads((root / "codex" / "config.toml").read_text())
|
|
|
|
codegraph = config["mcp_servers"]["codegraph"]
|
|
assert codegraph["command"] == "codegraph"
|
|
assert codegraph["args"] == ["serve", "--mcp"]
|
|
assert codegraph["env"]["CODEGRAPH_TELEMETRY"] == "0"
|
|
|
|
|
|
def test_extracts_thread_id_from_jsonl() -> None:
|
|
output = '\n'.join(
|
|
[
|
|
'{"type":"turn.started"}',
|
|
'{"type":"thread.started","thread_id":"abc-123"}',
|
|
"not json",
|
|
]
|
|
)
|
|
assert _session_id(output) == "abc-123"
|
|
|
|
|
|
def test_missing_thread_id_returns_none() -> None:
|
|
assert _session_id('{"type":"turn.completed"}') is None
|
|
|
|
|
|
def test_turns_skip_interactive_git_trust_check(tmp_path) -> None:
|
|
client = CodexClient(
|
|
codex_home=tmp_path / "codex",
|
|
schemas_dir=tmp_path / "schemas",
|
|
timeout_seconds=60,
|
|
research_model="gpt-5.6-luna",
|
|
research_reasoning="high",
|
|
context7_api_key=None,
|
|
)
|
|
|
|
args = client._turn_args("model", "medium", "agentci-read", "plan.json")
|
|
|
|
assert "--skip-git-repo-check" in args
|
|
|
|
|
|
def test_configures_research_agent_and_optional_context7_key(tmp_path) -> None:
|
|
client = CodexClient(
|
|
codex_home=tmp_path / "codex",
|
|
schemas_dir=tmp_path / "schemas",
|
|
timeout_seconds=60,
|
|
research_model="research-model",
|
|
research_reasoning="high",
|
|
context7_api_key="ctx7-secret",
|
|
)
|
|
|
|
agent = (tmp_path / "codex" / "agents" / "research.toml").read_text()
|
|
parsed = tomllib.loads(agent)
|
|
assert 'name = "research"' in agent
|
|
assert parsed["model"] == "research-model"
|
|
assert parsed["model_reasoning_effort"] == "high"
|
|
assert parsed["web_search"] == "live"
|
|
assert parsed["mcp_servers"]["codegraph"]["enabled"] is False
|
|
assert 'url = "https://mcp.context7.com/mcp"' in agent
|
|
assert 'url = "https://mcp.grep.app"' in agent
|
|
assert "ctx7-secret" not in agent
|
|
assert client._environment()["CONTEXT7_API_KEY"] == "ctx7-secret"
|
|
|
|
|
|
def test_exposes_development_tools_without_service_secrets(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("PATH", "/usr/bin")
|
|
monkeypatch.setenv("AGENTCI_PRIVATE_VALUE", "secret")
|
|
tools_bin = tmp_path / "tools" / "bin"
|
|
client = CodexClient(
|
|
codex_home=tmp_path / "codex",
|
|
schemas_dir=tmp_path / "schemas",
|
|
timeout_seconds=60,
|
|
research_model="gpt-5.6-luna",
|
|
research_reasoning="high",
|
|
context7_api_key=None,
|
|
tools_bin=tools_bin,
|
|
)
|
|
|
|
environment = client._environment()
|
|
|
|
assert environment["PATH"] == f"{tools_bin}:/usr/bin"
|
|
assert "AGENTCI_PRIVATE_VALUE" not in environment
|
|
|
|
|
|
async def test_invokes_codex_from_workflow_workspace(tmp_path, monkeypatch) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
captured: dict[str, object] = {}
|
|
|
|
class Process:
|
|
returncode = 0
|
|
|
|
async def communicate(self, prompt: bytes) -> tuple[bytes, bytes]:
|
|
assert prompt == b"prompt"
|
|
return b'{"type":"thread.started","thread_id":"thread"}', b""
|
|
|
|
async def create_subprocess_exec(*args, **kwargs):
|
|
captured["cwd"] = kwargs["cwd"]
|
|
output = Path(args[args.index("--output-last-message") + 1])
|
|
output.write_text( # noqa: ASYNC240 - tiny test-owned result file
|
|
'{"summary_markdown":"summary","tests":[]}'
|
|
)
|
|
return Process()
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", create_subprocess_exec)
|
|
client = CodexClient(
|
|
codex_home=tmp_path / "codex",
|
|
schemas_dir=tmp_path / "schemas",
|
|
timeout_seconds=60,
|
|
research_model="gpt-5.6-luna",
|
|
research_reasoning="high",
|
|
context7_api_key=None,
|
|
)
|
|
|
|
_, result = await client._invoke(
|
|
["codex", "exec", "-"],
|
|
"prompt",
|
|
AgentResult,
|
|
workspace=workspace,
|
|
)
|
|
|
|
assert captured["cwd"] == workspace
|
|
assert result.summary_markdown == "summary"
|