115 lines
3.6 KiB
Python
Vendored
115 lines
3.6 KiB
Python
Vendored
# -*- coding: utf-8 -*-
|
|
"""离线自检：_run_guarded 的超时治理（不 attach 任何游戏进程，不需要 frida）。
|
|
|
|
跑法（本机 python 无 frida ⇒ 自动走「无 Cancellable 降级」路径，正好验证降级）：
|
|
python _selftest_run_guarded.py
|
|
|
|
四项断言：
|
|
1. 正常调用 → 返回值正确、timeouts 归零、线程从账本摘除
|
|
2. 单次超时 → 抛「超时」、dead=True、卡住线程计入 stuck_thread_count
|
|
3. 连续超时达 STUCK_MAX_CONSECUTIVE → 抛「放弃该 script」
|
|
4. 新 script（模拟 load 成功清零）后计数从头开始
|
|
"""
|
|
import sys
|
|
import os
|
|
import time
|
|
import threading
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import loader_common as LC
|
|
|
|
|
|
class Fake(object):
|
|
"""只借用 Mod._run_guarded，不接触 frida。"""
|
|
|
|
def __init__(self):
|
|
self.name = "fake"
|
|
self.dead = False
|
|
self.timeouts = 0
|
|
|
|
_run_guarded = LC.Mod._run_guarded
|
|
|
|
|
|
def call(mod, fn, timeout):
|
|
box = {}
|
|
|
|
def run():
|
|
try:
|
|
box["r"] = fn()
|
|
except Exception as e:
|
|
box["e"] = e
|
|
|
|
return mod._run_guarded(run, box, "probe", timeout)
|
|
|
|
|
|
def main():
|
|
ok = 0
|
|
bad = 0
|
|
|
|
def check(label, cond, extra=""):
|
|
nonlocal ok, bad
|
|
if cond:
|
|
ok += 1
|
|
print(" [OK] %s %s" % (label, extra))
|
|
else:
|
|
bad += 1
|
|
print(" [FAIL] %s %s" % (label, extra))
|
|
|
|
print("STUCK_MAX_CONSECUTIVE =", LC.STUCK_MAX_CONSECUTIVE)
|
|
|
|
# 1) 正常调用
|
|
m = Fake()
|
|
r = call(m, lambda: 42, 2.0)
|
|
check("1 正常返回值", r == 42, "got=%r" % (r,))
|
|
check("1 timeouts 归零", m.timeouts == 0, "timeouts=%d" % m.timeouts)
|
|
check("1 未误判 dead", m.dead is False)
|
|
check("1 线程已摘除", LC.stuck_thread_count() == 0,
|
|
"stuck=%d" % LC.stuck_thread_count())
|
|
|
|
# 2) 单次超时（fn 睡 30 秒，保证 join 不回来）
|
|
m2 = Fake()
|
|
try:
|
|
call(m2, lambda: time.sleep(30), 0.5)
|
|
check("2 应抛超时", False, "没抛异常")
|
|
except RuntimeError as e:
|
|
check("2 抛超时", "超时" in str(e) and "连续超时" not in str(e), "msg=%s" % e)
|
|
check("2 超时即置 dead", m2.dead is True)
|
|
check("2 timeouts=1", m2.timeouts == 1, "timeouts=%d" % m2.timeouts)
|
|
check("2 卡住线程计入账本", LC.stuck_thread_count() == 1,
|
|
"stuck=%d" % LC.stuck_thread_count())
|
|
|
|
# 3) 连续超时到上限 ⇒ 快速失败
|
|
m3 = Fake()
|
|
msgs = []
|
|
for i in range(LC.STUCK_MAX_CONSECUTIVE):
|
|
try:
|
|
call(m3, lambda: time.sleep(30), 0.3)
|
|
except RuntimeError as e:
|
|
msgs.append(str(e))
|
|
check("3 抛了 %d 次" % LC.STUCK_MAX_CONSECUTIVE, len(msgs) == LC.STUCK_MAX_CONSECUTIVE,
|
|
"n=%d" % len(msgs))
|
|
check("3 最后一次是「放弃该 script」",
|
|
len(msgs) == LC.STUCK_MAX_CONSECUTIVE and "放弃该 script" in msgs[-1],
|
|
"last=%s" % (msgs[-1] if msgs else ""))
|
|
check("3 全部置 dead", m3.dead is True)
|
|
|
|
# 4) 新 script 清零（模拟 load() 成功）
|
|
m3.dead = False
|
|
m3.timeouts = 0
|
|
try:
|
|
call(m3, lambda: time.sleep(30), 0.3)
|
|
check("4 应抛超时", False)
|
|
except RuntimeError as e:
|
|
check("4 清零后重新计时（第 1 次只报超时）",
|
|
"超时" in str(e) and "放弃该 script" not in str(e), "msg=%s" % e)
|
|
check("4 timeouts=1", m3.timeouts == 1, "timeouts=%d" % m3.timeouts)
|
|
|
|
print()
|
|
print("结果: ok=%d fail=%d 卡住线程(残留, 进程退出才回收)=%d"
|
|
% (ok, bad, LC.stuck_thread_count()))
|
|
return 0 if bad == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|