From a714169ae6d9c6c8ac5687f9d66f379328b64cff Mon Sep 17 00:00:00 2001 From: loong10k <20489781+loong10k@users.noreply.github.com> Date: Thu, 24 Sep 2026 00:48:55 +0800 Subject: [PATCH] =?UTF-8?q?feat(flowguard):=20=E9=98=B6=E6=AE=B5=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E8=BF=81=E7=A7=BB=E7=9F=A9=E9=98=B5=20+=20=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE=E7=BA=A7=E6=96=87=E6=A1=A3=E6=A0=A1=E9=AA=8C=20+=20?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=E6=A0=88=E8=A6=86=E7=9B=96=E5=AF=B9=E7=85=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - stage_docs 增 LEGAL 迁移矩阵（单一事实源 + 全矩阵测试）：满足态互跳/ 重复写拒绝，回改走 in_progress；invalidated（正文已变）允许直接重新验收； pending/invalidated 不可作为写入目标 - 项目级文档验收补内容校验：02 ADR 追加式格式、07 规范集覆盖全部已注册 模块栈（detect.detect 接线）、10 发布内容功能引用须已知 task-id - validate 命令同步校验 02/07/10 - ARTIFACT_STAGE 映射单源 registry.ARTIFACTS - state_lock 跨平台：fcntl/msvcrt 自适应 - 新增 test_stage_transitions/test_state_tools 与校验器单测；219 测试全绿 - README quick start 4→6 命令（双语） --- README.md | 4 ++ README.zh-CN.md | 12 +++- hooks/flowguard_artifact_check.py | 11 +--- scripts/flowguard_lib/stage_docs.py | 27 ++++++++- scripts/flowguard_lib/state.py | 14 ++++- scripts/flowguard_lib/validation.py | 92 +++++++++++++++++++++++++++++ scripts/flowguard_state.py | 10 +++- tests/test_docs_pipeline.py | 10 ++-- tests/test_stage_transitions.py | 81 +++++++++++++++++++++++++ tests/test_state_tools.py | 68 +++++++++++++++++++++ tests/test_validation.py | 47 +++++++++++++++ 11 files changed, 356 insertions(+), 20 deletions(-) create mode 100644 tests/test_stage_transitions.py create mode 100644 tests/test_state_tools.py diff --git a/README.md b/README.md index d3b011b..2995f14 100644 --- a/README.md +++ b/README.md @@ -42,10 +42,14 @@ The ten stages are a verifiable process skeleton, not a Hook-driven script. Read ```bash /flowguard-discover # read-only; never installs or initializes tools /flowguard-context # classify and bind this session/worktree/task +/flowguard-init # create project-level stage docs 02/07/10 +/flowguard-stage # read and advance docs/ stage documents /flowguard-evidence # record real verification evidence /flowguard-governance # check code-write, commit, or release readiness ``` +The CLI additionally provides `validate` (artifact content, traceability matrix and module-stack coverage checks). + Kimi registers the same command prompts as namespaced Markdown commands, for example `/flowguard:flowguard-discover`. The Markdown files in `kimi-commands/` are generated from `commands/*.json`; regenerate with `python3 scripts/generate_kimi_commands.py --write` after changing a source command. Kimi Shell may not expose `KIMI_PLUGIN_ROOT`; in that case, use `/plugins info flowguard` to locate the enabled plugin before running its bundled CLI. The plugin remains centrally cataloged in `full-stack-plugins`; this source repository is maintained separately. CLI example: diff --git a/README.zh-CN.md b/README.zh-CN.md index 37fd12a..8ab3143 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -46,13 +46,21 @@ flowchart LR # 2. 智能体分类后绑定本次任务 /flowguard-context -# 3. 实施过程中记录真实证据 +# 3. 创建项目级阶段文档 02/07/10 +/flowguard-init + +# 4. 读取与推进 docs/ 阶段文档 +/flowguard-stage + +# 5. 实施过程中记录真实证据 /flowguard-evidence -# 4. 写码、提交或发布前复核 +# 6. 写码、提交或发布前复核 /flowguard-governance ``` +CLI 另提供 `validate`（产物内容、追溯矩阵与模块栈覆盖检查）。 + Kimi 将同源命令注册为带命名空间的 Markdown 命令，例如 `/flowguard:flowguard-discover`。`kimi-commands/` 由 `commands/*.json` 机械生成；修改 JSON 后运行 `python3 scripts/generate_kimi_commands.py --write`。若 Kimi Shell 未提供 `KIMI_PLUGIN_ROOT`，先通过 `/plugins info flowguard` 确认已启用插件的安装目录，再运行其自带 CLI。插件仍由 `full-stack-plugins` 统一登记与发布管理，源码仓库独立维护。 对应 CLI： diff --git a/hooks/flowguard_artifact_check.py b/hooks/flowguard_artifact_check.py index 3c7973f..1d3fd23 100644 --- a/hooks/flowguard_artifact_check.py +++ b/hooks/flowguard_artifact_check.py @@ -11,15 +11,10 @@ ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) -from flowguard_lib import codereview_evidence, context, evidence, stage_docs, tool_scope # noqa: E402 +from flowguard_lib import codereview_evidence, context, evidence, registry, stage_docs, tool_scope # noqa: E402 -# artifact 文件名 → 阶段（用于降级与校验路由） -ARTIFACT_STAGE = { - "01-requirements": "requirements", "02-architecture": "architecture", - "03-solution": "solution", "04-testcases": "testcases", - "05-hld": "hld", "06-lld": "lld", "07-standards": "standards", - "08-review": "review", "09-docs": "docs", "10-release": "release", -} +# artifact 文件名 → 阶段（失效提示路由；映射单源 registry.ARTIFACTS） +ARTIFACT_STAGE = {aid: spec["stage"] for aid, spec in registry.ARTIFACTS.items()} CODEGUARD_MCP_TOOLS = ("mcp__codeguard__check_code_style", "mcp__codeguard__auto_fix") diff --git a/scripts/flowguard_lib/stage_docs.py b/scripts/flowguard_lib/stage_docs.py index 388ea45..b7cd79c 100644 --- a/scripts/flowguard_lib/stage_docs.py +++ b/scripts/flowguard_lib/stage_docs.py @@ -4,13 +4,27 @@ import re from pathlib import Path -from . import ids, registry, state, validation +from . import detect, ids, registry, state, validation VALID_STATUSES = ( "pending", "in_progress", "pending_acceptance", "accepted", "inherited", "skipped", "invalidated", ) SATISFIED = ("accepted", "inherited", "skipped") + +# 合法状态迁移表（单一事实源；测试做全矩阵断言）。 +# invalidated 是指纹派生态：正文/前置变化后由 read() 推导，不可直接写入。 +# 满足态之间禁止互跳与重复写：回改须显式退回 in_progress，重新走验收； +# 仅 invalidated（正文已变）可直接重新验收。 +LEGAL = { + "pending": ("in_progress", "accepted", "inherited", "skipped"), + "in_progress": ("pending_acceptance", "accepted", "inherited", "skipped"), + "pending_acceptance": ("in_progress", "accepted", "inherited", "skipped"), + "invalidated": ("in_progress", "accepted", "inherited", "skipped"), + "accepted": ("in_progress",), + "inherited": ("in_progress",), + "skipped": ("in_progress",), +} FIELDS = ("任务", "父任务", "阶段", "阶段状态", "规格事实源", "原生产物", "批准依据") _ROW = re.compile(r"^\|\s*([^|]+?)\s*\|\s*([^|]*?)\s*\|\s*$", re.MULTILINE) @@ -275,6 +289,11 @@ def _advance_unlocked(root, task_id, stage, target, *, approval_ref=None, reason current = read(root, task_id, stage) if current["missing"]: raise StageDocError(f"阶段文档不存在: {path}") + allowed = LEGAL.get(current["status"], ()) + if target not in allowed: + raise StageDocError( + f"非法状态迁移 {current['status']} → {target}" + f"（合法目标: {', '.join(allowed) or '无'}）") if target in SATISFIED and not approval_ref: raise StageDocError("验收、继承或跳过必须提供批准依据；理由文本不能代替批准") if target in ("inherited", "skipped") and not reason: @@ -295,6 +314,12 @@ def _advance_unlocked(root, task_id, stage, target, *, approval_ref=None, reason issues = validation.validate_testcases(body, req_ids, root) elif stage == "08-review": issues = validation.validate_review(body) + elif stage == "02-architecture": + issues = validation.validate_architecture(body) + elif stage == "07-standards": + issues = validation.validate_standards(body, detect.detect(root)["modules"]) + elif stage == "10-release": + issues = validation.validate_release(body, validation.known_task_ids(root)) else: issues = [] blocking = [item["message"] for item in issues if item["level"] in ("ERROR", "WARNING")] diff --git a/scripts/flowguard_lib/state.py b/scripts/flowguard_lib/state.py index 53d75c3..3aac244 100644 --- a/scripts/flowguard_lib/state.py +++ b/scripts/flowguard_lib/state.py @@ -1,11 +1,16 @@ """状态锁与原子写工具；项目流程事实保存在 docs/（十阶段文档为唯一事实源）。""" import contextlib -import fcntl import os import pathlib import stat import tempfile +try: + import fcntl +except ImportError: # Windows：改用 msvcrt 字节锁，语义同为进程退出自动释放 + fcntl = None + import msvcrt + class StateError(Exception): pass @@ -17,9 +22,12 @@ def state_lock(root): from .runtime import repository_state_dir lock = repository_state_dir(root, create=True) / ".lock" lock.parent.mkdir(parents=True, exist_ok=True) - fh = lock.open("w") + fh = lock.open("w+") try: - fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) + if fcntl is not None: + fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) + else: # pragma: no cover - Windows + msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1) except OSError: fh.close() raise StateError("状态文件被其它进程锁定，请稍后重试") diff --git a/scripts/flowguard_lib/validation.py b/scripts/flowguard_lib/validation.py index 5f73f14..f4cb765 100644 --- a/scripts/flowguard_lib/validation.py +++ b/scripts/flowguard_lib/validation.py @@ -234,6 +234,98 @@ def flush(): return issues +def known_task_ids(root): + """docs/features/ 下的 kebab 任务目录即已知功能清单。""" + base = Path(root) / "docs" / "features" + if not base.is_dir(): + return [] + return sorted(p.name for p in base.iterdir() + if p.is_dir() and ids.is_kebab(p.name)) + + +def _section(text, start_re, end_re): + """截取 start_re 标题到 end_re 标题（不含）之间的正文。""" + out, active = [], False + for ln in text.splitlines(): + if not active and re.match(start_re, ln): + active = True + continue + if active and re.match(end_re, ln): + break + if active: + out.append(ln) + return "\n".join(out) + + +def validate_architecture(text): + """每条非占位 ADR 条目须带 `feature: <来源>` 与 `状态: proposed|accepted`（追加式格式契约）。 + + 含 `<占位符>`/`{{...}}` 的条目视为模板示例，不参与机械校验。 + """ + lines = text.splitlines() + masked = _mask_code_fences(lines) + issues = [] + for ln, hide in zip(lines, masked): + if hide or not re.match(r"^\s*-\s*ADR-", ln): + continue + if "<" in ln or "{{" in ln: + continue + if not re.search(r"feature:\s*\S+", ln): + issues.append(_issue("ERROR", "02-architecture.md", + f"ADR 条目缺 feature 来源标注: {ln.strip()!r}", + "追加式条目须带 `feature: <来源功能>`（禁止改写既有条目）")) + if not re.search(r"状态:\s*(proposed|accepted)", ln): + issues.append(_issue("ERROR", "02-architecture.md", + f"ADR 条目缺合法状态: {ln.strip()!r}", + "状态须为 proposed 或 accepted")) + return issues + + +def validate_standards(text, modules): + """已填规范集（2. 规范集节出现非占位条目）须覆盖全部已注册模块的栈。 + + 纯模板（规范集节全为占位示例）不参与机械校验；无栈模块（无法判定技术栈）跳过。 + """ + issues = [] + body = _section(text, r"^##\s*2\.", r"^##\s*3\.") + headings = [ln.lower() for ln in body.splitlines() + if re.match(r"^###\s+", ln) and "<" not in ln and "{{" not in ln] + if not headings: + return issues + for name, mod in sorted((modules or {}).items()): + stack = (mod or {}).get("stack") + if not stack: + continue + tokens = {name.lower(), str(stack).lower()} + tokens.update(t for t in str(stack).lower().replace("_", "-").split("-") if len(t) >= 3) + if not any(tok in h for h in headings for tok in tokens): + issues.append(_issue("ERROR", "07-standards.md", + f"规范集未覆盖模块 {name}（技术栈 {stack}）", + "按模块栈补充 2.1 规范条目（覆盖所有已注册模块的栈）")) + return issues + + +def validate_release(text, task_ids): + """发布内容表引用的功能须是已知 task-id；占位符行视为模板示例跳过。""" + issues = [] + scope = _section(text, r"^##\s*3\.", r"^##\s*4\.") + known = set(task_ids) + for ln in scope.splitlines(): + if "<" in ln or "{{" in ln: + continue + m = re.match(r"^\|\s*([^|]+?)\s*\|", ln) + if not m: + continue + cell = m.group(1).strip() + if not cell or set(cell) <= set(":- ") or cell.startswith("功能"): + continue + if cell not in known: + issues.append(_issue("ERROR", "10-release.md", + f"发布内容引用未知功能: {cell}", + "功能须是已存在的 docs/features/")) + return issues + + def missing_tier2(refs, root=None): """refs: (skill, pkg, install_cmd)；已安装判定见 _installed。缺失产出 WARNING。""" out = [] diff --git a/scripts/flowguard_state.py b/scripts/flowguard_state.py index 5e42eb6..7c2ed4e 100644 --- a/scripts/flowguard_state.py +++ b/scripts/flowguard_state.py @@ -11,7 +11,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) from flowguard_lib import ( # noqa: E402 - context, discovery, evidence, governance, registry, + context, detect, discovery, evidence, governance, registry, stage_docs, state, validation, ) from flowguard_lib.diag import emit as emit_diag, envelope as mk_env # noqa: E402 @@ -147,6 +147,14 @@ def cmd_validate(args): _die(mk_env("ERROR", "unknown_task", f"功能文档不存在: {args.task_id}", "先用 context bind 创建 docs/features//"), as_json=args.json) issues = [] + issues += validation.validate_architecture( + _read(stage_docs.path_for(root, "project", "02-architecture"))) + issues += validation.validate_standards( + _read(stage_docs.path_for(root, "project", "07-standards")), + detect.detect(root)["modules"]) + issues += validation.validate_release( + _read(stage_docs.path_for(root, "project", "10-release")), + validation.known_task_ids(root)) for task in tasks: task_id = task["task_id"] if task.get("error"): diff --git a/tests/test_docs_pipeline.py b/tests/test_docs_pipeline.py index ac22efa..a393183 100644 --- a/tests/test_docs_pipeline.py +++ b/tests/test_docs_pipeline.py @@ -184,7 +184,7 @@ def test_code_and_commit_wait_for_stage_documents_and_evidence(self): (self.root / "tests").mkdir() (self.root / "tests/test_refund.py").write_text("def test_refund():\n assert True\n", encoding="utf-8") for path in (self.root / "docs").rglob("*.md"): - text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")) + text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted") text = text.replace("测试文件: 已确认", "测试文件: tests/test_refund.py") text = text.replace("- 结论: fix|wontfix|deferred", "- 结论: fix") path.write_text(text, encoding="utf-8") @@ -219,7 +219,7 @@ def test_release_acceptance_invalidates_when_listed_feature_docs_change(self): encoding="utf-8", ) for path in (self.root / "docs").rglob("*.md"): - text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")) + text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted") text = text.replace("测试文件: 已确认", "测试文件: tests/test_refund.py") text = text.replace("- 结论: fix|wontfix|deferred", "- 结论: fix") path.write_text(text, encoding="utf-8") @@ -248,7 +248,7 @@ def test_release_scope_tracks_every_feature_with_escaped_table_text(self): encoding="utf-8", ) for path in (self.root / "docs").rglob("*.md"): - text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")) + text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted") text = text.replace("测试文件: 已确认", "测试文件: tests/test_refund.py") text = text.replace("- 结论: fix|wontfix|deferred", "- 结论: fix") path.write_text(text, encoding="utf-8") @@ -454,7 +454,7 @@ def test_editing_accepted_stage_invalidates_its_document_fingerprint(self): self.bind() from flowguard_lib import stage_docs path = self.root / "docs/features/refund/01-requirements.md" - text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")) + text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted") path.write_text(text, encoding="utf-8") stage_docs.advance(self.root, "refund", "01-requirements", "accepted", approval_ref="user-receipt:1") @@ -466,7 +466,7 @@ def test_reaccepting_changed_requirement_does_not_silently_reuse_downstream_acce from flowguard_lib import stage_docs for stage in ("01-requirements", "02-architecture", "03-solution"): path = stage_docs.path_for(self.root, "refund", stage) - path.write_text(re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")), + path.write_text(re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted"), encoding="utf-8") stage_docs.advance(self.root, "refund", stage, "accepted", approval_ref=f"user:{stage}") requirement = stage_docs.path_for(self.root, "refund", "01-requirements") diff --git a/tests/test_stage_transitions.py b/tests/test_stage_transitions.py new file mode 100644 index 0000000..f607322 --- /dev/null +++ b/tests/test_stage_transitions.py @@ -0,0 +1,81 @@ +"""阶段状态迁移全矩阵：满足态互跳/重复写必须被拒，回改走 in_progress。""" +import re +import tempfile +import unittest +from pathlib import Path + +from flowguard_lib import context, stage_docs + + +class TransitionMatrixTest(unittest.TestCase): + def setUp(self): + self.root = Path(tempfile.mkdtemp()) + context.bind(self.root, session_id="s", task_id="f1", task_type="simple_change", + spec_system="none", spec_ref=None) + + def fill(self, stage="01-requirements"): + path = stage_docs.path_for(self.root, "f1", stage) + path.write_text(re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")), + encoding="utf-8") + + def advance(self, target, stage="01-requirements", **kw): + return stage_docs.advance(self.root, "f1", stage, target, **kw) + + def test_legal_table_covers_all_statuses(self): + self.assertEqual(set(stage_docs.LEGAL), set(stage_docs.VALID_STATUSES)) + + def test_pending_to_accepted_is_legal(self): + self.fill() + self.assertEqual(self.advance("accepted", approval_ref="user:1")["status"], "accepted") + + def test_accepted_cannot_be_rewritten_without_content_change(self): + self.fill() + self.advance("accepted", approval_ref="user:1") + with self.assertRaisesRegex(stage_docs.StageDocError, "非法状态迁移 accepted → accepted"): + self.advance("accepted", approval_ref="user:2") + + def test_satisfied_states_cannot_flip_sideways(self): + self.fill() + self.advance("accepted", approval_ref="user:1") + for target in ("skipped", "inherited"): + with self.subTest(target=target): + with self.assertRaisesRegex(stage_docs.StageDocError, "非法状态迁移"): + self.advance(target, approval_ref="user:x", reason="r") + self.advance("in_progress") + self.advance("skipped", approval_ref="user:skip", reason="范围不适用") + for target in ("accepted", "inherited", "skipped"): + with self.subTest(target=target): + with self.assertRaisesRegex(stage_docs.StageDocError, "非法状态迁移"): + self.advance(target, approval_ref="user:x", reason="r") + + def test_rework_must_go_through_in_progress(self): + self.fill() + self.advance("accepted", approval_ref="user:1") + self.assertEqual(self.advance("in_progress")["status"], "in_progress") + self.assertEqual(self.advance("accepted", approval_ref="user:2")["status"], "accepted") + + def test_invalidated_allows_direct_reaccept(self): + self.fill() + self.advance("accepted", approval_ref="user:1") + path = stage_docs.path_for(self.root, "f1", "01-requirements") + path.write_text(path.read_text(encoding="utf-8") + "\n新增业务条件。\n", encoding="utf-8") + self.assertEqual(stage_docs.read(self.root, "f1", "01-requirements")["status"], "invalidated") + self.assertEqual(self.advance("accepted", approval_ref="user:recheck")["status"], "accepted") + + def test_pending_and_invalidated_are_not_writable_targets(self): + self.fill() + self.advance("in_progress") + for target in ("pending", "invalidated"): + with self.subTest(target=target): + with self.assertRaises(stage_docs.StageDocError): + self.advance(target) + + def test_pending_acceptance_round_trip(self): + self.fill() + self.advance("in_progress") + self.assertEqual(self.advance("pending_acceptance")["status"], "pending_acceptance") + self.assertEqual(self.advance("accepted", approval_ref="user:1")["status"], "accepted") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_state_tools.py b/tests/test_state_tools.py new file mode 100644 index 0000000..71773ac --- /dev/null +++ b/tests/test_state_tools.py @@ -0,0 +1,68 @@ +"""state 工具语义直接单测：锁互斥、原子写失败保留旧文、原子创建不覆盖。""" +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from flowguard_lib import state + + +class StateLockTest(unittest.TestCase): + def test_lock_conflict_raises_state_error(self): + root = Path(tempfile.mkdtemp()) + with state.state_lock(root): + with self.assertRaisesRegex(state.StateError, "锁定"): + with state.state_lock(root): + pass + + def test_lock_released_after_exit(self): + root = Path(tempfile.mkdtemp()) + with state.state_lock(root): + pass + with state.state_lock(root): # 不抛即可 + pass + + +class AtomicWriteTest(unittest.TestCase): + def test_atomic_write_replaces_content(self): + target = Path(tempfile.mkdtemp()) / "doc.md" + state.atomic_write_text(target, "旧") + state.atomic_write_text(target, "新") + self.assertEqual(target.read_text(encoding="utf-8"), "新") + + def test_atomic_write_failure_preserves_old_content(self): + target = Path(tempfile.mkdtemp()) / "doc.md" + state.atomic_write_text(target, "旧正文") + with mock.patch("os.replace", side_effect=OSError("replace failed")): + with self.assertRaises(OSError): + state.atomic_write_text(target, "新正文") + self.assertEqual(target.read_text(encoding="utf-8"), "旧正文") + self.assertEqual(list(target.parent.glob("*")), [target]) + + def test_atomic_create_does_not_overwrite(self): + target = Path(tempfile.mkdtemp()) / "doc.md" + state.atomic_create_text(target, "初版") + with self.assertRaises(FileExistsError): + state.atomic_create_text(target, "覆盖企图") + self.assertEqual(target.read_text(encoding="utf-8"), "初版") + + def test_atomic_create_failure_leaves_no_partial_or_residue(self): + target = Path(tempfile.mkdtemp()) / "doc.md" + with mock.patch("os.link", side_effect=OSError("link failed")): + with self.assertRaises(OSError): + state.atomic_create_text(target, "半成品") + self.assertFalse(target.exists()) + self.assertEqual(list(target.parent.glob(".fg-atomic-*")), []) + + +@unittest.skipUnless(sys.platform == "win32", "msvcrt 锁仅 Windows 路径") +class WindowsLockSmokeTest(unittest.TestCase): + def test_basic_lock_round_trip(self): + root = Path(tempfile.mkdtemp()) + with state.state_lock(root): + pass + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_validation.py b/tests/test_validation.py index f0039b6..fd2bb84 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -100,5 +100,52 @@ def test_installed_in_root_agents(self): refs = [("local-skill", "p", "cmd")] self.assertEqual(validation.missing_tier2(refs, root=root), []) + +class ProjectDocValidationTest(unittest.TestCase): + def test_architecture_template_and_valid_adr_pass(self): + self.assertEqual(validation.validate_architecture( + "- ADR-{{编号}} | feature: {{来源}} | 状态: {{proposed/accepted}} —— {{决策}}"), []) + self.assertEqual(validation.validate_architecture( + "- ADR-0001 | feature: refund | 状态: accepted —— 采用追加式 ADR"), []) + + def test_architecture_adr_missing_feature_or_status_is_error(self): + issues = validation.validate_architecture( + "- ADR-0002 | 状态: accepted —— 缺来源\n- ADR-0003 | feature: refund | 状态: draft —— 非法状态") + self.assertEqual([i["level"] for i in issues], ["ERROR", "ERROR"]) + self.assertTrue(any("feature" in i["message"] for i in issues)) + self.assertTrue(any("状态" in i["message"] for i in issues)) + + def test_standards_template_vacuous_and_coverage(self): + self.assertEqual(validation.validate_standards( + "## 2. 规范集\n\n### 2.1 {{模块 / 栈名}}\n- {{条目}}\n\n## 3. 增补\n", + {"order": {"stack": "java-spring"}}), []) + covered = "## 2. 规范集 (Standards)\n\n### 2.1 order / java-spring\n- 规约条目\n\n## 3. 增补\n" + self.assertEqual(validation.validate_standards(covered, + {"order": {"stack": "java-spring"}}), []) + issues = validation.validate_standards(covered, + {"order": {"stack": "java-spring"}, + "web": {"stack": "vue3"}}) + self.assertEqual([i["level"] for i in issues], ["ERROR"]) + self.assertIn("web", issues[0]["message"]) + # 无栈模块（无法判定技术栈）跳过 + self.assertEqual(validation.validate_standards(covered, {"app": {"stack": None}}), []) + + def test_release_template_vacuous_and_unknown_feature(self): + self.assertEqual(validation.validate_release( + "## 3. 发布内容\n\n| {{feature-id}} | {{✅ 已实现}} | {{...}} |\n\n## 4. 校验\n", + ["refund"]), []) + body = ("## 3. 发布内容 (Scope)\n\n" + "| 功能 (feature) | 状态 | 说明 |\n| :--- | :--- | :--- |\n" + "| refund | ✅ | ok |\n| ghost | ✅ | 幽灵 |\n\n## 4. 校验\n") + issues = validation.validate_release(body, ["refund"]) + self.assertEqual([i["level"] for i in issues], ["ERROR"]) + self.assertIn("ghost", issues[0]["message"]) + + def test_known_task_ids(self): + root = Path(tempfile.mkdtemp()) + (root / "docs" / "features" / "refund").mkdir(parents=True) + (root / "docs" / "features" / "Bad_ID").mkdir(parents=True) + self.assertEqual(validation.known_task_ids(root), ["refund"]) + if __name__ == "__main__": unittest.main()