ci(runner): add bounded hosted E2E support recheck

Co-Authored-By: Paperclip <noreply@paperclip.ing>

scripts/ci/test_source_e2e_mode.py

@@ -0,0 +1,136 @@
+"""Offline regressions for the reviewed, exclusive hosted support dispatch."""
+import os
+from pathlib import Path
+import re
+import shutil
+import sys
+import subprocess
+import tempfile
+import textwrap
+import unittest
+
+ROOT = Path(__file__).resolve().parents[2]
+WORKFLOW = '.github/workflows/docker-runner-check.yml'
+SOURCE = '5cd6d3d5237e3e15394fcc1c7e754c30a63c5169'
+LOCK = '38338a6867358440c5ab5993eaeb85fb501bc7df65d24acf52fbd63c880ee4ba'
+INTEL_SOURCE = '5eba61ece929dcfb2b0c1761825a2d9e557c2554'
+BASELINE = 'b58907578'
+
+
+def job(text, name):
+    return re.search(r'^  ' + name + r':\n.*?(?=^  [a-z_]+:\n|\Z)', text, re.M | re.S).group()
+
+
+def script(block, name):
+    step = block.split('      - name: ' + name + '\n', 1)[1].split('\n      - ', 1)[0]
+    return textwrap.dedent(step.split('        run: |\n', 1)[1]).rstrip() + '\n'
+
+
+class SupportMode(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.workflow = (ROOT / WORKFLOW).read_text()
+        cls.authorize = script(job(cls.workflow, 'authorize_manual'), 'Authorize an explicit maintainer image build')
+
+    def admission(self, **changes):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            fake = root / 'gh'
+            fake.write_text('''#!/usr/bin/env python3
+import os, sys
+path = sys.argv[2]
+if path == 'users/maintainer': print('1')
+elif path == 'repos/paperclipai/paperclip/git/ref/heads/coverage': print(os.environ['FAKE_TARGET'])
+elif path == 'repos/paperclipai/paperclip': print(os.environ['FAKE_VISIBILITY'])
+else: raise RuntimeError('Unexpected offline GitHub request')
+''')
+            fake.chmod(0o755)
+            env = dict(PATH=str(root) + os.pathsep + str(Path(shutil.which('jq')).parent) + os.pathsep + os.defpath,
+                       REPOSITORY='paperclipai/paperclip', REPOSITORY_ID='1170821064', ACTOR_ID='1',
+                       TRIGGERING_ACTOR='maintainer', ALLOWED_IDS='[1]', TARGET_BRANCH='coverage',
+                       EXPECTED_SOURCE_SHA=SOURCE, EXPECTED_LOCK_SHA256=LOCK, FAKE_TARGET=SOURCE,
+                       FAKE_VISIBILITY='public', VERIFY_RUNNER='false', VERIFY_SOURCE='false',
+                       SOURCE_ROOT_TESTS_ONLY='false', SOURCE_E2E_SUPPORT_ONLY='true', IMAGE_MODE='false',
+                       VERIFY_PI_INTEL='false', DIAGNOSE_AJV='false', OTHER_MODE='true',
+                       GITHUB_OUTPUT=str(root / 'output'))
+            env.update(changes)
+            result = subprocess.run(['bash', '-c', self.authorize], env=env, capture_output=True, text=True, timeout=5)
+            return result.returncode, (root / 'output').read_text() if (root / 'output').exists() else ''
+
+    def test_exact_reviewed_source_and_overlay_are_admitted(self):
+        self.assertEqual(self.admission(), (0, 'target_sha=' + SOURCE + '\n'))
+
+    def test_every_other_execution_mode_conflicts(self):
+        for mode in ('VERIFY_RUNNER', 'VERIFY_SOURCE', 'SOURCE_ROOT_TESTS_ONLY', 'IMAGE_MODE', 'VERIFY_PI_INTEL', 'DIAGNOSE_AJV'):
+            with self.subTest(mode=mode):
+                status, output = self.admission(**{mode: 'true'})
+                self.assertNotEqual(status, 0); self.assertEqual(output, '')
+
+    def test_wrong_missing_or_moving_source_and_lock_fail(self):
+        for changes in ({'EXPECTED_SOURCE_SHA': ''}, {'EXPECTED_SOURCE_SHA': INTEL_SOURCE},
+                        {'FAKE_TARGET': INTEL_SOURCE}, {'EXPECTED_LOCK_SHA256': ''},
+                        {'EXPECTED_LOCK_SHA256': 'a' * 64}, {'FAKE_VISIBILITY': 'private'}, {'ACTOR_ID': '2'}):
+            with self.subTest(changes=changes):
+                status, output = self.admission(**changes)
+                self.assertNotEqual(status, 0); self.assertEqual(output, '')
+
+    def test_original_intel_admission_and_entire_job_are_unchanged(self):
+        self.assertEqual(self.admission(SOURCE_E2E_SUPPORT_ONLY='false', VERIFY_PI_INTEL='true',
+                                       EXPECTED_SOURCE_SHA=INTEL_SOURCE, FAKE_TARGET=INTEL_SOURCE),
+                         (0, 'target_sha=' + INTEL_SOURCE + '\n'))
+        baseline = subprocess.check_output(['git', 'show', BASELINE + ':' + WORKFLOW], cwd=ROOT, text=True)
+        self.assertEqual(job(self.workflow, 'manual_pi_intel'), job(baseline, 'manual_pi_intel'))
+        for name in ('manual_runner_verify', 'manual_ajv_pack_diagnostic'):
+            self.assertEqual(job(self.workflow, name), job(baseline, name))
+
+    def test_focused_commands_are_complete_support_only_and_image_is_excluded(self):
+        source = job(self.workflow, 'manual_source_verify')
+        plan = script(source, 'Record immutable source and planned checks')
+        focused = plan.split('if [ "$SOURCE_E2E_SUPPORT_ONLY" = true ]; then', 1)[1].split('\nfi', 1)[0]
+        self.assertEqual(re.findall(r"^    '([^']+)'", focused, re.M), [
+            'pnpm test:e2e:runner:typecheck', 'pnpm test:e2e:runner:unit',
+            'node cli/node_modules/tsx/dist/cli.mjs tests/runner-e2e/launch.ts --list'])
+        self.assertIn('!inputs.source_e2e_support_only', job(self.workflow, 'manual_image'))
+        self.assertIn("inputs.source_e2e_support_only && 'source-e2e-support'", self.workflow)
+        self.assertNotIn('packages: write', source)
+        self.assertNotIn('secrets.', source)
+        self.assertIn('persist-credentials: false', source)
+        self.assertIn('1380 - $(date +%s) + checks_started_at', source)
+        self.assertIn('check_timeout=15m', source)
+        self.assertIn('inputs.source_e2e_support_only && 45', source)
+        self.assertIn('256 * 1024 * 1024', source)
+        self.assertIn('inputs.source_e2e_support_only && 7 || 14', source)
+
+    def test_complete_failure_evidence_is_admitted_and_unsafe_or_oversize_is_not(self):
+        source = script(job(self.workflow, 'manual_source_verify'), 'Admit complete focused evidence within the storage bound')
+        admission = source.split("<<'PYTHON'\n", 1)[1].rsplit('PYTHON', 1)[0]
+        for scenario in ('failed-check', 'oversize', 'symlink', 'absent'):
+            with self.subTest(scenario=scenario), tempfile.TemporaryDirectory() as temp:
+                root = Path(temp); evidence = root / 'remote-source-verification'; evidence.mkdir()
+                if scenario == 'failed-check':
+                    (evidence / 'check-1.log').write_text('real failure output\n')
+                    (evidence / 'check-1.status').write_text('1\n')
+                elif scenario == 'oversize':
+                    with (evidence / 'large.log').open('wb') as output: output.truncate(256 * 1024 * 1024 + 1)
+                elif scenario == 'symlink': (evidence / 'link').symlink_to(root / 'outside')
+                output = root / 'output'
+                result = subprocess.run([sys.executable, '-c', admission], cwd=root,
+                                        env={'GITHUB_OUTPUT': str(output)}, capture_output=True, timeout=5)
+                if scenario == 'failed-check':
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertEqual(output.read_text(), 'admitted=true\n')
+                    self.assertIn('real failure output', (evidence / 'check-1.log').read_text())
+                    self.assertTrue((evidence / 'evidence-inventory.json').is_file())
+                else:
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertFalse(output.exists())
+
+    def test_all_embedded_shell_scripts_parse(self):
+        for number, match in enumerate(re.finditer(r'^        run: \|\n((?:          .*\n|\n)+)', self.workflow, re.M)):
+            # The heredoc retains its Python as data for bash syntax validation.
+            with self.subTest(block=number):
+                subprocess.run(['bash', '-n'], input=textwrap.dedent(match.group(1)), text=True, check=True)
+
+
+if __name__ == '__main__':
+    unittest.main()