fix(host): _load_config_file 与 init 的 open() 改用 with 关闭

行号纠正:摘要的 2199/2443 是另一基线行号;origin/main
aebc623 上两处是 _load_config_file 的 json.load(open(path))
与 main() init 分支的 json.dump(init_cfg, open(...)),
2443 行在 aebc623 上是 if not args.repo: 并无 open。
只改资源关闭:open 参数、读写内容、错误语义(坏 JSON 仍
转 RuntimeError)均不变,不加 encoding、不改其它裸 open()。
回归:ConfigFileCloseTest(3 项),旧代码上 3 项全 fail、
新代码上全 pass(warnings 过滤 + gc 证明无 ResourceWarning)。
This commit is contained in:
muse authored and host committed 2026-11-03 02:49:25 +00:00
1 parent e768959c93
commit 1156d3c754
2 files changed
+77 -3

No files matched your search

+4 -3
View File
@@ -2210,7 +2210,8 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
def _load_config_file(path): def _load_config_file(path):
if path and os.path.exists(path): if path and os.path.exists(path):
try: try:
return json.load(open(path)) with open(path) as f:
return json.load(f)
except Exception as e: except Exception as e:
raise RuntimeError(f"配置文件读失败 {path}: {e}") raise RuntimeError(f"配置文件读失败 {path}: {e}")
return {} return {}
@@ -2454,8 +2455,8 @@ def main():
"max_daily_turns 是按 turn 数算的预算(协议不暴露 token 数);" "max_daily_turns 是按 turn 数算的预算(协议不暴露 token 数);"
"notify_command 是失败时跑的本地 shell 命令,可用 {task} {status}", "notify_command 是失败时跑的本地 shell 命令,可用 {task} {status}",
} }
json.dump(init_cfg, open(config_path, "w"), indent=2, with open(config_path, "w") as f:
ensure_ascii=False) json.dump(init_cfg, f, indent=2, ensure_ascii=False)
print(f"配置文件已写:{config_path},请按需改 repo/model 后再用") print(f"配置文件已写:{config_path},请按需改 repo/model 后再用")
return return
# 有效配置(init/config 之外的命令都用它) # 有效配置(init/config 之外的命令都用它)
+73
View File
@@ -80,5 +80,78 @@ class ReadLoopNonDictTest(unittest.TestCase):
self.assertEqual(srv.events[0]["method"], "thread/started") 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__": if __name__ == "__main__":
unittest.main() unittest.main()