From 052a9b7088638246c64b55b51d0ccb6f9f5f251a Mon Sep 17 00:00:00 2001 From: Raj Madhivanan Date: Thu, 17 Sep 2026 14:08:03 -0700 Subject: [PATCH 1/2] feat: report modification-control findings Why: The modification procedure defines evidence and authorization gates, but reviewers need a consistent mechanical check for missing records, stale approvals, and incomplete verification evidence. Introducing the checker in warn mode makes those gaps visible without unexpectedly blocking active work. What changed: - Added E1-E7 checks for authorization, impact analysis, requirement references, classification, approvals, verification evidence, and emergencies. - Added fail-closed GitHub data collection, pagination, response validation, and deterministic report-comment updates. - Added a constrained parser that validates the repository Change Request form. - Added a fork-safe pull-request workflow and an auditable mode file set to warn. - Added 78 focused policy, API, parser, workflow, and CLI tests. Impact: Policy findings are advisory in warn mode and do not fail pull requests. Collection or configuration errors still return an unable-to-run failure so missing evidence cannot be mistaken for a clean result. Runtime code, wire behavior, safety requirements, and coverage calculations are unchanged. Safety lifecycle: Modification authorization, impact analysis, verification evidence, and configuration management. Bears on IEC 61508-1:2010 sections 6.2 and 7.16, and IEC 61508-3:2010 sections 7.4, 7.8, and 7.9. Co-Authored-By: OpenCode --- .github/workflows/change-control.yml | 36 + docs/process/enforcement-mode | 1 + tools/change_control/__main__.py | 134 ++++ tools/change_control/checks.py | 385 ++++++++++ tools/change_control/fixtures/fake_gh.py | 46 ++ tools/change_control/issue_form.py | 140 ++++ tools/change_control/test_change_control.py | 801 ++++++++++++++++++++ 7 files changed, 1543 insertions(+) create mode 100644 .github/workflows/change-control.yml create mode 100644 docs/process/enforcement-mode create mode 100644 tools/change_control/__main__.py create mode 100644 tools/change_control/checks.py create mode 100755 tools/change_control/fixtures/fake_gh.py create mode 100644 tools/change_control/issue_form.py create mode 100644 tools/change_control/test_change_control.py diff --git a/.github/workflows/change-control.yml b/.github/workflows/change-control.yml new file mode 100644 index 00000000..f4c74998 --- /dev/null +++ b/.github/workflows/change-control.yml @@ -0,0 +1,36 @@ +--- +name: Change control + +on: + pull_request: + types: [opened, synchronize, reopened, labeled, unlabeled, edited] + pull_request_review: + types: [submitted, dismissed] + +concurrency: + group: change-control-${{ github.event.pull_request.number }} + cancel-in-progress: true + +permissions: + contents: read + issues: write + pull-requests: write + checks: read + actions: read + +jobs: + change-control: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - name: Run tests + run: python3 tools/change_control/self_test.py + - name: Check modification records + env: + GH_TOKEN: ${{ github.token }} + PR_NUMBER: ${{ github.event.pull_request.number }} + CAN_COMMENT: ${{ github.event.pull_request.head.repo.full_name == github.repository }} + run: |- + args=() + [ "$CAN_COMMENT" = 'true' ] || args+=(--no-comment) + python3 -m tools.change_control --repository '${{ github.repository }}' --pr "$PR_NUMBER" "${args[@]}" diff --git a/docs/process/enforcement-mode b/docs/process/enforcement-mode new file mode 100644 index 00000000..1ef71804 --- /dev/null +++ b/docs/process/enforcement-mode @@ -0,0 +1 @@ +warn diff --git a/tools/change_control/__main__.py b/tools/change_control/__main__.py new file mode 100644 index 00000000..e85f201c --- /dev/null +++ b/tools/change_control/__main__.py @@ -0,0 +1,134 @@ +# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Run change-control checks against pull-request state obtained through gh api.""" + +import argparse +import json +import shlex +import subprocess +import sys +from pathlib import Path + +from tools.safety_lint.model import LintError + +from .checks import cr_number, evaluate, load_mode, render_report, upsert_comment + + +class GhApi: + """Small fail-closed subprocess adapter around gh api.""" + + def __init__(self, command, repository): + self.command = shlex.split(command) + self.repository = repository + + def __call__(self, method, path, body=None, paginate=False, collection_key=None): + endpoint = path.replace('{repo}', self.repository) + command = [*self.command, 'api', endpoint] + if paginate: + command.extend(['--paginate', '--slurp']) + if method != 'GET': + command.extend(['--method', method]) + if body: + for key, value in body.items(): + command.extend(['--field', f'{key}={value}']) + result = subprocess.run(command, check=False, capture_output=True, text=True) + if result.returncode: + detail = result.stderr.strip() or 'no error detail' + raise RuntimeError(f'gh api failed for {endpoint}: {detail}') + try: + response = json.loads(result.stdout or '{}') + except json.JSONDecodeError as error: + raise RuntimeError(f'gh api returned invalid JSON for {endpoint}') from error + if not paginate: + return response + if not isinstance(response, list): + raise RuntimeError(f'paginated gh api response is not a page list for {endpoint}') + if collection_key: + merged = [] + for page in response: + if not isinstance(page, dict) or not isinstance(page.get(collection_key), list): + raise RuntimeError(f'paginated gh api response lacks {collection_key} for {endpoint}') + merged.extend(page[collection_key]) + return {collection_key: merged} + if not all(isinstance(page, list) for page in response): + raise RuntimeError(f'paginated gh api response contains a non-list page for {endpoint}') + return [item for page in response for item in page] + + +def _require(mapping, path): + value = mapping + for key in path: + if not isinstance(value, dict) or key not in value: + raise RuntimeError(f'partial GitHub response missing {".".join(path)}') + value = value[key] + return value + + +def collect(api, repository, pr_number): + """Collect the complete GitHub snapshot used by pure policy evaluation.""" + prefix = f'repos/{repository}' + pr = api('GET', f'{prefix}/pulls/{pr_number}') + _require(pr, ('user', 'login')) + head = _require(pr, ('head', 'sha')) + if 'body' not in pr or 'labels' not in pr: + raise RuntimeError('partial GitHub response missing PR body or labels') + cr = cr_number(pr['body'], repository) + issue = api('GET', f'{prefix}/issues/{cr}') if cr else {'labels': [], 'body': ''} + comments = api('GET', f'{prefix}/issues/{cr}/comments', paginate=True) if cr else [] + reviews = api('GET', f'{prefix}/pulls/{pr_number}/reviews', paginate=True) + files = api('GET', f'{prefix}/pulls/{pr_number}/files', paginate=True) + checks = api('GET', f'{prefix}/commits/{head}/check-runs', paginate=True, collection_key='check_runs') + workflows = api( + 'GET', + f'{prefix}/actions/runs?head_sha={head}', + paginate=True, + collection_key='workflow_runs', + ) + pr_comments = api('GET', f'{prefix}/issues/{pr_number}/comments', paginate=True) + for name, value in (('files', files), ('comments', comments), ('reviews', reviews), ('PR comments', pr_comments)): + if not isinstance(value, list): + raise RuntimeError(f'partial GitHub response: {name} is not a list') + return { + 'repository': repository, + 'pr': pr, + 'issue': issue, + 'issue_comments': comments, + 'reviews': reviews, + 'files': files, + 'check_runs': _require(checks, ('check_runs',)), + 'workflow_runs': _require(workflows, ('workflow_runs',)), + 'pr_comments': pr_comments, + } + + +def main(argv=None): + parser = argparse.ArgumentParser() + parser.add_argument('--root', default='.') + parser.add_argument('--repository', required=True) + parser.add_argument('--pr', required=True, type=int) + parser.add_argument('--gh', default='gh') + parser.add_argument('--no-comment', action='store_true') + args = parser.parse_args(argv) + try: + mode = load_mode(args.root) + api = GhApi(args.gh, args.repository) + data = collect(api, args.repository, args.pr) + results = evaluate(Path(args.root), data) + report = render_report(mode, results) + print(report) + findings = any(item.status == 'fail' for item in results) + if not args.no_comment: + try: + upsert_comment(api, args.pr, report, data['pr_comments']) + except RuntimeError as error: + print(f'change-control: comment publication warning: {error}', file=sys.stderr) + if mode == 'enforce': + return 2 + return 1 if findings and mode == 'enforce' else 0 + except (OSError, RuntimeError, ValueError, KeyError, LintError) as error: + print(f'change-control: cannot run: {error}', file=sys.stderr) + return 2 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/tools/change_control/checks.py b/tools/change_control/checks.py new file mode 100644 index 00000000..82cc033e --- /dev/null +++ b/tools/change_control/checks.py @@ -0,0 +1,385 @@ +# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Evaluate existence and ordering of modification records without judging adequacy.""" + +import re +from dataclasses import dataclass +from pathlib import Path + +from tools.safety_lint.parse_srs import parse_srs + +from .issue_form import parse_issue_form + +AUTHORIZERS = ('iliabaranov', 'rajasimman-madhivanan', 'davidt315') +CLASS_RANK = {'A': 1, 'B': 2, 'C': 3} +COMMENT_MARKER = '' + + +@dataclass(frozen=True) +class CheckResult: + """One caller-visible policy result.""" + + check_id: str + status: str + message: str + + +def load_mode(root): + """Read the auditable mode switch and reject every value except warn or enforce.""" + path = Path(root) / 'docs/process/enforcement-mode' + try: + value = path.read_text(encoding='utf-8') + except OSError as error: + raise RuntimeError(f'cannot read enforcement mode: {error}') from error + if value not in ('warn\n', 'enforce\n'): + raise RuntimeError('enforcement-mode must contain exactly warn or enforce followed by a newline') + return value.strip() + + +def _headings(path): + return [ + line.strip() for line in Path(path).read_text(encoding='utf-8').splitlines() if re.match(r'^#(?:\s|\d)', line) + ] + + +def impact_analysis_complete(text, headings): + """Check that headings occur in order and each has non-whitespace content beneath it.""" + positions = [] + cursor = 0 + for heading in headings: + match = re.search(rf'(?m)^{re.escape(heading)}\s*$', text[cursor:]) + if not match: + return False, [heading] + start = cursor + match.start() + end = cursor + match.end() + positions.append((heading, start, end)) + cursor = end + missing = [] + for index, (heading, _, end) in enumerate(positions): + next_start = positions[index + 1][1] if index + 1 < len(positions) else len(text) + if not text[end:next_start].strip(): + missing.append(heading) + return not missing, missing + + +def minimum_class(paths, wire_changed=False): + """Return the mechanical classification floor for changed repository paths.""" + floor = 'C' if wire_changed else 'A' + for path in paths: + if ( + path.startswith('pstop_c/') + or path in ('firmware/main/main.c', 'machn/main/main.c') + or path.startswith('docs/safety/') + ): + candidate = 'C' + elif path.startswith(('components/', 'common/')) or path in ( + 'firmware/sdkconfig.defaults', + 'machn/sdkconfig.defaults', + ): + candidate = 'B' + elif path.startswith('.github/workflows/') or path.startswith('scripts/'): + candidate = 'B' + else: + candidate = 'A' + if CLASS_RANK[candidate] > CLASS_RANK[floor]: + floor = candidate + return floor + + +def _labels(entity): + return {label['name'] if isinstance(label, dict) else label for label in entity.get('labels', [])} + + +def cr_number(body, repository): + """Return one local CR number, rejecting foreign or ambiguous candidates.""" + if repository.count('/') != 1: + return None + current_owner, current_name = repository.casefold().split('/') + numbers = set() + foreign = False + for line in (body or '').splitlines(): + local = re.fullmatch(r'\s*(?:closes|refs)\s+#(\d+)\s*', line, re.IGNORECASE) + if local: + numbers.add(int(local.group(1))) + continue + url = re.fullmatch( + r'\s*(?:(?:closes|refs)\s+)?https://github\.com/([^/\s]+)/([^/\s]+)/issues/(\d+)/?\s*', + line, + re.IGNORECASE, + ) + if url: + owner, name, number = url.groups() + if (owner.casefold(), name.casefold()) == (current_owner, current_name): + numbers.add(int(number)) + else: + foreign = True + return next(iter(numbers)) if len(numbers) == 1 and not foreign else None + + +def _issue_fields_complete(root, body): + form = parse_issue_form(Path(root) / '.github/ISSUE_TEMPLATE/change-request.yml') + missing = [] + for field in form['body']: + if not field['required']: + continue + label = field['label'] + match = re.search(rf'(?ms)^###\s+{re.escape(label)}\s*$\n(.*?)(?=^###\s|\Z)', body or '') + if not match or not match.group(1).strip() or match.group(1).strip() == '_No response_': + missing.append(field['id']) + return missing + + +def _find_ia(comments, headings): + for comment in comments: + body = comment.get('body', '') + if headings and headings[0] in body: + return body + return '' + + +def _find_short_ia(comments): + for comment in comments: + body = comment.get('body', '') + if all(re.search(phrase, body, re.IGNORECASE) for phrase in ('what changed', 'what it could affect', 'tests')): + return body + return '' + + +def _cited_sr_tokens(text): + return set(re.findall(r'\bSR-[A-Z]+-[0-9A-Za-z]+(?:-[0-9A-Za-z]+)*\b', text)) + + +def _is_authorization(comment, author=''): + login = comment.get('user', {}).get('login', '').lower() + return ( + login in AUTHORIZERS + and login != author.lower() + and bool(re.search(r'(?im)^\s*(?:decision:\s*)?authori[sz]ed\b', comment.get('body', ''))) + ) + + +def _named_tests(text): + sections = re.findall(r'(?ms)^# 7\. Verification plan for this change\s*$\n(.*?)(?=^# 8\.|\Z)', text) + names = [] + for section in sections: + names.extend(re.findall(r'`([^`]+)`', section)) + for line in section.splitlines(): + if line.lstrip().startswith(('-', '+')) and ':' in line: + value = line.split(':', 1)[1].strip() + if value and value.lower() not in ('none', 'n/a'): + names.append(value) + if line.strip().startswith('|'): + cells = [cell.strip() for cell in line.strip().strip('|').split('|')] + if len(cells) >= 2 and cells[0] not in ('Purpose', '---'): + value = cells[1] + if value and value not in ('---', 'None', 'N/A'): + names.extend(part.strip() for part in re.split(r'|,', value) if part.strip()) + if not sections: + match = re.search(r'(?im)^\s*(?:which\s+)?tests(?:\s+will\s+be\s+run)?\s*:\s*(.+)$', text) + if match: + names.extend(part.strip(' `') for part in match.group(1).split(',') if part.strip(' `')) + return list(dict.fromkeys(name.strip() for name in names if name.strip())) + + +def _evidence_matches(name, evidence_name): + """Match an IA entry to one complete check or workflow name, never a substring.""" + planned = ' '.join(name.casefold().split()) + observed = ' '.join(evidence_name.casefold().split()) + return bool(planned and observed and planned == observed) + + +def _approvers(data, require_head=True): + head = data['pr']['head']['sha'] + author = data['pr']['user']['login'].lower() + latest = {} + for review in data.get('reviews', []): + login = review.get('user', {}).get('login', '').lower() + if login: + latest[login] = review + return { + login + for login, review in latest.items() + if login in AUTHORIZERS + and login != author + and review.get('state') == 'APPROVED' + and (not require_head or review.get('commit_id') == head) + } + + +def evaluate(root, data): + """Evaluate E1-E7 against a complete, synthetic-or-live GitHub state snapshot.""" + root = Path(root) + pr = data['pr'] + labels = _labels(pr) + linked_cr = cr_number(pr.get('body', ''), data['repository']) + issue = data.get('issue', {}) + issue_labels = _labels(issue) + comments = data.get('issue_comments', []) + headings = _headings(root / 'docs/process/templates/IMPACT_ANALYSIS.md') + ia = _find_ia(comments, headings) + short_ia = _find_short_ia(comments) + emergency = 'emergency' in labels + if emergency and not ia: + ia = short_ia + results = [] + + if 'needs-change-request' in labels: + results.append(CheckResult('E1', 'pending', 'maintainer Change Request required before review begins')) + else: + author = pr['user']['login'] + required_authorizers = 2 if 'class-c' in labels or emergency else 1 + authorization_comments = {} + for comment in comments: + if _is_authorization(comment, author): + authorization_comments[comment['user']['login'].lower()] = comment + authorized_comment = len(authorization_comments) >= required_authorizers + authorization_times = [ + comment.get('created_at') for comment in authorization_comments.values() if comment.get('created_at') + ] + ia_times = [ + comment.get('created_at') + for comment in comments + if comment.get('body', '') == ia and comment.get('created_at') + ] + before_implementation = not pr.get('created_at') or ( + len(authorization_times) >= required_authorizers and max(authorization_times) <= pr['created_at'] + ) + after_analysis = not ia_times or ( + len(authorization_times) >= required_authorizers and max(ia_times) <= min(authorization_times) + ) + ordered = before_implementation and after_analysis + missing_fields = _issue_fields_complete(root, issue.get('body', '')) if linked_cr else ['change-request-link'] + okay = ( + linked_cr is not None + and 'change-request' in issue_labels + and 'status:authorized' in issue_labels + and authorized_comment + and ordered + and not missing_fields + ) + detail = ( + f'authorized Change Request has {required_authorizers} distinct pre-implementation authorizer(s)' + if okay + else f'Change Request missing, ambiguous, incomplete, or lacks {required_authorizers} distinct pre-implementation authorizer(s) ({", ".join(missing_fields)})' + ) + results.append(CheckResult('E1', 'pass' if okay else 'fail', detail)) + + if emergency and ia == short_ia and short_ia: + complete, empty = True, [] + else: + complete, empty = impact_analysis_complete(ia, headings) if ia else (False, ['Impact Analysis']) + results.append( + CheckResult( + 'E2', + 'pass' if complete else 'fail', + 'all IA sections exist and are nonblank; content truth and adequacy are not assessed' + if complete + else f'IA sections missing or blank: {", ".join(empty)}; content truth and adequacy are not assessed', + ) + ) + + canonical = {requirement.sr_id for requirement in parse_srs(root / 'docs/safety/SAFETY_REQUIREMENTS.md')} + cited = _cited_sr_tokens(ia) + invalid = sorted(cited - canonical) + results.append( + CheckResult( + 'E3', + 'fail' if invalid else 'pass', + f'invalid requirement IDs: {", ".join(invalid)}' if invalid else 'all cited requirement IDs exist', + ) + ) + + class_labels = sorted(label for label in labels if re.fullmatch(r'class-[abc]', label)) + paths = [item['filename'] for item in data.get('files', [])] + wire_changed = any(path.startswith('pstop_c/pstop/include/pstop/') for path in paths) + floor = minimum_class(paths, wire_changed) + if len(class_labels) != 1: + results.append(CheckResult('E4', 'fail', 'exactly one class-a, class-b, or class-c label is required')) + selected = None + else: + selected = class_labels[0][-1].upper() + under = CLASS_RANK[selected] < CLASS_RANK[floor] + results.append( + CheckResult( + 'E4', + 'fail' if under else 'pass', + f'Class {selected}; mechanical floor Class {floor}; checks existence/order, not classification adequacy', + ) + ) + + approvers = _approvers(data) + if selected == 'C': + status = 'pass' if len(approvers) >= 2 else 'fail' + results.append( + CheckResult('E5', status, f'current distinct non-author approving authorizers: {len(approvers)}/2') + ) + else: + results.append(CheckResult('E5', 'not-applicable', 'two-review requirement applies to Class C')) + + names = _named_tests(ia) + head = pr['head']['sha'] + evidence = { + item.get('name', '') + for item in data.get('check_runs', []) + if item.get('head_sha') == head and item.get('conclusion') == 'success' + } + evidence.update( + item.get('name', item.get('path', '')) + for item in data.get('workflow_runs', []) + if item.get('head_sha') == head and item.get('conclusion') == 'success' + ) + missing_tests = [name for name in names if not any(_evidence_matches(name, item) for item in evidence)] + explanation = 'check-run/workflow evidence cannot prove commands or tests inside a job executed' + if not names: + e6_status = 'fail' + e6_message = f'IA verification plan names no specific tests; {explanation}' + elif missing_tests: + e6_status = 'fail' + e6_message = f'missing head-SHA evidence: {", ".join(missing_tests)}; {explanation}' + else: + e6_status = 'pass' + e6_message = f'all named evidence matched; {explanation}' + results.append( + CheckResult( + 'E6', + e6_status, + e6_message, + ) + ) + + if emergency: + short_form = bool(ia and names and short_ia) + status = 'pass' if len(approvers) >= 2 and short_form else 'fail' + results.append( + CheckResult( + 'E7', + status, + 'emergency path requires two approvals and short-form IA; retrospective due within five working days of release', + ) + ) + else: + results.append(CheckResult('E7', 'not-applicable', 'PR is not labelled emergency')) + return results + + +def render_report(mode, results): + """Render one deterministic PR comment with the active mode visible.""" + lines = [f'mode: {mode}', '', '| Check | Result | Explanation |', '|---|---|---|'] + for item in results: + message = item.message.replace('|', '\\|') + lines.append(f'| {item.check_id} | {item.status} | {message} |') + lines.extend([ + '', + 'These checks verify artifact existence and ordering only, not truth, adequacy, or safety sufficiency.', + ]) + return '\n'.join(lines) + + +def upsert_comment(api, pr_number, report, comments): + """Create or update at most one marker-owned report comment.""" + body = f'{COMMENT_MARKER}\n{report}' + existing = next((comment for comment in comments if COMMENT_MARKER in comment.get('body', '')), None) + if existing: + api('PATCH', f'repos/{{repo}}/issues/comments/{existing["id"]}', {'body': body}) + else: + api('POST', f'repos/{{repo}}/issues/{pr_number}/comments', {'body': body}) diff --git a/tools/change_control/fixtures/fake_gh.py b/tools/change_control/fixtures/fake_gh.py new file mode 100755 index 00000000..990c599d --- /dev/null +++ b/tools/change_control/fixtures/fake_gh.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Deterministic gh replacement used only by change-control tests.""" + +import json +import os +import sys +from pathlib import Path + + +def main(): + """Serve API responses from the JSON file named by FAKE_GH_DATA.""" + data = json.loads(Path(os.environ['FAKE_GH_DATA']).read_text(encoding='utf-8')) + if data.get('exit_code'): + print(data.get('stderr', 'fake gh failure'), file=sys.stderr) + return int(data['exit_code']) + args = sys.argv[1:] + if not args or args[0] != 'api': + return 2 + endpoint = args[1] if len(args) > 1 else '' + method = 'GET' + if '--method' in args: + method = args[args.index('--method') + 1] + key = f'{method} {endpoint}' + calls = os.environ.get('FAKE_GH_CALLS') + if calls: + with Path(calls).open('a', encoding='utf-8') as stream: + stream.write(key + '\n') + response = data.get('responses', {}).get(key) + if response is None: + print(f'unconfigured fake gh request: {key}', file=sys.stderr) + return 1 + if isinstance(response, dict) and '__pages__' in response: + response = response['__pages__'] + elif '--slurp' in args: + response = [response] + if isinstance(response, str): + print(response) + else: + print(json.dumps(response)) + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/tools/change_control/issue_form.py b/tools/change_control/issue_form.py new file mode 100644 index 00000000..362a5261 --- /dev/null +++ b/tools/change_control/issue_form.py @@ -0,0 +1,140 @@ +# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Parse and validate the constrained YAML used by the Change Request issue form.""" + +import re +from pathlib import Path + + +def _scalar(value): + value = value.strip() + if not value: + return '' + if value in ('true', 'false'): + return value == 'true' + if value.isdigit(): + return int(value) + if value.startswith('[') and value.endswith(']'): + return [part.strip().strip('\'"') for part in value[1:-1].split(',') if part.strip()] + return value.strip('\'"') + + +def parse_issue_form(path): + """Return issue-form controls from the repository's deliberately limited YAML subset.""" + lines = Path(path).read_text(encoding='utf-8').splitlines() + if any('\t' in line for line in lines): + raise ValueError('tabs are not valid indentation') + document = {'body': []} + item = None + section = None + options = False + active_attribute_key = None + active_attribute_indent = None + in_body = False + for number, raw in enumerate(lines, 1): + if not raw.strip() or raw.lstrip().startswith('#') or raw.strip() == '---': + continue + indent = len(raw) - len(raw.lstrip(' ')) + text = raw.strip() + if indent == 0: + active_attribute_key = None + active_attribute_indent = None + match = re.fullmatch(r'([a-z_]+):(?:\s*(.*))?', text) + if not match: + raise ValueError(f'{path}:{number}: unsupported top-level YAML') + key, value = match.groups() + if key == 'body': + if value: + raise ValueError(f'{path}:{number}: body must be a sequence') + in_body = True + else: + if in_body: + raise ValueError(f'{path}:{number}: top-level key after body') + document[key] = _scalar(value or '') + continue + if not in_body: + raise ValueError(f'{path}:{number}: unexpected indentation') + if indent == 2 and text.startswith('- type: '): + item = {'type': _scalar(text[8:]), 'required': False, 'options': []} + document['body'].append(item) + section = None + options = False + active_attribute_key = None + active_attribute_indent = None + continue + if item is None: + raise ValueError(f'{path}:{number}: body entry must begin with type') + if indent == 4 and re.fullmatch(r'(id|attributes|validations):.*', text): + key, value = text.split(':', 1) + if key == 'id': + item['id'] = _scalar(value) + section = None + else: + if value.strip(): + raise ValueError(f'{path}:{number}: {key} must be a mapping') + section = key + options = False + active_attribute_key = None + active_attribute_indent = None + continue + if indent == 6 and section in ('attributes', 'validations'): + if ':' not in text: + raise ValueError(f'{path}:{number}: expected mapping value') + key, value = text.split(':', 1) + if section == 'validations' and key == 'required': + item['required'] = _scalar(value) + active_attribute_key = None + active_attribute_indent = None + elif section == 'attributes' and key == 'options': + if value.strip(): + raise ValueError(f'{path}:{number}: options must be a sequence') + options = True + active_attribute_key = None + active_attribute_indent = None + elif section == 'attributes': + item[key] = _scalar(value) + options = False + active_attribute_key = key + active_attribute_indent = indent + else: + raise ValueError(f'{path}:{number}: unsupported validation') + continue + if indent == 8 and options and text.startswith('- '): + item['options'].append(_scalar(text[2:])) + continue + if ( + section == 'attributes' + and not options + and active_attribute_key is not None + and indent > active_attribute_indent + ): + item[active_attribute_key] = f'{item[active_attribute_key]} {_scalar(text)}'.strip() + continue + raise ValueError(f'{path}:{number}: unsupported indentation or YAML construct') + validate_issue_form(document) + return document + + +def validate_issue_form(document): + """Reject forms that GitHub could silently replace with a blank issue page.""" + for key in ('name', 'description', 'title', 'labels'): + if not document.get(key): + raise ValueError(f'issue form missing top-level {key}') + body = document.get('body') + if not isinstance(body, list) or not body: + raise ValueError('issue form body must contain controls') + ids = [] + for field in body: + if field.get('type') not in ('input', 'textarea', 'dropdown'): + raise ValueError(f'unsupported issue form field type: {field.get("type")}') + if not field.get('id') or not field.get('label'): + raise ValueError('every issue form field needs id and label') + ids.append(field['id']) + if field['type'] == 'dropdown': + if not field.get('options') or not isinstance(field.get('default'), int): + raise ValueError('dropdown needs options and integer default') + if not 0 <= field['default'] < len(field['options']): + raise ValueError('dropdown default is outside options') + if len(ids) != len(set(ids)): + raise ValueError('duplicate issue form field id') + return document diff --git a/tools/change_control/test_change_control.py b/tools/change_control/test_change_control.py new file mode 100644 index 00000000..a517bb52 --- /dev/null +++ b/tools/change_control/test_change_control.py @@ -0,0 +1,801 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Spec-driven tests for the focused warn-mode change-control checker.""" + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) + +from tools.change_control.__main__ import GhApi # noqa: E402 +from tools.change_control.checks import ( # noqa: E402 + AUTHORIZERS, + evaluate, + impact_analysis_complete, + minimum_class, + upsert_comment, +) +from tools.change_control.issue_form import parse_issue_form, validate_issue_form # noqa: E402 + +FORM = ROOT / '.github/ISSUE_TEMPLATE/change-request.yml' + + +def snapshot(**overrides): + """Return a complete synthetic GitHub state shaped like the real API snapshot.""" + ia = (ROOT / 'docs/process/templates/IMPACT_ANALYSIS.md').read_text(encoding='utf-8') + filled = ia.replace('Modules changed:', 'Modules changed: tools/change_control').replace( + '| | |', '| None | None |' + ) + data = { + 'repository': 'polymathrobotics/protective-stop', + 'pr': { + 'user': {'login': 'contributor'}, + 'body': 'Closes #17', + 'labels': [{'name': 'class-b'}], + 'head': {'sha': 'abc123'}, + }, + 'files': [{'filename': 'tools/change_control/checks.py'}], + 'issue': { + 'labels': [{'name': 'change-request'}, {'name': 'status:authorized'}], + 'body': '### Reason for the change\nNeeded\n### Hazards that may be affected\nNone identified, because tooling only\n### Description of the proposed change\nTooling\n### Baseline affected\nmain\n### Proposed class\nB', + }, + 'issue_comments': [ + {'user': {'login': AUTHORIZERS[0]}, 'body': 'Authorized: proceed.'}, + {'user': {'login': 'analyst'}, 'body': filled}, + ], + 'reviews': [], + 'check_runs': [{'name': 'change-control', 'conclusion': 'success', 'head_sha': 'abc123'}], + 'workflow_runs': [], + } + data.update(overrides) + return data + + +class IssueFormTests(unittest.TestCase): + def _parse_field(self, attributes): + form = ( + 'name: Test\n' + 'description: Test form\n' + "title: '[Test] '\n" + 'labels: [test]\n' + 'body:\n' + ' - type: textarea\n' + ' id: field\n' + ' attributes:\n' + f'{attributes}' + ' validations:\n' + ' required: true\n' + ) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'form.yml' + path.write_text(form, encoding='utf-8') + return parse_issue_form(path)['body'][0] + + def test_issue_form_is_valid_yaml_and_parses(self): + """The checked-in issue form must parse as the deliberately supported YAML subset.""" + parsed = parse_issue_form(FORM) + self.assertEqual(parsed['name'], 'Change Request') + self.assertGreater(len(parsed['body']), 5) + + def test_required_fields_present(self): + """All five creation-time fields mandated by the procedure must be required.""" + fields = {field['id']: field for field in parse_issue_form(FORM)['body']} + self.assertTrue( + all(fields[name]['required'] for name in ('reason', 'hazards', 'description', 'baseline', 'class')) + ) + + def test_class_dropdown_defaults_to_c(self): + """An unclassified request must conservatively default to Class C.""" + fields = {field['id']: field for field in parse_issue_form(FORM)['body']} + self.assertEqual(fields['class']['options'][fields['class']['default']], 'C') + + def test_every_source_field_is_represented(self): + """The issue form must carry every Change Request source section without a duplicate field list.""" + ids = {field['id'] for field in parse_issue_form(FORM)['body']} + expected = { + 'reason', + 'hazards', + 'description', + 'baseline', + 'requester', + 'impact-analysis', + 'class', + 'authorization', + 'implementation', + 'gate-0', + 'gate-1', + 'review', + 'deviations', + 'release', + 'status', + } + self.assertEqual(ids, expected) + + def test_source_field_details_survive_issue_form_conversion(self): + """YAML conversion must retain source details needed to complete implementation and Gate 1 records.""" + fields = {field['id']: field for field in parse_issue_form(FORM)['body']} + self.assertIn('Yes / No, with link', fields['implementation']['description']) + self.assertIn('Run by', fields['gate-1']['description']) + self.assertIn('Forward -', fields['gate-1']['description']) + self.assertIn('Backward -', fields['gate-1']['description']) + + def test_plain_label_continuation_is_folded(self): + """A continued attributes label must be joined to its first line with one space.""" + field = self._parse_field(' label: First label line\n second label line\n') + self.assertEqual(field['label'], 'First label line second label line') + + def test_plain_description_continuation_is_folded(self): + """A continued attributes description must be joined to its first line with one space.""" + field = self._parse_field( + ' label: Field\n description: First description line\n second description line\n' + ) + self.assertEqual(field['description'], 'First description line second description line') + + def test_plain_placeholder_continuation_is_folded(self): + """A continued attributes placeholder must be joined to its first line with one space.""" + field = self._parse_field( + ' label: Field\n placeholder: First placeholder line\n second placeholder line\n' + ) + self.assertEqual(field['placeholder'], 'First placeholder line second placeholder line') + + def test_real_gate_one_description_is_complete(self): + """The checked-in folded Gate 1 description must retain its final continuation text.""" + fields = {field['id']: field for field in parse_issue_form(FORM)['body']} + self.assertTrue(fields['gate-1']['description'].endswith('requirements covered.')) + + def test_orphan_attribute_continuation_is_rejected(self): + """Indented text without an active scalar key must fail rather than disappear.""" + with self.assertRaises(ValueError): + self._parse_field(' orphan continuation\n label: Field\n') + + def test_option_continuation_is_rejected(self): + """Deeper text below an option must never be folded into an attributes scalar.""" + attributes = ' label: Field\n options:\n - First\n unsupported option continuation\n' + with self.assertRaises(ValueError): + self._parse_field(attributes) + + def test_validation_continuation_is_rejected(self): + """Deeper validation text must never be appended to the preceding attributes scalar.""" + attributes = ( + ' label: Field\n validations:\n required: true\n unsupported validation continuation\n' + ) + with self.assertRaises(ValueError): + self._parse_field(attributes) + + def test_malformed_issue_form_is_rejected(self): + """Malformed indentation must fail instead of silently degrading to a blank issue.""" + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'form.yml' + path.write_text('name: Bad\nbody:\n - type: input\n id: broken\n', encoding='utf-8') + with self.assertRaises(ValueError): + parse_issue_form(path) + + def test_blank_issue_fallback_risk_is_rejected(self): + """A form lacking required top-level metadata or body controls must be invalid.""" + with self.assertRaises(ValueError): + validate_issue_form({'name': 'Change Request', 'body': []}) + + +class RepositoryPolicyTests(unittest.TestCase): + def test_issue_chooser_keeps_blank_issues_and_links_security_policy(self): + """Public reports must remain available while safety defects are directed to the private policy.""" + config = (ROOT / '.github/ISSUE_TEMPLATE/config.yml').read_text(encoding='utf-8') + self.assertIn('blank_issues_enabled: true', config) + self.assertIn('https://github.com/polymathrobotics/protective-stop/security/policy', config) + + def test_labels_json_has_no_duplicate_names(self): + """The reproducible label definition must contain unique names.""" + labels = json.loads((ROOT / 'tools/change_control/labels.json').read_text(encoding='utf-8')) + names = [label['name'] for label in labels] + self.assertEqual(len(names), len(set(names))) + + def test_labels_json_contains_the_complete_settled_label_set(self): + """The reproducible data must contain every label settled by the modification procedure plan.""" + labels = json.loads((ROOT / 'tools/change_control/labels.json').read_text(encoding='utf-8')) + names = {label['name'] for label in labels} + self.assertEqual( + names, + { + 'change-request', + 'class-a', + 'class-b', + 'class-c', + 'emergency', + 'safety-defect', + 'wire-break', + 'needs-change-request', + 'status:proposed', + 'status:under-analysis', + 'status:authorized', + 'status:rejected', + 'status:in-implementation', + 'status:in-verification', + 'status:merged', + 'status:released', + }, + ) + + def test_label_sync_dry_run_is_deterministic_and_network_free(self): + """Two dry runs must produce identical plans without invoking GitHub.""" + environment = os.environ.copy() + environment['PATH'] = '/usr/bin:/bin' + first = subprocess.run( + ['scripts/sync_labels.sh', '--dry-run'], + cwd=ROOT, + env=environment, + check=False, + capture_output=True, + text=True, + ) + second = subprocess.run( + ['scripts/sync_labels.sh', '--dry-run'], + cwd=ROOT, + env=environment, + check=False, + capture_output=True, + text=True, + ) + self.assertEqual((first.returncode, second.returncode), (0, 0)) + self.assertEqual(first.stdout, second.stdout) + self.assertEqual(first.stdout.count('would sync label:'), 16) + + def test_label_sync_targets_the_intended_repository_explicitly(self): + """Label writes must not depend on ambiguous git-remote repository inference.""" + script = (ROOT / 'scripts/sync_labels.sh').read_text(encoding='utf-8') + self.assertIn('--repo "$REPOSITORY"', script) + + def test_every_label_referenced_in_the_procedure_exists_in_labels_json(self): + """Every backticked process label must exist in the reproducible label set.""" + import re + + procedure = (ROOT / 'docs/process/MODIFICATION_PROCEDURE.md').read_text(encoding='utf-8') + referenced = set( + re.findall( + r'`((?:class-[abc]|change-request|emergency|safety-defect|wire-break|needs-change-request|status:[a-z-]+))`', + procedure, + ) + ) + labels = { + item['name'] for item in json.loads((ROOT / 'tools/change_control/labels.json').read_text(encoding='utf-8')) + } + self.assertTrue(referenced) + self.assertEqual(referenced - labels, set()) + + def test_codeowners_parses_and_covers_root(self): + """One CODEOWNERS rule must cover the repository root with all verified authorizers.""" + lines = [ + line.split() + for line in (ROOT / '.github/CODEOWNERS').read_text(encoding='utf-8').splitlines() + if line and not line.startswith('#') + ] + self.assertEqual(lines, [['*', *('@' + name for name in AUTHORIZERS)]]) + + def test_procedure_authorizers_match_enforcement_and_codeowners(self): + """The procedure, enforcement code, and CODEOWNERS must name one identical authorizer set.""" + import re + + procedure = (ROOT / 'docs/process/MODIFICATION_PROCEDURE.md').read_text(encoding='utf-8') + named = set(re.findall(r'@(iliabaranov|rajasimman-madhivanan|davidt315)', procedure)) + self.assertEqual(named, set(AUTHORIZERS)) + + @unittest.skipUnless( + os.environ.get('GH_TOKEN'), 'GH_TOKEN absent; CODEOWNERS handle resolution test visibly skipped' + ) + def test_codeowners_handles_resolve(self): + """Every CODEOWNERS account must resolve through authenticated GitHub API access.""" + for handle in AUTHORIZERS: + result = subprocess.run(['gh', 'api', f'users/{handle}'], check=False, capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stderr) + + +class ImpactAndClassificationTests(unittest.TestCase): + def test_e2_rejects_heading_present_but_empty(self): + """An IA heading followed only by whitespace must fail completeness checking.""" + headings = ['# One', '# Two'] + complete, missing = impact_analysis_complete('# One\n \t\n# Two\nanswer\n', headings) + self.assertFalse(complete) + self.assertIn('# One', missing) + + def test_e2_accepts_na_content_without_judging_truth(self): + """The checker verifies nonblank content but does not judge whether N/A is adequate.""" + self.assertEqual(impact_analysis_complete('# One\nN/A\n# Two\nN/A\n', ['# One', '# Two']), (True, [])) + + def test_e3_rejects_invalid_sr_ids(self): + """A cited token shaped like an SR but outside the canonical grammar must be a finding.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\nSR-BOGUS-99\n' + results = evaluate(ROOT, data) + self.assertEqual(next(item for item in results if item.check_id == 'E3').status, 'fail') + + def test_e4_under_classification_is_a_finding(self): + """A class label below the path-derived floor must fail E4.""" + data = snapshot(files=[{'filename': 'docs/safety/HARA.md'}]) + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'fail') + + def test_e4_over_classification_is_not_a_finding(self): + """A class label above the path-derived floor must be accepted.""" + data = snapshot(files=[{'filename': 'README.md'}]) + data['pr']['labels'] = [{'name': 'class-c'}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'pass') + + def test_missing_class_label_is_a_finding(self): + """A PR without a classification label must fail classification checking.""" + data = snapshot() + data['pr']['labels'] = [] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'fail') + + def test_duplicate_class_labels_are_a_finding(self): + """Multiple classification labels are ambiguous and must fail E4.""" + data = snapshot() + data['pr']['labels'] = [{'name': 'class-a'}, {'name': 'class-b'}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'fail') + + def test_minimum_class_rules(self): + """Every settled path rule must produce its documented minimum classification floor.""" + self.assertEqual(minimum_class(['pstop_c/x.c']), 'C') + self.assertEqual(minimum_class(['firmware/main/main.c']), 'C') + self.assertEqual(minimum_class(['components/x.c']), 'B') + self.assertEqual(minimum_class(['firmware/sdkconfig.defaults']), 'B') + self.assertEqual(minimum_class(['.github/workflows/x.yml']), 'B') + + +class ApprovalAndLinkTests(unittest.TestCase): + def _class_c(self): + data = snapshot(files=[{'filename': 'docs/safety/HARA.md'}]) + data['pr']['labels'] = [{'name': 'class-c'}] + data['reviews'] = [ + {'user': {'login': AUTHORIZERS[0]}, 'state': 'APPROVED', 'commit_id': 'abc123'}, + {'user': {'login': AUTHORIZERS[1]}, 'state': 'APPROVED', 'commit_id': 'abc123'}, + ] + return data + + def test_class_c_two_distinct_approvals_pass(self): + """Class C requires two distinct current approving authorizers who are not the author.""" + self.assertEqual(next(item for item in evaluate(ROOT, self._class_c()) if item.check_id == 'E5').status, 'pass') + + def test_duplicate_reviews_count_once(self): + """Repeated approvals by one account must count as one approval.""" + data = self._class_c() + data['reviews'][1]['user']['login'] = AUTHORIZERS[0] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail') + + def test_author_approval_is_excluded(self): + """The PR author's own approval must never satisfy Class C approval.""" + data = self._class_c() + data['pr']['user']['login'] = AUTHORIZERS[0] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail') + + def test_stale_approval_is_excluded(self): + """An approval for a commit other than the PR head must not count.""" + data = self._class_c() + data['reviews'][1]['commit_id'] = 'oldsha' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail') + + def test_approval_without_commit_identity_is_excluded(self): + """An approval with no commit identity cannot establish review of the current diff.""" + data = self._class_c() + data['reviews'][1].pop('commit_id') + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail') + + def test_authorization_after_pr_is_an_ordering_finding(self): + """An authorization timestamp after implementation began must fail E1 ordering.""" + data = snapshot() + data['pr']['created_at'] = '2026-09-10T10:00:00Z' + data['issue_comments'][0]['created_at'] = '2026-09-10T11:00:00Z' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_authorization_before_impact_analysis_is_an_ordering_finding(self): + """Authorization must follow the completed Impact Analysis rather than merely precede implementation.""" + data = snapshot() + data['pr']['created_at'] = '2026-09-10T12:00:00Z' + data['issue_comments'][0]['created_at'] = '2026-09-10T10:00:00Z' + data['issue_comments'][1]['created_at'] = '2026-09-10T11:00:00Z' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_pr_author_cannot_authorize_own_change(self): + """The implementer must not satisfy the Change Request authorization requirement.""" + data = snapshot() + data['pr']['user']['login'] = AUTHORIZERS[0] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_class_c_requires_two_distinct_cr_authorizers_before_implementation(self): + """Class C implementation cannot start after only one Change Request authorization.""" + data = snapshot() + data['pr']['labels'] = [{'name': 'class-c'}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + data['issue_comments'].append({'user': {'login': AUTHORIZERS[1]}, 'body': 'Authorized: proceed.'}) + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass') + + def test_rejection_text_does_not_count_as_authorization(self): + """An authorizer saying a change is not authorized must not satisfy E1.""" + data = snapshot() + data['issue_comments'][0]['body'] = 'Rejected: this is not authorized.' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_emergency_requires_two_approvals_and_short_form_ia(self): + """Emergency E7 must require two current approvals and all three short-form IA subjects.""" + data = snapshot() + data['pr']['labels'].append({'name': 'emergency'}) + data['issue_comments'][1]['body'] = 'What changed: tooling\nWhat it could affect: process\nTests: change-control' + data['reviews'] = [ + {'user': {'login': AUTHORIZERS[0]}, 'state': 'APPROVED', 'commit_id': 'abc123'}, + {'user': {'login': AUTHORIZERS[1]}, 'state': 'APPROVED', 'commit_id': 'abc123'}, + ] + data['check_runs'] = [{'name': 'change-control', 'conclusion': 'success', 'head_sha': 'abc123'}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E7').status, 'pass') + + def test_ambiguous_change_request_link_fails(self): + """A bare issue number without Closes or Refs syntax must not be guessed as the CR.""" + data = snapshot() + data['pr']['body'] = 'Issue #17' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_multiple_change_request_links_fail(self): + """Multiple candidate CR links must fail rather than selecting one.""" + data = snapshot() + data['pr']['body'] = 'Closes #17\nRefs #18' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_same_repository_issue_url_is_accepted(self): + """A full issue URL for the current repository must identify its Change Request.""" + data = snapshot(repository='polymathrobotics/protective-stop') + data['pr']['body'] = 'https://github.com/polymathrobotics/protective-stop/issues/17' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass') + + def test_foreign_repository_same_issue_number_is_rejected(self): + """A foreign issue URL must not map an equal issue number into the current repository.""" + data = snapshot(repository='polymathrobotics/protective-stop') + data['pr']['body'] = 'Refs https://github.com/other/protective-stop/issues/17' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_foreign_repository_different_issue_number_is_rejected(self): + """A foreign issue URL must never select that number from the current repository.""" + data = snapshot(repository='polymathrobotics/protective-stop') + data['pr']['body'] = 'Closes https://github.com/other/project/issues/91' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_mixed_local_and_foreign_issue_links_are_rejected(self): + """A local CR candidate mixed with any foreign candidate must be treated as ambiguous.""" + data = snapshot(repository='polymathrobotics/protective-stop') + data['pr']['body'] = 'Closes #17\nRefs https://github.com/other/project/issues/91' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail') + + def test_duplicate_same_local_issue_link_is_accepted(self): + """Repeated equivalent local links must identify one unambiguous Change Request.""" + data = snapshot(repository='polymathrobotics/protective-stop') + data['pr']['body'] = 'Closes #17\nRefs #17' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass') + + def test_repository_identity_comparison_is_case_insensitive(self): + """GitHub owner and repository casing must not make a same-repository URL foreign.""" + data = snapshot(repository='PolyMathRobotics/Protective-Stop') + data['pr']['body'] = 'Refs https://github.com/POLYMATHROBOTICS/protective-stop/issues/17' + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass') + + def test_needs_change_request_label_exempts_e1(self): + """External PRs awaiting a maintainer CR must produce a pending E1 result.""" + data = snapshot() + data['pr']['labels'].append({'name': 'needs-change-request'}) + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pending') + + +class EvidenceAndCommentTests(unittest.TestCase): + def test_e6_requires_at_least_one_named_test(self): + """An IA that names no specific test must fail the plan-versus-execution check.""" + result = next(item for item in evaluate(ROOT, snapshot()) if item.check_id == 'E6') + self.assertEqual(result.status, 'fail') + + def test_e6_missing_workflow_evidence_names_test(self): + """A test named by the IA without head-SHA check evidence must fail and be named.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`missing-check`\n' + result = next(item for item in evaluate(ROOT, data) if item.check_id == 'E6') + self.assertEqual(result.status, 'fail') + self.assertIn('missing-check', result.message) + + def test_e6_ignores_evidence_for_other_sha(self): + """Evidence attached to an older commit must not satisfy the IA plan.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n' + data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': 'oldsha'}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail') + + def test_e6_ignores_check_evidence_without_head_sha(self): + """A successful check without explicit commit identity cannot satisfy the verification plan.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n' + data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success'}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail') + + def test_e6_ignores_workflow_evidence_with_null_head_sha(self): + """A successful workflow with null commit identity cannot satisfy the verification plan.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n' + data['check_runs'] = [] + data['workflow_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': None}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail') + + def test_e6_ignores_check_evidence_with_empty_head_sha(self): + """A successful check with empty commit identity cannot satisfy the verification plan.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n' + data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': ''}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail') + + def test_e6_accepts_workflow_evidence_with_exact_head_sha(self): + """A successful exactly named workflow explicitly attached to PR head must satisfy the plan.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n' + data['check_runs'] = [] + data['workflow_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': 'abc123'}] + self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'pass') + + def test_e6_does_not_accept_a_substring_check_name(self): + """A short IA test token must not match an unrelated longer check-run name.""" + data = snapshot() + data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host`\n' + data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': 'abc123'}] + result = next(item for item in evaluate(ROOT, data) if item.check_id == 'E6') + self.assertEqual(result.status, 'fail') + self.assertIn('host', result.message) + + def test_e6_reads_plain_table_cells(self): + """Plain test names in the IA verification table must be checked, not only backticked names.""" + data = snapshot() + data['issue_comments'][1]['body'] = data['issue_comments'][1]['body'].replace( + '| Tests that validate the change itself | |', + '| Tests that validate the change itself | missing-table-check |', + ) + result = next(item for item in evaluate(ROOT, data) if item.check_id == 'E6') + self.assertEqual(result.status, 'fail') + self.assertIn('missing-table-check', result.message) + + def test_upsert_comment_updates_existing_comment(self): + """An existing bot report must be updated instead of appending another comment.""" + calls = [] + upsert_comment( + lambda method, path, body=None: calls.append((method, path, body)), + 9, + 'report', + [{'id': 44, 'body': 'old'}], + ) + self.assertEqual(calls[0][0:2], ('PATCH', 'repos/{repo}/issues/comments/44')) + + def test_upsert_comment_creates_when_absent(self): + """A bot report must be created exactly once when no marker exists.""" + calls = [] + upsert_comment(lambda method, path, body=None: calls.append((method, path, body)), 9, 'report', []) + self.assertEqual( + calls, [('POST', 'repos/{repo}/issues/9/comments', {'body': '\nreport'})] + ) + + +class CliTests(unittest.TestCase): + def test_gh_api_flattens_all_paginated_list_pages(self): + """Policy evaluation must see every item returned across GitHub list pages.""" + with tempfile.TemporaryDirectory() as directory: + data = Path(directory) / 'gh.json' + data.write_text( + json.dumps({'responses': {'GET items': {'__pages__': [[{'id': 1}], [{'id': 2}]]}}}), + encoding='utf-8', + ) + with mock.patch.dict(os.environ, {'FAKE_GH_DATA': str(data)}): + api = GhApi(f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', 'acme/project') + self.assertEqual(api('GET', 'items', paginate=True), [{'id': 1}, {'id': 2}]) + + def test_gh_api_merges_all_paginated_collection_pages(self): + """Check and workflow evidence must include every GitHub response page.""" + with tempfile.TemporaryDirectory() as directory: + data = Path(directory) / 'gh.json' + pages = [{'check_runs': [{'id': 1}]}, {'check_runs': [{'id': 2}]}] + data.write_text( + json.dumps({'responses': {'GET checks': {'__pages__': pages}}}), + encoding='utf-8', + ) + with mock.patch.dict(os.environ, {'FAKE_GH_DATA': str(data)}): + api = GhApi(f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', 'acme/project') + self.assertEqual( + api('GET', 'checks', paginate=True, collection_key='check_runs'), + {'check_runs': [{'id': 1}, {'id': 2}]}, + ) + + def test_gh_api_rejects_non_list_paginated_page(self): + """A malformed list page must fail closed rather than hide omitted GitHub records.""" + with self.assertRaisesRegex(RuntimeError, 'non-list page'): + self._fake_api({'GET items': {'__pages__': [[{'id': 1}], {'id': 2}]}})( + 'GET', 'items', paginate=True + ) + + def test_gh_api_rejects_missing_collection_in_paginated_page(self): + """A malformed collection page must fail closed rather than produce partial evidence.""" + with self.assertRaisesRegex(RuntimeError, 'lacks check_runs'): + self._fake_api({'GET checks': {'__pages__': [{'check_runs': []}, {}]}})( + 'GET', 'checks', paginate=True, collection_key='check_runs' + ) + + def test_gh_api_rejects_invalid_json(self): + """A successful API process with malformed JSON must still be an unable-to-run error.""" + with self.assertRaisesRegex(RuntimeError, 'invalid JSON'): + self._fake_api({'GET items': 'not-json'})('GET', 'items') + + def test_workflows_disable_writes_for_fork_pull_requests(self): + """Fork pull requests must still run checks without attempting unavailable comment writes.""" + change = (ROOT / '.github/workflows/change-control.yml').read_text(encoding='utf-8') + self.assertIn('CAN_COMMENT', change) + self.assertIn('--no-comment', change) + + def test_marker_comment_workflows_cancel_superseded_pr_runs(self): + """The marker-comment writer must cancel stale runs in its per-PR concurrency group.""" + change = (ROOT / '.github/workflows/change-control.yml').read_text(encoding='utf-8') + self.assertIn('group: change-control-${{ github.event.pull_request.number }}', change) + self.assertEqual(change.count('cancel-in-progress: true'), 1) + + def test_warn_mode_exits_zero_with_findings(self): + """Warn mode must report findings while returning success to the caller.""" + result = self._run_cli('warn', {'responses': {}}) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('mode: warn', result.stdout) + + def test_enforce_mode_exits_one_with_findings(self): + """Enforce mode must return one for the same findings that warn mode tolerates.""" + result = self._run_cli('enforce', {'responses': {}}) + self.assertEqual(result.returncode, 1, result.stderr) + + def test_mode_file_rejects_unknown_value(self): + """An invalid mode must return two rather than silently defaulting to warn.""" + result = self._run_cli('maybe', {'responses': {}}) + self.assertEqual(result.returncode, 2) + + def test_missing_mode_file_returns_two(self): + """An unreadable mode source must return two rather than silently defaulting to warn.""" + with tempfile.TemporaryDirectory() as directory: + result = subprocess.run( + [ + sys.executable, + '-m', + 'tools.change_control', + '--root', + directory, + '--repository', + 'acme/project', + '--pr', + '7', + '--gh', + f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', + ], + cwd=ROOT, + check=False, + capture_output=True, + text=True, + ) + self.assertEqual(result.returncode, 2) + self.assertIn('cannot read enforcement mode', result.stderr) + + def test_gh_api_failure_returns_two(self): + """A failed gh subprocess must make the checker unable to run, not create policy findings.""" + result = self._run_cli('warn', {'exit_code': 1, 'stderr': 'API unavailable'}) + self.assertEqual(result.returncode, 2) + self.assertIn('API unavailable', result.stderr) + + def test_partial_json_returns_two(self): + """A partial GitHub response must fail closed as an execution error.""" + responses = self._responses() + responses['GET repos/acme/project/pulls/7'] = {'body': 'Closes #17'} + result = self._run_cli('warn', {'responses': responses}) + self.assertEqual(result.returncode, 2) + + def test_no_comment_avoids_write_api(self): + """Fork-safe change-control checks must not call the comment API when writes are unavailable.""" + result, calls = self._run_cli('warn', {'responses': self._responses()}, no_comment=True, record_calls=True) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertNotIn('POST repos/acme/project/issues/7/comments', calls) + + def test_warn_mode_survives_comment_permission_failure(self): + """Warn-mode findings remain visible in logs when GitHub denies advisory comment writes.""" + responses = self._responses() + responses.pop('POST repos/acme/project/issues/7/comments') + result = self._run_cli('warn', {'responses': responses}) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('comment publication warning', result.stderr) + + def test_enforce_mode_fails_closed_on_comment_permission_failure(self): + """Enforce mode must return unable-to-run when its required report comment cannot be published.""" + responses = self._responses() + responses.pop('POST repos/acme/project/issues/7/comments') + result = self._run_cli('enforce', {'responses': responses}) + self.assertEqual(result.returncode, 2) + self.assertIn('comment publication warning', result.stderr) + + def _fake_api(self, responses): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + data = Path(temporary.name) / 'gh.json' + data.write_text(json.dumps({'responses': responses}), encoding='utf-8') + patcher = mock.patch.dict(os.environ, {'FAKE_GH_DATA': str(data)}) + patcher.start() + self.addCleanup(patcher.stop) + return GhApi(f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', 'acme/project') + + def _responses(self): + data = snapshot() + data['pr']['labels'] = [] + return { + 'GET repos/acme/project/pulls/7': data['pr'], + 'GET repos/acme/project/pulls/7/files': data['files'], + 'GET repos/acme/project/issues/17': data['issue'], + 'GET repos/acme/project/issues/17/comments': data['issue_comments'], + 'GET repos/acme/project/pulls/7/reviews': data['reviews'], + 'GET repos/acme/project/commits/abc123/check-runs': {'check_runs': data['check_runs']}, + 'GET repos/acme/project/actions/runs?head_sha=abc123': {'workflow_runs': data['workflow_runs']}, + 'GET repos/acme/project/issues/7/comments': [], + 'POST repos/acme/project/issues/7/comments': {}, + } + + def _run_cli(self, mode, fake_data, no_comment=False, record_calls=False): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / 'docs/process').mkdir(parents=True) + (root / 'docs/process/enforcement-mode').write_text(mode + '\n', encoding='utf-8') + (root / 'docs/process/templates').mkdir() + shutil.copy2( + ROOT / 'docs/process/templates/IMPACT_ANALYSIS.md', root / 'docs/process/templates/IMPACT_ANALYSIS.md' + ) + (root / 'docs/safety').mkdir(parents=True) + shutil.copy2(ROOT / 'docs/safety/SAFETY_REQUIREMENTS.md', root / 'docs/safety/SAFETY_REQUIREMENTS.md') + (root / '.github/ISSUE_TEMPLATE').mkdir(parents=True) + shutil.copy2(FORM, root / '.github/ISSUE_TEMPLATE/change-request.yml') + data_path = root / 'gh.json' + data_path.write_text( + json.dumps( + fake_data + if fake_data.get('exit_code') + else {'responses': fake_data.get('responses') or self._responses()} + ), + encoding='utf-8', + ) + environment = os.environ.copy() + environment['FAKE_GH_DATA'] = str(data_path) + environment['PYTHONPATH'] = str(ROOT) + calls_path = root / 'gh-calls.txt' + if record_calls: + environment['FAKE_GH_CALLS'] = str(calls_path) + command = [ + sys.executable, + '-m', + 'tools.change_control', + '--root', + str(root), + '--repository', + 'acme/project', + '--pr', + '7', + '--gh', + f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', + ] + if no_comment: + command.append('--no-comment') + result = subprocess.run( + command, + cwd=ROOT, + env=environment, + check=False, + capture_output=True, + text=True, + ) + if record_calls: + calls = calls_path.read_text(encoding='utf-8') if calls_path.exists() else '' + return result, calls + return result + + +if __name__ == '__main__': + unittest.main(verbosity=2) From c16673847bfd4be660812b1b437305bb9fbe0152 Mon Sep 17 00:00:00 2001 From: Raj Madhivanan Date: Thu, 17 Sep 2026 14:31:58 -0700 Subject: [PATCH 2/2] test: align checker tests with repository hooks Why: The focused checker test module was split from an executable monolithic runner. It retained that runner's shebang even though the new module is imported by unittest and is intentionally not executable. Two lines also did not match the repository's Python formatter. What changed: - Removed the script shebang from the imported test module. - Applied the repository formatter's line wrapping. - Re-ran all 78 focused checker tests and the applicable pre-commit hooks. Impact: There is no checker, workflow, policy, or test-behavior change. The module now satisfies executable-bit and formatting validation when later stacked files cause repository-wide hooks to inspect it. Co-Authored-By: OpenCode --- tools/change_control/test_change_control.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/tools/change_control/test_change_control.py b/tools/change_control/test_change_control.py index a517bb52..68dd376a 100644 --- a/tools/change_control/test_change_control.py +++ b/tools/change_control/test_change_control.py @@ -1,4 +1,3 @@ -#!/usr/bin/env python3 # SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. # SPDX-License-Identifier: Apache-2.0 """Spec-driven tests for the focused warn-mode change-control checker.""" @@ -425,7 +424,9 @@ def test_emergency_requires_two_approvals_and_short_form_ia(self): """Emergency E7 must require two current approvals and all three short-form IA subjects.""" data = snapshot() data['pr']['labels'].append({'name': 'emergency'}) - data['issue_comments'][1]['body'] = 'What changed: tooling\nWhat it could affect: process\nTests: change-control' + data['issue_comments'][1]['body'] = ( + 'What changed: tooling\nWhat it could affect: process\nTests: change-control' + ) data['reviews'] = [ {'user': {'login': AUTHORIZERS[0]}, 'state': 'APPROVED', 'commit_id': 'abc123'}, {'user': {'login': AUTHORIZERS[1]}, 'state': 'APPROVED', 'commit_id': 'abc123'}, @@ -611,9 +612,7 @@ def test_gh_api_merges_all_paginated_collection_pages(self): def test_gh_api_rejects_non_list_paginated_page(self): """A malformed list page must fail closed rather than hide omitted GitHub records.""" with self.assertRaisesRegex(RuntimeError, 'non-list page'): - self._fake_api({'GET items': {'__pages__': [[{'id': 1}], {'id': 2}]}})( - 'GET', 'items', paginate=True - ) + self._fake_api({'GET items': {'__pages__': [[{'id': 1}], {'id': 2}]}})('GET', 'items', paginate=True) def test_gh_api_rejects_missing_collection_in_paginated_page(self): """A malformed collection page must fail closed rather than produce partial evidence."""