PaperClipAI/scripts/ci/test_source_e2e_mode.py

"""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()