Coverage for extra_codeowners/evaluator.py: 99%
133 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-13 09:46 +0000
1"""Pure evaluation of human-or-application CODEOWNER approval evidence."""
3from __future__ import annotations
5from collections.abc import Iterable
7from extra_codeowners.codeowners import CodeownersParseError, parse_codeowners
8from extra_codeowners.models import (
9 ActorKind,
10 EvaluationConclusion,
11 EvaluationInput,
12 EvaluationMessage,
13 EvaluationResult,
14 OwnerSetResult,
15 PathEvidence,
16 PullRequestReview,
17 ReviewState,
18)
19from extra_codeowners.policy import PolicyCompilationError, compile_policy
21_OPINIONATED_STATES = {
22 ReviewState.APPROVED,
23 ReviewState.CHANGES_REQUESTED,
24 ReviewState.DISMISSED,
25}
28def latest_opinionated_reviews(
29 reviews: Iterable[PullRequestReview],
30) -> tuple[PullRequestReview, ...]:
31 """Return each actor's latest approval, rejection, or dismissal.
33 Comments do not supersede an approval. Dismissed reviews do, ensuring an
34 older approval cannot become effective again after GitHub dismisses it.
35 """
37 latest: dict[tuple[ActorKind, int], PullRequestReview] = {}
38 for review in reviews:
39 if review.state not in _OPINIONATED_STATES:
40 continue
41 actor_key = (review.actor.kind, review.actor.user_id)
42 previous = latest.get(actor_key)
43 ordering = (review.submitted_at, review.review_id)
44 if previous is None or ordering > (previous.submitted_at, previous.review_id): 44 ↛ 38line 44 didn't jump to line 38 because the condition on line 44 was always true
45 latest[actor_key] = review
46 return tuple(
47 sorted(latest.values(), key=lambda review: (review.submitted_at, review.review_id))
48 )
51def _evaluation_paths(data: EvaluationInput) -> tuple[str, ...]:
52 paths: list[str] = []
53 for changed_file in data.changed_files:
54 if changed_file.previous_path is not None:
55 paths.append(changed_file.previous_path)
56 paths.append(changed_file.path)
57 return tuple(dict.fromkeys(paths))
60def _current_approvals(
61 data: EvaluationInput,
62) -> tuple[tuple[PullRequestReview, ...], tuple[EvaluationMessage, ...]]:
63 approvals: list[PullRequestReview] = []
64 warnings: list[EvaluationMessage] = []
65 for review in latest_opinionated_reviews(data.reviews):
66 if review.state is not ReviewState.APPROVED:
67 continue
68 if data.options.exact_head_reviews and review.commit_sha != data.head_sha:
69 warnings.append(
70 EvaluationMessage(
71 code="stale_approval_ignored",
72 message=(
73 f"approval {review.review_id} from @{review.actor.login} does not target "
74 "the current pull request head"
75 ),
76 )
77 )
78 continue
79 approvals.append(review)
80 return tuple(approvals), tuple(warnings)
83def _recognized_app_approvals(
84 data: EvaluationInput,
85 approvals: Iterable[PullRequestReview],
86) -> tuple[frozenset[str], tuple[EvaluationMessage, ...]]:
87 approved: set[str] = set()
88 warnings: list[EvaluationMessage] = []
89 for review in approvals:
90 actor = review.actor
91 if actor.kind is not ActorKind.APPLICATION:
92 continue
93 alias = next(
94 (
95 app_alias
96 for app_alias, enrolled in data.organization_policy.apps.items()
97 if enrolled.app_id == actor.app_id
98 and enrolled.slug == actor.app_slug
99 and enrolled.bot_user_id == actor.user_id
100 and actor.login == f"{enrolled.slug}[bot]"
101 ),
102 None,
103 )
104 if alias is None:
105 warnings.append(
106 EvaluationMessage(
107 code="untrusted_application_review_ignored",
108 message=(
109 f"approval {review.review_id} from @{actor.login} did not match an "
110 "organization-enrolled app_id, slug, bot login, and bot_user_id"
111 ),
112 )
113 )
114 continue
115 approved.add(alias)
116 return frozenset(approved), tuple(warnings)
119def _human_approvals_by_owner(
120 approvals: Iterable[PullRequestReview],
121) -> tuple[tuple[PullRequestReview, frozenset[str]], ...]:
122 resolved: list[tuple[PullRequestReview, frozenset[str]]] = []
123 for review in approvals:
124 if review.actor.kind is not ActorKind.HUMAN:
125 continue
126 identities = set(review.actor.owner_aliases)
127 if review.actor.direct_owner_eligible:
128 identities.add(f"@{review.actor.login.lower()}")
129 resolved.append((review, frozenset(identities)))
130 return tuple(resolved)
133def _error_result(
134 messages: tuple[EvaluationMessage, ...],
135 *,
136 warnings: tuple[EvaluationMessage, ...] = (),
137) -> EvaluationResult:
138 return EvaluationResult(
139 conclusion=EvaluationConclusion.FAILURE,
140 summary="Extra CODEOWNERS could not evaluate safely; approval is denied.",
141 errors=messages,
142 warnings=warnings,
143 )
146def evaluate(data: EvaluationInput) -> EvaluationResult:
147 """Evaluate one pull request using only supplied, already-fetched evidence."""
149 if not data.repository_policy.enabled:
150 return EvaluationResult(
151 conclusion=EvaluationConclusion.FAILURE,
152 summary="Repository policy is missing or disabled; approval is denied.",
153 errors=(
154 EvaluationMessage(
155 code="repository_policy_disabled",
156 message=(
157 "repository policy is missing or enabled=false; remove the required "
158 "check from the ruleset before disabling Extra CODEOWNERS"
159 ),
160 ),
161 ),
162 )
164 try:
165 document = parse_codeowners(data.codeowners_text)
166 except CodeownersParseError as error:
167 return _error_result(
168 tuple(
169 EvaluationMessage(code=issue.code, message=issue.render()) for issue in error.issues
170 )
171 )
173 try:
174 policy = compile_policy(
175 data.organization_policy,
176 data.repository_policy,
177 data.options,
178 )
179 except PolicyCompilationError as error:
180 return _error_result(
181 tuple(
182 EvaluationMessage(code=issue.code, message=issue.message) for issue in error.issues
183 )
184 )
186 warnings: list[EvaluationMessage] = []
187 if data.options.allow_insecure_changes:
188 warnings.append(
189 EvaluationMessage(
190 code="insecure_changes_enabled",
191 message=(
192 "the runtime allow-insecure-changes escape hatch disabled built-in "
193 "non-delegable paths; organization-added paths remain enforced"
194 ),
195 )
196 )
198 approvals, approval_warnings = _current_approvals(data)
199 warnings.extend(approval_warnings)
200 approved_apps, app_warnings = _recognized_app_approvals(data, approvals)
201 warnings.extend(app_warnings)
202 human_approvals = _human_approvals_by_owner(approvals)
204 owner_paths: dict[tuple[str, ...], list[str]] = {}
205 unowned_paths: list[str] = []
206 for path in _evaluation_paths(data):
207 owners = tuple(sorted(document.owners_for(path)))
208 if not owners:
209 unowned_paths.append(path)
210 continue
211 paths = owner_paths.setdefault(owners, [])
212 if path not in paths: 212 ↛ 206line 212 didn't jump to line 206 because the condition on line 212 was always true
213 paths.append(path)
215 requirements: list[OwnerSetResult] = []
216 for owners, paths in owner_paths.items():
217 owner_set = frozenset(owners)
218 satisfying_humans = [
219 review for review, identities in human_approvals if identities & owner_set
220 ]
221 if satisfying_humans:
222 actors = tuple(f"human:@{review.actor.login}" for review in satisfying_humans)
223 requirements.append(
224 OwnerSetResult(
225 owners=owners,
226 paths=tuple(paths),
227 satisfied=True,
228 satisfied_by=actors,
229 explanation=(
230 "approved by human CODEOWNER "
231 + ", ".join(f"@{review.actor.login}" for review in satisfying_humans)
232 ),
233 path_evidence=tuple(
234 PathEvidence(
235 path=path,
236 non_delegable=policy.is_non_delegable(path),
237 explanation="covered by the human CODEOWNER approval",
238 )
239 for path in paths
240 ),
241 )
242 )
243 continue
245 path_evidence: list[PathEvidence] = []
246 approving_apps: set[str] = set()
247 all_paths_satisfied = True
248 for path in paths:
249 if policy.is_non_delegable(path):
250 all_paths_satisfied = False
251 path_evidence.append(
252 PathEvidence(
253 path=path,
254 non_delegable=True,
255 explanation=(
256 "application delegation is disabled for this security-sensitive path"
257 ),
258 )
259 )
260 continue
262 decisions = policy.delegation_decisions(path, owner_set, data.labels)
263 eligible_apps = tuple(decision.app_alias for decision in decisions if decision.eligible)
264 path_approvals = tuple(sorted(set(eligible_apps) & approved_apps))
265 if path_approvals:
266 approving_apps.update(path_approvals)
267 explanation = "covered by approved application " + ", ".join(path_approvals)
268 else:
269 all_paths_satisfied = False
270 if eligible_apps:
271 explanation = "eligible application approval is missing: " + ", ".join(
272 eligible_apps
273 )
274 elif decisions:
275 blocked = "; ".join(
276 f"{decision.app_alias}: {', '.join(decision.reasons)}"
277 for decision in decisions
278 )
279 explanation = "delegation label conditions were not met: " + blocked
280 else:
281 explanation = "no application delegation covers this path and owner set"
282 path_evidence.append(
283 PathEvidence(
284 path=path,
285 non_delegable=False,
286 eligible_apps=eligible_apps,
287 approved_apps=path_approvals,
288 explanation=explanation,
289 )
290 )
292 if all_paths_satisfied:
293 aliases = tuple(sorted(approving_apps))
294 requirements.append(
295 OwnerSetResult(
296 owners=owners,
297 paths=tuple(paths),
298 satisfied=True,
299 satisfied_by=tuple(f"app:{alias}" for alias in aliases),
300 explanation="all paths are covered by approved application delegation",
301 path_evidence=tuple(path_evidence),
302 )
303 )
304 else:
305 requirements.append(
306 OwnerSetResult(
307 owners=owners,
308 paths=tuple(paths),
309 satisfied=False,
310 explanation=(
311 "requires an approval from a listed human CODEOWNER or an eligible "
312 "application for every path"
313 ),
314 path_evidence=tuple(path_evidence),
315 )
316 )
318 failed = sum(not requirement.satisfied for requirement in requirements)
319 if failed:
320 conclusion = EvaluationConclusion.FAILURE
321 summary = (
322 f"{failed} of {len(requirements)} distinct CODEOWNER requirements are unsatisfied."
323 )
324 elif requirements:
325 conclusion = EvaluationConclusion.SUCCESS
326 summary = f"All {len(requirements)} distinct CODEOWNER requirements are satisfied."
327 else:
328 conclusion = EvaluationConclusion.SUCCESS
329 summary = "No changed path has a CODEOWNER requirement."
331 return EvaluationResult(
332 conclusion=conclusion,
333 summary=summary,
334 requirements=tuple(requirements),
335 warnings=tuple(warnings),
336 unowned_paths=tuple(unowned_paths),
337 )