|
7 | 7 |
|
8 | 8 | import json |
9 | 9 | import re |
10 | | -from pathlib import Path |
| 10 | +from pathlib import Path, PurePosixPath |
11 | 11 | from typing import Any |
12 | 12 |
|
13 | 13 | DEFAULT_POLICY_PATH = Path(".github/codex_auto_merge_policy.json") |
@@ -70,16 +70,72 @@ def scan_diff(diff_text: str, path_patterns: list[re.Pattern[str]]) -> list[str] |
70 | 70 | return list(dict.fromkeys(violations)) |
71 | 71 |
|
72 | 72 |
|
| 73 | +def _safe_exact_paths(values: Any) -> set[str] | None: |
| 74 | + if not isinstance(values, list) or not values: |
| 75 | + return None |
| 76 | + paths: set[str] = set() |
| 77 | + for value in values: |
| 78 | + if not isinstance(value, str) or value != value.strip(): |
| 79 | + return None |
| 80 | + path = PurePosixPath(value) |
| 81 | + if ( |
| 82 | + not value |
| 83 | + or path.is_absolute() |
| 84 | + or path.as_posix() != value |
| 85 | + or ".." in path.parts |
| 86 | + or "\\" in value |
| 87 | + or any(character in value for character in "*?[]") |
| 88 | + or value in paths |
| 89 | + ): |
| 90 | + return None |
| 91 | + paths.add(value) |
| 92 | + return paths |
| 93 | + |
| 94 | + |
73 | 95 | def check_metadata(files: list[dict[str, Any]], policy: dict[str, Any]) -> list[str]: |
74 | 96 | issues: list[str] = [] |
| 97 | + approved_deleted_paths: set[str] = set() |
75 | 98 | max_files = policy.get("max_changed_files", 50) |
76 | 99 | max_lines = policy.get("max_changed_lines", 5000) |
| 100 | + changed_paths = { |
| 101 | + filename |
| 102 | + for f in files |
| 103 | + if isinstance((filename := f.get("filename")), str) |
| 104 | + } |
| 105 | + removed_paths = { |
| 106 | + f["filename"] |
| 107 | + for f in files |
| 108 | + if isinstance(f.get("filename"), str) |
| 109 | + and (f.get("status") or "").lower().strip() == "removed" |
| 110 | + } |
| 111 | + configured_bundles = policy.get("approved_change_bundles", []) |
| 112 | + if isinstance(configured_bundles, list): |
| 113 | + for bundle in configured_bundles: |
| 114 | + if not isinstance(bundle, dict): |
| 115 | + continue |
| 116 | + exact_changed_paths = _safe_exact_paths(bundle.get("exact_changed_paths")) |
| 117 | + exact_deleted_paths = _safe_exact_paths(bundle.get("exact_deleted_paths")) |
| 118 | + bundle_max_lines = bundle.get("max_changed_lines") |
| 119 | + if ( |
| 120 | + exact_changed_paths is None |
| 121 | + or exact_deleted_paths is None |
| 122 | + or not exact_deleted_paths.issubset(exact_changed_paths) |
| 123 | + or type(bundle_max_lines) is not int |
| 124 | + or bundle_max_lines < max_lines |
| 125 | + or changed_paths != exact_changed_paths |
| 126 | + or removed_paths != exact_deleted_paths |
| 127 | + ): |
| 128 | + continue |
| 129 | + approved_deleted_paths = exact_deleted_paths |
| 130 | + max_lines = bundle_max_lines |
| 131 | + break |
| 132 | + |
77 | 133 | total_added = sum(f.get("additions", 0) or 0 for f in files) |
78 | 134 | total_deleted = sum(f.get("deletions", 0) or 0 for f in files) |
79 | 135 | for f in files: |
80 | 136 | filename = f.get("filename", "?") |
81 | 137 | status = (f.get("status") or "").lower().strip() |
82 | | - if status == "removed": |
| 138 | + if status == "removed" and filename not in approved_deleted_paths: |
83 | 139 | issues.append(f"**File deleted**: `{filename}` — verify intentional") |
84 | 140 | elif status == "renamed": |
85 | 141 | issues.append(f"**File renamed**: `{f.get('previous_filename', '?')}` → `{filename}`") |
|
0 commit comments