diff --git a/.github/workflows/iba1-periodic-jet.yml b/.github/workflows/iba1-periodic-jet.yml index 1a9f1b3be..edceebf9c 100644 --- a/.github/workflows/iba1-periodic-jet.yml +++ b/.github/workflows/iba1-periodic-jet.yml @@ -5,6 +5,7 @@ name: IBA1 periodic jet (exact-periodic anisotropy replay) paths: - 'reviews/iba1_periodic_jet_claude_20261005/**' - '.github/workflows/iba1-periodic-jet.yml' + - 'tests/test_periodic_jet_workflow.py' - 'coefficients/side24_v1/ENCLOSURE.json' - 'coefficients/side24_v1/PROOF.md' - 'imports/lifetime_parent_20260925/UNIFORM_MATRIX_CAP_AND_LIFETIME.md' @@ -35,8 +36,26 @@ jobs: FLAGS: ${{ matrix.flags }} run: | set -euo pipefail + python $FLAGS -m unittest discover -s tests -p test_periodic_jet_workflow.py -v python -B -S - <<'PY' - import hashlib, json, os, pathlib, subprocess, sys + import hashlib, json, math, os, pathlib, subprocess, sys + def strict_json(raw): + def pairs(items): + out = {} + for key, value in items: + if key in out: + raise ValueError('duplicate JSON key: ' + key) + out[key] = value + return out + def constant(value): + raise ValueError('nonfinite JSON constant: ' + value) + def finite_float(value): + number = float(value) + if not math.isfinite(number): + raise ValueError('nonfinite JSON number: ' + value) + return number + return json.loads(raw, object_pairs_hook=pairs, + parse_constant=constant, parse_float=finite_float) root = pathlib.Path('reviews/iba1_periodic_jet_claude_20261005').resolve() meta = json.loads((root / 'SOURCE_FILES.json').read_text()) present = sorted(p.name for p in root.iterdir()) @@ -56,11 +75,42 @@ jobs: flags = os.environ['FLAGS'].split() checker = str(root / 'periodic_jet_check.py') run = subprocess.run([sys.executable] + flags + [checker], capture_output=True, cwd=root) - if run.returncode != 0 or run.stdout != (root / 'RESULTS.json').read_bytes(): + if run.returncode != 0 or run.stderr or run.stdout != (root / 'RESULTS.json').read_bytes(): raise ValueError('replay differs from RESULTS.json') + reference = strict_json(run.stdout) + if (reference['passed'] is not True + or not all(value is True for value in reference['checks'].values())): + raise ValueError('baseline must report every check passing') + intended = { + 'M1': {'R3_L24_anisotropy_measured', 'R3_L4_anisotropy_visible'}, + 'M2': {'R1_reference_d3_trace_frobenius', 'R1_reference_jet', + 'R4_c224_in_side24_interval'}, + 'M3': {'R4_c224_in_side24_interval'}, + 'M4': {'R1_reference_d3_trace_frobenius', + 'R6_transverse_frame_invariance_d3', 'R6_transverse_frame_invariance_d4'}, + } for label, code in (('M1', 1), ('M2', 1), ('M3', 1), ('M4', 1), ('M9', 2)): m = subprocess.run([sys.executable] + flags + [checker, '--mutant', label], capture_output=True, cwd=root) - if m.returncode != code: - raise ValueError('mutant %s exited %d, expected %d' % (label, m.returncode, code)) + if m.returncode != code or m.stderr: + raise ValueError('mutant %s: wrong exit or unexpected stderr' % label) + if label == 'M9': + if m.stdout != b'unknown mutant\n': + raise ValueError('M9: wrong unknown-label response') + print('M9: exact unknown-label rejection PASS') + continue + result = strict_json(m.stdout) + if (type(result) is not dict or result.keys() != reference.keys() + or result['passed'] is not False + or any(result[key] != reference[key] + for key in ('object', 'scientific_effect', 'precision_digits')) + or type(result['precision_digits']) is not int + or type(result['checks']) is not dict + or result['checks'].keys() != reference['checks'].keys() + or any(type(value) is not bool for value in result['checks'].values())): + raise ValueError('mutant %s: malformed rejection contract' % label) + failed = {key for key, value in result['checks'].items() if value is False} + if failed != intended[label]: + raise ValueError('mutant %s: wrong failed checks: %r' % (label, sorted(failed))) + print(json.dumps({'mutant': label, 'failed_checks': sorted(failed)}, sort_keys=True)) print('periodic jet replay: manifest, pins, replay and mutants PASS') PY diff --git a/tests/test_periodic_jet_workflow.py b/tests/test_periodic_jet_workflow.py new file mode 100644 index 000000000..2b9cc93ff --- /dev/null +++ b/tests/test_periodic_jet_workflow.py @@ -0,0 +1,195 @@ +"""Exercise the real embedded workflow driver with synthetic subprocess payloads. + +Fixture hashes authenticate only these fixtures. Real jet/source replay remains +separate hosted evidence; this suite does not test Gaussian mathematics. +""" +import hashlib +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import textwrap +import unittest + +ROOT = Path(__file__).resolve().parents[1] +WORKFLOW = ROOT / '.github/workflows/iba1-periodic-jet.yml' +PACKET = Path('reviews/iba1_periodic_jet_claude_20261005') +REJECTIONS = { + 'M1': ['R3_L24_anisotropy_measured', 'R3_L4_anisotropy_visible'], + 'M2': ['R1_reference_d3_trace_frobenius', 'R1_reference_jet', + 'R4_c224_in_side24_interval'], + 'M3': ['R4_c224_in_side24_interval'], + 'M4': ['R1_reference_d3_trace_frobenius', + 'R6_transverse_frame_invariance_d3', 'R6_transverse_frame_invariance_d4'], +} +CHECKS = sorted({k for v in REJECTIONS.values() for k in v} + | {'R2_positive_L%s_d%d' % (p, d) + for p in ('24', '8', '2pi', '4', '3') for d in (2, 3, 4)} + | {'R4_c224_equals_reference'} + | {'R5_quadrature_settled_L%s' % p for p in ('8', '2pi', '4', '3')}) +BASELINE = {'object': 'CL-IBA1-ITEM2-PERIODIC-JET-20261005-v1', + 'precision_digits': 270, 'scientific_effect': 'NONE', + 'cases': {}, 'transverse_rotation_max_abs_change_L4': {}, 'c_2': {}, + 'c_2_diagnostic_small_L': {}, 'checks': dict.fromkeys(CHECKS, True), + 'passed': True} + + +def driver(text): + start = " python -B -S - <<'PY'\n" + if text.count(start) != 1: + raise ValueError('expected one embedded driver') + return textwrap.dedent(text.split(start, 1)[1].split(' PY\n', 1)[0]) + + +def blob(data): + return hashlib.sha1(b'blob %d\0' % len(data) + data).hexdigest() + + +def fixture(root): + packet = root / PACKET + packet.mkdir(parents=True) + baseline = json.dumps(BASELINE, indent=1, sort_keys=True) + '\n' + source = '''import json, os, sys +from pathlib import Path +expected = json.loads(Path('RESULTS.json').read_text()) +rejections = REJECTIONS_LITERAL +label = sys.argv[2] if len(sys.argv) == 3 else 'baseline' +scenario = os.environ.get('PJ_TEST_SCENARIO', 'valid') +active = label == os.environ.get('PJ_TEST_STAGE', 'M1') +if label == 'baseline': + data = Path('RESULTS.json').read_text() + if active and scenario == 'baseline_changed': data += ' ' + if active and scenario == 'baseline_stderr': print('warning', file=sys.stderr) + print(data, end=''); sys.exit(1 if active and scenario == 'wrong_exit' else 0) +if label == 'M9': + if active and scenario == 'crash': raise RuntimeError('synthetic M9 crash') + if active and scenario == 'stderr': print('warning', file=sys.stderr) + print('wrong label' if active and scenario == 'm9_wrong' else 'unknown mutant', + end='' if active and scenario == 'm9_whitespace' else '\\n') + sys.exit(1 if active and scenario == 'wrong_exit' else 2) +expected['passed'] = False +for check in rejections[label]: expected['checks'][check] = False +if active: + if scenario == 'crash': raise RuntimeError('synthetic mutant crash') + if scenario == 'stderr': print('warning', file=sys.stderr) + if scenario == 'empty': sys.exit(1) + if scenario == 'garbage': print('not JSON'); sys.exit(1) + if scenario == 'passed_true': expected['passed'] = True + if scenario == 'passed_int': expected['passed'] = 0 + if scenario == 'wrong_reason': + expected['checks'] = dict.fromkeys(expected['checks'], True) + expected['checks']['R2_positive_L24_d2'] = False + if scenario == 'missing_check': expected['checks'].pop('R2_positive_L24_d2') + if scenario == 'extra_check': expected['checks']['unexpected'] = True + if scenario == 'check_int': expected['checks'][rejections[label][0]] = 0 + if scenario == 'extra_failure': expected['checks']['R2_positive_L24_d2'] = False + if scenario == 'missing_failure': expected['checks'][rejections[label][0]] = True + if scenario == 'wrong_object': expected['object'] = 'different packet' + if scenario == 'wrong_precision': expected['precision_digits'] = 270.0 + if scenario == 'wrong_effect': expected['scientific_effect'] = 'ACCEPT' + if scenario == 'extra_field': expected['unexpected'] = True + if scenario == 'missing_field': expected.pop('cases') + if scenario == 'checks_list': expected['checks'] = [] + if scenario == 'root_list': expected = [] + if scenario == 'nan': expected['cases']['bad'] = float('nan') +data = json.dumps(expected, sort_keys=True) +if active and scenario == 'duplicate': data = data.replace('"passed": false', '"passed": false, "passed": false') +if active and scenario == 'overflow': data = data.replace('"cases": {}', '"cases": {"bad": 1e999}') +if active and scenario == 'trailing': data += ' trailing text' +print(data) +sys.exit(0 if active and scenario == 'wrong_exit' else 1) +'''.replace('REJECTIONS_LITERAL', repr(REJECTIONS)) + payloads = {'NOTE.md': b'Synthetic protocol fixture, not proof evidence.\n', + 'RESULTS.json': baseline.encode(), 'periodic_jet_check.py': source.encode()} + records = [] + for name, data in payloads.items(): + (packet / name).write_bytes(data) + records.append({'path': name, 'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest(), 'git_blob': blob(data)}) + pin = root / 'synthetic-upstream.txt'; pin.write_bytes(b'synthetic upstream\n') + meta = {'files': records, 'pins_on_main': [{'path': pin.name, 'git_blob': blob(pin.read_bytes()), 'kind': 'pins_on_main'}]} + (packet / 'SOURCE_FILES.json').write_text(json.dumps(meta) + '\n') + return packet + + +class PeriodicJetWorkflowTests(unittest.TestCase): + def run_case(self, scenario='valid', stage='M1', mode=(), corrupt=None): + with tempfile.TemporaryDirectory() as td: + root = Path(td); packet = fixture(root) + if corrupt == 'payload': (packet / 'NOTE.md').write_bytes(b'changed\n') + if corrupt == 'pin': (root / 'synthetic-upstream.txt').write_bytes(b'changed\n') + if corrupt == 'extra': (packet / 'extra.txt').write_bytes(b'extra\n') + env = dict(os.environ, FLAGS=' '.join(['-B', *mode, '-S']), + PJ_TEST_SCENARIO=scenario, PJ_TEST_STAGE=stage) + return subprocess.run([sys.executable, '-B', *mode, '-S', '-c', driver(WORKFLOW.read_text())], + cwd=root, env=env, capture_output=True, timeout=15) + + def test_valid_protocol_both_modes(self): + self.assertEqual(len(CHECKS), 27) + for mode in ((), ('-O',)): + with self.subTest(mode=mode): + p = self.run_case(mode=mode) + self.assertEqual(p.returncode, 0, p.stderr.decode()) + self.assertIn(b'periodic jet replay: manifest, pins, replay and mutants PASS', p.stdout) + self.assertEqual(p.stderr, b'') + + def test_crashes_are_not_intended_rejections(self): + for mode in ((), ('-O',)): + for label in REJECTIONS: + with self.subTest(mode=mode, label=label): + p = self.run_case('crash', label, mode) + self.assertNotEqual(p.returncode, 0, 'workflow accepted a mutant crash') + + def check_mutant_contract(self, label): + scenarios = ('stderr', 'empty', 'garbage', 'passed_true', 'passed_int', 'wrong_reason', + 'missing_check', 'extra_check', 'check_int', 'extra_failure', 'missing_failure', + 'wrong_object', 'wrong_precision', 'wrong_effect', 'extra_field', 'missing_field', + 'checks_list', 'root_list', 'nan', 'duplicate', 'overflow', 'trailing', 'wrong_exit') + for mode in ((), ('-O',)): + for case in scenarios: + with self.subTest(mode=mode, label=label, scenario=case): + p = self.run_case(case, label, mode) + self.assertNotEqual(p.returncode, 0, f'workflow accepted {label}/{case}') + + def test_contract_M1(self): + self.check_mutant_contract('M1') + + def test_contract_M2(self): + self.check_mutant_contract('M2') + + def test_contract_M3(self): + self.check_mutant_contract('M3') + + def test_contract_M4(self): + self.check_mutant_contract('M4') + + def test_unknown_label_is_exact(self): + for mode in ((), ('-O',)): + for case in ('m9_wrong', 'm9_whitespace', 'stderr', 'wrong_exit', 'crash'): + with self.subTest(mode=mode, scenario=case): + self.assertNotEqual(self.run_case(case, 'M9', mode).returncode, 0) + + def test_baseline_and_custody_still_fail_closed(self): + for mode in ((), ('-O',)): + for case in ('baseline_changed', 'baseline_stderr', 'wrong_exit'): + with self.subTest(mode=mode, scenario=case): + self.assertNotEqual(self.run_case(case, 'baseline', mode).returncode, 0) + for corrupt in ('payload', 'pin', 'extra'): + with self.subTest(mode=mode, corruption=corrupt): + self.assertNotEqual(self.run_case(mode=mode, corrupt=corrupt).returncode, 0) + + def test_workflow_runs_regressions_and_retains_controls(self): + text = WORKFLOW.read_text() + self.assertIn("- 'tests/test_periodic_jet_workflow.py'", text) + self.assertIn('-m unittest discover -s tests -p test_periodic_jet_workflow.py -v', text) + self.assertIn("flags: ['-B -S', '-B -O -S']", text) + self.assertIn('set -euo pipefail', text) + self.assertIn('contents: read', text) + self.assertIn('persist-credentials: false', text) + self.assertNotIn('continue-on-error', text) + compile(driver(text), str(WORKFLOW), 'exec') + + +if __name__ == '__main__': + unittest.main()