|
| 1 | +"""第二批会话实测回归:误拦治理深化 + 可审计性(2026-09-22)。 |
| 2 | +
|
| 3 | +主题:判定不许"一刀切掩盖"——SC1071 剥离后重判同批真问题;豁免与门禁决策 |
| 4 | +必须可回溯(结构化日志);存量问题的豁免**必须有基线证据**(同工具同命令跑 |
| 5 | +改动前内容),无证据绝不豁免(verdict-integrity 既有原则,本批首次落地)。 |
| 6 | +""" |
| 7 | +from __future__ import annotations |
| 8 | + |
| 9 | +import json |
| 10 | +import os |
| 11 | +import subprocess |
| 12 | +import sys |
| 13 | +import tempfile |
| 14 | +import unittest |
| 15 | +from pathlib import Path |
| 16 | + |
| 17 | +PLUGIN = Path(__file__).resolve().parents[1] |
| 18 | +sys.path[:0] = [str(PLUGIN / "scripts"), str(PLUGIN / "hooks")] |
| 19 | + |
| 20 | +import gate_lib |
| 21 | +import pre_tool_git_guard |
| 22 | +import scope |
| 23 | +from verdict import FAIL, UNVERIFIED, finding_signatures, lint_verdict |
| 24 | + |
| 25 | + |
| 26 | +def _git(root: Path, *args: str) -> str: |
| 27 | + return subprocess.run(["git", *args], cwd=root, capture_output=True, |
| 28 | + check=True, text=True).stdout |
| 29 | + |
| 30 | + |
| 31 | +class GitRepoCase(unittest.TestCase): |
| 32 | + def setUp(self): |
| 33 | + self._td = tempfile.TemporaryDirectory(prefix="cg-b2-") |
| 34 | + self.addCleanup(self._td.cleanup) |
| 35 | + self.root = Path(self._td.name) / "repo" |
| 36 | + self.root.mkdir() |
| 37 | + _git(self.root, "init", "-q") |
| 38 | + _git(self.root, "config", "user.email", "t@t") |
| 39 | + _git(self.root, "config", "user.name", "t") |
| 40 | + |
| 41 | + def put(self, rel: str, text: str) -> Path: |
| 42 | + p = self.root / rel |
| 43 | + p.parent.mkdir(parents=True, exist_ok=True) |
| 44 | + p.write_text(text, encoding="utf-8") |
| 45 | + return p |
| 46 | + |
| 47 | + |
| 48 | +class DialectMaskingTests(unittest.TestCase): |
| 49 | + """SC1071 不得掩盖同批其余诊断;剥离后无剩余才算 UNVERIFIED。""" |
| 50 | + |
| 51 | + def test_sc1071_plus_real_issue_judges_on_real_issue(self): |
| 52 | + output = ( |
| 53 | + "In run.zsh line 1:\n#!/bin/zsh\n^-- SC1071 (error): ShellCheck only supports sh\n\n" |
| 54 | + "In lib.sh line 3:\n x=$y ^-- SC2086 (info): Double quote to prevent globbing\n" |
| 55 | + ) |
| 56 | + status, _why = lint_verdict(1, ["shellcheck", "run.zsh", "lib.sh"], output) |
| 57 | + self.assertEqual(FAIL, status, "剥离 SC1071 后 lib.sh 的真问题必须仍判 FAIL") |
| 58 | + |
| 59 | + def test_sc1071_alone_is_unverified(self): |
| 60 | + output = "In run.zsh line 1:\n#!/bin/zsh\n^-- SC1071 (error): ShellCheck only supports sh\n" |
| 61 | + status, _why = lint_verdict(1, ["shellcheck", "run.zsh"], output) |
| 62 | + self.assertEqual(UNVERIFIED, status) |
| 63 | + |
| 64 | + def test_finding_signatures_extract_codes(self): |
| 65 | + sigs = finding_signatures("a.py:1:1: F401 unused\nIn b.sh line 2:\n^-- SC2086 (info): quote") |
| 66 | + self.assertIn("F401", sigs) |
| 67 | + self.assertIn("SC2086", sigs) |
| 68 | + |
| 69 | + def test_finding_signatures_normalize_without_codes(self): |
| 70 | + sigs = finding_signatures("old.py:1:1: something went wrong here") |
| 71 | + self.assertTrue(sigs, "无规则码时也要产出可比较的签名") |
| 72 | + self.assertEqual(sigs, finding_signatures("other.py:9:3: something went wrong here"), |
| 73 | + "路径与行号变化不影响同一发现的签名") |
| 74 | + |
| 75 | + |
| 76 | +class BaselineStalenessTests(GitRepoCase): |
| 77 | + """存量问题豁免必须有基线证据:同工具跑 HEAD 版本得到相同发现才豁免。 |
| 78 | +
|
| 79 | + 用假 linter(确定性输出 CODE1)而非真 shellcheck——SC2086 是 info 级, |
| 80 | + 会被 --severity=warning 过滤掉,基线输出为空就永远比不出存量(实测踩过)。 |
| 81 | + """ |
| 82 | + |
| 83 | + def checker(self): |
| 84 | + return [sys.executable, "-c", |
| 85 | + ("import sys,pathlib;t=pathlib.Path(sys.argv[1]).read_text();" |
| 86 | + "sys.stdout.write('CODE1 bad\\n') if 'BAD' in t else None"), |
| 87 | + "{file}"] |
| 88 | + |
| 89 | + def test_pre_existing_finding_is_stale_with_baseline(self): |
| 90 | + self.put("lib.sh", "#!/bin/bash\nBAD\n") |
| 91 | + _git(self.root, "add", "-A") |
| 92 | + _git(self.root, "commit", "-qm", "base") |
| 93 | + self.put("lib.sh", "#!/bin/bash\nBAD\necho new\n") # 改动未触碰问题行 |
| 94 | + stale = gate_lib.baseline_stale_finding( |
| 95 | + self.root, "lib.sh", self.checker(), "CODE1 bad", |
| 96 | + ) |
| 97 | + self.assertEqual(stale, "存量问题(基线同命令同工具已存在)", |
| 98 | + "HEAD 版本就有同一发现 → 存量,豁免") |
| 99 | + |
| 100 | + def test_new_finding_is_not_stale(self): |
| 101 | + self.put("lib.sh", "#!/bin/bash\necho ok\n") |
| 102 | + _git(self.root, "add", "-A") |
| 103 | + _git(self.root, "commit", "-qm", "base") |
| 104 | + self.put("lib.sh", "#!/bin/bash\necho ok\nBAD\n") # 本次新增问题 |
| 105 | + stale = gate_lib.baseline_stale_finding( |
| 106 | + self.root, "lib.sh", self.checker(), "CODE1 bad", |
| 107 | + ) |
| 108 | + self.assertIsNone(stale, "基线没有的发现是新增违规,必须 FAIL") |
| 109 | + |
| 110 | + def test_no_baseline_refuses_excusal(self): |
| 111 | + self.put("new.sh", "#!/bin/bash\nBAD\n") |
| 112 | + _git(self.root, "add", "-A") |
| 113 | + _git(self.root, "commit", "-qm", "base") |
| 114 | + self.put("untracked.sh", "#!/bin/bash\nBAD\n") # 从未入库 → 无基线 |
| 115 | + stale = gate_lib.baseline_stale_finding( |
| 116 | + self.root, "untracked.sh", self.checker(), "CODE1 bad", |
| 117 | + ) |
| 118 | + self.assertIsNone(stale, "无基线证据绝不豁免(verdict-integrity 既有原则)") |
| 119 | + |
| 120 | + |
| 121 | +class GateDecisionLogTests(GitRepoCase): |
| 122 | + """门禁决策必须落盘可回溯(哪个语言、什么命令、rc、结论)。""" |
| 123 | + |
| 124 | + def test_decision_log_records_language_command_and_status(self): |
| 125 | + home = Path(self._td.name) / "home" |
| 126 | + os.environ["CODEGUARD_HOME"] = str(home) |
| 127 | + try: |
| 128 | + gate_lib.record_gate_decision( |
| 129 | + self.root, "shell", ["shellcheck", "x.sh"], 1, "FAIL", "检查发现违规") |
| 130 | + gate_lib.record_gate_decision( |
| 131 | + self.root, "java", ["mvn", "-B", "verify"], 1, "UNVERIFIED", "工具链执行异常") |
| 132 | + log = (home / "gate-decisions.jsonl").read_text(encoding="utf-8").splitlines() |
| 133 | + self.assertEqual(2, len(log)) |
| 134 | + first = json.loads(log[0]) |
| 135 | + self.assertEqual("shell", first["lang"]) |
| 136 | + self.assertEqual(["shellcheck", "x.sh"], first["cmd"]) |
| 137 | + self.assertEqual(1, first["rc"]) |
| 138 | + self.assertEqual("FAIL", first["status"]) |
| 139 | + self.assertEqual(self.root.name, first["repo"]) |
| 140 | + self.assertIn("ts", first) |
| 141 | + finally: |
| 142 | + os.environ.pop("CODEGUARD_HOME", None) |
| 143 | + |
| 144 | + |
| 145 | +class PathspecResolutionTests(GitRepoCase): |
| 146 | + """git add 的 pathspec 必须用 git 自己的解析(glob/目录/等价语法)。 |
| 147 | +
|
| 148 | + 实现已让位:pre_tool_git_guard.py 正被并行会话活跃修改(其 chain_skip_gate |
| 149 | + 与本批 inline_skip_gate 互补),避免竞争写入;待其稳定后把 match_pathspec |
| 150 | + 接进 staging_intent 再解除跳过。 |
| 151 | + """ |
| 152 | + |
| 153 | + @unittest.skip("pending: pre_tool_git_guard.py 并行修改中,match_pathspec 待接线") |
| 154 | + def test_glob_pathspec_expands_to_real_files(self): |
| 155 | + self.put("a.py", "print(1)\n") |
| 156 | + self.put("pkg/b.py", "print(2)\n") |
| 157 | + matched = pre_tool_git_guard.match_pathspec(self.root, self.root, "*.py") |
| 158 | + self.assertCountEqual(["a.py", "pkg/b.py"], matched, "basename glob 命中任意层级") |
| 159 | + matched = pre_tool_git_guard.match_pathspec(self.root, self.root, "pkg/*.py") |
| 160 | + self.assertEqual(["pkg/b.py"], matched) |
| 161 | + |
| 162 | + @unittest.skip("pending: pre_tool_git_guard.py 并行修改中,match_pathspec 待接线") |
| 163 | + def test_directory_pathspec_expands(self): |
| 164 | + self.put("pkg/a.py", "x=1\n") |
| 165 | + self.put("pkg/sub/b.py", "y=2\n") |
| 166 | + matched = pre_tool_git_guard.match_pathspec(self.root, self.root, "pkg") |
| 167 | + self.assertCountEqual(["pkg/a.py", "pkg/sub/b.py"], matched) |
| 168 | + |
| 169 | + @unittest.skip("pending: pre_tool_git_guard.py 并行修改中,match_pathspec 待接线") |
| 170 | + def test_nonmatching_pathspec_returns_empty(self): |
| 171 | + self.put("a.py", "print(1)\n") |
| 172 | + self.assertEqual([], pre_tool_git_guard.match_pathspec(self.root, self.root, "*.rs")) |
| 173 | + |
| 174 | + |
| 175 | +class RuffTargetVersionTests(GitRepoCase): |
| 176 | + """ruff 注入配置时按项目声明的 Python 版本自适应 target-version。""" |
| 177 | + |
| 178 | + def test_requires_python_drives_target_version(self): |
| 179 | + self.put("pyproject.toml", '[project]\nname = "x"\nrequires-python = ">=3.8"\n') |
| 180 | + args = scope.ruff_config_args(["ruff", "check", "."], self.root) |
| 181 | + self.assertIn("--target-version", args) |
| 182 | + self.assertIn("py38", args) |
| 183 | + |
| 184 | + def test_python_version_file_drives_target_version(self): |
| 185 | + self.put(".python-version", "3.9\n") |
| 186 | + args = scope.ruff_config_args(["ruff", "check", "."], self.root) |
| 187 | + self.assertIn("py39", args) |
| 188 | + |
| 189 | + def test_no_declaration_keeps_default(self): |
| 190 | + args = scope.ruff_config_args(["ruff", "check", "."], self.root) |
| 191 | + self.assertNotIn("--target-version", args) |
| 192 | + self.assertIn("--config", args) |
| 193 | + |
| 194 | + |
| 195 | +class EscapeHatchNoticeTests(GitRepoCase): |
| 196 | + """内联豁免放行时必须向用户明示,不得静默绕过。 |
| 197 | +
|
| 198 | + 同样待 pre_tool_git_guard.py 并行修改稳定后接线(提醒输出在 main() 的 |
| 199 | + inline_skip_gate 分支加 hookSpecificOutput)。 |
| 200 | + """ |
| 201 | + |
| 202 | + @unittest.skip("pending: pre_tool_git_guard.py 并行修改中,豁免提醒待接线") |
| 203 | + def test_inline_skip_emits_user_visible_notice(self): |
| 204 | + self.put("a.py", "print(1)\n") |
| 205 | + _git(self.root, "add", "-A") |
| 206 | + _git(self.root, "commit", "-qm", "base") |
| 207 | + self.put("a.py", "print(2)\n") |
| 208 | + _git(self.root, "add", "-A") |
| 209 | + os.environ["CODEGUARD_HOME"] = str(Path(self._td.name) / "home") |
| 210 | + try: |
| 211 | + import contextlib |
| 212 | + import io |
| 213 | + from unittest.mock import patch |
| 214 | + buf = io.StringIO() |
| 215 | + payload = {"tool_input": { |
| 216 | + "command": f"git -C {self.root} -c codeguard.skipGate=true commit -m x"}} |
| 217 | + with patch.object(pre_tool_git_guard, "read_payload", return_value=payload), \ |
| 218 | + patch.object(pre_tool_git_guard, "ensure_user_path"), \ |
| 219 | + contextlib.redirect_stdout(buf): |
| 220 | + rc = pre_tool_git_guard.main() |
| 221 | + self.assertEqual(0, rc) |
| 222 | + out = buf.getvalue() |
| 223 | + self.assertIn("codeguard", out) |
| 224 | + self.assertTrue("豁免" in out or "skip" in out.lower(), |
| 225 | + "放行必须向用户明示豁免发生") |
| 226 | + finally: |
| 227 | + os.environ.pop("CODEGUARD_HOME", None) |
| 228 | + |
| 229 | + |
| 230 | +class OpenspecValidateResilienceTests(GitRepoCase): |
| 231 | + """openspec validate 内部异常不得逃逸成 fail-open 静默吞。""" |
| 232 | + |
| 233 | + def test_unexpected_exception_is_skipped_not_raised(self): |
| 234 | + self.put("openspec/config.yaml", "schema: spec-driven\n") |
| 235 | + from unittest.mock import patch |
| 236 | + with patch.object(gate_lib.shutil, "which", return_value="/usr/bin/fake-openspec"), \ |
| 237 | + patch.object(gate_lib.subprocess, "run", side_effect=NameError("boom")): |
| 238 | + result = gate_lib._openspec_validate(self.root) |
| 239 | + self.assertIn("skipped", result) |
| 240 | + self.assertTrue(result["failures"] == []) |
| 241 | + self.assertIn("openspec", (result["skipped"] or "")) |
| 242 | + |
| 243 | + |
| 244 | +class SkipGateRetryTests(GitRepoCase): |
| 245 | + """git config 读取必须低超时重试,两次失败才按未豁免 + 审计处理。""" |
| 246 | + |
| 247 | + def test_retry_after_timeout_then_success(self): |
| 248 | + _git(self.root, "config", "codeguard.skipGate", "true") |
| 249 | + calls = {"n": 0} |
| 250 | + real_run = gate_lib.subprocess.run |
| 251 | + |
| 252 | + def flaky(*args, **kwargs): |
| 253 | + calls["n"] += 1 |
| 254 | + if calls["n"] == 1: |
| 255 | + raise gate_lib.subprocess.TimeoutExpired("git", 3) |
| 256 | + return real_run(*args, **kwargs) |
| 257 | + |
| 258 | + from unittest.mock import patch |
| 259 | + os.environ["CODEGUARD_HOME"] = str(Path(self._td.name) / "home") |
| 260 | + try: |
| 261 | + with patch.object(gate_lib.subprocess, "run", side_effect=flaky): |
| 262 | + self.assertTrue(gate_lib.skip_gate_via_git_config(self.root), |
| 263 | + "第一次超时后重试成功必须算豁免") |
| 264 | + state = (Path(self._td.name) / "home" / "session_state.json") |
| 265 | + self.assertFalse(state.exists(), "最终成功不应留 read-error 审计") |
| 266 | + finally: |
| 267 | + os.environ.pop("CODEGUARD_HOME", None) |
| 268 | + |
| 269 | + def test_two_failures_audited(self): |
| 270 | + from unittest.mock import patch |
| 271 | + home = Path(self._td.name) / "home" |
| 272 | + os.environ["CODEGUARD_HOME"] = str(home) |
| 273 | + try: |
| 274 | + with patch.object(gate_lib.subprocess, "run", |
| 275 | + side_effect=gate_lib.subprocess.TimeoutExpired("git", 3)): |
| 276 | + self.assertFalse(gate_lib.skip_gate_via_git_config(self.root)) |
| 277 | + state = json.loads((home / "session_state.json").read_text(encoding="utf-8")) |
| 278 | + self.assertIn("skipGate-read-error", state["_skip"]["kinds"]) |
| 279 | + finally: |
| 280 | + os.environ.pop("CODEGUARD_HOME", None) |
| 281 | + |
| 282 | + |
| 283 | +if __name__ == "__main__": |
| 284 | + unittest.main() |
0 commit comments