103 lines
4.2 KiB
Python
103 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from datatest.ai import CodexCLIAdapter
|
|
|
|
|
|
class CodexAdapterTests(unittest.TestCase):
|
|
def test_uses_isolated_read_only_structured_invocation(self) -> None:
|
|
adapter = CodexCLIAdapter(executable="/usr/bin/true")
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
schema = Path(directory) / "schema.json"
|
|
schema.write_text('{"type":"object"}', encoding="utf-8")
|
|
|
|
def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
|
|
output_index = command.index("--output-last-message") + 1
|
|
Path(command[output_index]).write_text('{"ok":true}', encoding="utf-8")
|
|
self.assertIn("--ignore-user-config", command)
|
|
self.assertIn("--ephemeral", command)
|
|
self.assertEqual(command[command.index("--sandbox") + 1], "read-only")
|
|
self.assertEqual(command[command.index("--output-schema") + 1], str(schema.resolve()))
|
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
|
|
with patch("datatest.ai.subprocess.run", side_effect=fake_run):
|
|
result = adapter.run_structured("解析需求", {"content": "demo"}, schema)
|
|
|
|
self.assertEqual(result, {"ok": True})
|
|
|
|
def test_all_structured_output_objects_are_strict(self) -> None:
|
|
schemas = [
|
|
"requirement-extraction.schema.json",
|
|
"test-case.schema.json",
|
|
"case-agent-response.schema.json",
|
|
"failure-analysis.schema.json",
|
|
]
|
|
|
|
def validate_node(node: object, path: str) -> None:
|
|
if isinstance(node, dict):
|
|
if node.get("type") == "object":
|
|
self.assertIs(node.get("additionalProperties"), False, path)
|
|
properties = set(node.get("properties", {}))
|
|
self.assertEqual(set(node.get("required", [])), properties, path)
|
|
for key, value in node.items():
|
|
validate_node(value, f"{path}.{key}")
|
|
elif isinstance(node, list):
|
|
for index, value in enumerate(node):
|
|
validate_node(value, f"{path}[{index}]")
|
|
|
|
for name in schemas:
|
|
schema = json.loads((Path(__file__).resolve().parents[1] / "schemas" / name).read_text())
|
|
validate_node(schema, name)
|
|
|
|
def test_streams_jsonl_events_and_reads_schema_bound_final_message(self) -> None:
|
|
adapter = CodexCLIAdapter(executable="/usr/bin/codex")
|
|
received: list[dict[str, object]] = []
|
|
captured_command: list[str] = []
|
|
|
|
class FakeProcess:
|
|
def __init__(self) -> None:
|
|
self.stdin = io.StringIO()
|
|
self.stdout = io.StringIO(
|
|
'{"type":"thread.started","thread_id":"demo"}\n'
|
|
'{"type":"turn.started"}\n'
|
|
'{"type":"turn.completed","usage":{}}\n'
|
|
)
|
|
self.stderr = io.StringIO("")
|
|
|
|
def wait(self) -> int:
|
|
return 0
|
|
|
|
def kill(self) -> None:
|
|
pass
|
|
|
|
def fake_popen(command: list[str], **_: object) -> FakeProcess:
|
|
captured_command.extend(command)
|
|
output_index = command.index("--output-last-message") + 1
|
|
Path(command[output_index]).write_text('{"ok":true}', encoding="utf-8")
|
|
return FakeProcess()
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
schema = Path(directory) / "schema.json"
|
|
schema.write_text('{"type":"object"}', encoding="utf-8")
|
|
with patch("datatest.ai.subprocess.Popen", side_effect=fake_popen):
|
|
result = adapter.run_structured_streaming(
|
|
"调整案例", {"message": "demo"}, schema, received.append
|
|
)
|
|
|
|
self.assertIn("--json", captured_command)
|
|
self.assertEqual([event["type"] for event in received], [
|
|
"thread.started", "turn.started", "turn.completed",
|
|
])
|
|
self.assertEqual(result, {"ok": True})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|