"""Temporary Git repositories only; production verification performs no writes.""" import contextlib import hashlib import io import json from pathlib import Path import subprocess import tempfile import unittest import verify_git_payload as v def data(value): return (json.dumps(value, sort_keys=True, indent=2) + '\n').encode() def row(name, raw): return dict(path=name, bytes=len(raw), sha256=v.sha(raw)) class TemporaryGit(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.repo = Path(self.temp.name).resolve() self.git('init', '-q') self.git('config', 'user.name', 'Synthetic Test') self.git('config', 'user.email', 'synthetic@example.invalid') self.git('config', 'core.autocrlf', 'false') self.pin = self.fixture() self.stage() def tearDown(self): self.temp.cleanup() def git(self, *args, input=None): result = subprocess.run(['git', '-C', str(self.repo), *args], input=input, stdout=subprocess.PIPE, stderr=subprocess.PIPE) self.assertEqual(result.returncode, 0, result.stderr.decode('utf-8', 'replace')) return result.stdout def write(self, path, raw): target = self.repo / path; target.parent.mkdir(parents=True, exist_ok=True); target.write_bytes(raw) def stage(self): self.git('add', '--', v.MODEL, v.REGISTRY) def fixture(self, status='PASS-EXACT-LIGHT-CONTINUATION-ENGINEERING-NOT-LEARNING'): payload, source_rows = {}, [] source_sha, input_sha = {}, {} for name in v.SOURCES: raw = ('# synthetic source ' + name + '\n').encode() sealed = 'protocol/source-snapshot/' + name payload[sealed] = raw self.write(v.MODEL + '/' + name, raw) source_sha[v.PROJECT + '/' + v.MODEL + '/' + name] = v.sha(raw) input_sha[v.PROJECT + '/' + v.EXPERIMENT + '/' + sealed] = v.sha(raw) source_rows.append(dict(model_source=v.MODEL + '/' + name, sealed_source=v.EXPERIMENT + '/' + sealed, copy='source-snapshot/' + name, bytes=len(raw), sha256=v.sha(raw))) payload['protocol/preregistered.json'] = data(dict(schema='phase4an-engineering-registration-v1', training=False, source_sha256=source_sha, input_sha256=input_sha)) payload['audit/final-status.json'] = data(dict(status=status)) payload['audit/engineering-analysis.json'] = data(dict(status=status)) payload['audit/small.json'] = data({'synthetic': True}) payload['runs/a/state.npz'] = b'not copied bulk' payload['runs/a/queue.jsonl'] = b'not copied jsonl\n' payload['extension-build/provenance.json'] = b'not copied build' rows = [row(name, raw) for name, raw in sorted(payload.items())] manifest = data(dict(root=v.PROJECT + '/' + v.EXPERIMENT, files=rows)) checksums = ''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows) checksums = (checksums + v.sha(manifest) + ' manifest.json\n').encode() all_data = {**payload, 'manifest.json': manifest, 'checksums.sha256': checksums} index, omitted = [], [] for name, raw in all_data.items(): current = row(name, raw); destination, reason = v._selection(current) if destination is None: omitted.append(dict(current, reason=reason)) else: self.write(v.REGISTRY + '/' + destination, raw) index.append(dict(source=v.EXPERIMENT + '/' + name, copy=destination, bytes=len(raw), sha256=v.sha(raw))) self.write(v.REGISTRY + '/evidence-index.json', data(index)) self.write(v.REGISTRY + '/source-identity.json', data(source_rows)) self.write(v.REGISTRY + '/seal-verification.json', data(dict(pass_all=True, root=v.PROJECT + '/' + v.EXPERIMENT, manifest_sha256=v.sha(manifest), checksums_sha256=v.sha(checksums), payload_file_count=len(rows), payload_bytes=sum(r['bytes'] for r in rows)))) directory = self.repo / v.REGISTRY receipt = dict(schema='phase4an-engineering-publication-v1', publication_complete=True, registry=v.REGISTRY, project_root=v.PROJECT, bulk_remains_at=v.EXPERIMENT, declared_status=status, model_directory_sealed=False, used_source_count=21, model_files_copied=False, complete_bulk_backup=False, training=False, candidate_generated=False, learning_pass_claimed=False, source_identity='source-identity.json', experiment_manifest_sha256=v.sha(manifest), experiment_checksums_sha256=v.sha(checksums), preregistration_sha256=v.sha(payload['protocol/preregistered.json']), final_sha256=v.sha(payload['audit/final-status.json']), analysis_sha256=v.sha(payload['audit/engineering-analysis.json']), copied_evidence_files=len(index), copied_evidence_bytes=sum(r['bytes'] for r in index), experiment_payload_not_copied=omitted, files=[row(path.relative_to(directory).as_posix(), path.read_bytes()) for path in sorted(directory.rglob('*')) if path.is_file() and path.name != 'publication-receipt.json']) raw = data(receipt); self.write(v.REGISTRY + '/publication-receipt.json', raw) return v.sha(raw) def refresh_receipt(self, **updates): path = self.repo / v.REGISTRY / 'publication-receipt.json' receipt = json.loads(path.read_bytes()); receipt.update(updates) receipt['files'] = [row(r['path'], (self.repo / v.REGISTRY / r['path']).read_bytes()) for r in receipt['files']] raw = data(receipt); path.write_bytes(raw); self.pin = v.sha(raw); self.stage() def test_index_and_commit_pass_without_touching_index(self): before = (self.repo / '.git/index').read_bytes() result = v.verify(self.repo, self.pin, staged=True) self.assertTrue(result['pass_all']) self.assertFalse(result['working_tree_payload_used']) self.assertFalse(result['experiment_bulk_verified_by_this_tool']) self.assertEqual(result['model_source_files'], 21) self.assertEqual((self.repo / '.git/index').read_bytes(), before) self.git('commit', '-qm', 'synthetic publication') committed = v.verify(self.repo, self.pin, commit='HEAD') self.assertEqual(committed['source'], 'commit') self.assertEqual(committed['commit'], self.git('rev-parse', 'HEAD').decode().strip()) def test_working_tree_never_substitutes_for_staged_or_committed_blobs(self): target = v.MODEL + '/' + v.SOURCES[0] self.git('commit', '-qm', 'valid synthetic') original = (self.repo / target).read_bytes() self.write(target, b'bad working tree but index is valid\n') self.assertTrue(v.verify(self.repo, self.pin, staged=True)['pass_all']) self.git('add', '--', target) self.write(target, original) with self.assertRaisesRegex(ValueError, 'SHA/bytes'): v.verify(self.repo, self.pin, staged=True) self.assertTrue(v.verify(self.repo, self.pin, commit='HEAD')['pass_all']) def test_external_receipt_sha_and_exclusive_source_mode_are_required(self): for kwargs, pin in [({'staged': True}, '0' * 64), ({}, self.pin), ({'staged': True, 'commit': 'HEAD'}, self.pin)]: with self.subTest(kwargs=kwargs), self.assertRaises(ValueError): v.verify(self.repo, pin, **kwargs) def test_extra_scientific_draft_and_registry_file_are_rejected(self): for name in (v.MODEL + '/PROTOCOL_DRAFT.md', v.REGISTRY + '/unlisted.json'): self.write(name, b'not in registered inventory\n'); self.git('add', '--', name) with self.subTest(name=name), self.assertRaises(ValueError): v.verify(self.repo, self.pin, staged=True) self.git('rm', '--cached', '--', name) (self.repo / name).unlink() def test_missing_source_and_nonregular_index_entry_rejected(self): path = v.MODEL + '/' + v.SOURCES[0] self.git('rm', '--cached', '--', path) with self.assertRaises(ValueError): v.verify(self.repo, self.pin, staged=True) self.git('add', '--', path) oid = self.git('hash-object', '-w', '--stdin', input=b'elsewhere\n').decode().strip() self.git('update-index', '--cacheinfo', '120000,' + oid + ',' + path) with self.assertRaisesRegex(ValueError, 'nonregular'): v.verify(self.repo, self.pin, staged=True) def test_source_identity_semantics_checked_even_under_new_receipt_pin(self): path = self.repo / v.REGISTRY / 'source-identity.json' identities = json.loads(path.read_bytes()); identities[0]['sha256'] = 'f' * 64 path.write_bytes(data(identities)); self.refresh_receipt() with self.assertRaisesRegex(ValueError, 'Source/snapshot'): v.verify(self.repo, self.pin, staged=True) def test_evidence_index_source_and_escaping_receipt_paths_fail(self): path = self.repo / v.REGISTRY / 'evidence-index.json' index = json.loads(path.read_bytes()); index[0]['source'] = 'wrong/source.py' path.write_bytes(data(index)); self.refresh_receipt() with self.assertRaisesRegex(ValueError, 'Evidence index'): v.verify(self.repo, self.pin, staged=True) receipt_path = self.repo / v.REGISTRY / 'publication-receipt.json' receipt = json.loads(receipt_path.read_bytes()); receipt['files'][0]['path'] = '../escape.json' raw = data(receipt); receipt_path.write_bytes(raw); self.pin = v.sha(raw); self.stage() with self.assertRaisesRegex(ValueError, 'escaping'): v.verify(self.repo, self.pin, staged=True) def test_checksums_manifest_self_hash_is_not_just_trusted(self): path = self.repo / v.REGISTRY / 'evidence/checksums.sha256' text = path.read_text(); lines = text.splitlines(); lines[-1] = '0' * 64 + ' manifest.json' raw = ('\n'.join(lines) + '\n').encode(); path.write_bytes(raw) self.refresh_receipt(experiment_checksums_sha256=v.sha(raw)) with self.assertRaisesRegex(ValueError, 'Checksum inventory'): v.verify(self.repo, self.pin, staged=True) def test_declared_fail_status_is_preserved_not_relabelled_science_pass(self): self.pin = self.fixture(status='FAIL-NUMERICAL'); self.stage() result = v.verify(self.repo, self.pin, staged=True) self.assertTrue(result['pass_all']) self.assertEqual(result['declared_status'], 'FAIL-NUMERICAL') self.assertTrue(result['publication_integrity_is_not_scientific_pass']) self.assertFalse(result['scientific_gate_executed']) def test_duplicate_JSON_field_in_receipt_is_rejected(self): path = self.repo / v.REGISTRY / 'publication-receipt.json' raw = path.read_bytes().replace(b'{', b'{"publication_complete":true,', 1) path.write_bytes(raw); self.pin = v.sha(raw); self.stage() with self.assertRaisesRegex(ValueError, 'Duplicate JSON'): v.verify(self.repo, self.pin, staged=True) def test_cli_emits_compact_summary_without_full_payload(self): out = io.StringIO() with contextlib.redirect_stdout(out): status = v.main(['--repo', str(self.repo), '--receipt-sha256', self.pin, '--staged']) self.assertEqual(status, 0) result = json.loads(out.getvalue()) self.assertTrue(result['pass_all']) self.assertNotIn('verified_blobs', result) self.assertLess(len(out.getvalue()), 1800) if __name__ == '__main__': unittest.main()