host
488aedbcad
(cherry picked from commit 1db02545f74c96b3b92be4fc32954659cec1f814) Reconcile-conflict: A14b E402 imports referencing cloud.handler_core (does not exist here; core methods live in server.py) — kept HEAD imports with incoming's noqa: E402 markers, dropped the unused core_mod import (verified unused elsewhere in both files).
636 lines
33 KiB
Python
636 lines
33 KiB
Python
"""Coverage batch v2: check_transplant.py in-process tests.
|
||
|
||
Why this file exists: the existing test_skill_transplant_prep.py
|
||
drives the script through subprocess, which coverage cannot see --
|
||
so the tool measured 11.4% (31/271) despite being behaviour-tested.
|
||
These tests import the module in-process and call the same code
|
||
paths directly, with real assertions on drafts and error strings
|
||
(the subprocess tests stay untouched and still pass).
|
||
|
||
Fixtures are the same construction as test_skill_transplant_prep
|
||
(built from the script's own fragment constants, not copies of them),
|
||
plus small variants that trip each loud-failure branch.
|
||
"""
|
||
|
||
import contextlib
|
||
import importlib.util
|
||
import io
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
SKILL_DIR = REPO_ROOT / "cloud" / "skills" / "software-engineering"
|
||
SCRIPT = SKILL_DIR / "tools" / "check_transplant.py"
|
||
|
||
|
||
def _load_script():
|
||
spec = importlib.util.spec_from_file_location(
|
||
"check_transplant_v2cov", str(SCRIPT))
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
CT = _load_script()
|
||
|
||
|
||
def _source_fixture():
|
||
m = CT
|
||
parts = [
|
||
"---\nname: software-engineering\n---\n",
|
||
"## 角色线程矩阵\n",
|
||
"| 角色 | 线程数 | 负责目录 | 产出 |",
|
||
"| QA / 测试 | 1 | `tests/` | 测试用例 |",
|
||
"| 独立审计 | 1 | 只写审计记录 | 审计结论 |",
|
||
"| 验收 / 发布 | 协调员兼任 | `docs/release/` | 验收清单 |",
|
||
"## 失败、续跑与成本\n",
|
||
"- 成本控制：分阶段本身就是成本控制。",
|
||
"## 独立审计与自动推回\n",
|
||
"本节是审计加固第 3 步的技能机制，与代码 `cloud/audit_pushback.py`",
|
||
"（推回决策纯函数）双轨落地，只写文档不算落地。只补审计机制，",
|
||
"不改上面已有规则的语义：G5 缺陷回路、合并拆分、G6 用户验收",
|
||
"的原意都不变。" + m.AUDIT_WARNING,
|
||
"### 独立审计角色：不写代码，只独立重跑、抽查证据、单独签字\n",
|
||
"- 审计线程不写业务代码、不改待办、不给合并决定、不执行合并。",
|
||
"### 自动推回：审计不过关，系统自动推回，不等人工点头\n",
|
||
"- 推回上限与升级：推回修复最多三轮" + m.PUSHBACK_LIMIT_PAREN + "。",
|
||
"- 与已有回路的关系：G5 缺陷回路复用为修复执行通道，推回派生的修复子任务",
|
||
" 就按 G5 方式执行，只补“自动触发 + 合并发布阻塞”这两块缺失；",
|
||
m.G7_FULLAUTO_BLOCK,
|
||
"## 推倒重做：超过三次后的升级分支（不是并行回路）\n",
|
||
"- 与上一节自动推回的关系再写一遍以免混淆：自动推回（1–3 轮修补）",
|
||
" 在前、推倒重做（超过 3 次）在后，是同一链路的前后两段；",
|
||
" 与 G5 缺陷回路的关系是复用修复执行通道（重做任务本身仍按",
|
||
" 阶段派生、仍走 G5 回归复核）" + m.REBUILD_TAIL,
|
||
"## 三条收紧：分流、保留资产、审计的审计\n",
|
||
"- 收紧一：阈值不能死板，第三次失败先判根因。",
|
||
m.TIGHTEN3_NOTIFY,
|
||
"## QA 全链路：测试环节跑通各个和所有业务链路\n",
|
||
"本节是“独立审计与 QA”部分的补充验收标准，与自动推回、推倒重做、",
|
||
"三条收紧并列，不覆盖已有规则。" + m.QA_INTRO,
|
||
"### 每个客户端交付前自测：三项缺一不算自测完\n",
|
||
"- 单元测试 + 对照契约的集成测试 + 手工冒烟清单，三项缺一不算自测完。",
|
||
"## 诚实边界：原生客户端\n",
|
||
"- 在只有 Linux 主机的环境里，真原生 iOS 需要 Xcode。",
|
||
"## 校验（协调员每过一个门禁自查一次）\n",
|
||
"- 分阶段走了吗？G3.5 没过就开了 G4b / G4c 吗？",
|
||
"- 审计与推回走了吗？审计签字人是否独立于执行人、协调员批准人、",
|
||
" QA 测试人？安全签字是不是无条件通过？审计 fail 后有没有自动推回",
|
||
" 并阻塞合并/发布？模块失败超过 3 次后有没有按推倒重做分支处理",
|
||
" （先判根因、只推倒该模块、旧实现归档、避坑清单带进新包），",
|
||
" 而不是无限修补或悄悄放行？",
|
||
"- QA 全链路走了吗？每端自测三项齐了吗？PRD 验收标准全有对应用例吗？",
|
||
" agent 专属六项齐了吗？跨端矩阵覆盖全部已选端了吗，未测端写明原因了吗？",
|
||
" 未编译端是不是只写了“代码评审 + Web 核心验证”？QA 矩阵签字与",
|
||
" 独立审计签字两个签字位都在且不是同一人吗？推回包里失败链路齐了吗，",
|
||
" 重审重跑失败链路了吗？",
|
||
"有一项为否，就停在当前门禁，不进入下一阶段。",
|
||
"## 附录：角色模板（播种随附版）\n",
|
||
"下面是 `templates/roles/` 的随附全文。",
|
||
"---",
|
||
"## 附录角色：08-shared-core（来源 templates/roles/08-shared-core.md）\n",
|
||
"# 角色模板：共享核心\n",
|
||
"- 不在共享逻辑里写某端专属代码。",
|
||
"---",
|
||
"## 附录角色：09-auditor（独立审计，新增）\n",
|
||
"# 角色模板：独立审计（1 个线程）\n",
|
||
"你是本项目的独立审计线程。",
|
||
]
|
||
return "\n".join(parts) + "\n"
|
||
|
||
|
||
def _target_fixture(target_id):
|
||
markers = " ".join(CT.MARKERS[target_id])
|
||
last = ("## 附录角色：10-iteration（来源 templates/roles/10-iteration.md，v0.5 新增）"
|
||
if target_id == "v05"
|
||
else "## 附录角色：08-shared-core（来源 templates/roles/08-shared-core.md）")
|
||
parts = [
|
||
"# software-engineering：多客户端软件工程编排\n",
|
||
"## 角色线程矩阵\n",
|
||
"| 角色 | 线程数 | 负责目录 | 产出 |",
|
||
"| QA / 测试 | 1 | `tests/` | 测试用例 |",
|
||
"| 验收 / 发布 | 协调员兼任 | `docs/release/` | 验收清单 |",
|
||
"## 失败、续跑与成本\n",
|
||
"- 成本控制：分阶段本身就是成本控制。",
|
||
"## 诚实边界：原生客户端\n",
|
||
"- 在只有 Linux 主机的环境里，真原生 iOS 需要 Xcode。",
|
||
"## 校验（协调员每过一个门禁自查一次）\n",
|
||
"- 分阶段走了吗？G3.5 没过就开了 G4b / G4c 吗？",
|
||
"- 标记行：" + markers,
|
||
"| G1 | PRD 冻结 | PRD 按 `templates/artifacts/prd-outline.md` 写完，用户故事有验收标准 | 回产品线程改 |",
|
||
"问卷模板在 `templates/artifacts/intake-questions.md`。最关键的六问：",
|
||
"- 验收清单模板在 `templates/artifacts/acceptance-checklist.md`。",
|
||
"按 `templates/artifacts/prd-outline.md` 写一份 PRD，至少包含：",
|
||
" 协议模板见 `templates/artifacts/` 下的 agent 任务契约模板。",
|
||
"1. 按 `templates/artifacts/acceptance-checklist.md` 生成本项目的验收清单，",
|
||
"有一项为否，就停在当前门禁，不进入下一阶段。",
|
||
"## 附录：角色模板（播种随附版）\n",
|
||
last + "\n",
|
||
"# 角色模板：共享核心\n",
|
||
"- 不在共享逻辑里写某端专属代码。",
|
||
]
|
||
return "\n".join(parts) + "\n"
|
||
|
||
|
||
def _main_lines_ending_with_tighten3():
|
||
"""Main lines whose last paragraph is the tighten-3 notify text.
|
||
|
||
Exercises transform_main's v04 else-branch (notify text without a
|
||
trailing newline), which the standard fixture cannot reach because
|
||
the QA section follows the notify paragraph there. Built as a
|
||
lines list directly: transform_main only needs main lines, while
|
||
extract_source would also demand matrix/checklist/appendix.
|
||
"""
|
||
m = CT
|
||
return [
|
||
"## 独立审计与自动推回",
|
||
"前言。" + m.AUDIT_WARNING,
|
||
"### 独立审计角色：不写代码，只独立重跑、抽查证据、单独签字",
|
||
"### 自动推回：审计不过关，系统自动推回，不等人工点头",
|
||
"上限" + m.PUSHBACK_LIMIT_PAREN + "。",
|
||
m.G7_FULLAUTO_BLOCK,
|
||
"## 推倒重做：超过三次后的升级分支（不是并行回路）",
|
||
"关系" + m.REBUILD_TAIL,
|
||
"## 三条收紧：分流、保留资产、审计的审计",
|
||
m.QA_INTRO,
|
||
m.TIGHTEN3_NOTIFY,
|
||
]
|
||
|
||
|
||
class ExtractTestCase(unittest.TestCase):
|
||
def test_extract_happy(self):
|
||
hunks = CT.extract_source(_source_fixture())
|
||
self.assertEqual(hunks["matrix_row"], "| 独立审计 | 1 | 只写审计记录 | 审计结论 |")
|
||
self.assertEqual(hunks["main"][0], "## 独立审计与自动推回")
|
||
self.assertTrue(hunks["audit_item"][0].startswith("- 审计与推回走了吗？"))
|
||
self.assertTrue(hunks["audit_item"][-1].endswith("悄悄放行？"))
|
||
self.assertTrue(hunks["qa_item"][0].startswith("- QA 全链路走了吗？"))
|
||
self.assertTrue(hunks["qa_item"][-1].endswith("重审重跑失败链路了吗？"))
|
||
self.assertEqual(hunks["appendix_body"][0], "# 角色模板：独立审计（1 个线程）")
|
||
|
||
def test_extract_matrix_zero_and_two(self):
|
||
base = _source_fixture()
|
||
no_matrix = base.replace("| 独立审计 | 1 | 只写审计记录 | 审计结论 |\n", "")
|
||
with self.assertRaises(ValueError) as ctx:
|
||
CT.extract_source(no_matrix)
|
||
self.assertIn("auditor matrix row", str(ctx.exception))
|
||
two = base.replace("| 验收 / 发布 |",
|
||
"| 独立审计 | 1 | 只写审计记录 | 审计结论 |\n| 验收 / 发布 |")
|
||
with self.assertRaises(ValueError):
|
||
CT.extract_source(two)
|
||
|
||
def test_extract_main_after_honesty(self):
|
||
base = _source_fixture()
|
||
# Move the single honesty heading to before the main heading:
|
||
# delete the original, then insert one copy ahead of main, so
|
||
# both _find_exact calls still find exactly one hit and only
|
||
# the order check can fire.
|
||
moved = base.replace("## 诚实边界：原生客户端\n", "", 1)
|
||
moved = moved.replace("## 独立审计与自动推回\n",
|
||
"## 诚实边界：原生客户端\n## 独立审计与自动推回\n", 1)
|
||
with self.assertRaises(ValueError) as ctx:
|
||
CT.extract_source(moved)
|
||
self.assertIn("does not precede", str(ctx.exception))
|
||
|
||
def test_extract_missing_headings_and_unterminated_item(self):
|
||
base = _source_fixture()
|
||
with self.assertRaises(ValueError):
|
||
CT.extract_source(base.replace("## 独立审计与自动推回\n", "## 其它节\n", 1))
|
||
with self.assertRaises(ValueError):
|
||
CT.extract_source(base.replace("## 附录角色：09-auditor（独立审计，新增）\n", "", 1))
|
||
with self.assertRaises(ValueError):
|
||
CT.extract_source(base.replace("- 审计与推回走了吗？", "- 审计走了吗？", 1))
|
||
with self.assertRaises(ValueError):
|
||
CT.extract_source(base.replace("- QA 全链路走了吗？", "- QA 走了吗？", 1))
|
||
unterminated = base.replace("而不是无限修补或悄悄放行？",
|
||
"而不是无限修补。")
|
||
with self.assertRaises(ValueError) as ctx:
|
||
CT.extract_source(unterminated)
|
||
self.assertIn("never ends", str(ctx.exception))
|
||
|
||
def test_extract_trims_trailing_blank_lines_of_main(self):
|
||
# Main slice ending in blank lines: the while-pop loop runs
|
||
# and the returned main has no trailing blank.
|
||
base = _source_fixture()
|
||
variant = base.replace("## 诚实边界：原生客户端\n",
|
||
"\n\n## 诚实边界：原生客户端\n", 1)
|
||
hunks = CT.extract_source(variant)
|
||
self.assertNotEqual(hunks["main"][-1], "")
|
||
self.assertEqual(hunks["main"][0], "## 独立审计与自动推回")
|
||
|
||
def test_find_exact_and_lines_and_fail(self):
|
||
self.assertEqual(CT._lines("a\nb"), ["a", "b"])
|
||
lines = ["x", "y", "x"]
|
||
with self.assertRaises(ValueError):
|
||
CT._find_exact(lines, lambda line: line == "z", "missing")
|
||
with self.assertRaises(ValueError):
|
||
CT._find_exact(lines, lambda line: line == "x", "duplicated")
|
||
err = io.StringIO()
|
||
with contextlib.redirect_stderr(err):
|
||
self.assertEqual(CT.fail("boom"), 1)
|
||
self.assertIn("FAIL: boom", err.getvalue())
|
||
|
||
|
||
class TransformTestCase(unittest.TestCase):
|
||
def _main(self, source=None):
|
||
return CT.extract_source(source or _source_fixture())["main"]
|
||
|
||
def test_transform_v03(self):
|
||
lines, stopped = CT.transform_main(self._main(), "v03")
|
||
text = "\n".join(lines)
|
||
self.assertEqual(stopped, [])
|
||
self.assertNotIn("G7 热修复回路并行", text)
|
||
self.assertNotIn("全自动 CI 自动修复", text)
|
||
self.assertIn("验收失败导致的推倒要先通知用户", text)
|
||
self.assertIn("## QA 全链路", text)
|
||
self.assertNotIn("移植警告", text)
|
||
self.assertIn("推回上限独立于其它修复回路", text)
|
||
self.assertTrue(text.rstrip().endswith("。") or "三项缺一不算自测完" in text)
|
||
|
||
def test_transform_v04(self):
|
||
lines, stopped = CT.transform_main(self._main(), "v04")
|
||
text = "\n".join(lines)
|
||
self.assertEqual(len(stopped), 1)
|
||
self.assertIn("收紧三", stopped[0])
|
||
self.assertNotIn("G7 热修复回路并行", text)
|
||
self.assertNotIn("验收失败导致的推倒要先通知用户", text)
|
||
self.assertIn("全自动 CI 自动修复并行", text)
|
||
self.assertIn("## QA 全链路", text)
|
||
|
||
def test_transform_v05(self):
|
||
lines, stopped = CT.transform_main(self._main(), "v05")
|
||
text = "\n".join(lines)
|
||
self.assertEqual(len(stopped), 2)
|
||
self.assertNotIn("## QA 全链路", text)
|
||
self.assertNotIn("验收失败导致的推倒要先通知用户", text)
|
||
self.assertNotIn("移植警告", text)
|
||
self.assertIn("## 三条收紧", text)
|
||
|
||
def test_transform_v04_tighten3_without_trailing_newline(self):
|
||
lines, stopped = CT.transform_main(
|
||
_main_lines_ending_with_tighten3(), "v04")
|
||
text = "\n".join(lines)
|
||
self.assertEqual(len(stopped), 1)
|
||
self.assertNotIn("验收失败导致的推倒要先通知用户", text)
|
||
self.assertNotIn("G7 热修复回路并行", text)
|
||
|
||
def test_transform_unknown_target(self):
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(self._main(), "v99")
|
||
|
||
def test_transform_fragment_count_errors(self):
|
||
base = _source_fixture()
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(
|
||
base.replace(CT.G7_FULLAUTO_BLOCK, ""))["main"], "v03")
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(
|
||
base.replace(CT.PUSHBACK_LIMIT_PAREN, ""))["main"], "v03")
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(
|
||
base.replace(CT.REBUILD_TAIL, ""))["main"], "v03")
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(
|
||
base.replace(CT.G7_FULLAUTO_BLOCK, ""))["main"], "v04")
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(
|
||
base.replace(CT.AUDIT_WARNING, ""))["main"], "v03")
|
||
|
||
def test_transform_v05_qa_missing_and_tighten3_missing(self):
|
||
base = _source_fixture()
|
||
no_qa = base.replace("## QA 全链路：测试环节跑通各个和所有业务链路\n", "", 1)
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(no_qa)["main"], "v05")
|
||
no_tighten = base.replace(CT.TIGHTEN3_NOTIFY, "")
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(no_tighten)["main"], "v05")
|
||
with self.assertRaises(ValueError):
|
||
CT.transform_main(CT.extract_source(no_tighten)["main"], "v04")
|
||
|
||
|
||
class ApplyTestCase(unittest.TestCase):
|
||
def test_apply_all_targets_in_process(self):
|
||
for target_id in ("v03", "v04", "v05"):
|
||
with self.subTest(target_id=target_id):
|
||
draft, report = CT.apply_transplant(
|
||
_source_fixture(), _target_fixture(target_id), target_id)
|
||
self.assertNotIn("templates/artifacts", draft)
|
||
self.assertEqual(draft.count("| 独立审计 |"), 1)
|
||
self.assertEqual(
|
||
CT.check_transplanted(_target_fixture(target_id),
|
||
draft, target_id), [])
|
||
number = "11" if target_id == "v05" else "09"
|
||
self.assertIn("## 附录角色：%s-auditor" % number, draft)
|
||
self.assertTrue(any("matrix row" in line for line in report))
|
||
if target_id == "v03":
|
||
self.assertIn("验收失败导致的推倒要先通知用户", draft)
|
||
self.assertNotIn("G7 热修复回路并行", draft)
|
||
if target_id == "v04":
|
||
self.assertIn("STOPPED", "\n".join(report))
|
||
if target_id == "v05":
|
||
self.assertNotIn("## QA 全链路", draft)
|
||
self.assertNotIn("09-auditor", draft)
|
||
|
||
def test_apply_error_branches(self):
|
||
base_t = _target_fixture("v03")
|
||
base_s = _source_fixture()
|
||
with self.assertRaises(ValueError):
|
||
CT.apply_transplant(base_s, base_t.replace(
|
||
"| QA / 测试 |", "| QA |"), "v03")
|
||
with self.assertRaises(ValueError):
|
||
CT.apply_transplant(base_s, base_t.replace(
|
||
"## 诚实边界：原生客户端", "## 诚实分界线", 1), "v03")
|
||
with self.assertRaises(ValueError):
|
||
CT.apply_transplant(base_s, base_t.replace(
|
||
"有一项为否", "有一项不对", 1), "v03")
|
||
already = base_t.replace(
|
||
"## 附录：角色模板（播种随附版）",
|
||
"## 附录角色：09-auditor（来源 x）\n## 附录：角色模板（播种随附版）")
|
||
with self.assertRaises(ValueError):
|
||
CT.apply_transplant(base_s, already, "v03")
|
||
duplicated = base_t.replace(
|
||
"问卷模板在 `templates/artifacts/intake-questions.md`。",
|
||
"问卷模板在 `templates/artifacts/intake-questions.md`。"
|
||
"问卷模板在 `templates/artifacts/intake-questions.md`。")
|
||
with self.assertRaises(ValueError):
|
||
CT.apply_transplant(base_s, duplicated, "v03")
|
||
missing_artifact = base_t.replace(
|
||
"问卷模板在 `templates/artifacts/intake-questions.md`。",
|
||
"问卷模板已删。")
|
||
with self.assertRaises(ValueError):
|
||
CT.apply_transplant(base_s, missing_artifact, "v03")
|
||
|
||
|
||
class AnchorsTestCase(unittest.TestCase):
|
||
def _anchors(self, source, target, target_id):
|
||
buf = io.StringIO()
|
||
with contextlib.redirect_stdout(buf):
|
||
errors = CT.check_anchors(source, target, target_id)
|
||
return errors, buf.getvalue()
|
||
|
||
def test_anchors_happy_all_targets(self):
|
||
for target_id in ("v03", "v04", "v05"):
|
||
with self.subTest(target_id=target_id):
|
||
errors, out = self._anchors(_source_fixture(),
|
||
_target_fixture(target_id),
|
||
target_id)
|
||
self.assertEqual(errors, [])
|
||
self.assertIn("anchors:", out)
|
||
|
||
def test_anchors_source_unextractable(self):
|
||
errors, _out = self._anchors("no hunks here",
|
||
_target_fixture("v03"), "v03")
|
||
self.assertEqual(len(errors), 1)
|
||
self.assertIn("not extractable", errors[0])
|
||
|
||
def test_anchors_each_error(self):
|
||
base_t = _target_fixture("v03")
|
||
base_s = _source_fixture()
|
||
cases = [
|
||
(base_t.replace("| QA / 测试 |", "| QA |"),
|
||
"target QA matrix row"),
|
||
(base_t.replace("## 诚实边界：原生客户端", "## 诚实分界线", 1),
|
||
"target honesty heading"),
|
||
(base_t.replace("有一项为否", "有一项不对", 1),
|
||
"target checklist end"),
|
||
(base_t.replace("## 附录：角色模板（播种随附版）",
|
||
"## 附录模板", 1),
|
||
"target appendix intro"),
|
||
(base_t.replace("## 附录角色：08-shared-core", "## 附录角色：07-x", 1),
|
||
"lacks last chapter"),
|
||
(base_t.replace("## 附录：角色模板（播种随附版）",
|
||
"## 附录角色：09-auditor（来源 x）\n"
|
||
"## 附录：角色模板（播种随附版）"),
|
||
"already contains"),
|
||
(base_t.replace("单端复核", "单端检查", 1),
|
||
"preserved marker missing"),
|
||
]
|
||
for target, needle in cases:
|
||
with self.subTest(needle=needle):
|
||
errors, _out = self._anchors(base_s, target, "v03")
|
||
self.assertTrue(any(needle in e for e in errors),
|
||
errors)
|
||
|
||
|
||
class CheckTransplantedTestCase(unittest.TestCase):
|
||
def _draft(self, target_id):
|
||
draft, _report = CT.apply_transplant(
|
||
_source_fixture(), _target_fixture(target_id), target_id)
|
||
return draft
|
||
|
||
def test_check_happy_all_targets(self):
|
||
for target_id in ("v03", "v04", "v05"):
|
||
with self.subTest(target_id=target_id):
|
||
self.assertEqual(
|
||
CT.check_transplanted(_target_fixture(target_id),
|
||
self._draft(target_id),
|
||
target_id), [])
|
||
|
||
def test_check_marker_and_matrix_errors(self):
|
||
draft = self._draft("v03")
|
||
target = _target_fixture("v03")
|
||
dropped = "\n".join(line for line in draft.split("\n")
|
||
if not line.startswith("- 标记行："))
|
||
errors = CT.check_transplanted(target, dropped, "v03")
|
||
self.assertTrue(any("preserved marker lost" in e for e in errors))
|
||
no_row = draft.replace("| 独立审计 | 1 | 只写审计记录 | 审计结论 |\n", "")
|
||
errors = CT.check_transplanted(target, no_row, "v03")
|
||
self.assertTrue(any("auditor matrix row" in e for e in errors))
|
||
two_rows = draft.replace("| 独立审计 | 1 |",
|
||
"| 独立审计 | 1 |\n| 独立审计 | 1 |", 1)
|
||
errors = CT.check_transplanted(target, two_rows, "v03")
|
||
self.assertTrue(any("auditor matrix row" in e for e in errors))
|
||
|
||
def test_check_main_header_missing(self):
|
||
draft = self._draft("v03")
|
||
target = _target_fixture("v03")
|
||
for header in CT.MAIN_HEADERS:
|
||
with self.subTest(header=header):
|
||
mutated = draft.replace(header + "\n", "", 1)
|
||
errors = CT.check_transplanted(target, mutated, "v03")
|
||
self.assertTrue(any("main header missing" in e
|
||
for e in errors), errors)
|
||
|
||
def test_check_stopped_paragraphs_per_target(self):
|
||
# v03: QA must be present, G7/full-auto absent, tighten3 present.
|
||
draft = self._draft("v03")
|
||
target = _target_fixture("v03")
|
||
no_qa = draft.replace("## QA 全链路：测试环节跑通各个和所有业务链路\n", "", 1)
|
||
self.assertTrue(any("QA section missing" in e for e in
|
||
CT.check_transplanted(target, no_qa, "v03")))
|
||
with_g7 = draft + "\nG7 热修复回路并行\n"
|
||
self.assertTrue(any("STOPPED G7" in e for e in
|
||
CT.check_transplanted(target, with_g7, "v03")))
|
||
with_fullauto = draft + "\n全自动 CI 自动修复\n"
|
||
self.assertTrue(any("STOPPED full-auto" in e for e in
|
||
CT.check_transplanted(target, with_fullauto, "v03")))
|
||
no_tighten = draft.replace(CT.TIGHTEN3_NOTIFY, "")
|
||
self.assertTrue(any("tighten3 notify missing" in e for e in
|
||
CT.check_transplanted(target, no_tighten, "v03")))
|
||
# v04: QA present, G7/tighten3 absent.
|
||
draft4 = self._draft("v04")
|
||
target4 = _target_fixture("v04")
|
||
no_qa4 = draft4.replace("## QA 全链路：测试环节跑通各个和所有业务链路\n", "", 1)
|
||
self.assertTrue(any("QA section missing" in e for e in
|
||
CT.check_transplanted(target4, no_qa4, "v04")))
|
||
with_tighten4 = draft4 + "\n" + CT.TIGHTEN3_NOTIFY + "\n"
|
||
self.assertTrue(any("STOPPED tighten3" in e for e in
|
||
CT.check_transplanted(target4, with_tighten4, "v04")))
|
||
with_g7_4 = draft4 + "\nG7 热修复回路并行\n"
|
||
self.assertTrue(any("STOPPED G7" in e for e in
|
||
CT.check_transplanted(target4, with_g7_4, "v04")))
|
||
# v05: QA section/item and tighten3 must all be absent.
|
||
draft5 = self._draft("v05")
|
||
target5 = _target_fixture("v05")
|
||
with_qa5 = draft5 + "\n## QA 全链路：测试环节跑通各个和所有业务链路\n"
|
||
self.assertTrue(any("STOPPED QA section" in e for e in
|
||
CT.check_transplanted(target5, with_qa5, "v05")))
|
||
with_qa_item5 = draft5 + "\n- QA 全链路走了吗？\n"
|
||
self.assertTrue(any("STOPPED QA checklist" in e for e in
|
||
CT.check_transplanted(target5, with_qa_item5, "v05")))
|
||
with_tighten5 = draft5 + "\n" + CT.TIGHTEN3_NOTIFY + "\n"
|
||
self.assertTrue(any("STOPPED tighten3" in e for e in
|
||
CT.check_transplanted(target5, with_tighten5, "v05")))
|
||
|
||
def test_check_appendix_stale_artifacts_and_no_overwrite(self):
|
||
draft = self._draft("v03")
|
||
target = _target_fixture("v03")
|
||
no_heading = draft.replace(
|
||
"## 附录角色：09-auditor（来源 templates/roles/09-auditor.md）",
|
||
"## 附录角色：09-auditor（来源 templates/roles/09-auditor2.md）")
|
||
self.assertTrue(any("appendix heading" in e for e in
|
||
CT.check_transplanted(target, no_heading, "v03")))
|
||
old_form = draft + "\n09-auditor（独立审计，新增）\n"
|
||
self.assertTrue(any("old appendix heading" in e for e in
|
||
CT.check_transplanted(target, old_form, "v03")))
|
||
for stale in ("移植警告", "以生产 v0.2 技能为准", "技能正文 diff 全是加法"):
|
||
with self.subTest(stale=stale):
|
||
mutated = draft + "\n%s\n" % stale
|
||
self.assertTrue(any("stale transplant warning" in e for e in
|
||
CT.check_transplanted(target, mutated, "v03")))
|
||
with_artifact = draft + "\n见 `templates/artifacts/x.md`。\n"
|
||
self.assertTrue(any("templates/artifacts" in e for e in
|
||
CT.check_transplanted(target, with_artifact, "v03")))
|
||
# No-overwrite: drop one target line that no other check pins.
|
||
dropped_honesty = draft.replace("## 诚实边界：原生客户端\n", "", 1)
|
||
self.assertTrue(any("no-overwrite" in e for e in
|
||
CT.check_transplanted(target, dropped_honesty, "v03")))
|
||
lines = draft.split("\n")
|
||
idx_a = lines.index("## 失败、续跑与成本")
|
||
idx_b = lines.index("## 诚实边界：原生客户端")
|
||
lines[idx_a], lines[idx_b] = lines[idx_b], lines[idx_a]
|
||
reordered = "\n".join(lines)
|
||
self.assertTrue(any("no-overwrite" in e for e in
|
||
CT.check_transplanted(target, reordered, "v03")))
|
||
|
||
|
||
class MainCliTestCase(unittest.TestCase):
|
||
def _write(self, tmp, source, target, target_id="v03"):
|
||
src = Path(tmp) / "source.md"
|
||
tgt = Path(tmp) / "target.md"
|
||
src.write_text(source, encoding="utf-8")
|
||
tgt.write_text(target, encoding="utf-8")
|
||
return src, tgt
|
||
|
||
def _run(self, argv):
|
||
out, err = io.StringIO(), io.StringIO()
|
||
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
||
code = CT.main(argv)
|
||
return code, out.getvalue(), err.getvalue()
|
||
|
||
def test_main_anchors_ok_and_failed(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src, tgt = self._write(tmp, _source_fixture(),
|
||
_target_fixture("v03"))
|
||
code, out, _err = self._run(["anchors", "--source244", str(src),
|
||
"--target", str(tgt),
|
||
"--target-id", "v03"])
|
||
self.assertEqual(code, 0)
|
||
self.assertIn("anchors OK", out)
|
||
bad_tgt = Path(tmp) / "bad.md"
|
||
bad_tgt.write_text(_target_fixture("v03").replace(
|
||
"| QA / 测试 |", "| QA |"), encoding="utf-8")
|
||
code, _out, err = self._run(["anchors", "--source244", str(src),
|
||
"--target", str(bad_tgt),
|
||
"--target-id", "v03"])
|
||
self.assertEqual(code, 1)
|
||
self.assertIn("FAILED", err)
|
||
|
||
def test_main_apply_out_and_stdout(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src, tgt = self._write(tmp, _source_fixture(),
|
||
_target_fixture("v04"))
|
||
out_path = Path(tmp) / "draft.md"
|
||
code, _out, err = self._run(["apply", "--source244", str(src),
|
||
"--target", str(tgt),
|
||
"--target-id", "v04",
|
||
"--out", str(out_path)])
|
||
self.assertEqual(code, 0)
|
||
self.assertIn("STOPPED", err)
|
||
self.assertIn("## 附录角色：09-auditor", out_path.read_text())
|
||
code, out, _err = self._run(["apply", "--source244", str(src),
|
||
"--target", str(tgt),
|
||
"--target-id", "v04"])
|
||
self.assertEqual(code, 0)
|
||
self.assertIn("## 附录角色：09-auditor", out)
|
||
|
||
def test_main_check_transplanted_ok_and_failed(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src, tgt = self._write(tmp, _source_fixture(),
|
||
_target_fixture("v05"))
|
||
draft, _report = CT.apply_transplant(_source_fixture(),
|
||
_target_fixture("v05"), "v05")
|
||
draft_path = Path(tmp) / "draft.md"
|
||
draft_path.write_text(draft, encoding="utf-8")
|
||
code, out, _err = self._run(["check-transplanted",
|
||
"--source244", str(src),
|
||
"--target", str(tgt),
|
||
"--target-id", "v05",
|
||
"--transplanted", str(draft_path)])
|
||
self.assertEqual(code, 0)
|
||
self.assertIn("check-transplanted OK", out)
|
||
draft_path.write_text(draft + "\n见 `templates/artifacts/x.md`。\n",
|
||
encoding="utf-8")
|
||
code, _out, err = self._run(["check-transplanted",
|
||
"--source244", str(src),
|
||
"--target", str(tgt),
|
||
"--target-id", "v05",
|
||
"--transplanted", str(draft_path)])
|
||
self.assertEqual(code, 1)
|
||
self.assertIn("FAILED", err)
|
||
|
||
def test_main_read_error_and_value_error_are_code_2(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src, tgt = self._write(tmp, _source_fixture(),
|
||
_target_fixture("v03"))
|
||
code, _out, err = self._run(["anchors", "--source244",
|
||
str(Path(tmp) / "missing.md"),
|
||
"--target", str(tgt),
|
||
"--target-id", "v03"])
|
||
self.assertEqual(code, 2)
|
||
self.assertIn("cannot read input", err)
|
||
bad_src = Path(tmp) / "bad.md"
|
||
bad_src.write_text("no hunks here", encoding="utf-8")
|
||
code, _out, _err = self._run(["apply", "--source244", str(bad_src),
|
||
"--target", str(tgt),
|
||
"--target-id", "v03"])
|
||
self.assertEqual(code, 2)
|
||
|
||
def test_main_argparse_errors_exit_2(self):
|
||
# argparse writes usage to stderr itself; swallow it so the
|
||
# full-suite log stays readable.
|
||
with contextlib.redirect_stderr(io.StringIO()):
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
CT.main(["apply", "--source244", "s", "--target", "t",
|
||
"--target-id", "v99"])
|
||
self.assertEqual(ctx.exception.code, 2)
|
||
with contextlib.redirect_stderr(io.StringIO()):
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
CT.main([])
|
||
self.assertEqual(ctx.exception.code, 2)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|