261 lines
13 KiB
Python
261 lines
13 KiB
Python
|
# 🔴 变异验证 · 2026-10-04 · 会话级忙判据两个新缺陷
|
|||
|
|
#
|
|||
|
|
# ⚠️ 纪律（来自 references/00-动手前必过.md ③④）：
|
|||
|
|
# ① **先验基线绿**：基线 PASS 17/FAIL 0（忙判据那条用例）。
|
|||
|
|
# ② **锚在唯一位置**：`def _session_busy_conv` / `def _conv_ts` 唯一定位，⛔ 不锚"文本出现过"。
|
|||
|
|
# ③ **每个变异先验「被检对象真被改了」**：比对长度+md5，⛔ 没真改就算**失败**（不许跳过分母）。
|
|||
|
|
# ④ ⛔ **跳过的变异一律计入失败**。
|
|||
|
|
#
|
|||
|
|
# 用法：python mut_run.py
|
|||
|
|
import hashlib
|
|||
|
|
import importlib.util
|
|||
|
|
import io
|
|||
|
|
import os
|
|||
|
|
import re
|
|||
|
|
import shutil
|
|||
|
|
import subprocess
|
|||
|
|
import sys
|
|||
|
|
import tempfile
|
|||
|
|
|
|||
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|||
|
|
COLLABD = os.path.join(HERE, "collabd.py")
|
|||
|
|
SELFTEST = os.path.join(HERE, "selftest.py")
|
|||
|
|
PY = sys.executable
|
|||
|
|
|
|||
|
|
|
|||
|
|
def md5(p):
|
|||
|
|
return hashlib.md5(open(p, "rb").read()).hexdigest()
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ── 变异体定义 ────────────────────────────────────────────────────────────
|
|||
|
|
# (编号, 目标判据, 人类可读描述, 变异函数)
|
|||
|
|
def mut_order_revert(src):
|
|||
|
|
"""M1 针对 ⑭：**丢掉 key 的时间戳分量** = 退回"纯遍历顺序"⇒ 昨天盖掉今天。
|
|||
|
|
|
|||
|
|
⚠️ 踩坑记录（这条本身就是变异验证的教训）：第一版写成
|
|||
|
|
`key[1] >= last_t[1]`（只把比较从元组换成 seq）⇒ **等价变异、恒绿**。
|
|||
|
|
原因：样本里每个文件内部时间戳本来就升序，**任何"取最后一条"的写法都等价**。
|
|||
|
|
⇒ 变异必须动**语义**（时间戳参与/不参与），⛔ 不能只动写法。
|
|||
|
|
"""
|
|||
|
|
old = " key = (ts if ts is not None else 0.0, seq)"
|
|||
|
|
new = " key = (0.0, seq) # MUTANT: 丢掉时间戳 = 退回纯遍历顺序"
|
|||
|
|
assert old in src, "锚点1 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_freshness_off(src):
|
|||
|
|
"""M2 针对 ⑮：把新鲜度闸阈值放大到"永不过期"（缺陷原样）。"""
|
|||
|
|
old = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH:"
|
|||
|
|
new = " if False: # MUTANT: 关掉新鲜度闸"
|
|||
|
|
assert old in src, "锚点2 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_tz_local(src):
|
|||
|
|
"""M3 针对 ⑰：时间戳改按**本地时区**解析（mktime）⇒ 东八区差 8h ⇒ 恒判过期。"""
|
|||
|
|
old = " t = float(calendar.timegm((int(m.group(\"Y\")), int(m.group(\"M\")), int(m.group(\"D\")),\n int(m.group(\"h\")), int(m.group(\"m\")), int(m.group(\"s\")), 0, 0, 0)))"
|
|||
|
|
new = (" import calendar as _c # MUTANT: 按本地时区解析\n"
|
|||
|
|
" _st = (int(m.group(\"Y\")), int(m.group(\"M\")), int(m.group(\"D\")),\n"
|
|||
|
|
" int(m.group(\"h\")), int(m.group(\"m\")), int(m.group(\"s\")), 0, 0, -1)\n"
|
|||
|
|
" t = float(time.mktime(_st))")
|
|||
|
|
assert old in src, "锚点3 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_tz_double_offset(src):
|
|||
|
|
"""M9 针对 ⑰：额外+8h 偏移（另一种时区错法）⇒ 「2 分钟前」变「8 小时后」⇒ 恒不过期。"""
|
|||
|
|
old = " ms = m.group(\"ms\")\n return t + (float(\"0.\" + ms) if ms else 0.0)"
|
|||
|
|
new = (" ms = m.group(\"ms\")\n"
|
|||
|
|
" return t + 28800.0 + (float(\"0.\" + ms) if ms else 0.0) # MUTANT: 多加 8h")
|
|||
|
|
assert old in src, "锚点9 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_freshness_too_hard(src):
|
|||
|
|
"""M4 针对 ⑯：新鲜度闸写太狠（窗口=0）⇒ 连正在跑的也判 idle（反向回归）。"""
|
|||
|
|
old = "(time.time() - last_t[0]) > SRSM_FRESH:"
|
|||
|
|
new = "(time.time() - last_t[0]) > 0: # MUTANT: 闸写太狠"
|
|||
|
|
assert old in src, "锚点4 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_ts_none(src):
|
|||
|
|
"""M5 针对 ⑰：时间戳一律返回 None（解析失败）⇒ 保守当 busy ⇒ ⑯ 报红。"""
|
|||
|
|
old = " m = _CONV_TS.match(ln)"
|
|||
|
|
new = " return None # MUTANT: 一律解析失败\n m = _CONV_TS.match(ln)"
|
|||
|
|
assert old in src, "锚点5 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_busy_always(src):
|
|||
|
|
"""M8 针对 ⑯：`idle` 也被当成 busy（陈旧闸写太狠的另一种写法）⇒ ⑯/⑭ 报红。"""
|
|||
|
|
old = " if last_to in (\"working\", \"planning\"):"
|
|||
|
|
new = " if True: # MUTANT: 不分状态一律当在跑"
|
|||
|
|
assert old in src, "锚点8 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_order_reverse_files(src):
|
|||
|
|
"""M6 针对 ⑭：把 `_conv_log` 的返回顺序反过来（[昨天, 今天]）⇒ 顺序依赖立刻现形。"""
|
|||
|
|
old = " for cand in (\"%s.log\" % sid, \"%s.log\" % str(sid).replace(\"-\", \"\")):"
|
|||
|
|
new = " for cand in (\"%s.log\" % sid,): # MUTANT: 去掉无连字符形态"
|
|||
|
|
assert old in src, "锚点6 未命中"
|
|||
|
|
s = src.replace(old, new, 1)
|
|||
|
|
return s.replace(" for d in (time.strftime(\"%Y-%m-%d\"),\n time.strftime(\"%Y-%m-%d\", time.localtime(time.time() - 86400))):",
|
|||
|
|
" for d in (time.strftime(\"%Y-%m-%d\", time.localtime(time.time() - 86400)),\n time.strftime(\"%Y-%m-%d\")):", 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def mut_freshness_only_first_file(src):
|
|||
|
|
"""M7 针对 ⑮：陈旧判定只在**第一个文件**生效 ⇒ 单文件陈旧样本仍判 busy（⑮ 报红）。"""
|
|||
|
|
old = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH:"
|
|||
|
|
new = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH and len(_conv_log(sid)) > 1: # MUTANT: 陈旧判定只在多文件时生效"
|
|||
|
|
assert old in src, "锚点7 未命中"
|
|||
|
|
return src.replace(old, new, 1)
|
|||
|
|
|
|||
|
|
|
|||
|
|
MUTS = [
|
|||
|
|
("M1", "⑭跨日取最新", mut_order_revert),
|
|||
|
|
("M2", "⑮新鲜度闸", mut_freshness_off),
|
|||
|
|
("M3", "⑰UTC时区", mut_tz_local),
|
|||
|
|
("M4", "⑯反向回归", mut_freshness_too_hard),
|
|||
|
|
("M5", "⑰时间戳解析", mut_ts_none),
|
|||
|
|
("M6", "⑭跨日取最新", mut_order_reverse_files),
|
|||
|
|
("M7", "⑮新鲜度闸", mut_freshness_only_first_file),
|
|||
|
|
("M8", "⑯反向回归", mut_busy_always),
|
|||
|
|
("M9", "⑰UTC时区", mut_tz_double_offset),
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def run_selftest(workdir):
|
|||
|
|
"""⚠️ **必须用工作目录里的那份 `selftest.py`**，⛔ 不能用技能目录的原版绝对路径。
|
|||
|
|
|
|||
|
|
🔴🔴 2026-10-04 当场栽过（写在这里当护栏，别删）：`selftest.py` 里
|
|||
|
|
`HERE = Path(__file__).resolve().parent`，而 `imp()` 加载的是 **`HERE / "collabd.py"`**
|
|||
|
|
⇒ 只改 `cwd` 没用、只传原版绝对路径也没用 ⇒ **测的一直是原版 ⇒ 恒绿 ⇒ 7/7 全是假象**。
|
|||
|
|
⇒ 判据：`HERE` 指向 `workdir` 时才算数（下面 `assert` 硬查）。
|
|||
|
|
"""
|
|||
|
|
st = os.path.join(workdir, "selftest.py")
|
|||
|
|
assert os.path.abspath(os.path.dirname(st)) == os.path.abspath(workdir), "测的不是工作目录里的 selftest"
|
|||
|
|
p = subprocess.run([PY, st, "-k", "忙判据"],
|
|||
|
|
capture_output=True, text=True, encoding="utf-8",
|
|||
|
|
errors="replace", cwd=workdir, timeout=600)
|
|||
|
|
out = p.stdout + p.stderr
|
|||
|
|
# 🔴 护栏：确认真加载的是 workdir 里那份（防止以后再有人改回绝对路径）
|
|||
|
|
assert workdir.replace("\\", "/") in out.replace("\\", "/"), "输出里看不到工作目录 ⇒ 加载的不是变异体"
|
|||
|
|
return out
|
|||
|
|
|
|||
|
|
|
|||
|
|
def verdict(out):
|
|||
|
|
"""返回 (是否报红, 是否真找到用例)。⛔ 没找到用例 = 验证作废，不算绿。"""
|
|||
|
|
ran = ("忙判据" in out) and ("投递忙判据" in out)
|
|||
|
|
if not ran:
|
|||
|
|
return None, False
|
|||
|
|
# 该用例整体 FAIL ⇔ 汇总行 PASS 0/ FAIL >=1，或用例行前是 ✗
|
|||
|
|
m = re.search(r"合计：PASS (\d+) / FAIL (\d+)", out)
|
|||
|
|
if not m:
|
|||
|
|
return None, True
|
|||
|
|
return (int(m.group(2)) >= 1), True
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
print("=" * 74)
|
|||
|
|
print("变异验证 · 会话级忙判据（每个新判据 ≥2 个变异体，⛔ 跳过计入失败）")
|
|||
|
|
print("=" * 74)
|
|||
|
|
orig_src = open(COLLABD, "rb").read().decode("utf-8")
|
|||
|
|
orig_md5 = md5(COLLABD)
|
|||
|
|
work = tempfile.mkdtemp(prefix="mutrun-")
|
|||
|
|
shutil.copy2(COLLABD, os.path.join(work, "collabd.py"))
|
|||
|
|
shutil.copy2(SELFTEST, os.path.join(work, "selftest.py"))
|
|||
|
|
|
|||
|
|
# 替身：把 collabd 换成变异体
|
|||
|
|
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
|||
|
|
f.write(orig_src)
|
|||
|
|
|
|||
|
|
base_out = run_selftest(work)
|
|||
|
|
base_red, base_ran = verdict(base_out)
|
|||
|
|
print("\n【基线】mutant 未植入时的原版：")
|
|||
|
|
print(" 找到用例=%s 报红=%s => %s"
|
|||
|
|
% (base_ran, base_red, "✅ 绿（可谈抓得住）" if (base_ran and not base_red) else "❌ 基线不绿，验证作废"))
|
|||
|
|
if not (base_ran and not base_red):
|
|||
|
|
print("\n❌ 基线不绿 ⇒ 停在此处（恒红比漏网更坏：会让所有变异都'红'，验证作废）。")
|
|||
|
|
return 2
|
|||
|
|
|
|||
|
|
results = []
|
|||
|
|
for mid, target, fn in MUTS:
|
|||
|
|
# 还原原版再植入
|
|||
|
|
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
|||
|
|
f.write(orig_src)
|
|||
|
|
try:
|
|||
|
|
mutated = fn(orig_src)
|
|||
|
|
except AssertionError as e:
|
|||
|
|
results.append((mid, target, "锚点未命中:%s" % e, False))
|
|||
|
|
print(" ⛔ %-3s %-12s 锚点未命中 ⇒ 计入失败" % (mid, target))
|
|||
|
|
continue
|
|||
|
|
# ③ 验「被检对象真被改了」：长度差 + md5 + 语法
|
|||
|
|
if mutated == orig_src:
|
|||
|
|
results.append((mid, target, "变异体与原版相同（replace 没生效）", False))
|
|||
|
|
print(" ⛔ %-3s %-12s 真没被改 ⇒ 计入失败" % (mid, target))
|
|||
|
|
continue
|
|||
|
|
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
|||
|
|
f.write(mutated)
|
|||
|
|
dlen = len(mutated) - len(orig_src)
|
|||
|
|
mm = md5(os.path.join(work, "collabd.py"))
|
|||
|
|
cp = subprocess.run([PY, "-m", "py_compile", os.path.join(work, "collabd.py")],
|
|||
|
|
capture_output=True, text=True, encoding="utf-8")
|
|||
|
|
changed = (mm != orig_md5) and (dlen != 0)
|
|||
|
|
if not changed:
|
|||
|
|
results.append((mid, target, "被检对象未真改", False))
|
|||
|
|
print(" ⛔ %-3s %-12s md5/长度未变 ⇒ 计入失败" % (mid, target))
|
|||
|
|
continue
|
|||
|
|
if cp.returncode != 0:
|
|||
|
|
results.append((mid, target, "变异体语法错（测的是崩溃不是判据）", False))
|
|||
|
|
print(" ⛔ %-3s %-12s 语法错 ⇒ 计入失败" % (mid, target))
|
|||
|
|
continue
|
|||
|
|
out = run_selftest(work)
|
|||
|
|
# 🔴🔴 最关键的一道：**确认被检代码里真的有变异标记**。
|
|||
|
|
# 上一轮 7/7 全"假绿"就是栽在这：变异体压根没被加载，测的是原版。
|
|||
|
|
# ⛔ 只看"跑绿/跑红"不够 —— 必须证明**跑的那份**含变异。
|
|||
|
|
marker = "# MUTANT:"
|
|||
|
|
if marker not in mutated:
|
|||
|
|
results.append((mid, target, "变异体缺标记（无法证明被加载）", False))
|
|||
|
|
print(" ⛔ %-3s %-12s 变异体无标记 ⇒ 计入失败" % (mid, target))
|
|||
|
|
continue
|
|||
|
|
if marker not in open(os.path.join(work, "collabd.py"), encoding="utf-8").read():
|
|||
|
|
results.append((mid, target, "工作目录里没有变异体", False))
|
|||
|
|
print(" ⛔ %-3s %-12s 工作目录无变异体 ⇒ 计入失败" % (mid, target))
|
|||
|
|
continue
|
|||
|
|
red, ran = verdict(out)
|
|||
|
|
if red is None:
|
|||
|
|
results.append((mid, target, "没跑到用例（验证作废）", False))
|
|||
|
|
print(" ⛔ %-3s %-12s 没跑到用例 ⇒ 计入失败" % (mid, target))
|
|||
|
|
continue
|
|||
|
|
ok = bool(red)
|
|||
|
|
results.append((mid, target, "报红" if ok else "❌未报红(假绿)", ok))
|
|||
|
|
print(" %s %-3s %-12s 真改(len%+d) ⇒ %s"
|
|||
|
|
% ("✅" if ok else "❌", mid, target, dlen, "报红 ✅" if ok else "未报红 ❌ 假绿"))
|
|||
|
|
|
|||
|
|
# 还原
|
|||
|
|
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
|||
|
|
f.write(orig_src)
|
|||
|
|
|
|||
|
|
print("\n" + "-" * 74)
|
|||
|
|
npass = sum(1 for r in results if r[3])
|
|||
|
|
ntot = len(results)
|
|||
|
|
print("变异汇总：%d/%d 报红" % (npass, ntot))
|
|||
|
|
# 每个新判据的覆盖
|
|||
|
|
cov = {}
|
|||
|
|
for mid, target, msg, ok in results:
|
|||
|
|
cov.setdefault(target, []).append((mid, ok))
|
|||
|
|
print("\n每个新判据的变异体覆盖（≥2 个）：")
|
|||
|
|
for target, lst in cov.items():
|
|||
|
|
okc = sum(1 for _, o in lst if o)
|
|||
|
|
flag = "✅" if len(lst) >= 2 and okc == len(lst) else "❌"
|
|||
|
|
print(" %s %-12s 变异体 %d 个，报红 %d 个 %s"
|
|||
|
|
% (flag, target, len(lst), okc, [m for m, _ in lst]))
|
|||
|
|
shutil.rmtree(work, ignore_errors=True)
|
|||
|
|
print("\n" + ("✅ 全部变异体按预期报红" if npass == ntot and ntot >= 2 else "❌ 有变异体未报红/被跳过"))
|
|||
|
|
return 0 if npass == ntot else 1
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(main())
|