diff --git a/.github/workflows/coverage-delta.yml b/.github/workflows/coverage-delta.yml new file mode 100644 index 00000000..9132e5f3 --- /dev/null +++ b/.github/workflows/coverage-delta.yml @@ -0,0 +1,35 @@ +--- +name: Coverage delta + +on: + pull_request: + +concurrency: + group: coverage-delta-${{ github.event.pull_request.number }} + cancel-in-progress: true + +permissions: + contents: read + issues: write + pull-requests: write + +jobs: + coverage-delta: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + - name: Compare safety-linter citations + env: + GH_TOKEN: ${{ github.token }} + 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.coverage_delta \ + --base '${{ github.event.pull_request.base.sha }}' \ + --head '${{ github.event.pull_request.head.sha }}' \ + --repository '${{ github.repository }}' \ + --pr '${{ github.event.pull_request.number }}' \ + "${args[@]}" diff --git a/tools/change_control/coverage_delta.py b/tools/change_control/coverage_delta.py new file mode 100644 index 00000000..6b032f40 --- /dev/null +++ b/tools/change_control/coverage_delta.py @@ -0,0 +1,161 @@ +# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Compare deterministic safety-linter output between two Git revisions.""" + +import argparse +import json +import os +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +from .__main__ import GhApi + +MARKER = '' + + +def compare_reports(base, head): + """Render coverage and citation changes without assigning safety significance.""" + if head.get('unavailable'): + raise RuntimeError(f'head coverage unavailable: {head["unavailable"]}') + if base.get('unavailable'): + return ( + 'Coverage before: unavailable. The base predates the stacked dependency on ' + f'change-0001 (`tools/safety_lint`): {base["unavailable"]}\n' + f'Coverage after: {head.get("coverage", {}).get("cited_tests", "?")}/{head.get("coverage", {}).get("total", "?")} cited.' + ) + before = base.get('coverage', {}) + after = head.get('coverage', {}) + lines = [ + f'Coverage before: {before.get("cited_tests", "?")}/{before.get("total", "?")} cited.', + f'Coverage after: {after.get("cited_tests", "?")}/{after.get("total", "?")} cited.', + ] + base_citations = base.get('citations', {}) + head_citations = head.get('citations', {}) + for sr_id in sorted(set(base_citations) | set(head_citations)): + old = set(base_citations.get(sr_id, [])) + new = set(head_citations.get(sr_id, [])) + if old - new: + lines.append(f'- {sr_id} lost citation(s): {", ".join(sorted(old - new))}') + if new - old: + lines.append(f'- {sr_id} gained citation(s): {", ".join(sorted(new - old))}') + old_unresolved = { + (item.get('check_id'), item.get('subject'), item.get('message')) + for item in base.get('findings', []) + if item.get('check_id') in ('C3', 'C4') + } + new_unresolved = { + (item.get('check_id'), item.get('subject'), item.get('message')) + for item in head.get('findings', []) + if item.get('check_id') in ('C3', 'C4') + } + for _, subject, message in sorted(new_unresolved - old_unresolved): + lines.append(f'- Newly unresolvable citation for {subject}: {message}') + if len(lines) == 2: + lines.append('- No citation gains, losses, or newly unresolvable citations.') + lines.append( + 'Limitation: this is deterministic citation resolution, not evidence that a cited test executed or passed.' + ) + return '\n'.join(lines) + + +def run_linter_at_tree(worktree): + """Run the revision's own unchanged linter and add its parsed citation map.""" + if not (worktree / 'tools/safety_lint/__main__.py').is_file(): + return {'unavailable': 'tools/safety_lint is absent at this revision'} + child_environment = os.environ.copy() + for name in ('GH_TOKEN', 'GITHUB_TOKEN'): + child_environment.pop(name, None) + result = subprocess.run( + [sys.executable, '-m', 'tools.safety_lint', '--json'], + cwd=worktree, + env=child_environment, + check=False, + capture_output=True, + text=True, + ) + if result.returncode == 2: + raise RuntimeError(result.stderr.strip() or 'safety linter could not run') + try: + report = json.loads(result.stdout) + except json.JSONDecodeError as error: + raise RuntimeError('safety linter emitted invalid JSON') from error + citation_code = ( + 'import json; from tools.safety_lint.runner import analyze; ' + 'print(json.dumps({r.sr_id: sorted(set(r.test_refs)) for r in analyze(".").trace}, sort_keys=True))' + ) + citations = subprocess.run( + [sys.executable, '-c', citation_code], + cwd=worktree, + env=child_environment, + check=False, + capture_output=True, + text=True, + ) + if citations.returncode: + raise RuntimeError(citations.stderr.strip() or 'cannot extract linter citations') + report['citations'] = json.loads(citations.stdout) + return report + + +def report_at_revision(root, revision): + """Run the unchanged checked-in linter at one detached revision.""" + temporary = Path(tempfile.mkdtemp(prefix='pstop-coverage-delta-')) + try: + result = subprocess.run( + ['git', 'worktree', 'add', '--detach', str(temporary), revision], + cwd=root, + check=False, + capture_output=True, + text=True, + ) + if result.returncode: + raise RuntimeError(result.stderr.strip() or f'cannot materialize revision {revision}') + return run_linter_at_tree(temporary) + finally: + subprocess.run( + ['git', 'worktree', 'remove', '--force', str(temporary)], cwd=root, check=False, capture_output=True + ) + shutil.rmtree(temporary, ignore_errors=True) + + +def upsert_coverage_comment(api, repository, pr, report): + """Create or update the single marker-owned deterministic coverage comment.""" + comments = api('GET', f'repos/{repository}/issues/{pr}/comments', paginate=True) + existing = next((comment for comment in comments if MARKER in comment.get('body', '')), None) + body = f'{MARKER}\n{report}' + if existing: + api('PATCH', f'repos/{repository}/issues/comments/{existing["id"]}', {'body': body}) + else: + api('POST', f'repos/{repository}/issues/{pr}/comments', {'body': body}) + + +def main(argv=None): + parser = argparse.ArgumentParser() + parser.add_argument('--root', default='.') + parser.add_argument('--base', required=True) + parser.add_argument('--head', default='HEAD') + 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: + root = Path(args.root).resolve() + report = compare_reports(report_at_revision(root, args.base), report_at_revision(root, args.head)) + print(report) + if not args.no_comment: + try: + upsert_coverage_comment(GhApi(args.gh, args.repository), args.repository, args.pr, report) + except RuntimeError as error: + print(f'coverage-delta: comment publication warning: {error}', file=sys.stderr) + return 0 + except (OSError, RuntimeError, ValueError) as error: + print(f'coverage-delta: cannot run: {error}', file=sys.stderr) + return 2 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/tools/change_control/test_coverage_delta.py b/tools/change_control/test_coverage_delta.py new file mode 100644 index 00000000..d5d26964 --- /dev/null +++ b/tools/change_control/test_coverage_delta.py @@ -0,0 +1,148 @@ +# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Spec-driven tests for deterministic safety-citation coverage delta reporting.""" + +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.coverage_delta import ( # noqa: E402 + compare_reports, + run_linter_at_tree, + upsert_coverage_comment, +) + + +class CoverageDeltaTests(unittest.TestCase): + def test_coverage_delta_detects_lost_citation(self): + """Deleting real cited evidence in a scratch repository must name the affected requirement.""" + with tempfile.TemporaryDirectory() as directory: + scratch = Path(directory) / 'repository' + shutil.copytree(ROOT / 'tools/safety_lint/fixtures/repository', scratch) + (scratch / 'tools').mkdir(exist_ok=True) + shutil.copytree( + ROOT / 'tools/safety_lint', + scratch / 'tools/safety_lint', + ignore=shutil.ignore_patterns('__pycache__'), + ) + base = run_linter_at_tree(scratch) + (scratch / 'tests/test_unique_probe.py').unlink() + head = run_linter_at_tree(scratch) + delta = compare_reports(base, head) + self.assertIn('SR-R-01', delta) + self.assertIn('unresolvable', delta.lower()) + + def test_revision_linter_subprocesses_cannot_observe_actions_credentials(self): + """Code from an untrusted revision must receive no Actions token while producing a normal report.""" + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + package = root / 'tools/safety_lint' + package.mkdir(parents=True) + (root / 'tools/__init__.py').write_text('', encoding='utf-8') + (package / '__init__.py').write_text('', encoding='utf-8') + guard = ( + "if os.environ.get('GH_TOKEN') or os.environ.get('GITHUB_TOKEN'): raise RuntimeError('token leaked')" + ) + (package / '__main__.py').write_text( + 'import json, os\n' + f'{guard}\n' + "print(json.dumps({'coverage': {'cited_tests': 1, 'total': 1}, 'findings': []}))\n", + encoding='utf-8', + ) + (package / 'runner.py').write_text( + 'import os\nfrom types import SimpleNamespace\n' + f'{guard}\n' + "def analyze(root): return SimpleNamespace(trace=[SimpleNamespace(sr_id='SR-X-01', test_refs=['probe'])])\n", + encoding='utf-8', + ) + with mock.patch.dict(os.environ, {'GH_TOKEN': 'gh-sentinel', 'GITHUB_TOKEN': 'github-sentinel'}): + report = run_linter_at_tree(root) + self.assertEqual(os.environ['GH_TOKEN'], 'gh-sentinel') + self.assertEqual(os.environ['GITHUB_TOKEN'], 'github-sentinel') + self.assertEqual(report['coverage'], {'cited_tests': 1, 'total': 1}) + self.assertEqual(report['citations'], {'SR-X-01': ['probe']}) + + def test_base_without_linter_exposes_stacked_dependency(self): + """A base predating change-0001 must be reported as unavailable, never treated as zero coverage.""" + delta = compare_reports( + {'unavailable': 'tools/safety_lint absent'}, + {'coverage': {'total': 1, 'cited_tests': 1}, 'findings': [], 'citations': {}}, + ) + self.assertIn('stacked dependency', delta.lower()) + + def test_head_without_linter_is_an_execution_error(self): + """Missing head coverage must fail explicitly rather than render unknown values as a delta.""" + with self.assertRaisesRegex(RuntimeError, 'head coverage unavailable'): + compare_reports( + {'coverage': {'total': 1, 'cited_tests': 1}, 'findings': [], 'citations': {}}, + {'unavailable': 'tools/safety_lint absent'}, + ) + + def test_coverage_comment_updates_instead_of_appending(self): + """Coverage delta must update its marker-owned comment rather than append on every run.""" + writes = [] + + def api(method, path, body=None, paginate=False): + if method == 'GET': + self.assertTrue(paginate) + return [{'id': 12, 'body': '\nold'}] + writes.append((method, path, body)) + return {} + + upsert_coverage_comment(api, 'acme/project', 7, 'new') + self.assertEqual(writes[0][0:2], ('PATCH', 'repos/acme/project/issues/comments/12')) + + def test_coverage_comment_marker_on_second_page_is_updated(self): + """Comment lookup must paginate so a marker beyond page one is updated rather than duplicated.""" + calls = [] + + def api(method, path, body=None, paginate=False): + calls.append((method, path, body, paginate)) + if method == 'GET': + self.assertTrue(paginate) + return [{'id': 1, 'body': 'first page'}, {'id': 12, 'body': '\nold'}] + return {} + + upsert_coverage_comment(api, 'acme/project', 7, 'new') + self.assertIn( + ('PATCH', 'repos/acme/project/issues/comments/12', {'body': '\nnew'}, False), calls + ) + self.assertFalse(any(call[0] == 'POST' for call in calls)) + + def test_coverage_cli_no_comment_avoids_write_api(self): + """Fork-safe coverage reporting must support stdout-only operation when write tokens are unavailable.""" + result = subprocess.run( + [sys.executable, '-m', 'tools.change_control.coverage_delta', '--help'], + cwd=ROOT, + check=False, + capture_output=True, + text=True, + ) + self.assertEqual(result.returncode, 0) + self.assertIn('--no-comment', result.stdout) + + +class WorkflowTests(unittest.TestCase): + def test_workflows_disable_writes_for_fork_pull_requests(self): + """Fork pull requests must still report coverage without attempting unavailable comment writes.""" + coverage = (ROOT / '.github/workflows/coverage-delta.yml').read_text(encoding='utf-8') + self.assertIn('CAN_COMMENT', coverage) + self.assertIn('--no-comment', coverage) + + def test_marker_comment_workflows_cancel_superseded_pr_runs(self): + """The marker-comment writer must cancel stale runs in its per-PR concurrency group.""" + coverage = (ROOT / '.github/workflows/coverage-delta.yml').read_text(encoding='utf-8') + self.assertIn('group: coverage-delta-${{ github.event.pull_request.number }}', coverage) + self.assertEqual(coverage.count('cancel-in-progress: true'), 1) + + +if __name__ == '__main__': + unittest.main(verbosity=2)