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%），未补新测试。
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
"""host-state 并发保存回归测试（2026-10-29 断点丢失实弹）。
|
|
|
|
旧写法两处同时失效：tmp 文件名只含 pid（容器内 host.py 全是
|
|
pid 1，并发保存写同一个 tmp 交错出坏 JSON）；flock 在 runsc
|
|
容器间不互斥（实测双双 0 秒拿到排他锁）。坏 JSON 又被
|
|
_read_state_file 静默当成空 state，下一次保存合法地重写，
|
|
全部任务条目一起丢、版本从 390 跌回个位数。
|
|
|
|
本文件锁死三条修法：tmp 名每个写者唯一、mkdir 原子锁跨进程
|
|
互斥、磁盘已损坏时保存直接拒绝覆盖。
|
|
"""
|
|
|
|
import glob
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
import host
|
|
|
|
|
|
def _save_one(args):
|
|
path, name = args
|
|
state = {"tasks": {name: {"status": "running"}}, "automations": []}
|
|
host.save_state(path, state)
|
|
return name
|
|
|
|
|
|
def _tmp_name(_ignored):
|
|
return host._state_tmp_path("/tmp/x/host-state.json")
|
|
|
|
|
|
class StateTmpNameTest(unittest.TestCase):
|
|
def test_tmp_names_unique_in_process(self):
|
|
names = {host._state_tmp_path("/tmp/x/host-state.json")
|
|
for _ in range(100)}
|
|
self.assertEqual(len(names), 100)
|
|
|
|
def test_tmp_names_unique_across_processes(self):
|
|
ctx = multiprocessing.get_context("fork")
|
|
with ctx.Pool(8) as pool:
|
|
names = pool.map(_tmp_name, range(8))
|
|
self.assertEqual(len(set(names)), 8)
|
|
|
|
|
|
class StateMkdirLockTest(unittest.TestCase):
|
|
def test_concurrent_saves_all_tasks_survive(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "host-state.json")
|
|
ctx = multiprocessing.get_context("fork")
|
|
names = [f"task-{i}" for i in range(8)]
|
|
with ctx.Pool(8) as pool:
|
|
pool.map(_save_one, [(path, n) for n in names])
|
|
with open(path, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
self.assertEqual(set(data["tasks"]), set(names))
|
|
self.assertEqual(data["version"], 8)
|
|
self.assertFalse(glob.glob(path + ".tmp.*"))
|
|
self.assertFalse(os.path.exists(path + ".lockdir"))
|
|
|
|
def test_stale_lock_is_recovered(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "host-state.json")
|
|
lock_dir = path + ".lockdir"
|
|
os.mkdir(lock_dir)
|
|
old = time.time() - 1000
|
|
os.utime(lock_dir, (old, old))
|
|
host.save_state(path, {"tasks": {"a": {}}, "automations": []})
|
|
self.assertFalse(os.path.exists(lock_dir))
|
|
with open(path, encoding="utf-8") as f:
|
|
self.assertIn("a", json.load(f)["tasks"])
|
|
|
|
def test_fresh_lock_timeout_raises_instead_of_writing_unlocked(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "host-state.json")
|
|
os.mkdir(path + ".lockdir")
|
|
old_timeout = host._STATE_LOCK_TIMEOUT
|
|
host._STATE_LOCK_TIMEOUT = 0.3
|
|
try:
|
|
with self.assertRaises(RuntimeError):
|
|
host.save_state(path, {"tasks": {"a": {}},
|
|
"automations": []})
|
|
finally:
|
|
host._STATE_LOCK_TIMEOUT = old_timeout
|
|
self.assertFalse(os.path.exists(path))
|
|
|
|
|
|
class StateCorruptRefusalTest(unittest.TestCase):
|
|
def test_corrupt_disk_refuses_to_overwrite(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "host-state.json")
|
|
garbage = b"{not json, torn write"
|
|
with open(path, "wb") as f:
|
|
f.write(garbage)
|
|
with self.assertRaises(RuntimeError):
|
|
host.save_state(path, {"tasks": {"a": {}},
|
|
"automations": []})
|
|
with open(path, "rb") as f:
|
|
self.assertEqual(f.read(), garbage)
|
|
self.assertFalse(os.path.exists(path + ".bak"))
|
|
self.assertFalse(os.path.exists(path + ".lockdir"))
|
|
|
|
def test_wrong_shaped_disk_refuses_to_overwrite(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "host-state.json")
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump({"tasks": ["not", "an", "object"],
|
|
"automations": []}, f)
|
|
with self.assertRaises(RuntimeError):
|
|
host.save_state(path, {"tasks": {"a": {}},
|
|
"automations": []})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|