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()