feat(reports): classify git fetch as a git ref mutation (closes #297) #344
@@ -2816,3 +2816,119 @@ def assess_email_disclosure(
|
||||
for email in emails
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Git ref mutations (#297)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_GIT_REF_MUTATIONS_FIELD_RE = re.compile(
|
||||
r"^\s*git ref mutations\s*:\s*(.+?)\s*$", re.I | re.M)
|
||||
_GIT_WORKTREE_MUTATIONS_NONE_RE = re.compile(
|
||||
r"^\s*git/worktree mutations\s*:\s*none\b", re.I | re.M)
|
||||
|
||||
_GIT_REF_MUTATING_FIRST_WORDS = frozenset(
|
||||
{"fetch", "pull", "push", "merge", "update-ref"})
|
||||
_GIT_REF_MUTATING_PREFIXES = ("remote update", "branch -d", "branch -D",
|
||||
"reset --hard")
|
||||
|
||||
|
||||
def _command_text(entry):
|
||||
if isinstance(entry, dict):
|
||||
return str(entry.get("command") or "").strip()
|
||||
return str(entry or "").strip()
|
||||
|
||||
|
||||
def _git_subcommand_line(command):
|
||||
tokens = command.split()
|
||||
if not tokens or tokens[0] != "git":
|
||||
return None
|
||||
return " ".join(tokens[1:])
|
||||
|
||||
|
||||
def git_ref_mutating_commands(command_log):
|
||||
"""Return logged commands that update git refs (#297).
|
||||
|
||||
``git fetch`` and friends edit no files but move refs; they are never
|
||||
read-only diagnostics.
|
||||
"""
|
||||
out = []
|
||||
for entry in command_log or []:
|
||||
command = _command_text(entry)
|
||||
rest = _git_subcommand_line(command)
|
||||
if rest is None:
|
||||
continue
|
||||
first = rest.split()[0] if rest.split() else ""
|
||||
if first in _GIT_REF_MUTATING_FIRST_WORDS:
|
||||
out.append(command)
|
||||
continue
|
||||
if any(rest.startswith(prefix) for prefix in _GIT_REF_MUTATING_PREFIXES):
|
||||
out.append(command)
|
||||
return out
|
||||
|
||||
|
||||
def _fetch_targets(ref_commands):
|
||||
"""Extract ``<remote>/<branch>`` targets from git fetch commands."""
|
||||
targets = []
|
||||
for command in ref_commands:
|
||||
rest = _git_subcommand_line(command) or ""
|
||||
tokens = rest.split()
|
||||
if not tokens or tokens[0] != "fetch":
|
||||
continue
|
||||
args = [t for t in tokens[1:] if not t.startswith("-")]
|
||||
if len(args) >= 2:
|
||||
targets.append(f"{args[0]}/{args[1]}")
|
||||
return targets
|
||||
|
||||
|
||||
def assess_git_ref_mutation_report(report_text, *, command_log=None):
|
||||
"""#297: ref updates are Git ref mutations, not read-only diagnostics.
|
||||
|
||||
When the command log holds ref-updating commands, the report must carry
|
||||
a non-none ``Git ref mutations`` entry (naming ``fetched
|
||||
<remote>/<branch>`` for targeted fetches), may not claim ``Git/worktree
|
||||
mutations: None``, and may not list the command as read-only.
|
||||
"""
|
||||
text = report_text or ""
|
||||
lower = text.lower()
|
||||
reasons = []
|
||||
ref_commands = git_ref_mutating_commands(command_log)
|
||||
fetch_targets = _fetch_targets(ref_commands)
|
||||
|
||||
if ref_commands:
|
||||
field = _GIT_REF_MUTATIONS_FIELD_RE.search(text)
|
||||
value = field.group(1).strip().lower() if field else None
|
||||
if not value or value == "none":
|
||||
reasons.append(
|
||||
"ref-updating commands ran but the report has no "
|
||||
"'Git ref mutations' entry"
|
||||
)
|
||||
for target in fetch_targets:
|
||||
if "fetched" not in lower or target.lower() not in lower:
|
||||
reasons.append(
|
||||
"git fetch ran; report must state "
|
||||
f"'Git ref mutations: fetched {target}'"
|
||||
)
|
||||
if _GIT_WORKTREE_MUTATIONS_NONE_RE.search(text):
|
||||
reasons.append(
|
||||
"'Git/worktree mutations: None' rejected: ref-updating "
|
||||
"commands ran"
|
||||
)
|
||||
for line in text.splitlines():
|
||||
if "read-only" not in line.lower():
|
||||
continue
|
||||
for command in ref_commands:
|
||||
if command.lower() in line.lower():
|
||||
reasons.append(
|
||||
"read-only diagnostics may not include ref-updating "
|
||||
f"command '{command}'"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": reasons,
|
||||
"ref_mutating_commands": ref_commands,
|
||||
"fetch_targets": fetch_targets,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
"""Git ref mutations must be reported, never hidden as diagnostics (#297).
|
||||
|
||||
``git fetch`` updates remote-tracking refs even though it edits no files.
|
||||
Reports that ran fetch (or any ref-updating command) must carry a
|
||||
``Git ref mutations`` entry and may not claim ``Git/worktree mutations:
|
||||
None`` or classify the fetch as read-only diagnostics.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import review_proofs
|
||||
|
||||
|
||||
def _assess(report, commands):
|
||||
return review_proofs.assess_git_ref_mutation_report(
|
||||
report, command_log=commands)
|
||||
|
||||
|
||||
class TestGitRefMutationDetection(unittest.TestCase):
|
||||
def test_fetch_only_review_with_proper_ledger_passes(self):
|
||||
report = "\n".join([
|
||||
"Git ref mutations: fetched prgs/master",
|
||||
"Review decision: APPROVE",
|
||||
])
|
||||
res = _assess(report, ["git fetch prgs master"])
|
||||
self.assertTrue(res["proven"], res["reasons"])
|
||||
self.assertEqual(res["ref_mutating_commands"], ["git fetch prgs master"])
|
||||
|
||||
def test_fetch_without_ref_mutation_line_is_blocked(self):
|
||||
report = "Review decision: APPROVE\nMutations: None\n"
|
||||
res = _assess(report, ["git fetch prgs master"])
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(
|
||||
any("git ref mutations" in r.lower() for r in res["reasons"]))
|
||||
|
||||
def test_fetch_with_remote_branch_requires_fetched_target_in_report(self):
|
||||
report = "Git ref mutations: some refs updated\n"
|
||||
res = _assess(report, ["git fetch prgs master"])
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(
|
||||
any("prgs/master" in r for r in res["reasons"]))
|
||||
|
||||
def test_git_worktree_mutations_none_rejected_when_fetch_occurred(self):
|
||||
report = "\n".join([
|
||||
"Git ref mutations: fetched prgs/master",
|
||||
"Git/worktree mutations: None",
|
||||
])
|
||||
res = _assess(report, ["git fetch prgs master"])
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(
|
||||
any("git/worktree mutations" in r.lower() for r in res["reasons"]))
|
||||
|
||||
def test_fetch_classified_as_read_only_diagnostics_rejected(self):
|
||||
report = "\n".join([
|
||||
"Git ref mutations: fetched prgs/master",
|
||||
"Read-only diagnostics: git fetch prgs master, git status",
|
||||
])
|
||||
res = _assess(report, ["git fetch prgs master"])
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(
|
||||
any("read-only" in r.lower() for r in res["reasons"]))
|
||||
|
||||
def test_worktree_creation_is_not_a_ref_mutation(self):
|
||||
report = "Worktree mutations: created branches/review-pr9\n"
|
||||
res = _assess(report, ["git worktree add branches/review-pr9 prgs/master"])
|
||||
self.assertTrue(res["proven"], res["reasons"])
|
||||
self.assertEqual(res["ref_mutating_commands"], [])
|
||||
|
||||
def test_merge_command_counts_as_ref_mutation(self):
|
||||
report = "Review decision: APPROVE\n"
|
||||
res = _assess(report, ["git merge --no-ff feat/x"])
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertIn("git merge --no-ff feat/x", res["ref_mutating_commands"])
|
||||
|
||||
def test_cleanup_branch_delete_counts_as_ref_mutation(self):
|
||||
report = "Cleanup mutations: deleted branch feat/x\n"
|
||||
res = _assess(report, ["git branch -d feat/x"])
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertIn("git branch -d feat/x", res["ref_mutating_commands"])
|
||||
|
||||
def test_no_mutation_blocked_handoff_passes(self):
|
||||
report = "\n".join([
|
||||
"Git/worktree mutations: None",
|
||||
"Current status: blocked handoff, no mutations performed",
|
||||
])
|
||||
res = _assess(report, ["git status --porcelain", "git log --oneline -1"])
|
||||
self.assertTrue(res["proven"], res["reasons"])
|
||||
self.assertEqual(res["ref_mutating_commands"], [])
|
||||
|
||||
def test_command_log_dict_entries_supported(self):
|
||||
report = "Git ref mutations: fetched prgs/master\n"
|
||||
res = _assess(report, [{"command": "git fetch prgs master"}])
|
||||
self.assertTrue(res["proven"], res["reasons"])
|
||||
self.assertEqual(
|
||||
res["ref_mutating_commands"], ["git fetch prgs master"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user