host
4cd9b5eaa4
新增 run_tests.py：三套件统一入口，只加仓库根（deploy 套件 再加 deploy/ 目录在根之前），各套件走独立子进程；verify 作 业改走运行器，coverage 作业因子进程测不到覆盖率保持原 discover 命令。tests/ 13 个文件的 sys.path 前置 import 全部 改回正常 import（含 test_host_trace 的 try/except 回落）； deploy 的 deploy 目录插入与 verify-multiuser 的 spec 加载按 陷阱 2 如实保留（仅删 verify-multiuser 冗余的根插入）。 .coveragerc omit 加 */run_tests.py，使新运行器文件不计入 生产覆盖率。门槛未动（仍 75%），未补新测试。
62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
"""host.py goal-resume 断点丢失守卫：state 里没有任务时拒绝不炸。
|
||
|
||
2026-10-29 长任务实弹：断点续跑在 state 里找不到任务时抛
|
||
KeyError traceback，worker 把无结论异常退出判成 interrupted，
|
||
自动恢复再排队、再 KeyError，12 秒一轮空转到 50 次上限。
|
||
守卫后：打印拒绝行、正常返回，state 文件不动、线程不恢复。
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from contextlib import redirect_stdout
|
||
from unittest import mock
|
||
import host
|
||
|
||
|
||
class FakeServer(object):
|
||
def __init__(self):
|
||
self.resumed = []
|
||
|
||
def initialize(self):
|
||
pass
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
def resume_thread(self, thread_id):
|
||
self.resumed.append(thread_id)
|
||
|
||
|
||
class GoalResumeMissingTest(unittest.TestCase):
|
||
def test_missing_task_refuses_without_traceback(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state_path = os.path.join(tmp, "host-state.json")
|
||
cfg_path = os.path.join(tmp, "host-config.json")
|
||
with open(state_path, "w", encoding="utf-8") as f:
|
||
json.dump({"tasks": {"g": {"name": "g",
|
||
"thread_id": "x",
|
||
"status": "goal-running"}},
|
||
"automations": []}, f)
|
||
with open(cfg_path, "w", encoding="utf-8") as f:
|
||
json.dump({}, f)
|
||
fake = FakeServer()
|
||
argv = ["host.py", "--state", state_path,
|
||
"--config", cfg_path,
|
||
"goal-resume", "--task", "missing-task"]
|
||
out = io.StringIO()
|
||
with mock.patch.object(sys, "argv", argv):
|
||
with mock.patch.object(host, "Server",
|
||
lambda **kw: fake):
|
||
with redirect_stdout(out):
|
||
host.main()
|
||
self.assertIn("goal-resume 拒绝", out.getvalue())
|
||
self.assertIn("missing-task", out.getvalue())
|
||
self.assertEqual(fake.resumed, [])
|
||
with open(state_path, encoding="utf-8") as f:
|
||
saved = json.load(f)
|
||
self.assertEqual(list(saved["tasks"].keys()), ["g"])
|