diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 1f6e33c..49933e5 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -36,7 +36,9 @@ jobs: python-version: "3.12" - name: Run governance tests - run: uv run python -m unittest discover -s org-tools/governance/tests + run: >- + uv run --with markdown-it-py==3.0.0 --with markdown==3.10.2 + python -m unittest discover -s org-tools/governance/tests - name: Run label-sync tests run: uv run python -m unittest discover -s org-tools/label-sync diff --git a/org-tools/README.md b/org-tools/README.md index 59640b4..8cfe1a4 100644 --- a/org-tools/README.md +++ b/org-tools/README.md @@ -31,7 +31,8 @@ You can run all unit tests locally from the root of the repository: ```bash # Run all tests -uv run python3 -m unittest discover -s org-tools/governance/tests && \ +uv run --with markdown-it-py==3.0.0 --with markdown==3.10.2 \ + python3 -m unittest discover -s org-tools/governance/tests && \ uv run python3 -m unittest discover -s org-tools/label-sync && \ uv run python3 -m unittest discover -s org-tools/triage/tests && \ uv run --with jsonschema==4.26.0 --with markdown-it-py==3.0.0 --with markdown==3.10.2 \ diff --git a/org-tools/governance/README.md b/org-tools/governance/README.md index 5812b22..4d2d801 100644 --- a/org-tools/governance/README.md +++ b/org-tools/governance/README.md @@ -19,16 +19,23 @@ graph TD C[GovernanceConfigParser] D[GovernanceConfigValidator] end + subgraph normative_detector.py + H[detect_normative_changes] + end + I[requirements_extractor] B --> C B --> D B --> E - C -->|Reads Config| F(python-sdk-rules.yml) + C -->|Reads Config| F(ucp-rules.yml) D -->|Static Sanity Analysis| F + E -->|Checks normative clauses in opted-in rules| H + H -->|Delegates parsing & classification| I E -->|Evaluates approvals| G[GitHub API] ``` - **`pr_validator.py`**: The entrypoint script and validator logic. - - `PullRequestValidator`: Fetches PR metadata (changed files, reviews, authors) from GitHub, queries team memberships, and evaluates reviews against the governance rules. + - `PullRequestValidator`: Fetches PR metadata (changed files, reviews, authors, and — for Markdown files matched by a rule with `normative_escalation: true` — diffs/contents) from GitHub, queries team memberships, escalates those Markdown files to at least Level 3 (Technical Council) when they modify normative RFC 2119 clauses, and evaluates reviews against the governance rules. +- **`normative_detector.py`**: Normative RFC 2119 clause change detector for Markdown documents (`*.md`, `*.markdown`). Delegates Markdown parsing, RFC 2119 classification (`MUST`, `MUST NOT`, `SHALL`, `SHALL NOT`, `SHOULD`, `SHOULD NOT`, `MAY`, and prose `REQUIRED`, `RECOMMENDED`, `OPTIONAL`), exclusion filtering, and clause digest normalization directly to `org-tools/requirements-extractor`, while also flagging unbolded uppercase RFC 2119 keywords in prose. - **`governance_config_parser.py`**: Helper module for loading rules and analyzing configurations. - `GovernanceConfigParser`: Parses YAML configuration files into typed python dataclasses (`Team`, `GovernanceRule`, `GovernanceConfig`, etc.). - `GovernanceConfigValidator`: Performs static analysis on the repository files to detect rule design flaws (overlapping rules or files falling into default fallbacks). @@ -71,6 +78,7 @@ A list of review criteria matching specific file glob patterns: - `patterns`: File paths or wildcards (supports `*`, `**`, `?` globs) matching target files. - `excludes`: Optional glob patterns to exclude from the rule. +- `normative_escalation`: Optional boolean (default `false`). When `true`, Markdown files matched by this rule are escalated to at least Level 3 (Technical Council) if they add, remove, or modify normative RFC 2119 clauses (see [Normative Clause L3 Escalation](#step-2-pr-evaluation-logic) below). In `ucp-rules.yml`, only `documentation_spec` (`docs/specification/**`) opts in. - `requires`: The specific approval targets: - `team`: Requires an approval from an exact team. - `min_team`: Requires an approval from the specified team or any team higher in the hierarchy. @@ -113,12 +121,18 @@ When evaluating a pull request: 1. **Filter Author**: The PR author's own approvals are ignored. 2. **Match Rules**: Compares all modified files in the PR against the rules to collect the matching `RuleRequirement` sets. If a file matches no rule, the `fallback` requirements are applied. -3. **Resolve Approvals**: +3. **Normative Clause L3 Escalation (opt-in per rule)**: For any changed Markdown document (`*.md`, `*.markdown`) matched by a rule with `normative_escalation: true`, `normative_detector.py` (backed by `org-tools/requirements-extractor`) compares the base and head normative RFC 2119 clauses (excluding code blocks, inline code spans, HTML blocks/comments, headings, table headers, RFC 2119 boilerplate definitions, and schema field optionality annotations). + - If normative clauses (`MUST`, `MUST NOT`, `SHALL`, `SHALL NOT`, `SHOULD`, `SHOULD NOT`, `MAY`, or prose `REQUIRED`, `RECOMMENDED`, `OPTIONAL`) were added, removed, or modified, any requirement on that file with a tier below Level 3 (`level < 3`, such as `devops-maintainers` or `maintainers`) is automatically escalated to **Level 3 (Technical Council)** while preserving `min_approvals`. + - Domain-specific paths (`shopping`, `food`, `lodging`, `payment`/`payments`) escalate to the corresponding Domain Tech Council (`shopping-tech-council`, `food-tech-council`, `lodging-tech-council`, `payments-tech-council`), while cross-domain documents escalate to any Technical Council (Level 3) or higher. + - The escalation reason and touched RFC 2119 keywords are explicitly surfaced in both the Global PR Summary (`* **Reason:** ⚠️ ...`) and the File-by-File Breakdown (`* **Escalation:** ⚠️ ...`). + - **Fail-closed**: If neither the base/head file contents nor the diff patch can be retrieved from GitHub (for example, very large files or diffs), the file is escalated as if its normative clauses changed, and the reason states that the clauses could not be analyzed. + - Markdown files matched only by rules without the flag (such as general docs, `README.md`, or `AGENTS.md`), files that fall back to the `fallback` requirements, and repositories whose rules files do not opt in are not subject to normative clause escalation. +4. **Resolve Approvals**: - Retrieves active PR reviews from GitHub. - Checks if the reviewer is a proxy reviewer. - Queries team memberships for reviewers from the GitHub API. - Maps team memberships to hierarchical levels. -4. **Evaluate Gates**: Validates that the approvals fulfill every required target. Prints results and writes a review summary to `summary.txt`. +5. **Evaluate Gates**: Validates that the approvals fulfill every required target. Prints results and writes a review summary to `summary.txt`. --- @@ -150,9 +164,14 @@ python3 scripts/pr_validator.py \ The test suite mock-patches the GitHub API and local git files to allow fast, isolated test runs: ```bash -# Run pr_validator tests -python3 org-tools/governance/tests/test_pr_validator.py - -# Run config parser and validator tests -python3 org-tools/governance/tests/test_governance_config_parser.py +# Run all governance unit tests +uv run --with markdown-it-py==3.0.0 --with markdown==3.10.2 \ + python3 -m unittest discover -s org-tools/governance/tests + +# Or run individual test suites +uv run --with markdown-it-py==3.0.0 --with markdown==3.10.2 \ + python3 org-tools/governance/tests/test_pr_validator.py +uv run --with markdown-it-py==3.0.0 --with markdown==3.10.2 \ + python3 org-tools/governance/tests/test_normative_detector.py +uv run python3 org-tools/governance/tests/test_governance_config_parser.py ``` diff --git a/org-tools/governance/docs/validation_report.md b/org-tools/governance/docs/validation_report.md index 6da171e..136ae2a 100644 --- a/org-tools/governance/docs/validation_report.md +++ b/org-tools/governance/docs/validation_report.md @@ -63,18 +63,20 @@ Each requirement in the summary is formatted with detailed progress: ```markdown - **[Min Approvals] approval(s)** from [Target Team / Hierarchical Level] + - **Reason:** ⚠️ [Optional escalation reason when normative RFC 2119 clauses are modified] - **Met:** [🟢 Yes / 🔴 No] ([Approved Count]/[Min Approvals] approved - approved by: [Reviewers]) - **Pending:** Needs [Remaining Count] approval(s) from [Target Team] ([Eligible Count] eligible reviewer(s) assigned). [Hint] ``` #### Key Fields: -1. **Requirement Line**: Describes the rule (e.g., `* **2 approvals** from team 'shopping-tech-council'` or `* **1 approval** from team 'shopping-tech-council' or higher in the UCP governance hierarchy`). -2. **Met Line**: +1. **Requirement Line**: Describes the rule (e.g., `* **2 approvals** from team 'shopping-tech-council'`, `* **1 approval** from team 'shopping-tech-council' or higher in the UCP governance hierarchy`, or `* **2 approvals** from any Technical Council (Level 3) or higher in the UCP governance hierarchy`). +2. **Reason Line** (only shown when escalated): Explains why a lower-level requirement (such as Level 1 `devops-maintainers` or Level 2 `maintainers`) was automatically escalated to **Level 3 (Technical Council)** because a Markdown document added, removed, or modified normative RFC 2119 clauses (`MUST`, `MUST NOT`, `SHALL`, `SHALL NOT`, `SHOULD`, `SHOULD NOT`, `MAY`, `REQUIRED`, `RECOMMENDED`, `OPTIONAL`). +3. **Met Line**: - **🟢 Yes**: The requirement is fully satisfied. - **🔴 No**: The requirement is not yet satisfied. - **approved by**: Displays a comma-separated list of the authorized users whose reviews satisfied this requirement. -3. **Pending Line** (only shown if **Met** is **🔴 No**): +4. **Pending Line** (only shown if **Met** is **🔴 No**): - Shows the number of remaining approvals needed. - Displays the number of **eligible reviewers** currently assigned/requested on the PR who can satisfy this requirement. - **Hint:** Explains the current status and next steps: @@ -92,6 +94,7 @@ This section shows the validation status of every single changed file. This is e ```markdown - **File:** `path/to/file.py` - **Status:** [🟢 SATISFIED / 🔴 UNSATISFIED] + - **Escalation:** ⚠️ [Optional escalation reason when normative RFC 2119 clauses are modified] - **Requirements:** [List of individual requirements for this file] ``` @@ -101,6 +104,7 @@ This section shows the validation status of every single changed file. This is e 1. **Rule Matching**: The validator compares the file path against the glob patterns defined in your governance configuration file (e.g., `rules/python-sdk-rules.yml`). 2. **Cumulative Rules**: If a file matches multiple rules, it must satisfy the requirements of **all** matching rules. 3. **Fallback**: If a file does not match any custom rule in the configuration, the catch-all `fallback` requirement is applied. +4. **Normative Clause Escalation (Level 3 Technical Council, opt-in per rule)**: If a Markdown document (`*.md`, `*.markdown`) matched by a rule with `normative_escalation: true` (in `ucp`, `docs/specification/**`) adds, removes, or modifies normative RFC 2119 clauses (`MUST`, `MUST NOT`, `SHALL`, `SHALL NOT`, `SHOULD`, `SHOULD NOT`, `MAY`, or prose `REQUIRED`, `RECOMMENDED`, `OPTIONAL`), any lower-level requirement (`level < 3`) for that file is automatically escalated to at least **Level 3 (Technical Council)** while keeping its required `min_approvals` count. Requirements already at Level 3 or Level 4 (`governance-council`) remain unchanged. If the file contents and diff cannot be retrieved, the file is escalated as a precaution. Files matched only by rules without the flag are unaffected. --- diff --git a/org-tools/governance/rules/ucp-rules.yml b/org-tools/governance/rules/ucp-rules.yml index 64acb3e..2eaea48 100644 --- a/org-tools/governance/rules/ucp-rules.yml +++ b/org-tools/governance/rules/ucp-rules.yml @@ -31,6 +31,9 @@ rules: - name: "documentation_spec" patterns: - "docs/specification/**" + # Escalate Markdown edits that add, remove, or modify normative RFC 2119 + # clauses to Level 3 (Technical Council). + normative_escalation: true requires: # Grants access to maintainers (L2), tech councils (L3) and # governance-council (L4) diff --git a/org-tools/governance/scripts/governance_config_parser.py b/org-tools/governance/scripts/governance_config_parser.py index fc47051..436e304 100644 --- a/org-tools/governance/scripts/governance_config_parser.py +++ b/org-tools/governance/scripts/governance_config_parser.py @@ -31,6 +31,9 @@ class GovernanceRule: patterns: List[str] requires_all: List[RuleRequirement] excluded_patterns: List[str] = field(default_factory=list) + # Opt-in: escalate Markdown files matched by this rule to Level 3 when + # they add, remove, or modify normative RFC 2119 clauses. + normative_escalation: bool = False # Internal compiled pattern fields _compiled_patterns: List[re.Pattern] = field(init=False, repr=False) @@ -104,6 +107,12 @@ def _get_requirements_for_file(self, file_path: str) -> List[RuleRequirement]: return merge_requirements(requirements) + def is_normative_escalation_file(self, file_path: str) -> bool: + """Return True if any rule matching the file opts into normative clause escalation.""" + return any( + rule.normative_escalation and rule.matches(file_path) for rule in self.rules + ) + def get_applicable_requirements( self, changed_files: List[str] ) -> Dict[str, List[RuleRequirement]]: @@ -211,12 +220,19 @@ def _parse_rule(self, rule_data: dict, teams: Dict[str, Team]) -> GovernanceRule if isinstance(excludes, str): excludes = [excludes] + normative_escalation = rule_data.get("normative_escalation", False) + if not isinstance(normative_escalation, bool): + raise ValueError( + f"Rule '{name}' has a non-boolean 'normative_escalation' value." + ) + requires_all = self._parse_requirements(rule_data.get("requires", []), teams) return GovernanceRule( name=name, patterns=patterns, requires_all=requires_all, excluded_patterns=excludes, + normative_escalation=normative_escalation, ) def _resolve_team( diff --git a/org-tools/governance/scripts/normative_detector.py b/org-tools/governance/scripts/normative_detector.py new file mode 100644 index 0000000..2ef4680 --- /dev/null +++ b/org-tools/governance/scripts/normative_detector.py @@ -0,0 +1,409 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Normative RFC 2119 clause change detector for Markdown documents. + +Delegates Markdown parsing, RFC 2119 classification, exclusion rules, and +clause normalization directly to ``requirements_extractor`` +(``org-tools/requirements-extractor``) while also flagging unbolded uppercase +RFC 2119 keywords in prose so that any addition, removal, or modification of +normative clauses in Markdown files is escalated to at least Level 3 (Technical +Council) approval. +""" + +import dataclasses +import re +import sys +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path + +_EXTRACTOR_SCRIPTS_DIR = str( + Path(__file__).resolve().parents[2] / "requirements-extractor" / "scripts" +) +if _EXTRACTOR_SCRIPTS_DIR not in sys.path: + sys.path.insert(0, _EXTRACTOR_SCRIPTS_DIR) + +from requirements_extractor import ( # noqa: E402 + classifier, + config, + identity, + models, + parser, +) + +_LEVEL_ORDER: dict[str, int] = { + level.value: idx for idx, level in enumerate(models.Level) +} + +_CODE_SPAN_RE = re.compile(r"`[^`]*`") + +# Matches a leading field or status label on the opening sentence of a list +# item, e.g. `` `ec_version`: ``, `` `credential` (string, **OPTIONAL**): ``, +# or `` **complete_in_progress**: ``. +_BULLET_LABEL_RE = re.compile( + r"^\s*(" + r"(?:\*\*`[^`]+`\*\*|`[^`]+`|\*\*[^*:]+\*\*|[A-Za-z0-9_.-]+)" + r"(?:\s*\([^)]*\))?" + r")\s*:" +) + + +@dataclass(frozen=True) +class NormativeClause: + """A normative RFC 2119 requirement clause extracted from Markdown prose.""" + + level: str + keyword: str + bolded: bool + text: str + compound_parent: str | None = None + normalized: str = "" + text_digest: str = "" + line_start: int = 1 + line_end: int = 1 + + @property + def comparison_key(self) -> tuple[str, str, bool, str | None, str, str]: + """Tuple identity used to detect any semantic or emphasis change to a normative clause.""" + normalized_parent = ( + identity.normalize(self.compound_parent) if self.compound_parent else None + ) + normalized_text = self.normalized or identity.normalize(self.text) + return ( + self.level, + self.keyword, + self.bolded, + normalized_parent, + normalized_text, + self.text_digest, + ) + + +@dataclass(frozen=True) +class NormativeChangeResult: + """Result of checking a file for normative RFC 2119 clause changes.""" + + has_normative_changes: bool + added_clauses: list[NormativeClause] = field(default_factory=list) + removed_clauses: list[NormativeClause] = field(default_factory=list) + + @property + def affected_keywords(self) -> list[str]: + """Sorted list of distinct RFC 2119 keywords touched by the change.""" + rank_by_keyword: dict[str, int] = {} + for clause in (*self.added_clauses, *self.removed_clauses): + rank = _LEVEL_ORDER.get(clause.level, 99) + if ( + clause.keyword not in rank_by_keyword + or rank < rank_by_keyword[clause.keyword] + ): + rank_by_keyword[clause.keyword] = rank + return sorted(rank_by_keyword, key=lambda k: (rank_by_keyword[k], k)) + + def format_reason(self, target_team_desc: str | None = None) -> str: + """Human-readable explanation for why L3 approval is required.""" + keywords = self.affected_keywords + kw_str = f" ({', '.join(f'`{k}`' for k in keywords)})" if keywords else "" + target_suffix = f" ({target_team_desc})" if target_team_desc else "" + return ( + f"Normative RFC 2119 clauses{kw_str} were added, removed, or modified " + f"in this document; requires at least Level 3 Technical Council " + f"approval{target_suffix}." + ) + + +def is_markdown_file(file_path: str) -> bool: + """Return True if the file path points to a Markdown document.""" + return file_path.lower().endswith((".md", ".markdown")) + + +def _replace_outside_code_spans(text: str, old: str, new: str) -> str: + """Replace ``old`` with ``new`` everywhere outside inline backtick code spans.""" + pieces: list[str] = [] + last = 0 + for span in _CODE_SPAN_RE.finditer(text): + pieces.append(text[last : span.start()].replace(old, new)) + pieces.append(span.group(0)) + last = span.end() + pieces.append(text[last:].replace(old, new)) + return "".join(pieces) + + +def _with_bullet_label( + clause: NormativeClause, bullet_label: str | None +) -> NormativeClause: + """Attach a leading list-item field/status label to ``compound_parent`` and recompute ``text_digest``.""" + if not bullet_label: + return clause + parent = ( + f"{clause.compound_parent} | {bullet_label}" + if clause.compound_parent + else bullet_label + ) + normalized = clause.normalized or identity.normalize(clause.text) + digest = identity.text_digest(parent, normalized) + return dataclasses.replace( + clause, + compound_parent=parent, + normalized=normalized, + text_digest=digest, + ) + + +def _to_normative_clause( + ob: models.Clause, *, bolded: bool, text_override: str | None = None +) -> NormativeClause: + """Convert a classified ``requirements_extractor.models.Clause`` to ``NormativeClause``.""" + text = text_override if text_override is not None else ob.text + level_str = ob.level.value if ob.level else "" + keyword_str = ob.annotation.value if ob.annotation else level_str + normalized = identity.normalize(text) + digest = identity.text_digest(ob.compound_parent, normalized) + return NormativeClause( + level=level_str, + keyword=keyword_str, + bolded=bolded, + text=text, + compound_parent=ob.compound_parent, + normalized=normalized, + text_digest=digest, + line_start=ob.source.line_start, + line_end=ob.source.line_end, + ) + + +def _classify_unbolded_candidate(clause: models.Clause) -> list[NormativeClause]: + """Classify a candidate clause with unbolded uppercase RFC 2119 keywords via ``classifier.classify``.""" + masked = parser.mask_code_spans(clause.text) + hits = classifier.find_keywords(masked, config.ALL_KEYWORDS) + unbolded_hits = [hit for hit in hits if not hit.bolded] + if not unbolded_hits: + return [] + + emphasized_text = clause.text + for hit in reversed(unbolded_hits): + emphasized_text = ( + f"{emphasized_text[: hit.start]}**{emphasized_text[hit.start : hit.end]}**" + f"{emphasized_text[hit.end :]}" + ) + + emphasized_clause = dataclasses.replace(clause, text=emphasized_text) + classified = classifier.classify(emphasized_clause, models.ExtractionReport()) + if not classified: + return [] + + results: list[NormativeClause] = [] + for ob in classified: + if not ob.split_from_compound: + restored_text = clause.text + else: + restored_text = ob.text + for hit in unbolded_hits: + raw_kw = clause.text[hit.start : hit.end] + restored_text = _replace_outside_code_spans( + restored_text, f"**{raw_kw}**", raw_kw + ) + results.append( + _to_normative_clause(ob, bolded=False, text_override=restored_text) + ) + return results + + +def _classify_clause(clause: models.Clause) -> list[NormativeClause]: + """Classify a candidate ``models.Clause`` into zero or more ``NormativeClause`` items.""" + report = models.ExtractionReport() + obligations = classifier.classify(clause, report) + if obligations: + return [_to_normative_clause(ob, bolded=ob.bolded) for ob in obligations] + + if report.candidates_without_emphasis: + return _classify_unbolded_candidate(clause) + + # If a list item has no RFC 2119 keyword of its own, check whether its + # colon-terminated stem (compound_parent) is a normative obligation. + if ( + clause.compound_parent + and clause.source.block_type is models.BlockType.LIST_ITEM + ): + parent_candidate = dataclasses.replace( + clause, + text=clause.compound_parent, + compound_parent=None, + source=dataclasses.replace( + clause.source, block_type=models.BlockType.PARAGRAPH + ), + ) + parent_clauses = _classify_clause(parent_candidate) + if parent_clauses: + parent = parent_clauses[0] + normalized = identity.normalize(clause.text) + digest = identity.text_digest(clause.compound_parent, normalized) + return [ + NormativeClause( + level=parent.level, + keyword=parent.keyword, + bolded=parent.bolded, + text=clause.text, + compound_parent=clause.compound_parent, + normalized=normalized, + text_digest=digest, + line_start=clause.source.line_start, + line_end=clause.source.line_end, + ) + ] + + return [] + + +def extract_normative_clauses( + markdown_text: str, file_path: str = "document.md" +) -> list[NormativeClause]: + """Extract all normative RFC 2119 clauses from a Markdown document.""" + if not markdown_text: + return [] + candidates = parser.parse_text(markdown_text, file_path) + clauses: list[NormativeClause] = [] + active_list_span: tuple[int, int] | None = None + active_bullet_label: str | None = None + + for candidate in candidates: + bullet_label_for_clause: str | None = None + if candidate.source.block_type is models.BlockType.LIST_ITEM: + span = (candidate.source.line_start, candidate.source.line_end) + if span != active_list_span: + active_list_span = span + label_match = _BULLET_LABEL_RE.match(candidate.text) + active_bullet_label = label_match.group(1) if label_match else None + else: + bullet_label_for_clause = active_bullet_label + else: + active_list_span = None + active_bullet_label = None + + for extracted in _classify_clause(candidate): + clauses.append(_with_bullet_label(extracted, bullet_label_for_clause)) + return clauses + + +def _compare_clause_lists( + before_clauses: list[NormativeClause], + after_clauses: list[NormativeClause], +) -> tuple[list[NormativeClause], list[NormativeClause]]: + """Return (added_or_modified_after, removed_or_modified_before) clauses.""" + before_counts = Counter(c.comparison_key for c in before_clauses) + after_counts = Counter(c.comparison_key for c in after_clauses) + + added_diff = after_counts - before_counts + removed_diff = before_counts - after_counts + + added: list[NormativeClause] = [] + for clause in after_clauses: + key = clause.comparison_key + if added_diff[key] > 0: + added.append(clause) + added_diff[key] -= 1 + + removed: list[NormativeClause] = [] + for clause in before_clauses: + key = clause.comparison_key + if removed_diff[key] > 0: + removed.append(clause) + removed_diff[key] -= 1 + + return added, removed + + +def _extract_hunks_from_patch(patch: str) -> list[tuple[str, str]]: + """Parse a unified diff patch into (before_text, after_text) per hunk.""" + if not patch: + return [] + + hunks: list[tuple[str, str]] = [] + before_lines: list[str] = [] + after_lines: list[str] = [] + in_hunk = False + + for line in patch.replace("\r\n", "\n").replace("\r", "\n").split("\n"): + if line.startswith("@@"): + if in_hunk and (before_lines or after_lines): + hunks.append(("\n".join(before_lines), "\n".join(after_lines))) + before_lines = [] + after_lines = [] + in_hunk = True + continue + if not in_hunk: + continue + if line.startswith("\\ No newline at end of file"): + continue + if line.startswith("-") and not line.startswith("---"): + before_lines.append(line[1:]) + elif line.startswith("+") and not line.startswith("+++"): + after_lines.append(line[1:]) + elif line.startswith(" "): + before_lines.append(line[1:]) + after_lines.append(line[1:]) + else: + before_lines.append(line) + after_lines.append(line) + + if in_hunk and (before_lines or after_lines): + hunks.append(("\n".join(before_lines), "\n".join(after_lines))) + + return hunks + + +def detect_normative_changes( + file_path: str, + base_content: str | None = None, + head_content: str | None = None, + patch: str | None = None, +) -> NormativeChangeResult: + """Detect whether a file change adds, removes, or modifies normative clauses. + + Only Markdown files (`*.md`, `*.markdown`) are evaluated. When `base_content` + or `head_content` is provided, full-document comparison is performed so + multi-line paragraphs, code fences, colon-list stems, and table headers + have complete structural context. Falls back to hunk-by-hunk unified diff + comparison when only `patch` is available. + """ + if not is_markdown_file(file_path): + return NormativeChangeResult(has_normative_changes=False) + + if base_content is not None or head_content is not None: + before_clauses = extract_normative_clauses(base_content or "", file_path) + after_clauses = extract_normative_clauses(head_content or "", file_path) + added, removed = _compare_clause_lists(before_clauses, after_clauses) + return NormativeChangeResult( + has_normative_changes=bool(added or removed), + added_clauses=added, + removed_clauses=removed, + ) + + if patch: + all_added: list[NormativeClause] = [] + all_removed: list[NormativeClause] = [] + for before_hunk, after_hunk in _extract_hunks_from_patch(patch): + before_clauses = extract_normative_clauses(before_hunk, file_path) + after_clauses = extract_normative_clauses(after_hunk, file_path) + added, removed = _compare_clause_lists(before_clauses, after_clauses) + all_added.extend(added) + all_removed.extend(removed) + return NormativeChangeResult( + has_normative_changes=bool(all_added or all_removed), + added_clauses=all_added, + removed_clauses=all_removed, + ) + + return NormativeChangeResult(has_normative_changes=False) diff --git a/org-tools/governance/scripts/pr_models.py b/org-tools/governance/scripts/pr_models.py index 8cbc595..4d6e90b 100644 --- a/org-tools/governance/scripts/pr_models.py +++ b/org-tools/governance/scripts/pr_models.py @@ -104,6 +104,7 @@ class RuleRequirement: min_approvals: int team: Team | None = None min_team: Team | None = None + escalation_reason: str | None = None def __post_init__(self): if self.min_approvals is None or not isinstance(self.min_approvals, int): @@ -144,8 +145,20 @@ def merge_requirements(requirements: list[RuleRequirement]) -> list[RuleRequirem merged[key] = req else: existing = merged[key] - if req.min_approvals > existing.min_approvals: - merged[key] = req + max_approvals = max(existing.min_approvals, req.min_approvals) + if existing.escalation_reason and req.escalation_reason: + if req.escalation_reason not in existing.escalation_reason: + reason = f"{existing.escalation_reason}; {req.escalation_reason}" + else: + reason = existing.escalation_reason + else: + reason = existing.escalation_reason or req.escalation_reason + merged[key] = RuleRequirement( + min_approvals=max_approvals, + team=existing.team, + min_team=existing.min_team, + escalation_reason=reason, + ) return list(merged.values()) @@ -177,6 +190,10 @@ class PullRequest: reviews: list[Review] assigned_user_names: list[str] = field(default_factory=list) assigned_team_names: list[str] = field(default_factory=list) + file_patches: dict[str, str] = field(default_factory=dict) + file_contents: dict[str, tuple[str | None, str | None]] = field( + default_factory=dict + ) @classmethod def create( @@ -188,6 +205,8 @@ def create( reviews: list[Review], assigned_user_names: list[str] = None, assigned_team_names: list[str] = None, + file_patches: dict[str, str] | None = None, + file_contents: dict[str, tuple[str | None, str | None]] | None = None, ) -> "PullRequest": # Normalize to lowercase. GitHub usernames and team slugs are case-insensitive. # Reference: https://docs.github.com/en/rest/teams/teams?apiVersion=2026-03-10#get-a-team-by-name @@ -205,6 +224,8 @@ def create( reviews=reviews, assigned_user_names=assigned_u, assigned_team_names=assigned_t, + file_patches=dict(file_patches) if file_patches else {}, + file_contents=dict(file_contents) if file_contents else {}, ) @property @@ -275,6 +296,7 @@ class FileValidationStatus: file_path: str requirement_statuses: list[RequirementStatus] is_satisfied: bool + escalation_reason: str | None = None @dataclass(frozen=True) diff --git a/org-tools/governance/scripts/pr_validator.py b/org-tools/governance/scripts/pr_validator.py index 8e1bdcc..8ecaf45 100755 --- a/org-tools/governance/scripts/pr_validator.py +++ b/org-tools/governance/scripts/pr_validator.py @@ -14,9 +14,12 @@ # limitations under the License. # /// script +# requires-python = ">=3.11" # dependencies = [ # "PyGithub", # "PyYAML", +# "markdown-it-py==3.0.0", +# "markdown==3.10.2", # ] # /// """Pull Request validator using governance rules and team memberships under the Venn Diagram model.""" @@ -24,14 +27,16 @@ import argparse import os import sys +from collections.abc import Callable +from pathlib import Path from github import Auth, Github, GithubException - from governance_config_parser import ( GovernanceConfig, GovernanceConfigParser, RuleRequirement, ) +from normative_detector import detect_normative_changes, is_markdown_file from pr_models import ( FileValidationStatus, MergeableReason, @@ -39,6 +44,7 @@ RequirementStatus, Review, ReviewState, + Team, TeamMemberships, User, ValidationErrorReason, @@ -47,6 +53,29 @@ ) from validation_logger import ValidationLogger +NORMATIVE_ESCALATION_LEVEL = 3 +GENERIC_TECH_COUNCIL_TEAM = Team( + name="technical-council", level=NORMATIVE_ESCALATION_LEVEL +) +DOMAIN_TECH_COUNCIL_BY_SEGMENT: dict[str, str] = { + "shopping": "shopping-tech-council", + "food": "food-tech-council", + "lodging": "lodging-tech-council", + "payment": "payments-tech-council", + "payments": "payments-tech-council", +} + + +def requires_normative_check(config: GovernanceConfig, file_path: str) -> bool: + """Return True if a changed file must be checked for normative RFC 2119 clause changes. + + Only Markdown files matched by a rule that opts in via + `normative_escalation: true` are checked. + """ + return is_markdown_file(file_path) and config.is_normative_escalation_file( + file_path + ) + class GitHubClient: """Handles all interactions with the GitHub API.""" @@ -87,17 +116,86 @@ def fetch_team_memberships( members_by_team=members_by_team, teams=config.teams ) - def fetch_pull_request(self, repo_name: str, pr_number: int) -> PullRequest: + def _fetch_file_content( + self, repo: object, path: str, ref: str | None + ) -> str | None: + """Fetch decoded UTF-8 file content from GitHub at a specific git ref.""" + if not ref or not isinstance(ref, str): + return None + try: + content_file = repo.get_contents(path, ref=ref) + if isinstance(content_file, list): + return None + decoded = getattr(content_file, "decoded_content", None) + if isinstance(decoded, bytes): + return decoded.decode("utf-8", errors="replace") + except Exception: + # Callers treat missing content as "unable to analyze" and fail closed. + return None + return None + + def _fetch_base_and_head_contents( + self, repo: object, pr_file: object, base_sha: str | None, head_sha: str | None + ) -> tuple[str | None, str | None] | None: + """Fetch (base, head) contents for a changed file. + + Returns None unless every side that should exist (base unless the file was + added, head unless it was removed) was fetched successfully, so that a + partial fetch is never mistaken for an added or deleted document. + """ + status = getattr(pr_file, "status", None) + prev_filename = getattr(pr_file, "previous_filename", None) or pr_file.filename + + base_text: str | None = None + if status != "added": + base_text = self._fetch_file_content(repo, prev_filename, base_sha) + if base_text is None: + return None + + head_text: str | None = None + if status != "removed": + head_text = self._fetch_file_content(repo, pr_file.filename, head_sha) + if head_text is None: + return None + + return base_text, head_text + + def fetch_pull_request( + self, + repo_name: str, + pr_number: int, + normative_check: Callable[[str], bool] | None = None, + ) -> PullRequest: """Fetch PR files, reviews, and requested reviewers from GitHub. - Maps them to the PullRequest dataclass. + Maps them to the PullRequest dataclass. For changed files accepted by + `normative_check`, also fetches the unified diff patch and the base/head + file contents used for normative RFC 2119 clause detection. """ try: repo = self.g.get_repo(repo_name) pr = repo.get_pull(pr_number) - # Fetch changed files - changed_files = [f.filename for f in pr.get_files()] + pr_files = list(pr.get_files()) + changed_files = [f.filename for f in pr_files] + file_patches: dict[str, str] = {} + file_contents: dict[str, tuple[str | None, str | None]] = {} + + if normative_check is not None: + base_sha = getattr(getattr(pr, "base", None), "sha", None) + head_sha = getattr(getattr(pr, "head", None), "sha", None) + + for f in pr_files: + if not normative_check(f.filename): + continue + patch_val = getattr(f, "patch", None) + if isinstance(patch_val, str): + file_patches[f.filename] = patch_val + contents = self._fetch_base_and_head_contents( + repo, f, base_sha, head_sha + ) + if contents is not None: + file_contents[f.filename] = contents # Fetch reviews reviews = [] @@ -129,6 +227,8 @@ def fetch_pull_request(self, repo_name: str, pr_number: int) -> PullRequest: reviews=reviews, assigned_user_names=assigned_user_names, assigned_team_names=assigned_team_names, + file_patches=file_patches, + file_contents=file_contents, ) except GithubException as e: raise RuntimeError(f"Failed to fetch PR info from GitHub: {e}") from e @@ -142,6 +242,105 @@ def __init__(self, config: GovernanceConfig, memberships: TeamMemberships): self.config = config self.memberships = memberships + def _resolve_l3_target_team(self, file_path: str) -> Team: + """Resolve the Level 3 Technical Council target team for a file path.""" + segments = Path(file_path).as_posix().lower().split("/") + for segment in segments: + council_slug = DOMAIN_TECH_COUNCIL_BY_SEGMENT.get(segment) + if council_slug and council_slug in self.config.teams: + team = self.config.teams[council_slug] + if team.level >= NORMATIVE_ESCALATION_LEVEL: + return team + return self.config.teams.get("technical-council", GENERIC_TECH_COUNCIL_TEAM) + + @staticmethod + def _escalation_reason(file_path: str, cause: str, sources: list[str]) -> str: + """Build the human-readable reason for a Level 3 normative escalation.""" + return ( + f"{cause} in `{file_path}` (escalated from {', '.join(sources)} " + f"to Level 3 Technical Council)." + ) + + def _apply_normative_escalations( + self, + pr: PullRequest, + requirements_by_file: dict[str, list[RuleRequirement]], + ) -> tuple[dict[str, list[RuleRequirement]], dict[str, str]]: + """Escalate lower-level requirements to Level 3 for opted-in Markdown files that touch normative clauses. + + Fails closed: if neither the base/head contents nor the diff patch of an + opted-in Markdown file are available, the file is escalated as if it + modified normative clauses. + """ + updated_by_file: dict[str, list[RuleRequirement]] = {} + escalation_reasons_by_file: dict[str, str] = {} + + for file_path, reqs in requirements_by_file.items(): + if not requires_normative_check(self.config, file_path): + updated_by_file[file_path] = reqs + continue + + contents = pr.file_contents.get(file_path) + patch = pr.file_patches.get(file_path) + if contents is None and not patch: + cause = ( + "Normative RFC 2119 clauses could not be analyzed because the " + "file contents and diff were unavailable" + ) + else: + base_content, head_content = contents or (None, None) + change_result = detect_normative_changes( + file_path=file_path, + base_content=base_content, + head_content=head_content, + patch=patch, + ) + if not change_result.has_normative_changes: + updated_by_file[file_path] = reqs + continue + keywords = change_result.affected_keywords + kw_str = ( + f" ({', '.join(f'`{k}`' for k in keywords)})" if keywords else "" + ) + cause = ( + f"Normative RFC 2119 clauses{kw_str} were added, removed, " + f"or modified" + ) + + target_l3_team = self._resolve_l3_target_team(file_path) + new_reqs: list[RuleRequirement] = [] + escalated_from_descriptions: list[str] = [] + + for req in reqs: + source_team = req.team or req.min_team + if ( + source_team is not None + and source_team.level < NORMATIVE_ESCALATION_LEVEL + ): + source_desc = f"'{source_team.name}' (Level {source_team.level})" + if source_desc not in escalated_from_descriptions: + escalated_from_descriptions.append(source_desc) + new_reqs.append( + RuleRequirement( + min_approvals=req.min_approvals, + min_team=target_l3_team, + escalation_reason=self._escalation_reason( + file_path, cause, [source_desc] + ), + ) + ) + else: + new_reqs.append(req) + + updated_by_file[file_path] = merge_requirements(new_reqs) + + if escalated_from_descriptions: + escalation_reasons_by_file[file_path] = self._escalation_reason( + file_path, cause, escalated_from_descriptions + ) + + return updated_by_file, escalation_reasons_by_file + def validate(self, pr: PullRequest) -> ValidationResult: """Validate a Pull Request against the governance rules.""" # 1. Draft PR Check @@ -167,8 +366,11 @@ def validate(self, pr: PullRequest) -> ValidationResult: mergeable_reason=MergeableReason.NO_CHANGED_FILES, ) - # 5. Rule Matching + # 5. Rule Matching & Normative Clause L3 Escalation requirements_by_file = self.config.get_applicable_requirements(pr.changed_files) + requirements_by_file, escalation_reasons_by_file = ( + self._apply_normative_escalations(pr, requirements_by_file) + ) unmerged_requirements = [] for reqs in requirements_by_file.values(): unmerged_requirements.extend(reqs) @@ -193,6 +395,7 @@ def validate(self, pr: PullRequest) -> ValidationResult: approver_usernames, assigned_usernames, author=pr.author, + escalation_reasons_by_file=escalation_reasons_by_file, ) # 9. Changes Requested Check @@ -265,6 +468,7 @@ def _evaluate_requirement( min_approvals=min_approvals, team=req.team, min_team=req.min_team, + escalation_reason=req.escalation_reason, ) require_total_approvals = req.min_approvals @@ -351,9 +555,11 @@ def _evaluate_file_statuses( approver_usernames: set[str], assigned_usernames: set[str], author: str | None = None, + escalation_reasons_by_file: dict[str, str] | None = None, ) -> list[FileValidationStatus]: """Evaluates and generates validation statuses for each changed file in the PR.""" file_statuses = [] + escalation_map = escalation_reasons_by_file or {} for file, file_requirements in requirements_by_file.items(): file_req_statuses = self._evaluate_requirements( file_requirements, approver_usernames, assigned_usernames, author=author @@ -365,6 +571,7 @@ def _evaluate_file_statuses( file_path=file, requirement_statuses=file_req_statuses, is_satisfied=file_satisfied, + escalation_reason=escalation_map.get(file), ) ) return file_statuses @@ -421,7 +628,11 @@ def run_validation(args: argparse.Namespace) -> ValidationResult: # 3. Fetch team memberships & PR details memberships = github_client.fetch_team_memberships(args.org, config) - pr = github_client.fetch_pull_request(args.repo, args.pr) + pr = github_client.fetch_pull_request( + args.repo, + args.pr, + normative_check=lambda path: requires_normative_check(config, path), + ) # 4. Validate PR validator = PullRequestValidator(config, memberships) diff --git a/org-tools/governance/scripts/validation_logger.py b/org-tools/governance/scripts/validation_logger.py index 919bfab..2560b1a 100644 --- a/org-tools/governance/scripts/validation_logger.py +++ b/org-tools/governance/scripts/validation_logger.py @@ -93,6 +93,8 @@ def generate_report(self) -> str: ) lines.append(f"* **File:** `{f_status.file_path}`") lines.append(f" * **Status:** {status_icon}") + if f_status.escalation_reason: + lines.append(f" * **Escalation:** ⚠️ {f_status.escalation_reason}") lines.append(" * **Requirements:**") for status in f_status.requirement_statuses: formatted_status = self._format_requirement_status( @@ -123,14 +125,22 @@ def _format_requirement_status( ) -> str: """Formats the status details of a single requirement with Markdown styling.""" req = status.requirement - team_desc = ( - f"team '{req.team.name}'" - if req.team - else f"team '{req.min_team.name}' or higher in the UCP governance hierarchy" - ) + if req.team: + team_desc = f"team '{req.team.name}'" + elif req.min_team and req.min_team.name == "technical-council": + team_desc = "any Technical Council (Level 3) or higher in the UCP governance hierarchy" + else: + team_desc = ( + f"team '{req.min_team.name}' or higher in the UCP governance hierarchy" + ) # A) The requirement description req_line = f"{indent}* **{req.min_approvals} approval{'s' if req.min_approvals > 1 else ''}** from {team_desc}" + reason_line = ( + f"\n{indent} * **Reason:** ⚠️ {req.escalation_reason}" + if req.escalation_reason + else "" + ) # B) Which requirements have been met met_str = "🟢 Yes" if status.is_satisfied else "🔴 No" @@ -144,7 +154,7 @@ def _format_requirement_status( # C) Which still need to be met (Pending if not satisfied, omitted if satisfied) if status.is_satisfied: - return f"{req_line}\n{met_line}" + return f"{req_line}{reason_line}\n{met_line}" missing = req.min_approvals - status.approved_count if missing <= 0: @@ -164,4 +174,4 @@ def _format_requirement_status( f"reviewer{'s' if status.assigned_count != 1 else ''} assigned). {hint}" ) - return f"{req_line}\n{met_line}\n{pending_line}" + return f"{req_line}{reason_line}\n{met_line}\n{pending_line}" diff --git a/org-tools/governance/tests/test_governance_config_parser.py b/org-tools/governance/tests/test_governance_config_parser.py index fe92174..d4b92d7 100644 --- a/org-tools/governance/tests/test_governance_config_parser.py +++ b/org-tools/governance/tests/test_governance_config_parser.py @@ -169,6 +169,39 @@ def test_parse_invalid_root_type(self): parser._parse(["not", "a", "map"]) self.assertIn("must be a YAML map", str(ctx.exception)) + def test_parse_normative_escalation_flag(self): + """Test parsing of the optional per-rule 'normative_escalation' flag.""" + parser = GovernanceConfigParser() + yaml_data = { + "team_hierarchy": {"devops": 1, "maintainers": 2}, + "fallback": {"requires": [{"min_team": "maintainers", "min_approvals": 1}]}, + "rules": [ + { + "name": "spec", + "patterns": ["docs/specification/**"], + "normative_escalation": True, + "requires": [{"min_team": "maintainers", "min_approvals": 2}], + }, + { + "name": "docs", + "patterns": ["docs/**"], + "excludes": ["docs/specification/**"], + "requires": [{"min_team": "devops", "min_approvals": 1}], + }, + ], + } + config = parser._parse(yaml_data) + self.assertTrue(config.rules[0].normative_escalation) + self.assertFalse(config.rules[1].normative_escalation) + self.assertTrue(config.is_normative_escalation_file("docs/specification/a.md")) + self.assertFalse(config.is_normative_escalation_file("docs/guides/a.md")) + self.assertFalse(config.is_normative_escalation_file("README.md")) + + yaml_data["rules"][0]["normative_escalation"] = "yes" + with self.assertRaises(ValueError) as ctx: + parser._parse(yaml_data) + self.assertIn("non-boolean 'normative_escalation'", str(ctx.exception)) + def test_min_approvals_validation(self): """Test validations for min_approvals values.""" # 1. min_approvals missing/None diff --git a/org-tools/governance/tests/test_normative_detector.py b/org-tools/governance/tests/test_normative_detector.py new file mode 100644 index 0000000..f013e0b --- /dev/null +++ b/org-tools/governance/tests/test_normative_detector.py @@ -0,0 +1,463 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the normative RFC 2119 clause extractor and change detector.""" + +import os +import sys +import unittest + +sys.path.insert( + 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../scripts")) +) + +from normative_detector import ( + detect_normative_changes, + extract_normative_clauses, + is_markdown_file, +) + + +class TestNormativeParserAndClassifier(unittest.TestCase): + """Tests for Markdown prose extraction and RFC 2119 classification.""" + + def test_is_markdown_file(self): + """Test Markdown file extension detection.""" + self.assertTrue(is_markdown_file("README.md")) + self.assertTrue(is_markdown_file("docs/specification/checkout.MD")) + self.assertTrue(is_markdown_file("notes.markdown")) + self.assertFalse(is_markdown_file("scripts/pr_validator.py")) + self.assertFalse(is_markdown_file("source/schemas/checkout.json")) + + def test_all_seven_bolded_obligation_keywords(self): + """Test extraction of all 7 bolded RFC 2119 obligation levels.""" + md = ( + "1. The server **MUST** validate the token.\n" + "2. The server **MUST NOT** expose secrets.\n" + "3. The client **SHALL** use HTTPS.\n" + "4. The client **SHALL NOT** reuse nonces.\n" + "5. The response **SHOULD** include a timestamp.\n" + "6. The response **SHOULD NOT** exceed 1MB.\n" + "7. The caller **MAY** cache the result.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 7) + self.assertEqual( + [c.level for c in clauses], + [ + "MUST", + "MUST NOT", + "SHALL", + "SHALL NOT", + "SHOULD", + "SHOULD NOT", + "MAY", + ], + ) + self.assertTrue(all(c.bolded for c in clauses)) + + def test_unbolded_uppercase_keywords_detected(self): + """Test extraction of unbolded uppercase RFC 2119 keywords in prose.""" + md = ( + "The server MUST validate the request.\n\n" + "The server MUST NOT log credentials.\n\n" + "The client SHOULD retry on 503.\n\n" + "This header is REQUIRED for authentication.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 4) + self.assertEqual( + [(c.level, c.keyword, c.bolded) for c in clauses], + [ + ("MUST", "MUST", False), + ("MUST NOT", "MUST NOT", False), + ("SHOULD", "SHOULD", False), + ("MUST", "REQUIRED", False), + ], + ) + + def test_lowercase_and_partial_words_ignored(self): + """Test that lowercase modal verbs and partial word matches are ignored.""" + md = ( + "The server must validate the request and should log errors.\n\n" + "You may optionality check MUSTANG or MAYBE or SHOULDER.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(clauses, []) + + def test_annotation_keywords_promoted_in_prose(self): + """Test that REQUIRED, RECOMMENDED, and OPTIONAL in prose promote to MUST/SHOULD/MAY.""" + md = ( + "TLS 1.3 support is **REQUIRED** for all endpoints.\n\n" + "Idempotency keys are **RECOMMENDED** for POST requests.\n\n" + "Webhook signatures are **OPTIONAL** in sandbox mode.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 3) + self.assertEqual( + [(c.level, c.keyword, c.bolded) for c in clauses], + [ + ("MUST", "REQUIRED", True), + ("SHOULD", "RECOMMENDED", True), + ("MAY", "OPTIONAL", True), + ], + ) + + def test_schema_field_annotations_excluded(self): + """Test that schema field optionality headers like `field` (string, REQUIRED): are excluded.""" + md = ( + "`ec_version` (string, **REQUIRED**):\n\n" + "`checkout_id` (string, REQUIRED).\n\n" + "**`metadata`** (object, **OPTIONAL**):\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(clauses, []) + + def test_rfc2119_boilerplate_excluded(self): + """Test that standard RFC 2119 boilerplate sentences citing >= 4 keywords are excluded.""" + md = ( + 'The key words "**MUST**", "**MUST NOT**", "**REQUIRED**", ' + '"**SHALL**", "**SHALL NOT**", "**SHOULD**", "**SHOULD NOT**", ' + '"**RECOMMENDED**", "**MAY**", and "**OPTIONAL**" in this document ' + "are to be interpreted as described in RFC 2119.\n\n" + 'The key words "MUST", "MUST NOT", "SHOULD", and "MAY" are normative.\n' + ) + clauses = extract_normative_clauses(md) + self.assertEqual(clauses, []) + + def test_fenced_and_indented_code_blocks_ignored(self): + """Test that fenced code blocks and indented code blocks are ignored.""" + md = ( + "```json\n" + '{"note": "The server **MUST** ignore this block"}\n' + "```\n\n" + "~~~python\n" + "# The client MUST ignore tilde fences too\n" + "~~~\n\n" + "Paragraph before indented code.\n\n" + " if x **MUST** be 1:\n" + " return REQUIRED\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(clauses, []) + + def test_inline_code_spans_ignored_and_punctuation_inside_code_preserved(self): + """Test that keywords and punctuation inside inline backticks do not trigger or split.""" + md = ( + "The status value `MUST` or `REQUIRED` is a literal string.\n\n" + "The payload `{transfer: [port2]}` **MUST** be serialized as JSON.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 1) + self.assertEqual( + clauses[0].text, + "The payload `{transfer: [port2]}` **MUST** be serialized as JSON.", + ) + + def test_headings_and_html_blocks_ignored(self): + """Test that ATX headings, Setext headings, and HTML blocks/comments are ignored.""" + md = ( + "# The Server **MUST** Ignore Headings\n\n" + "Setext Heading **MUST** Be Ignored\n" + "==================================\n\n" + "\n\n" + "
The server **MUST** ignore HTML blocks
\n\n" + "The server **MUST** strip
inline HTML tags.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 1) + self.assertEqual(clauses[0].text, "The server **MUST** strip inline HTML tags.") + + def test_abbreviations_and_decimals_preserved(self): + """Test that abbreviations like e.g. and decimals like 1.3. are not corrupted.""" + md = ( + "Clients **MUST** support modern protocols, e.g. TLS version 1.3.\n" + "They **SHOULD** also complete step 2.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 2) + self.assertEqual( + clauses[0].text, + "Clients **MUST** support modern protocols, e.g. TLS version 1.3.", + ) + self.assertEqual( + clauses[1].text, + "They **SHOULD** also complete step 2.", + ) + + def test_colon_stem_attaches_to_list_items_and_resets_after_code_block(self): + """Test colon-terminated stem attachment and reset on intervening code fence.""" + md = ( + "When processing a request, the server:\n\n" + "- **MUST** verify the signature.\n" + "- **SHOULD** log the request ID.\n\n" + "When building a config:\n\n" + "```yaml\n" + "key: value\n" + "```\n\n" + "- **MUST** use UTF-8.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 3) + self.assertEqual( + clauses[0].compound_parent, + "When processing a request, the server:", + ) + self.assertEqual( + clauses[1].compound_parent, + "When processing a request, the server:", + ) + self.assertIsNone(clauses[2].compound_parent) + + def test_table_cells_get_row_and_column_context(self): + """Test GFM pipe table extraction and row | column compound_parent context.""" + md = ( + "| Field | Requirement | Notes |\n" + "| --- | --- | --- |\n" + "| `id` | The server **MUST** populate this field. | Primary key |\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 1) + self.assertEqual(clauses[0].level, "MUST") + self.assertEqual(clauses[0].compound_parent, "`id` | Requirement") + + def test_compound_sentence_splitting_and_collapsing(self): + """Test compound sentence splitting on conjunctions and collapsing without conjunctions.""" + md = ( + "Clients **MUST** send the header and **SHOULD** retry on timeout.\n\n" + "When a client **MUST** retry, it **SHOULD** use exponential backoff.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 3) + self.assertEqual(clauses[0].level, "MUST") + self.assertEqual(clauses[0].text, "Clients **MUST** send the header") + self.assertEqual(clauses[1].level, "SHOULD") + self.assertEqual(clauses[1].text, "Clients **SHOULD** retry on timeout.") + # Second sentence has no coordinating conjunction between MUST and SHOULD -> collapses to MUST + self.assertEqual(clauses[2].level, "MUST") + self.assertEqual( + clauses[2].text, + "When a client **MUST** retry, it **SHOULD** use exponential backoff.", + ) + + def test_compound_conjunctions_inside_code_spans_not_split(self): + """Test that conjunctions inside inline code spans do not split compound sentences.""" + md = ( + "If a header **MUST** equal `a;b` (**MAY** also equal `c`), reject others.\n\n" + "Clients MUST send `**MUST**` and SHOULD log retries.\n" + ) + clauses = extract_normative_clauses(md) + self.assertEqual(len(clauses), 3) + self.assertEqual(clauses[0].level, "MUST") + self.assertEqual( + clauses[0].text, + "If a header **MUST** equal `a;b` (**MAY** also equal `c`), reject others.", + ) + self.assertEqual(clauses[1].text, "Clients MUST send `**MUST**`") + self.assertEqual(clauses[2].text, "Clients SHOULD log retries.") + + +class TestDetectNormativeChanges(unittest.TestCase): + """Tests for detect_normative_changes across full content and diff patches.""" + + def test_non_markdown_file_ignored(self): + """Test that non-Markdown files never trigger normative clause changes.""" + res = detect_normative_changes( + file_path="src/server.py", + base_content="# The server MUST do X\n", + head_content="# The server MUST do Y\n", + ) + self.assertFalse(res.has_normative_changes) + + def test_non_normative_edit_in_same_paragraph_does_not_trigger(self): + """Test that editing a non-normative sentence next to a normative one does not trigger.""" + base = ( + "This section provides an overview of checkout.\n" + "The server **MUST** return a 200 OK response.\n" + ) + head = ( + "This section provides a detailed overview of checkout sessions.\n" + "The server **MUST** return a 200 OK response.\n" + ) + res = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head, + ) + self.assertFalse(res.has_normative_changes) + + def test_wrapped_line_edit_in_normative_sentence_triggers(self): + """Test that editing a continuation line of a normative sentence triggers escalation.""" + base = "The server **MUST** return\na 200 OK response within 5 seconds.\n" + head = "The server **MUST** return\na 201 Created response within 10 seconds.\n" + res = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head, + ) + self.assertTrue(res.has_normative_changes) + self.assertEqual(res.affected_keywords, ["MUST"]) + + def test_pure_line_wrap_reflow_does_not_trigger_but_emphasis_change_triggers(self): + """Test that pure hard-wrap reflow matches requirements_extractor while emphasis edits trigger.""" + base = "The server **MUST** return\na 200 OK response.\n" + head = "The server **MUST** return a 200 OK response.\n" + res = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head, + ) + self.assertFalse(res.has_normative_changes) + + head_unbolded = "The server MUST return a 200 OK response.\n" + res_emphasis = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head_unbolded, + ) + self.assertTrue(res_emphasis.has_normative_changes) + self.assertEqual(res_emphasis.affected_keywords, ["MUST"]) + + def test_editing_colon_stem_above_normative_list_triggers(self): + """Test that modifying the colon stem introducing normative bullets triggers escalation.""" + base = ( + "When handling a checkout request, the server:\n\n" + "- **MUST** validate the cart items.\n" + ) + head = ( + "When handling an order request, the server:\n\n" + "- **MUST** validate the cart items.\n" + ) + res = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head, + ) + self.assertTrue(res.has_normative_changes) + self.assertEqual(res.affected_keywords, ["MUST"]) + + def test_editing_list_item_under_normative_stem_triggers(self): + """Test that modifying a bullet item under a normative colon stem triggers escalation.""" + base = ( + "The server **MUST** validate all of the following fields:\n\n" + "- `checkout_id`\n" + "- `merchant_id`\n" + ) + head = ( + "The server **MUST** validate all of the following fields:\n\n" + "- `checkout_id`\n" + "- `store_id`\n" + ) + res = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head, + ) + self.assertTrue(res.has_normative_changes) + self.assertEqual(res.affected_keywords, ["MUST"]) + + def test_patch_fallback_mode_detects_changes(self): + """Test that unified diff patch fallback detects normative changes when full content is absent.""" + patch = ( + "@@ -10,3 +10,3 @@\n" + " Context line before.\n" + "-The client **SHOULD** retry failed requests.\n" + "+The client **MUST** retry failed requests.\n" + " Context line after.\n" + ) + res = detect_normative_changes( + "docs/specification/overview.md", + patch=patch, + ) + self.assertTrue(res.has_normative_changes) + self.assertEqual(res.affected_keywords, ["MUST", "SHOULD"]) + + def test_link_retargeting_and_non_keyword_formatting_do_not_trigger(self): + """Test that link URL retargeting, attr_list attributes, and non-keyword italics do not trigger, while link label or code-span wildcard changes do.""" + base = ( + "Payloads **MUST** conform to [RFC 8259](http://tools.ietf.org/html/rfc8259) " + "for capability `dev.ucp.*` - with *valid* UTF-8.\n" + ) + head_cosmetic = ( + 'Payloads **MUST** conform to [RFC 8259](https://datatracker.ietf.org/doc/html/rfc8259){ target="_blank" } ' + "for capability `dev.ucp.*` \u2014 with valid UTF-8.\n" + ) + res_cosmetic = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head_cosmetic, + ) + self.assertFalse(res_cosmetic.has_normative_changes) + + head_code_wildcard_changed = ( + "Payloads **MUST** conform to [RFC 8259](http://tools.ietf.org/html/rfc8259) " + "for capability `dev.ucp.shopping` - with *valid* UTF-8.\n" + ) + res_wildcard = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head_code_wildcard_changed, + ) + self.assertTrue(res_wildcard.has_normative_changes) + + head_link_label_changed = ( + "Payloads **MUST** conform to [RFC 9110](http://tools.ietf.org/html/rfc8259) " + "for capability `dev.ucp.*` - with *valid* UTF-8.\n" + ) + res_label = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head_link_label_changed, + ) + self.assertTrue(res_label.has_normative_changes) + + def test_label_led_bullet_field_or_status_change_triggers(self): + """Test that renaming a field/status label on a multi-sentence bullet triggers while non-normative bullet prose does not.""" + base = ( + "* **complete_in_progress**: The Complete Checkout request was accepted. " + "The response **MUST NOT** contain order.\n" + "* `credential` (string, **OPTIONAL**): Bearer token. " + "**MUST** be set if `auth` is present.\n" + ) + head_non_normative_prose = ( + "* **complete_in_progress**: The Complete Checkout request was accepted by the Business. " + "The response **MUST NOT** contain order.\n" + "* `credential` (string, **OPTIONAL**): Opaque bearer token. " + "**MUST** be set if `auth` is present.\n" + ) + self.assertFalse( + detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head_non_normative_prose, + ).has_normative_changes + ) + + head_renamed_field = ( + "* **complete_in_progress**: The Complete Checkout request was accepted. " + "The response **MUST NOT** contain order.\n" + "* `api_key` (string, **OPTIONAL**): Bearer token. " + "**MUST** be set if `auth` is present.\n" + ) + res_renamed = detect_normative_changes( + "docs/specification/checkout.md", + base_content=base, + head_content=head_renamed_field, + ) + self.assertTrue(res_renamed.has_normative_changes) + self.assertEqual(res_renamed.affected_keywords, ["MUST"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/org-tools/governance/tests/test_pr_models.py b/org-tools/governance/tests/test_pr_models.py index a5cf5db..f9b9a20 100644 --- a/org-tools/governance/tests/test_pr_models.py +++ b/org-tools/governance/tests/test_pr_models.py @@ -140,6 +140,28 @@ def test_merge_requirements(self): self.assertEqual(devops_req.target_key, (RequirementTargetType.TEAM, "devops")) self.assertEqual(admin_req.target_key, (RequirementTargetType.TEAM, "admins")) + def test_merge_requirements_preserves_escalation_reasons(self): + """Test that merge_requirements preserves and combines distinct escalation reasons.""" + tc_team = Team.create(name="technical-council", level=3) + req1 = RuleRequirement( + min_approvals=1, + min_team=tc_team, + escalation_reason="Escalated in docs/a.md", + ) + req2 = RuleRequirement( + min_approvals=2, + min_team=tc_team, + escalation_reason="Escalated in docs/b.md", + ) + + merged = merge_requirements([req1, req2]) + self.assertEqual(len(merged), 1) + self.assertEqual(merged[0].min_approvals, 2) + self.assertEqual( + merged[0].escalation_reason, + "Escalated in docs/a.md; Escalated in docs/b.md", + ) + if __name__ == "__main__": unittest.main() diff --git a/org-tools/governance/tests/test_pr_validator.py b/org-tools/governance/tests/test_pr_validator.py index 77e689e..1c35d99 100644 --- a/org-tools/governance/tests/test_pr_validator.py +++ b/org-tools/governance/tests/test_pr_validator.py @@ -1010,6 +1010,286 @@ def test_author_comments_not_counted_in_assigned(self): self.assertEqual(status.assigned_count, 0) # Author comments do not count self.assertEqual(status.approvers, ["maint1"]) + def _normative_config(self, requires=None, normative_escalation=True): + """Build a config with a single 'docs/specification/**' rule for normative escalation tests.""" + spec_rule = GovernanceRule( + name="Documentation Specification", + patterns=["docs/specification/**"], + requires_all=requires + or [ + RuleRequirement(min_approvals=2, min_team=self.hierarchy["maintainers"]) + ], + normative_escalation=normative_escalation, + ) + general_docs_rule = GovernanceRule( + name="General Docs", + patterns=["docs/**", "README.md", "AGENTS.md"], + excluded_patterns=["docs/specification/**"], + requires_all=[ + RuleRequirement(min_approvals=1, min_team=self.hierarchy["devops"]) + ], + ) + return GovernanceConfig( + teams=self.hierarchy, + rules=[spec_rule, general_docs_rule], + fallback=self.fallback, + proxy_reviewers=self.proxy_reviewers, + ) + + def test_normative_clause_escalates_domain_markdown_to_domain_l3(self): + """Test that touching normative clauses in an opted-in domain doc escalates L2 requirement to that domain's L3 Tech Council.""" + validator = PullRequestValidator(self._normative_config(), self.memberships) + + file_path = "docs/specification/shopping/checkout.md" + base_md = "# Checkout\n\nThis document describes the checkout flow.\n" + head_md = ( + "# Checkout\n\n" + "This document describes the checkout flow.\n" + "The server **MUST** return a valid session ID and **SHOULD** include totals.\n" + ) + + # 2 L2 maintainer approvals (maint1, maint2) are insufficient after L3 escalation + pr_l2_only = PullRequest.create( + number=10, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[ + Review(user="maint1", state=ReviewState.APPROVED), + Review(user="maint2", state=ReviewState.APPROVED), + ], + file_contents={file_path: (base_md, head_md)}, + ) + res_fail = validator.validate(pr_l2_only) + self.assertFalse(res_fail.is_mergeable) + self.assertEqual(res_fail.error, ValidationErrorReason.INSUFFICIENT_APPROVALS) + self.assertEqual(len(res_fail.requirement_statuses), 1) + req_status = res_fail.requirement_statuses[0] + self.assertEqual( + req_status.requirement.min_team, + self.hierarchy["shopping-tech-council"], + ) + self.assertEqual(req_status.requirement.min_approvals, 2) + self.assertIn("`MUST`", req_status.requirement.escalation_reason) + self.assertIn("`SHOULD`", req_status.requirement.escalation_reason) + self.assertIn(file_path, req_status.requirement.escalation_reason) + self.assertEqual( + res_fail.file_statuses[0].escalation_reason, + req_status.requirement.escalation_reason, + ) + + # 2 L3+ approvals satisfy the escalated requirement + pr_l3_ok = PullRequest.create( + number=10, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[ + Review(user="tc-member1", state=ReviewState.APPROVED), + Review(user="tc-member2", state=ReviewState.APPROVED), + ], + file_contents={file_path: (base_md, head_md)}, + ) + res_ok = validator.validate(pr_l3_ok) + self.assertTrue(res_ok.is_mergeable) + self.assertEqual(res_ok.mergeable_reason, MergeableReason.RULES_SATISFIED) + + def test_normative_clause_escalates_general_markdown_to_generic_l3(self): + """Test that touching normative clauses in an opted-in cross-domain doc escalates to generic L3 Technical Council.""" + validator = PullRequestValidator( + self._normative_config( + requires=[ + RuleRequirement( + min_approvals=1, min_team=self.hierarchy["maintainers"] + ) + ] + ), + self.memberships, + ) + file_path = "docs/specification/overview.md" + patch_diff = ( + "@@ -1,2 +1,3 @@\n" + " # Overview\n" + "+Implementations MUST support UTF-8 encoding.\n" + ) + + # Rule requires 1 maintainer (L2); normative edit escalates to L3 technical-council + pr_fail = PullRequest.create( + number=11, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[Review(user="maint1", state=ReviewState.APPROVED)], + file_patches={file_path: patch_diff}, + ) + res_fail = validator.validate(pr_fail) + self.assertFalse(res_fail.is_mergeable) + req_status = res_fail.requirement_statuses[0] + self.assertEqual(req_status.requirement.min_team.name, "technical-council") + self.assertEqual(req_status.requirement.min_team.level, 3) + self.assertEqual(req_status.requirement.min_approvals, 1) + self.assertIn("`MUST`", req_status.requirement.escalation_reason) + + # Any L3 Tech Council member (e.g., food-tc-member1) satisfies the escalated requirement + pr_ok = PullRequest.create( + number=11, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[Review(user="food-tc-member1", state=ReviewState.APPROVED)], + file_patches={file_path: patch_diff}, + ) + res_ok = validator.validate(pr_ok) + self.assertTrue(res_ok.is_mergeable) + + def test_normative_clause_escalation_only_applies_to_opted_in_rules(self): + """Test that Markdown files matched only by rules without 'normative_escalation' are never escalated.""" + validator = PullRequestValidator(self._normative_config(), self.memberships) + base_md = "# Guide\n\nOverview.\n" + head_md = "# Guide\n\nAgents **MUST** run the linter before committing.\n" + + for file_path in ( + "README.md", + "AGENTS.md", + "docs/guides/getting-started.md", + "CHANGELOG.md", # Falls back to fallback requirements + ): + with self.subTest(file_path=file_path): + pr = PullRequest.create( + number=14, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[Review(user="maint1", state=ReviewState.APPROVED)], + file_contents={file_path: (base_md, head_md)}, + ) + res = validator.validate(pr) + self.assertTrue(res.is_mergeable) + self.assertIsNone( + res.requirement_statuses[0].requirement.escalation_reason + ) + self.assertIsNone(res.file_statuses[0].escalation_reason) + + def test_normative_clause_escalation_disabled_without_opt_in(self): + """Test that the same spec file is not escalated when its rule does not opt in.""" + validator = PullRequestValidator( + self._normative_config(normative_escalation=False), self.memberships + ) + file_path = "docs/specification/overview.md" + pr = PullRequest.create( + number=15, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[ + Review(user="maint1", state=ReviewState.APPROVED), + Review(user="maint2", state=ReviewState.APPROVED), + ], + file_contents={file_path: ("# Overview\n", "Clients **MUST** retry.\n")}, + ) + res = validator.validate(pr) + self.assertTrue(res.is_mergeable) + self.assertEqual( + res.requirement_statuses[0].requirement.min_team, + self.hierarchy["maintainers"], + ) + + def test_normative_check_fails_closed_when_content_unavailable(self): + """Test that an opted-in Markdown file with no contents and no patch is escalated to L3.""" + validator = PullRequestValidator(self._normative_config(), self.memberships) + file_path = "docs/specification/overview.md" + pr = PullRequest.create( + number=16, + author="author1", + is_draft=False, + changed_files=[file_path, "docs/guides/intro.md"], + reviews=[ + Review(user="maint1", state=ReviewState.APPROVED), + Review(user="maint2", state=ReviewState.APPROVED), + ], + ) + res = validator.validate(pr) + self.assertFalse(res.is_mergeable) + self.assertEqual(res.error, ValidationErrorReason.INSUFFICIENT_APPROVALS) + + statuses_by_file = {fs.file_path: fs for fs in res.file_statuses} + spec_status = statuses_by_file[file_path] + self.assertIn("could not be analyzed", spec_status.escalation_reason) + self.assertIn(file_path, spec_status.escalation_reason) + spec_req = spec_status.requirement_statuses[0].requirement + self.assertEqual(spec_req.min_team.name, "technical-council") + self.assertEqual(spec_req.min_approvals, 2) + + # Files outside opted-in rules are not escalated even without content + self.assertIsNone(statuses_by_file["docs/guides/intro.md"].escalation_reason) + + def test_non_normative_markdown_edit_does_not_escalate(self): + """Test that editing non-normative prose in an opted-in Markdown document keeps L2 requirement.""" + validator = PullRequestValidator(self._normative_config(), self.memberships) + file_path = "docs/specification/overview.md" + base_md = "# Overview\n\nThe server **MUST** use TLS 1.3.\n\nTypo in intro.\n" + head_md = ( + "# Overview\n\nThe server **MUST** use TLS 1.3.\n\nFixed typo in intro.\n" + ) + + pr = PullRequest.create( + number=12, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[ + Review(user="maint1", state=ReviewState.APPROVED), + Review(user="maint2", state=ReviewState.APPROVED), + ], + file_contents={file_path: (base_md, head_md)}, + ) + res = validator.validate(pr) + self.assertTrue(res.is_mergeable) + self.assertEqual( + res.requirement_statuses[0].requirement.min_team, + self.hierarchy["maintainers"], + ) + self.assertIsNone(res.requirement_statuses[0].requirement.escalation_reason) + self.assertIsNone(res.file_statuses[0].escalation_reason) + + def test_l4_requirement_not_downgraded_on_normative_change(self): + """Test that files already requiring L4 Governance Council remain at L4 when normative clauses change.""" + validator = PullRequestValidator( + self._normative_config( + requires=[ + RuleRequirement( + min_approvals=2, team=self.hierarchy["governance-council"] + ) + ] + ), + self.memberships, + ) + + file_path = "docs/specification/reference.md" + pr = PullRequest.create( + number=13, + author="author1", + is_draft=False, + changed_files=[file_path], + reviews=[ + Review(user="tc-member1", state=ReviewState.APPROVED), + Review(user="tc-member2", state=ReviewState.APPROVED), + ], + file_contents={ + file_path: ( + "# Reference\n", + "# Reference\n\nClients **MUST** authenticate.\n", + ) + }, + ) + res = validator.validate(pr) + self.assertFalse(res.is_mergeable) + self.assertEqual( + res.requirement_statuses[0].requirement.team, + self.hierarchy["governance-council"], + ) + self.assertIsNone(res.requirement_statuses[0].requirement.escalation_reason) + class TestFetchTeamMemberships(unittest.TestCase): """Tests for GitHubClient.fetch_team_memberships method.""" @@ -1131,6 +1411,124 @@ def test_fetch_team_not_found_handled(self): {Team("shopping-tech-council", 3): set()}, ) + def test_fetch_pull_request_populates_markdown_contents_and_patches(self): + """Test fetch_pull_request fetches patches and base/head Markdown contents.""" + mock_g = MagicMock() + mock_repo = MagicMock() + mock_pr = MagicMock() + mock_g.get_repo.return_value = mock_repo + mock_repo.get_pull.return_value = mock_pr + + mock_pr.number = 42 + mock_pr.user.login = "Alice" + mock_pr.draft = False + mock_pr.base.sha = "base-sha-123" + mock_pr.head.sha = "head-sha-456" + + md_file = MagicMock() + md_file.filename = "docs/spec.md" + md_file.status = "modified" + md_file.previous_filename = None + md_file.patch = "@@ -1 +1 @@\n-old\n+The server **MUST** work." + + py_file = MagicMock() + py_file.filename = "src/app.py" + py_file.status = "modified" + py_file.previous_filename = None + py_file.patch = "@@ -1 +1 @@\n-x = 1\n+x = 2" + + mock_pr.get_files.return_value = [md_file, py_file] + mock_pr.get_reviews.return_value = [] + mock_pr.get_review_requests.return_value = ([], []) + + def get_contents_side_effect(path, ref=None): + content_obj = MagicMock() + if path == "docs/spec.md" and ref == "base-sha-123": + content_obj.decoded_content = b"old\n" + return content_obj + if path == "docs/spec.md" and ref == "head-sha-456": + content_obj.decoded_content = b"The server **MUST** work.\n" + return content_obj + raise MockGithubException(status=404, data={"message": "Not Found"}) + + mock_repo.get_contents.side_effect = get_contents_side_effect + + github_client = GitHubClient(mock_g) + pr_checked = github_client.fetch_pull_request( + "Universal-Commerce-Protocol/ucp", + 42, + normative_check=lambda path: path.endswith(".md"), + ) + + self.assertEqual(pr_checked.changed_files, ["docs/spec.md", "src/app.py"]) + # Patches and contents are only collected for files accepted by normative_check + self.assertEqual(list(pr_checked.file_patches), ["docs/spec.md"]) + self.assertEqual( + pr_checked.file_contents, + {"docs/spec.md": ("old\n", "The server **MUST** work.\n")}, + ) + + # Without normative_check, no Markdown contents or patches are fetched + mock_repo.get_contents.reset_mock() + pr_unchecked = github_client.fetch_pull_request( + "Universal-Commerce-Protocol/samples", 42 + ) + self.assertEqual(pr_unchecked.changed_files, ["docs/spec.md", "src/app.py"]) + self.assertEqual(pr_unchecked.file_patches, {}) + self.assertEqual(pr_unchecked.file_contents, {}) + mock_repo.get_contents.assert_not_called() + + def test_fetch_pull_request_omits_partial_contents(self): + """Test that a modified file whose base or head fetch fails has no recorded contents.""" + mock_g = MagicMock() + mock_repo = MagicMock() + mock_pr = MagicMock() + mock_g.get_repo.return_value = mock_repo + mock_repo.get_pull.return_value = mock_pr + + mock_pr.number = 43 + mock_pr.user.login = "Alice" + mock_pr.draft = False + mock_pr.base.sha = "base-sha" + mock_pr.head.sha = "head-sha" + + modified = MagicMock() + modified.filename = "docs/modified.md" + modified.status = "modified" + modified.previous_filename = None + modified.patch = None + + added = MagicMock() + added.filename = "docs/added.md" + added.status = "added" + added.previous_filename = None + added.patch = None + + mock_pr.get_files.return_value = [modified, added] + mock_pr.get_reviews.return_value = [] + mock_pr.get_review_requests.return_value = ([], []) + + def get_contents_side_effect(path, ref=None): + if ref == "head-sha": + content_obj = MagicMock() + content_obj.decoded_content = b"Clients **MUST** retry.\n" + return content_obj + # Base fetches fail (e.g., file too large for the contents API) + raise MockGithubException(status=403, data={"message": "Too large"}) + + mock_repo.get_contents.side_effect = get_contents_side_effect + + pr = GitHubClient(mock_g).fetch_pull_request( + "Universal-Commerce-Protocol/ucp", 43, normative_check=lambda _: True + ) + + # The modified file's base is missing, so it must not be treated as newly added + self.assertNotIn("docs/modified.md", pr.file_contents) + self.assertEqual( + pr.file_contents["docs/added.md"], (None, "Clients **MUST** retry.\n") + ) + self.assertEqual(pr.file_patches, {}) + class TestPRValidatorMain(unittest.TestCase): """Tests for the main function of pr_validator.""" diff --git a/org-tools/governance/tests/test_validation_logger.py b/org-tools/governance/tests/test_validation_logger.py index cee5bd7..51f8a2b 100644 --- a/org-tools/governance/tests/test_validation_logger.py +++ b/org-tools/governance/tests/test_validation_logger.py @@ -477,6 +477,77 @@ def test_write_summary_with_hierarchical_clearance(self, mock_open): ) + HELP_SUFFIX self.assertEqual(expected, written_content) + @patch("builtins.open", new_callable=unittest.mock.mock_open) + def test_write_summary_with_normative_escalation_reason(self, mock_open): + """Test write_summary outputs escalation reason in both Global Summary and File Breakdown.""" + min_team = Team(name="technical-council", level=3) + reason = ( + "Normative RFC 2119 clauses (`MUST`, `SHOULD`) were added, removed, " + "or modified in `docs/specification/overview.md` (escalated from " + "'maintainers' (Level 2) to Level 3 Technical Council)." + ) + req = RuleRequirement( + min_approvals=2, min_team=min_team, escalation_reason=reason + ) + status = RequirementStatus( + requirement=req, + approved_count=0, + assigned_count=0, + is_satisfied=False, + ) + file_status = FileValidationStatus( + file_path="docs/specification/overview.md", + requirement_statuses=[status], + is_satisfied=False, + escalation_reason=reason, + ) + res = ValidationResult( + is_mergeable=False, + error=ValidationErrorReason.INSUFFICIENT_APPROVALS, + requirement_statuses=[status], + file_statuses=[file_status], + ) + logger = ValidationLogger(res) + logger.write_summary("dummy.txt") + + handle = mock_open() + written_content = "".join(call.args[0] for call in handle.write.call_args_list) + + expected = ( + "# ⚖️ Governance Gate Validation Report\n" + "\n" + "## 🔴 FAILURE: Insufficient approvals.\n" + "---\n" + "\n" + "### 🌍 1. GLOBAL PR SUMMARY\n" + "\n" + "Merged requirements across all changed files:\n" + "\n" + "* **2 approvals** from any Technical Council (Level 3) or higher in the UCP governance hierarchy\n" + f" * **Reason:** ⚠️ {reason}\n" + " * **Met:** 🔴 No (0/2 approved)\n" + " * **Pending:** Needs 2 approvals from any Technical Council (Level 3) or higher in the UCP governance hierarchy " + "(0 eligible reviewers assigned). Waiting for 2 more reviewer(s) to be assigned from " + "any Technical Council (Level 3) or higher in the UCP governance hierarchy.\n" + "\n" + "\n" + "---\n" + "\n" + "### 📂 2. FILE-BY-FILE BREAKDOWN\n" + "\n" + "* **File:** `docs/specification/overview.md`\n" + " * **Status:** 🔴 UNSATISFIED\n" + f" * **Escalation:** ⚠️ {reason}\n" + " * **Requirements:**\n" + " * **2 approvals** from any Technical Council (Level 3) or higher in the UCP governance hierarchy\n" + f" * **Reason:** ⚠️ {reason}\n" + " * **Met:** 🔴 No (0/2 approved)\n" + " * **Pending:** Needs 2 approvals from any Technical Council (Level 3) or higher in the UCP governance hierarchy " + "(0 eligible reviewers assigned). Waiting for 2 more reviewer(s) to be assigned from " + "any Technical Council (Level 3) or higher in the UCP governance hierarchy.\n" + ) + HELP_SUFFIX + self.assertEqual(expected, written_content) + if __name__ == "__main__": unittest.main()