codex-app-server/tests/test_host_trace.py

"""实时链路（host 侧）单测：通知摘要器 + 链路文件追加。

不启动 app-server：Server 实例用 object.__new__ 绕开 __init__
（__init__ 会拉起 codex 子进程），只测与进程无关的方法。
"""

import json
import os
import sys
import tempfile
import unittest
from pathlib import Path

try:
    import host
    from host import Server, TRACE_MAX_LINES, _summarize_trace_item
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    import host
    from host import Server, TRACE_MAX_LINES, _summarize_trace_item


def _msg(item, ts=1700000000000):
    return {"method": "item/completed",
            "params": {"item": item, "completedAtMs": ts}}


class SummarizeTraceItemTestCase(unittest.TestCase):
    def test_exec_truncates_command(self):
        entries = _summarize_trace_item(
            "item/completed",
            {"item": {"type": "commandExecution", "id": "c1",
                      "command": "x" * 600, "cwd": "/w",
                      "exitCode": 0, "durationMs": 12},
             "completedAtMs": 1700000000000})
        self.assertEqual(len(entries), 1)
        self.assertEqual(entries[0]["kind"], "exec")
        self.assertEqual(len(entries[0]["command"]), 500)
        self.assertEqual(entries[0]["exitCode"], 0)
        self.assertEqual(entries[0]["ts"], 1700000000000)

    def test_file_change_split_per_change(self):
        entries = _summarize_trace_item(
            "item/completed",
            {"item": {"type": "fileChange", "id": "f1", "status": "completed",
                      "changes": [{"path": "a.py", "kind": "update"},
                                  {"path": "b.py", "kind": "add"}]},
             "startedAtMs": 1700000000001})
        self.assertEqual([e["kind"] for e in entries], ["file", "file"])
        self.assertEqual(entries[0]["id"], "f1:a.py")
        self.assertEqual(entries[1]["path"], "b.py")

    def test_message_reasoning_tool_call_plan(self):
        cases = [
            ({"type": "agentMessage", "id": "m1", "phase": "final",
              "text": "done"}, "message"),
            ({"type": "reasoning", "id": "r1",
              "summary": ["a", "b"]}, "reasoning"),
            ({"type": "mcpToolCall", "id": "t1", "server": "s",
              "tool": "read", "status": "completed"}, "tool"),
            ({"type": "functionCallOutput", "id": "o1",
              "name": "read_file"}, "call"),
            ({"type": "plan", "id": "p1", "text": "step"}, "plan"),
        ]
        for item, kind in cases:
            entries = _summarize_trace_item("item/completed",
                                            {"item": item})
            self.assertEqual(len(entries), 1)
            self.assertEqual(entries[0]["kind"], kind)

    def test_skipped_notifications(self):
        self.assertEqual(_summarize_trace_item(
            "item/completed", {"item": {"type": "userMessage", "id": "u1",
                                        "text": "hi"}}), [])
        self.assertEqual(_summarize_trace_item(
            "item/completed", {"item": {"type": "agentMessage", "id": "m2",
                                        "text": ""}}), [])
        self.assertEqual(_summarize_trace_item("item/completed", {}), [])
        self.assertEqual(_summarize_trace_item(
            "item/completed", {"item": {"type": "unknownThing",
                                        "id": "z"}}), [])


class TraceFileTestCase(unittest.TestCase):
    def _bare_server(self):
        srv = object.__new__(Server)
        srv.trace_path = None
        srv._trace_lines = 0
        return srv

    def test_set_trace_file_and_append(self):
        with tempfile.TemporaryDirectory() as d:
            state_path = os.path.join(d, "host-state.json")
            srv = self._bare_server()
            srv.set_trace_file(state_path, "abc/def?id=1")
            self.assertEqual(
                srv.trace_path,
                os.path.join(d, "host-traces", "abc-def-id-1.jsonl"))
            srv._append_trace(_msg({"type": "commandExecution", "id": "c1",
                                    "command": "ls", "cwd": "/w",
                                    "exitCode": 0, "durationMs": 3}))
            srv._append_trace(_msg({"type": "userMessage", "id": "u1",
                                    "text": "hi"}))
            with open(srv.trace_path, encoding="utf-8") as f:
                lines = f.read().strip().split("\n")
            self.assertEqual(len(lines), 1)
            self.assertEqual(json.loads(lines[0])["kind"], "exec")

    def test_set_trace_file_failure_keeps_none(self):
        with tempfile.TemporaryDirectory() as d:
            # host-traces 已被普通文件占住，建目录失败时不抛、不影响回合。
            blocker = os.path.join(d, "host-traces")
            with open(blocker, "w", encoding="utf-8") as f:
                f.write("x")
            srv = self._bare_server()
            srv.set_trace_file(os.path.join(d, "host-state.json"), "t1")
            self.assertIsNone(srv.trace_path)

    def test_append_stops_at_max_lines(self):
        with tempfile.TemporaryDirectory() as d:
            srv = self._bare_server()
            srv.trace_path = os.path.join(d, "t.jsonl")
            srv._trace_lines = TRACE_MAX_LINES - 1
            srv._append_trace(_msg({"type": "commandExecution", "id": "c1",
                                    "command": "a"}))
            srv._append_trace(_msg({"type": "commandExecution", "id": "c2",
                                    "command": "b"}))
            with open(srv.trace_path, encoding="utf-8") as f:
                lines = f.read().strip().split("\n")
            self.assertEqual(len(lines), 1)
            self.assertEqual(json.loads(lines[0])["id"], "c1")


if __name__ == "__main__":
    unittest.main()