Digital Drosophila Archive
206 lines
10 KiB
Python
206 lines
10 KiB
Python
"""Only temporary synthetic seals are used; production publication is never run."""
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import publication
|
|
|
|
|
|
def _sha(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _write(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value), encoding='utf-8')
|
|
|
|
|
|
def _seal(root):
|
|
rows = [dict(path=path.relative_to(root).as_posix(), bytes=path.stat().st_size, sha256=_sha(path))
|
|
for path in sorted(root.rglob('*')) if path.is_file() and
|
|
path.name not in ('manifest.json', 'checksums.sha256') and '__pycache__' not in path.parts]
|
|
_write(root / 'manifest.json', dict(root=str(root), files=rows))
|
|
lines = [row['sha256'] + ' ' + row['path'] for row in rows]
|
|
lines.append(_sha(root / 'manifest.json') + ' manifest.json')
|
|
(root / 'checksums.sha256').write_text('\n'.join(lines) + '\n', encoding='utf-8')
|
|
|
|
|
|
class PublicationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory(prefix='phase4ai-publication-unit-')
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name).resolve()
|
|
self.model = self.root / publication.MODEL_RELATIVE
|
|
self.experiment = self.root / publication.EXPERIMENT_RELATIVE
|
|
self.registry = self.root / publication.REGISTRY_RELATIVE
|
|
self.model.mkdir(parents=True)
|
|
self.experiment.mkdir(parents=True)
|
|
(self.model / 'module.py').write_text('# synthetic\n', encoding='utf-8')
|
|
status = dict(status=publication.FINAL_STATUS, pass_all=True, candidate_generated=False,
|
|
learning_achieved=False, new_brain_simulation=False, GPU_used=False)
|
|
_write(self.experiment / 'audit/final-status.json', status)
|
|
_write(self.experiment / 'trials/example/summary.json', dict(synthetic=True))
|
|
_write(self.experiment / 'trials/example/apl-summary.json', dict(synthetic=True))
|
|
(self.experiment / 'REPORT.md').write_text('Synthetic evidence.\n', encoding='utf-8')
|
|
(self.experiment / 'unit-tests.log').write_text('Synthetic tests passed.\n', encoding='utf-8')
|
|
(self.experiment / 'counts.npy').write_bytes(b'not-a-real-array')
|
|
_seal(self.model)
|
|
_seal(self.experiment)
|
|
|
|
def test_all_payload_and_checksum_self_hash_are_verified(self):
|
|
result = publication.verify_seal(self.experiment)
|
|
self.assertTrue(result['pass_all'])
|
|
self.assertEqual(result['payload_file_count'], 6)
|
|
self.assertEqual(result['manifest_sha256'], _sha(self.experiment / 'manifest.json'))
|
|
|
|
def test_publication_only_copies_small_evidence_and_leaves_sources_unchanged(self):
|
|
before = {str(path): _sha(path) for base in (self.model, self.experiment)
|
|
for path in base.rglob('*') if path.is_file()}
|
|
outcome = publication.publish(self.root)
|
|
self.assertTrue(outcome['publication_complete'])
|
|
self.assertFalse(outcome['scientific_gate_executed_by_publisher'])
|
|
self.assertTrue((self.registry / 'evidence/trials/example/apl-summary.json').is_file())
|
|
self.assertTrue((self.registry / 'evidence/unit-tests.log').is_file())
|
|
self.assertFalse((self.registry / 'evidence/counts.npy').exists())
|
|
self.assertFalse((self.registry / 'model/module.py').exists())
|
|
self.assertTrue((self.registry / 'model/manifest.json').is_file())
|
|
receipt = json.loads((self.registry / 'publication-receipt.json').read_text())
|
|
self.assertFalse(receipt['complete_bulk_backup'])
|
|
self.assertFalse(receipt['scientific_gate_certified_by_publication'])
|
|
for row in receipt['files']:
|
|
path = self.registry / row['path']
|
|
self.assertEqual(path.stat().st_size, row['bytes'])
|
|
self.assertEqual(_sha(path), row['sha256'])
|
|
for name, checksum in before.items():
|
|
self.assertEqual(_sha(Path(name)), checksum)
|
|
|
|
def test_existing_registry_is_never_overwritten_or_resumed(self):
|
|
self.registry.mkdir(parents=True)
|
|
sentinel = self.registry / 'keep.txt'
|
|
sentinel.write_text('keep', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'Registry already exists'):
|
|
publication.publish(self.root)
|
|
self.assertEqual(sentinel.read_text(), 'keep')
|
|
self.assertEqual(list(self.registry.iterdir()), [sentinel])
|
|
|
|
def test_bulk_payload_corruption_blocks_before_registry_creation(self):
|
|
(self.experiment / 'counts.npy').write_bytes(b'corrupt')
|
|
with self.assertRaisesRegex(ValueError, 'Payload byte count mismatch'):
|
|
publication.publish(self.root)
|
|
self.assertFalse(self.registry.exists())
|
|
|
|
def test_same_size_payload_hash_corruption_is_detected(self):
|
|
path = self.experiment / 'counts.npy'
|
|
path.write_bytes(b'x' * path.stat().st_size)
|
|
with self.assertRaisesRegex(ValueError, 'Payload SHA-256 mismatch'):
|
|
publication.verify_seal(self.experiment)
|
|
|
|
def test_missing_payload_or_unlisted_file_is_detected(self):
|
|
(self.experiment / 'extra.txt').write_text('unlisted', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'unlisted/missing'):
|
|
publication.verify_seal(self.experiment)
|
|
|
|
def test_wrong_manifest_self_checksum_is_detected(self):
|
|
path = self.experiment / 'checksums.sha256'
|
|
lines = path.read_text().splitlines()
|
|
lines[-1] = '0' * 64 + ' manifest.json'
|
|
path.write_text('\n'.join(lines) + '\n')
|
|
with self.assertRaisesRegex(ValueError, 'manifest self-checksum'):
|
|
publication.verify_seal(self.experiment)
|
|
|
|
def test_duplicate_checksum_path_is_detected(self):
|
|
path = self.experiment / 'checksums.sha256'
|
|
text = path.read_text()
|
|
path.write_text(text + text.splitlines()[0] + '\n')
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate checksum'):
|
|
publication.verify_seal(self.experiment)
|
|
|
|
def test_manifest_root_mismatch_is_detected(self):
|
|
path = self.experiment / 'manifest.json'
|
|
value = json.loads(path.read_text())
|
|
value['root'] = str(self.model)
|
|
_write(path, value)
|
|
with self.assertRaisesRegex(ValueError, 'another root'):
|
|
publication.verify_seal(self.experiment)
|
|
|
|
def test_manifest_duplicate_path_and_traversal_are_rejected(self):
|
|
manifest = self.experiment / 'manifest.json'
|
|
original = json.loads(manifest.read_text())
|
|
for bad_path in ('../escape.txt', '/absolute', 'a/../escape', 'a\\escape', 'C:/escape'):
|
|
with self.subTest(path=bad_path):
|
|
value = json.loads(json.dumps(original))
|
|
value['files'][0]['path'] = bad_path
|
|
_write(manifest, value)
|
|
with self.assertRaises(ValueError):
|
|
publication.verify_seal(self.experiment)
|
|
original['files'].append(original['files'][0].copy())
|
|
_write(manifest, original)
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate manifest path'):
|
|
publication.verify_seal(self.experiment)
|
|
|
|
def test_explicitly_excluded_pycache_is_not_published(self):
|
|
cache = self.experiment / '__pycache__/unused.pyc'
|
|
cache.parent.mkdir()
|
|
cache.write_bytes(b'cache')
|
|
self.assertEqual(publication.verify_seal(self.experiment)['excluded_pycache_files'], 1)
|
|
publication.publish(self.root)
|
|
self.assertFalse((self.registry / 'evidence/__pycache__').exists())
|
|
|
|
def test_file_size_limit_skips_large_json_without_omitting_verification(self):
|
|
huge = self.experiment / 'huge.json'
|
|
huge.write_bytes(b' ' * (publication.MAX_EVIDENCE_BYTES + 1))
|
|
_seal(self.experiment)
|
|
publication.publish(self.root)
|
|
self.assertFalse((self.registry / 'evidence/huge.json').exists())
|
|
receipt = json.loads((self.registry / 'publication-receipt.json').read_text())
|
|
omitted = {row['path']: row for row in receipt['experiment_payload_not_copied']}
|
|
self.assertEqual(omitted['huge.json']['reason'], 'exceeds_4_MiB')
|
|
|
|
def test_unexpected_final_status_or_scope_flag_blocks_publication(self):
|
|
path = self.experiment / 'audit/final-status.json'
|
|
original = json.loads(path.read_text())
|
|
for field, value in [('status', 'OTHER-PASS'), ('pass_all', False), ('learning_achieved', True),
|
|
('GPU_used', True), ('new_brain_simulation', True), ('candidate_generated', True)]:
|
|
with self.subTest(field=field):
|
|
bad = dict(original)
|
|
bad[field] = value
|
|
_write(path, bad)
|
|
_seal(self.experiment)
|
|
with self.assertRaises(ValueError):
|
|
publication.publish(self.root)
|
|
self.assertFalse(self.registry.exists())
|
|
|
|
def test_copy_failure_is_retained_without_success_receipt(self):
|
|
with mock.patch.object(publication, '_copy_verified', side_effect=OSError('synthetic interruption')):
|
|
with self.assertRaisesRegex(OSError, 'synthetic interruption'):
|
|
publication.publish(self.root)
|
|
self.assertTrue(self.registry.is_dir())
|
|
self.assertTrue((self.registry / 'publication-failure.json').is_file())
|
|
self.assertFalse((self.registry / 'publication-receipt.json').exists())
|
|
with self.assertRaisesRegex(ValueError, 'Registry already exists'):
|
|
publication.publish(self.root)
|
|
|
|
def test_source_change_during_copy_prevents_success(self):
|
|
original = publication._copy_verified
|
|
changed = False
|
|
|
|
def copy_then_modify(source, target, byte_count, checksum):
|
|
nonlocal changed
|
|
original(source, target, byte_count, checksum)
|
|
if not changed:
|
|
changed = True
|
|
(self.experiment / 'counts.npy').write_bytes(b'deliberately-changed-after-initial-verification')
|
|
|
|
with mock.patch.object(publication, '_copy_verified', side_effect=copy_then_modify):
|
|
with self.assertRaisesRegex(ValueError, 'Payload byte count mismatch'):
|
|
publication.publish(self.root)
|
|
self.assertFalse((self.registry / 'publication-receipt.json').exists())
|
|
self.assertTrue((self.registry / 'publication-failure.json').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|