codex-app-server/tests/test_host_bugfix_host.py

"""host.py 覆盖率 bugfix 轨（host 分工）回归测试。

只覆盖本分工的三项：_read_loop 非 dict 消息、配置/init 的
open() 关闭、daemon 循环的 --state 透传。不碰其它轨。
"""
import contextlib
import io
import json
import queue
import sys
import threading
import unittest
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT))

import host


def _make_read_loop_server(lines, pending_id=7):
    """不拉起真实 app-server 的 Server：只给 _read_loop 供料。

    proc.stdout 直接用字节行列表；_read_loop 对它做
    `for raw in self.proc.stdout`，与真实 PIPE 的逐行迭代
    语义一致。脏消息后仍能送达正常响应，即证明读循环
    没有在脏消息处提前退出（旧代码会在第一条非 dict 处
    抛 AttributeError/TypeError，被外层 except 吞掉后退出）。
    """
    srv = host.Server.__new__(host.Server)
    srv.approval = "deny"
    srv.pending_lock = threading.Lock()
    srv.pending = {pending_id: queue.Queue()}
    srv.events_lock = threading.Lock()
    srv.events = []
    srv.turn_queues = {}
    srv.trace_path = None
    srv._trace_lines = 0

    class FakeProc:
        def __init__(self, stdout_lines):
            self.stdout = stdout_lines

    srv.proc = FakeProc(lines)
    return srv


class ReadLoopNonDictTest(unittest.TestCase):
    def test_non_dict_json_then_normal_response(self):
        lines = [
            b'""\n',            # 合法 JSON，值是 str（旧代码致死）
            b'"hello"\n',       # 合法 JSON，值是 str
            b'123\n',           # 合法 JSON，值是 int
            b'[1,2]\n',         # 合法 JSON，值是 list
            b'null\n',          # 合法 JSON，值是 None
            b'true\n',          # 合法 JSON，值是 bool
            b'not-json{\n',     # 畸形 JSON：旧代码本已 continue
            b'{"jsonrpc": "2.0", "id": 7, '
            b'"result": {"ok": true}}\n',
        ]
        srv = _make_read_loop_server(lines)
        err = io.StringIO()
        with contextlib.redirect_stderr(err):
            srv._read_loop()
        resp = srv.pending[7].get(timeout=1)
        self.assertEqual(resp["result"], {"ok": True})
        # 脏消息没有被当成通知塞进 events，也没有被当成请求处理。
        self.assertEqual(srv.events, [])
        self.assertIn("skip non-dict message", err.getvalue())

    def test_non_dict_json_then_normal_notification(self):
        lines = [
            b'""\n',
            b'{"jsonrpc": "2.0", "method": "thread/started", '
            b'"params": {"threadId": "t1"}}\n',
        ]
        srv = _make_read_loop_server(lines)
        srv._read_loop()
        self.assertEqual(len(srv.events), 1)
        self.assertEqual(srv.events[0]["method"], "thread/started")


class ConfigFileCloseTest(unittest.TestCase):
    """第 2 项：_load_config_file 与 init 的 open() 必须关闭。

    行号纠正：任务摘要的 2199/2443 是另一基线（coverage85
    worktree）的行号；origin/main aebc623 上两处分别是
    _load_config_file 的 json.load(open(path)) 与 main() init
    分支的 json.dump(init_cfg, open(config_path, "w"))，
    2443 行在 aebc623 上是 `if not args.repo:`，并无 open。
    本测试只针对这两处生产路径，不扩大到其它裸 open()。
    """

    def test_load_config_file_no_resource_warning(self):
        import gc
        import os
        import tempfile
        import warnings
        with tempfile.TemporaryDirectory() as tmp:
            cfg_path = os.path.join(tmp, "cfg.json")
            with open(cfg_path, "w", encoding="utf-8") as f:
                json.dump({"model": "x"}, f)
            with warnings.catch_warnings(record=True) as caught:
                warnings.simplefilter("always", ResourceWarning)
                value = host._load_config_file(cfg_path)
                gc.collect()
            self.assertEqual(value, {"model": "x"})
            self.assertEqual(
                [str(w.message) for w in caught
                 if issubclass(w.category, ResourceWarning)], [])

    def test_load_config_file_bad_json_still_raises_and_closes(self):
        import gc
        import os
        import tempfile
        import warnings
        with tempfile.TemporaryDirectory() as tmp:
            cfg_path = os.path.join(tmp, "cfg.json")
            with open(cfg_path, "w", encoding="utf-8") as f:
                f.write("{not json")
            with warnings.catch_warnings(record=True) as caught:
                warnings.simplefilter("always", ResourceWarning)
                with self.assertRaises(RuntimeError):
                    host._load_config_file(cfg_path)
                gc.collect()
            self.assertEqual(
                [str(w.message) for w in caught
                 if issubclass(w.category, ResourceWarning)], [])

    def test_init_writes_config_no_resource_warning(self):
        import gc
        import os
        import tempfile
        import warnings
        from unittest import mock
        with tempfile.TemporaryDirectory() as tmp:
            cfg_path = os.path.join(tmp, "new-config.json")
            argv = ["host.py", "--config", cfg_path, "init"]
            out = io.StringIO()
            with warnings.catch_warnings(record=True) as caught:
                warnings.simplefilter("always", ResourceWarning)
                with mock.patch.object(sys, "argv", argv):
                    with contextlib.redirect_stdout(out):
                        host.main()
                gc.collect()
            with open(cfg_path, encoding="utf-8") as f:
                written = json.load(f)
            self.assertEqual(written["state"], "host-state.json")
            self.assertEqual(written["model"], "gpt-5.6-terra")
            self.assertIn("配置文件已写", out.getvalue())
            self.assertEqual(
                [str(w.message) for w in caught
                 if issubclass(w.category, ResourceWarning)], [])


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