-
Notifications
You must be signed in to change notification settings - Fork 26
Expand file tree
/
Copy pathwhen_evaluator.py
More file actions
74 lines (62 loc) · 3.52 KB
/
Copy pathwhen_evaluator.py
File metadata and controls
74 lines (62 loc) · 3.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
from __future__ import annotations
import fnmatch
import logging
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from src.rules.models import RuleWhen
logger = logging.getLogger(__name__)
def should_apply_rule(when: RuleWhen | None, event_data: dict[str, Any]) -> tuple[bool, str]:
"""
Return whether the rule should be evaluated for this event, and a reason if skipped.
Args:
when: Parsed RuleWhen block (or None when the rule has no predicates).
event_data: Enriched event data, expected to include `contributor_context`
and `changed_files` when the predicates reference them.
Returns:
A tuple of (applies, reason). ``applies`` is True when all named
predicates in ``when`` hold (or ``when`` is empty/None). ``reason``
is a human-readable explanation when the rule is skipped, or an
empty string when the rule applies. If a predicate is present but
its required context is missing, the rule is applied (fail-open)
and a warning is logged — skipping silently on missing data would
hide misconfiguration.
"""
if when is None:
return True, ""
contributor_ctx = event_data.get("contributor_context") or {}
if when.contributor is not None:
if not contributor_ctx:
logger.warning("when.contributor set but contributor_context missing — applying rule")
elif contributor_ctx.get("merged_pr_count") is None:
# API failure: we cannot tell whether the author is first-time or trusted.
# Fail-open (apply the rule) so a transient Search API outage does not
# silently disable stricter checks for newcomers.
logger.warning(f"when.contributor='{when.contributor}' set but merged_pr_count is unknown — applying rule")
else:
predicate = when.contributor.strip().lower()
if predicate == "first_time":
if not contributor_ctx.get("is_first_time", False):
return False, "contributor is not first-time"
elif predicate == "trusted":
if not contributor_ctx.get("trusted", False):
return False, "contributor is not trusted"
else:
logger.warning(f"Unknown contributor predicate '{when.contributor}' — ignoring")
if when.pr_count_below is not None:
if not contributor_ctx:
logger.warning("when.pr_count_below set but contributor_context missing — applying rule")
else:
merged_count = contributor_ctx.get("merged_pr_count")
if merged_count is None:
logger.warning("when.pr_count_below set but merged_pr_count is None — applying rule")
elif merged_count >= when.pr_count_below:
return False, f"contributor has {merged_count} merged PRs (threshold: {when.pr_count_below})"
if when.files_match is not None:
patterns: list[str] = [when.files_match] if isinstance(when.files_match, str) else list(when.files_match)
changed_files = event_data.get("changed_files") or []
filenames = [f.get("filename", "") for f in changed_files if isinstance(f, dict) and f.get("filename")]
# TODO: swap fnmatch for pathspec gitwildmatch once the expression parser lands.
# fnmatch's `*` matches `/`, so `src/*.py` wrongly matches `src/sub/x.py`.
if not any(fnmatch.fnmatch(name, pat) for name in filenames for pat in patterns):
return False, f"no changed files match pattern {patterns}"
return True, ""