|
1 | 1 | #!/usr/bin/env python3 |
2 | | -"""Falsifiable controls for snapshot admission and the one-time bootstrap exception.""" |
| 2 | +"""Falsifiable controls for event selection, snapshot admission and bootstrap.""" |
3 | 3 | import importlib.util |
4 | 4 | import json |
| 5 | +import os |
5 | 6 | from pathlib import Path |
| 7 | +import re |
6 | 8 | import subprocess |
7 | 9 | import sys |
8 | 10 | import tempfile |
| 11 | +import textwrap |
9 | 12 | import unittest |
10 | 13 | from unittest.mock import patch |
11 | 14 |
|
@@ -54,6 +57,58 @@ def run(command, **kwargs): |
54 | 57 | '--tool', str(self.root / 'tool'), '--output', str(self.root / 'reports')]), patch.object(ci.subprocess, 'run', side_effect=run): |
55 | 58 | return ci.main() |
56 | 59 |
|
| 60 | + def select_event(self, event, head, before='', pr_base='', pr_head='', manual_base=''): |
| 61 | + workflow = (Path(__file__).resolve().parents[1] / '.github/workflows/ci.yml').read_text() |
| 62 | + selections = re.findall(r'(?m)^ case "\$EVENT_NAME" in\n(?:.*\n)*? esac\n', workflow) |
| 63 | + self.assertEqual(1, len(selections), 'Exercise the actual workflow event selector') |
| 64 | + script = 'set -euo pipefail\n' + textwrap.dedent(selections[0]) |
| 65 | + script += '\nprintf "%s\\n%s\\n" "$openapi_base" "$openapi_head"\n' |
| 66 | + result = REAL_RUN(['bash', '-c', script], cwd=self.root, check=True, capture_output=True, text=True, |
| 67 | + env={**os.environ, 'EVENT_NAME': event, 'HEAD_SHA': head, 'PUSH_BEFORE_SHA': before, |
| 68 | + 'PR_BASE_SHA': pr_base, 'PR_HEAD_SHA': pr_head, 'MANUAL_BASE_REF': manual_base}) |
| 69 | + return result.stdout.splitlines() |
| 70 | + |
| 71 | + def test_new_branch_push_uses_the_selected_heads_verified_parent(self): |
| 72 | + parent = self.save(CURRENT) |
| 73 | + head = self.save({**CURRENT, 'info': {**CURRENT['info'], 'title': 'Selected head'}}) |
| 74 | + self.save({**CURRENT, 'info': {**CURRENT['info'], 'title': 'Later checkout'}}) |
| 75 | + base_ref, head_ref = self.select_event('push', head, before='0' * 40) |
| 76 | + self.assertEqual(parent, ci.commit(base_ref)) |
| 77 | + self.assertEqual(head, head_ref) |
| 78 | + self.assertEqual(0, self.run_ci(base_ref, head_ref)) |
| 79 | + recorded = json.loads((self.root / 'reports/commits.json').read_text()) |
| 80 | + self.assertEqual((parent, head, False), (recorded['base'], recorded['head'], recorded['firstBaseline'])) |
| 81 | + |
| 82 | + def test_existing_branch_push_preserves_before_across_multiple_commits(self): |
| 83 | + before = self.save(CURRENT) |
| 84 | + self.save({**CURRENT, 'info': {**CURRENT['info'], 'title': 'Intermediate'}}) |
| 85 | + head = self.save({**CURRENT, 'info': {**CURRENT['info'], 'title': 'Head'}}) |
| 86 | + self.assertEqual([before, head], self.select_event('push', head, before=before)) |
| 87 | + |
| 88 | + def test_pull_request_preserves_its_base_and_head(self): |
| 89 | + head = self.save(CURRENT) |
| 90 | + self.assertEqual([self.base, head], self.select_event('pull_request', 'unused-runner-head', |
| 91 | + before='0' * 40, pr_base=self.base, pr_head=head)) |
| 92 | + |
| 93 | + def test_manual_run_preserves_its_explicit_base_ref(self): |
| 94 | + self.git('branch', 'manual-base', self.base) |
| 95 | + head = self.save(CURRENT) |
| 96 | + base_ref, head_ref = self.select_event('workflow_dispatch', head, manual_base='manual-base') |
| 97 | + self.assertEqual('manual-base', base_ref) |
| 98 | + self.assertEqual(self.base, ci.commit(base_ref)) |
| 99 | + self.assertEqual(head, head_ref) |
| 100 | + |
| 101 | + def test_unknown_event_is_refused(self): |
| 102 | + with self.assertRaises(subprocess.CalledProcessError) as error: |
| 103 | + self.select_event('pull_request_target', self.base) |
| 104 | + self.assertIn('Unsupported OpenAPI event', error.exception.stderr) |
| 105 | + |
| 106 | + def test_new_branch_push_without_a_parent_is_refused(self): |
| 107 | + base_ref, head_ref = self.select_event('push', self.base, before='0' * 40) |
| 108 | + with patch.object(ci, 'bootstrap') as bootstrap, self.assertRaises(subprocess.CalledProcessError): |
| 109 | + self.run_ci(base_ref, head_ref) |
| 110 | + bootstrap.assert_not_called() |
| 111 | + |
57 | 112 | def test_missing_head_snapshot_is_refused_before_bootstrap(self): |
58 | 113 | with patch.object(ci, 'bootstrap') as bootstrap, self.assertRaisesRegex(ValueError, 'Head snapshot'): |
59 | 114 | self.run_ci(self.base, self.base) |
|
0 commit comments