170 lines
5.5 KiB
Python
170 lines
5.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""fix_selfcheck_exit.py —— 修复 selfcheck.py「打印 HAS FAILURES 但 rc=0」的门禁失真缺陷。
|
||
|
||
背景
|
||
----
|
||
自查脚本 scripts/selfcheck.py 原先在 __main__ 分支里无条件 sys.exit(0)（或不显式退出），
|
||
导致存在失败项时进程退出码仍为 0。QC / CI 以退出码为唯一门禁依据，等于门禁形同虚设。
|
||
|
||
本脚本对 selfcheck.py 的 `if __name__ == "__main__":` 尾块做**幂等外科替换**：
|
||
1. 捕获 main() 的返回值 / SystemExit.code / 未捕获异常 -> 基础退出码
|
||
2. 同时 tee 全部 stdout，扫描失败特征串（HAS FAILURES / FAILED / [FAIL] / 有失败项）
|
||
3. 基础退出码为 0 但输出含失败特征 -> 强制改写为 1
|
||
4. 注入标记 SELFCHECK_EXIT_PATCHED，重复执行不会二次改写
|
||
|
||
用法（机构工作空间根目录）::
|
||
|
||
python3 modules/pbl_blueprint/scripts/fix_selfcheck_exit.py # 应用补丁
|
||
python3 modules/pbl_blueprint/scripts/fix_selfcheck_exit.py --check # 只检查是否已打补丁
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
TARGET = os.path.join(HERE, "selfcheck.py")
|
||
MARKER = "SELFCHECK_EXIT_PATCHED"
|
||
|
||
MAIN_RE = re.compile(r'^if\s+__name__\s*==\s*[\'"]__main__[\'"]\s*:\s*$', re.MULTILINE)
|
||
|
||
PATCH_BLOCK = '''if __name__ == "__main__": # {marker} —— 退出码门禁：有失败项必须非 0
|
||
# 本块由 scripts/fix_selfcheck_exit.py 幂等注入，勿手工编辑。
|
||
import sys as _sys
|
||
import io as _io
|
||
import re as _re
|
||
import traceback as _tb
|
||
|
||
class _Tee(object):
|
||
"""同时写原始 stdout 与捕获缓冲，保证控制台输出不变。"""
|
||
|
||
def __init__(self, *streams):
|
||
self._streams = streams
|
||
|
||
def write(self, s):
|
||
for _st in self._streams:
|
||
try:
|
||
_st.write(s)
|
||
except Exception:
|
||
pass
|
||
return len(s)
|
||
|
||
def flush(self):
|
||
for _st in self._streams:
|
||
try:
|
||
_st.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self._streams[0], name)
|
||
|
||
_real_stdout = _sys.stdout
|
||
_buf = _io.StringIO()
|
||
_sys.stdout = _Tee(_real_stdout, _buf)
|
||
_rc = 0
|
||
try:
|
||
_entry = None
|
||
for _name in ("main", "run", "selfcheck", "run_selfcheck"):
|
||
_entry = globals().get(_name)
|
||
if callable(_entry):
|
||
break
|
||
if callable(_entry):
|
||
_rv = _entry()
|
||
if isinstance(_rv, bool):
|
||
_rc = 0 if _rv else 1
|
||
elif isinstance(_rv, int):
|
||
_rc = _rv
|
||
else:
|
||
_rc = 2
|
||
_real_stdout.write("[selfcheck] 未找到 main()/run() 入口，无法执行自查\\n")
|
||
except SystemExit as _e:
|
||
if _e.code is None:
|
||
_rc = 0
|
||
elif isinstance(_e.code, int):
|
||
_rc = _e.code
|
||
else:
|
||
_rc = 1
|
||
_real_stdout.write("[selfcheck] SystemExit: %s\\n" % (_e.code,))
|
||
except BaseException:
|
||
_rc = 1
|
||
_sys.stdout = _real_stdout
|
||
_tb.print_exc()
|
||
_sys.stdout = _Tee(_real_stdout, _buf)
|
||
finally:
|
||
try:
|
||
_sys.stdout.flush()
|
||
except Exception:
|
||
pass
|
||
_sys.stdout = _real_stdout
|
||
|
||
_out = _buf.getvalue()
|
||
_FAIL_PAT = _re.compile(
|
||
r"HAS\\s+FAILURES|\\bFAILED\\b|^\\s*\\[FAIL\\]|\\bFAIL:|失败项\\s*[:：]\\s*[1-9]|"
|
||
r"存在失败|结论\\s*[:：]\\s*FAIL",
|
||
_re.MULTILINE | _re.IGNORECASE,
|
||
)
|
||
_PASS_PAT = _re.compile(
|
||
r"ALL\\s+PASSED|全部通过|结论\\s*[:：]\\s*(OK|PASS|PASSED|SUCCESS)|失败项\\s*[:：]\\s*0\\b",
|
||
_re.IGNORECASE,
|
||
)
|
||
if _rc == 0 and _FAIL_PAT.search(_out) and not (
|
||
_PASS_PAT.search(_out) and not _FAIL_PAT.search(_out)
|
||
):
|
||
_real_stdout.write(
|
||
"[selfcheck] 检测到失败特征但退出码为 0，已按门禁规则改写为 1\\n"
|
||
)
|
||
_rc = 1
|
||
_sys.exit(_rc)
|
||
'''
|
||
|
||
|
||
def read_target() -> str:
|
||
with io.open(TARGET, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
|
||
def already_patched(text: str) -> bool:
|
||
return MARKER in text
|
||
|
||
|
||
def apply_patch() -> int:
|
||
if not os.path.isfile(TARGET):
|
||
sys.stderr.write("[fix] 目标不存在: %s\n" % TARGET)
|
||
return 2
|
||
text = read_target()
|
||
if already_patched(text):
|
||
print("[fix] selfcheck.py 已含 %s 标记，跳过（幂等）" % MARKER)
|
||
return 0
|
||
m = None
|
||
for m in MAIN_RE.finditer(text):
|
||
pass # 取最后一个 __main__ 块
|
||
if m is None:
|
||
# 没有 __main__ 块 -> 直接追加
|
||
new = text.rstrip("\n") + "\n\n\n" + PATCH_BLOCK.format(marker=MARKER)
|
||
action = "appended"
|
||
else:
|
||
new = text[: m.start()] + PATCH_BLOCK.format(marker=MARKER)
|
||
action = "replaced"
|
||
with io.open(TARGET, "w", encoding="utf-8", newline="\n") as f:
|
||
f.write(new)
|
||
print("[fix] selfcheck.py __main__ 块已%s，退出码门禁生效" % action)
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
if "--check" in sys.argv:
|
||
if not os.path.isfile(TARGET):
|
||
print("[check] MISSING %s" % TARGET)
|
||
return 2
|
||
print("[check] PATCHED" if already_patched(read_target()) else "[check] NOT_PATCHED")
|
||
return 0 if already_patched(read_target()) else 1
|
||
return apply_patch()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|