120 lines
4.9 KiB
Python
120 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""M11b-2c-A3 交付自洽性重放校验器（可复跑，QC 复核用）。
|
||
|
||
背景（QC 退回 #10）：上一轮引擎在交付时又代为提交了一次补丁器修正（51c3f26），
|
||
说明 agent 交付的仓库状态并非最终自洽状态——补丁器与定稿 tx_write.py 不一致，
|
||
靠引擎收口补齐。本脚本把「HEAD 即最终态」变成可机械复验的断言。
|
||
|
||
做法（不污染工作区，全部在临时目录内完成）：
|
||
1. 从 git 取 HEAD 版本的补丁器 scripts/m11b2c_a3_patch_tx_write.py；
|
||
2. 从 git 取 5542cf5^（修复前）版本的 pbl_runtime_ext/tx_write.py 原文；
|
||
3. 在临时目录里按补丁器自身的 TARGET 相对路径摆好文件，直接 import 并调用其
|
||
main()（等价于在机构工作空间根执行该补丁器）；
|
||
4. 把重放产物与 HEAD 版本的 tx_write.py 逐字节比对（cmp + sha256 双口径）；
|
||
5. 一致 -> REPLAY_IDENTICAL: YES，退出码 0；不一致 -> NO，退出码 1。
|
||
|
||
用法（在模块仓库内或机构工作空间根均可）：
|
||
python3 scripts/m11b2c_a3_replay_check.py
|
||
"""
|
||
import hashlib
|
||
import io
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
PATCHER = "scripts/m11b2c_a3_patch_tx_write.py"
|
||
TX_WRITE = "pbl_runtime_ext/tx_write.py"
|
||
FIX_COMMIT = "5542cf5d5182130aa929e5aa37d0c22a043cee25"
|
||
|
||
|
||
def _git_rev_parse(rev):
|
||
return _git("rev-parse", rev).strip()
|
||
|
||
|
||
def _git(*args):
|
||
out = subprocess.run(["git", "-C", REPO] + list(args),
|
||
check=True, stdout=subprocess.PIPE)
|
||
return out.stdout.decode("utf-8")
|
||
|
||
|
||
def _sha(path):
|
||
with io.open(path, "rb") as f:
|
||
return hashlib.sha256(f.read()).hexdigest()
|
||
|
||
|
||
def main():
|
||
print("== M11b-2c-A3 补丁器/定稿一致性重放校验 ==")
|
||
print("repo : %s" % REPO)
|
||
head = _git_rev_parse("HEAD")
|
||
print("HEAD : %s" % head)
|
||
print("HEAD subject : %s" % _git("log", "-1", "--format=%s").strip())
|
||
print("fix commit : %s" % FIX_COMMIT)
|
||
print("patcher@HEAD : %s" % _git_rev_parse("HEAD:%s" % PATCHER))
|
||
print("tx_write@HEAD : %s" % _git_rev_parse("HEAD:%s" % TX_WRITE))
|
||
print("tx_write@fix^ : %s" % _git_rev_parse("%s^:%s" % (FIX_COMMIT, TX_WRITE)))
|
||
|
||
tmp = tempfile.mkdtemp(prefix="m11b2c_a3_replay_")
|
||
rc = 1
|
||
try:
|
||
# 1) 取 HEAD 版补丁器
|
||
patcher_dir = os.path.join(tmp, "inst", "modules", "pbl_runtime_ext", "scripts")
|
||
target_dir = os.path.join(tmp, "inst", "modules", "pbl_runtime_ext", "pbl_runtime_ext")
|
||
os.makedirs(patcher_dir)
|
||
os.makedirs(target_dir)
|
||
with io.open(os.path.join(patcher_dir, os.path.basename(PATCHER)), "w",
|
||
encoding="utf-8") as f:
|
||
f.write(_git("show", "HEAD:%s" % PATCHER))
|
||
|
||
# 2) 取修复前 tx_write.py 原文，摆到补丁器 TARGET 的相对位置
|
||
pre_path = os.path.join(target_dir, "tx_write.py")
|
||
with io.open(pre_path, "w", encoding="utf-8") as f:
|
||
f.write(_git("show", "%s^:%s" % (FIX_COMMIT, TX_WRITE)))
|
||
head_path = os.path.join(tmp, "tx_write_HEAD.py")
|
||
with io.open(head_path, "w", encoding="utf-8") as f:
|
||
f.write(_git("show", "HEAD:%s" % TX_WRITE))
|
||
|
||
pre_sha = _sha(pre_path)
|
||
head_sha = _sha(head_path)
|
||
print("pre-fix sha256 : %s" % pre_sha)
|
||
print("HEAD sha256 : %s" % head_sha)
|
||
|
||
# 3) 在临时目录内执行 HEAD 版补丁器（TARGET 是相对路径，故 cwd 必须是 inst/）
|
||
sys.path.insert(0, patcher_dir)
|
||
mod_name = "m11b2c_a3_patch_tx_write"
|
||
if mod_name in sys.modules:
|
||
del sys.modules[mod_name]
|
||
patcher = __import__(mod_name)
|
||
cwd = os.getcwd()
|
||
os.chdir(os.path.join(tmp, "inst"))
|
||
try:
|
||
prc = patcher.main()
|
||
finally:
|
||
os.chdir(cwd)
|
||
print("patcher exit rc : %d (0=已应用 4 处补丁, 2=片段命中数异常)" % prc)
|
||
if prc != 0:
|
||
print("REPLAY_IDENTICAL: NO（补丁器在修复前原文上执行失败，rc=%d）" % prc)
|
||
return 1
|
||
|
||
# 4) 逐字节比对
|
||
replayed_sha = _sha(pre_path)
|
||
print("replayed sha256 : %s" % replayed_sha)
|
||
cmp_rc = subprocess.run(["cmp", pre_path, head_path]).returncode
|
||
print("cmp 退出码 : %d (0=逐字节一致)" % cmp_rc)
|
||
if replayed_sha == head_sha and cmp_rc == 0:
|
||
print("REPLAY_IDENTICAL: YES（补丁器重放产物 == HEAD 定稿，逐字节一致）")
|
||
print("结论：HEAD 即最终自洽态，补丁器与定稿 tx_write.py 一致，无待引擎收口内容。")
|
||
rc = 0
|
||
else:
|
||
print("REPLAY_IDENTICAL: NO（补丁器与定稿不一致，仓库状态非最终态）")
|
||
rc = 1
|
||
return rc
|
||
finally:
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|