Seal Phase 4AM whole-brain plastic interface audit without learning claim
This commit is contained in:
Digital Drosophila Archive
@@ -0,0 +1,411 @@
|
||||
"""Temporary synthetic seals only; never publish/register any real evidence."""
|
||||
import hashlib
|
||||
import copy
|
||||
import json
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import publish_phase4am as publication
|
||||
|
||||
|
||||
HELPER = Path(__file__).resolve().parents[1] / 'pn-recruitment-audit-v1/publication.py'
|
||||
if not HELPER.is_file():
|
||||
HELPER = publication.ROOT / publication.HELPER_RELATIVE
|
||||
|
||||
|
||||
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, indent=2), 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):
|
||||
temporary = tempfile.TemporaryDirectory(prefix='phase4am-publication-synthetic-')
|
||||
self.addCleanup(temporary.cleanup)
|
||||
self.root = Path(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, not a model\n', encoding='utf-8')
|
||||
self.config = copy.deepcopy(publication.EXPECTED_CONFIG)
|
||||
write(self.model / 'config.json', self.config)
|
||||
self.planned = []
|
||||
for cohort in publication.COHORTS:
|
||||
for mode in publication.MODES:
|
||||
for odor in range(6):
|
||||
for seed in range(11, 16):
|
||||
name = f'{cohort}-{mode}-odor{odor}-seed{seed}'
|
||||
for repeat in ((0, 1) if seed == 11 else (0,)):
|
||||
self.planned.append(dict(name=name, cohort=cohort, mode=mode,
|
||||
odor_id=f'{cohort}-odor{odor}', seed=seed, repeat=repeat))
|
||||
self.registration = dict(config=self.config, planned=self.planned, training=False,
|
||||
candidate_generated=False, before_new_brain_trajectories=True)
|
||||
write(self.experiment / 'protocol/preregistered.json', self.registration)
|
||||
for stage in publication.STAGES:
|
||||
completed = []
|
||||
for row in self.planned:
|
||||
key = row['name'] + '-r' + str(row['repeat'])
|
||||
folder = self.experiment / 'gpu' / stage / key
|
||||
case = self.root / 'experiments/phase4ak-apl-timing-validation/cases' / row['cohort'] / row['name']
|
||||
mode = 'zero' if stage == 'zero' else ('repeat' if row['repeat'] else 'self')
|
||||
left = (self.root / 'experiments/phase4ak-apl-timing-validation/gpu' / row['cohort'] / key
|
||||
if stage == 'zero' else (self.experiment / 'gpu' / stage / (row['name'] + '-r0')
|
||||
if row['repeat'] else folder))
|
||||
write(folder / 'comparison.json', dict(pass_all=True, overlay_name=stage, mode=mode,
|
||||
case_dir=str(case), left=str(left), right=str(folder)))
|
||||
item = folder / 'comparison.json'
|
||||
write(folder / 'output-hashes.json', [dict(path=item.name, bytes=item.stat().st_size, sha256=sha(item))])
|
||||
completed.append(dict(name=row['name'], repeat=row['repeat'],
|
||||
output_hashes_sha256=sha(folder / 'output-hashes.json'), comparison_sha256=sha(item)))
|
||||
write(self.experiment / 'audit' / (stage + '-complete.json'), dict(stage=stage, pass_all=True,
|
||||
trials=144, primary_trials=120, technical_repeats=24, completed=completed, training=False))
|
||||
table = self.experiment / 'summaries/per-MBON-static-effects.parquet'
|
||||
table.parent.mkdir(parents=True)
|
||||
table.write_bytes(b'synthetic table placeholder, never parsed as science')
|
||||
self.science = dict(status=publication.FINAL_STATUS, primary_trials_per_stage=120,
|
||||
technical_repeats_per_stage=24, zero_overlay_exact=True, static25_repeat_exact=True,
|
||||
training=False, learned_candidate=False, conditional_AL_not_a_nonzero_arithmetic_oracle=True,
|
||||
scope=publication.SCIENCE_SCOPE, per_MBON_sha256=sha(table),
|
||||
timing=[dict(stage=s, case=r['name'], wall_seconds=.01, simulation_seconds=.005)
|
||||
for s in publication.STAGES for r in self.planned if r['repeat'] == 0],
|
||||
blocks={c + '/' + m: dict(trials=30, populations={p: dict(trials_event_change=0,
|
||||
trials_count_change=0, mean_count_l1=0., mean_recruited_cells=0., mean_lost_cells=0.)
|
||||
for p in publication.POPULATIONS}) for c in publication.COHORTS for m in publication.MODES})
|
||||
write(self.experiment / 'audit/scientific-summary.json', self.science)
|
||||
self.status = dict(status=publication.FINAL_STATUS, primary_trials_per_stage=120, repeats_per_stage=24,
|
||||
zero_overlay_exact=True, static25_repeat_exact=True, old_integrity_pass=True,
|
||||
frozen_five_reference_pass=True, frozen_eligibility_and_mask_unchanged=True, base_weights_unchanged=True,
|
||||
training=False, learned_candidate=False,
|
||||
scientific_summary_sha256=sha(self.experiment / 'audit/scientific-summary.json'))
|
||||
write(self.experiment / 'audit/final-status.json', self.status)
|
||||
write(self.experiment / 'regression/frozen-five.json', dict(pass_all=True,
|
||||
trials=[dict(trial=name, exact={k: True for k in ('step', 'neuron_idx', 'counts', 'v', 'g')})
|
||||
for name in publication.REFERENCE_NAMES]))
|
||||
write(self.experiment / 'regression/old-integrity-after.json', {'synthetic-old-tree': {'pass': True}})
|
||||
write(self.experiment / 'nested/case/summary.json', {'synthetic': True})
|
||||
(self.experiment / 'REPORT.md').write_text('Synthetic zero-effect diagnostic.\n', encoding='utf-8')
|
||||
(self.experiment / 'tests.log').write_text('Synthetic log\n', encoding='utf-8')
|
||||
for suffix in ('.npy', '.npz', '.parquet'):
|
||||
(self.experiment / ('bulk' + suffix)).write_bytes(b'synthetic bulk bytes, not scientific data')
|
||||
seal(self.model)
|
||||
seal(self.experiment)
|
||||
|
||||
def publish(self):
|
||||
return publication.publish(self.root, HELPER)
|
||||
|
||||
def check_declarations(self):
|
||||
seal(self.experiment)
|
||||
rows = {r['path']: r for r in json.loads((self.experiment / 'manifest.json').read_text())['files']}
|
||||
return publication._status_evidence(publication.load_helpers(HELPER), self.experiment, rows, self.model)
|
||||
|
||||
def write_science(self, science):
|
||||
write(self.experiment / 'audit/scientific-summary.json', science)
|
||||
write(self.experiment / 'audit/final-status.json', dict(self.status,
|
||||
scientific_summary_sha256=sha(self.experiment / 'audit/scientific-summary.json')))
|
||||
|
||||
def test_stage_receipts_require_both_full_144_inventories(self):
|
||||
for stage in publication.STAGES:
|
||||
path = self.experiment / 'audit' / (stage + '-complete.json')
|
||||
original = json.loads(path.read_text())
|
||||
for change in ({'trials': 143}, {'primary_trials': 119}, {'technical_repeats': 23},
|
||||
{'completed': original['completed'][:-1]}, {'pass_all': False}, {'stage': 'other'}):
|
||||
with self.subTest(stage=stage, change=list(change)):
|
||||
write(path, dict(original, **change))
|
||||
with self.assertRaises(ValueError):
|
||||
self.check_declarations()
|
||||
write(path, original)
|
||||
|
||||
def test_stage_receipt_hash_and_order_cannot_hide_wrong_case(self):
|
||||
path = self.experiment / 'audit/static25-complete.json'
|
||||
original = json.loads(path.read_text())
|
||||
for bad in ('hash', 'duplicate', 'reorder'):
|
||||
receipt = copy.deepcopy(original)
|
||||
if bad == 'hash':
|
||||
receipt['completed'][0]['comparison_sha256'] = '0' * 64
|
||||
elif bad == 'duplicate':
|
||||
receipt['completed'][1] = copy.deepcopy(receipt['completed'][0])
|
||||
else:
|
||||
receipt['completed'][0], receipt['completed'][1] = receipt['completed'][1], receipt['completed'][0]
|
||||
with self.subTest(bad=bad):
|
||||
write(path, receipt)
|
||||
with self.assertRaises(ValueError):
|
||||
self.check_declarations()
|
||||
|
||||
def test_reanchored_comparison_cannot_change_stage_case_or_repeat_mode(self):
|
||||
stage_path = self.experiment / 'audit/static25-complete.json'
|
||||
original_stage = json.loads(stage_path.read_text())
|
||||
plan = self.planned[0]
|
||||
folder = self.experiment / 'gpu/static25' / (plan['name'] + '-r0')
|
||||
comparison = folder / 'comparison.json'
|
||||
inventory = folder / 'output-hashes.json'
|
||||
original_report = json.loads(comparison.read_text())
|
||||
for key, value in (('mode', 'repeat'), ('overlay_name', 'zero'), ('case_dir', '/wrong-case'),
|
||||
('left', '/wrong-reference'), ('right', '/wrong-output')):
|
||||
with self.subTest(key=key):
|
||||
write(comparison, dict(original_report, **{key: value}))
|
||||
write(inventory, [dict(path=comparison.name, bytes=comparison.stat().st_size, sha256=sha(comparison))])
|
||||
receipt = copy.deepcopy(original_stage)
|
||||
receipt['completed'][0].update(comparison_sha256=sha(comparison), output_hashes_sha256=sha(inventory))
|
||||
write(stage_path, receipt)
|
||||
with self.assertRaisesRegex(ValueError, 'wrong case/stage/repeat identity'):
|
||||
self.check_declarations()
|
||||
|
||||
def test_preregistered_scientific_coverage_and_first_seed_repeats_required(self):
|
||||
for mode in ('missing', 'duplicate', 'wrong_repeat_seed', 'training'):
|
||||
reg = copy.deepcopy(self.registration)
|
||||
if mode == 'missing':
|
||||
reg['planned'].pop()
|
||||
elif mode == 'duplicate':
|
||||
reg['planned'][-1] = copy.deepcopy(reg['planned'][0])
|
||||
elif mode == 'wrong_repeat_seed':
|
||||
next(r for r in reg['planned'] if r['repeat'] == 1)['seed'] += 1
|
||||
else:
|
||||
reg['training'] = True
|
||||
with self.subTest(mode=mode):
|
||||
write(self.experiment / 'protocol/preregistered.json', reg)
|
||||
with self.assertRaises(ValueError):
|
||||
self.check_declarations()
|
||||
|
||||
def test_scientific_population_metrics_and_explicit_scopes_required(self):
|
||||
for defect in ('missing_population', 'nonfinite', 'negative', 'bool_count', 'more_than30',
|
||||
'count_without_event', 'AL_oracle_claim', 'learning', 'timing_missing', 'table_hash'):
|
||||
science = copy.deepcopy(self.science)
|
||||
pops = science['blocks'][next(iter(science['blocks']))]['populations']
|
||||
if defect == 'missing_population':
|
||||
pops.pop('motor')
|
||||
elif defect in ('nonfinite', 'negative'):
|
||||
pops['KC']['mean_count_l1'] = float('nan') if defect == 'nonfinite' else -1.
|
||||
elif defect in ('bool_count', 'more_than30'):
|
||||
pops['KC']['trials_event_change'] = True if defect == 'bool_count' else 31
|
||||
elif defect == 'count_without_event':
|
||||
pops['KC']['trials_count_change'] = 1
|
||||
elif defect == 'AL_oracle_claim':
|
||||
science['conditional_AL_not_a_nonzero_arithmetic_oracle'] = False
|
||||
elif defect == 'learning':
|
||||
science['learning_achieved'] = True
|
||||
elif defect == 'timing_missing':
|
||||
science['timing'].pop()
|
||||
else:
|
||||
science['per_MBON_sha256'] = '0' * 64
|
||||
with self.subTest(defect=defect):
|
||||
self.write_science(science)
|
||||
with self.assertRaises(ValueError):
|
||||
self.check_declarations()
|
||||
|
||||
def test_five_refs_require_original_names_and_nonempty_exact_keys(self):
|
||||
path = self.experiment / 'regression/frozen-five.json'
|
||||
original = json.loads(path.read_text())
|
||||
for defect in ('duplicate_name', 'unknown_name', 'empty_exact', 'missing_exact', 'non_bool'):
|
||||
refs = copy.deepcopy(original)
|
||||
if defect == 'duplicate_name':
|
||||
refs['trials'][1]['trial'] = refs['trials'][0]['trial']
|
||||
elif defect == 'unknown_name':
|
||||
refs['trials'][0]['trial'] = 'invented-reference'
|
||||
elif defect == 'empty_exact':
|
||||
refs['trials'][0]['exact'] = {}
|
||||
elif defect == 'missing_exact':
|
||||
refs['trials'][0]['exact'].pop('g')
|
||||
else:
|
||||
refs['trials'][0]['exact']['g'] = 1
|
||||
with self.subTest(defect=defect):
|
||||
write(path, refs)
|
||||
with self.assertRaises(ValueError):
|
||||
self.check_declarations()
|
||||
|
||||
def test_completed_zero_effect_diagnostic_is_not_regraded_as_learning(self):
|
||||
before = {str(p): sha(p) for base in (self.model, self.experiment) for p in base.rglob('*') if p.is_file()}
|
||||
result = self.publish()
|
||||
self.assertTrue(result['publication_complete'])
|
||||
receipt = json.loads((self.registry / 'publication-receipt.json').read_text(encoding='utf-8'))
|
||||
self.assertEqual(receipt['declared_audit_status'], publication.FINAL_STATUS)
|
||||
self.assertFalse(receipt['static_effect_size_gate_defined'])
|
||||
self.assertEqual(set(receipt['declared_stage_coverage']), {'zero', 'static25'})
|
||||
self.assertEqual(receipt['declared_audit_counts'], {'primary_trials_per_stage': 120, 'repeats_per_stage': 24})
|
||||
self.assertFalse(receipt['learning_achieved'])
|
||||
self.assertTrue(receipt['publication_success_is_not_scientific_pass'])
|
||||
self.assertEqual(receipt['scientific_summary_sha256'], self.status['scientific_summary_sha256'])
|
||||
for suffix in ('.npy', '.npz', '.parquet'):
|
||||
self.assertFalse((self.registry / ('evidence/bulk' + suffix)).exists())
|
||||
for row in receipt['files']:
|
||||
self.assertEqual(sha(self.registry / row['path']), row['sha256'])
|
||||
self.assertEqual(before, {str(p): sha(p) for base in (self.model, self.experiment) for p in base.rglob('*') if p.is_file()})
|
||||
|
||||
def test_changed_config_blocks_before_destination(self):
|
||||
for key, value in (('phase', '4AK'), ('primary_trials', 119), ('candidate_edges', 14231),
|
||||
('max_fractional_depression', .5), ('training', True), ('virtual_gate_tick', 901),
|
||||
('apl_delay_steps', 0)):
|
||||
with self.subTest(key=key):
|
||||
write(self.model / 'config.json', dict(self.config, **{key: value}))
|
||||
seal(self.model)
|
||||
with self.assertRaisesRegex(ValueError, 'configuration/design'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_summary_hash_counts_and_scope_must_match(self):
|
||||
for changed in ({'scientific_summary_sha256': '0' * 64}, {'primary_trials_per_stage': 119},
|
||||
{'repeats_per_stage': 23}, {'learned_candidate': True}, {'zero_overlay_exact': 'true'},
|
||||
{'old_integrity_pass': False}, {'base_weights_unchanged': False}):
|
||||
with self.subTest(changed=changed):
|
||||
write(self.experiment / 'audit/final-status.json', dict(self.status, **changed))
|
||||
seal(self.experiment)
|
||||
with self.assertRaises(ValueError):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_missing_or_corrupt_scientific_summary_is_rejected(self):
|
||||
path = self.experiment / 'audit/scientific-summary.json'
|
||||
path.unlink()
|
||||
seal(self.experiment)
|
||||
with self.assertRaisesRegex(ValueError, 'Required finalized evidence absent'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_complete_status_requires_all_four_complete_blocks(self):
|
||||
value = dict(self.science, blocks={})
|
||||
write(self.experiment / 'audit/scientific-summary.json', value)
|
||||
write(self.experiment / 'audit/final-status.json', dict(self.status,
|
||||
scientific_summary_sha256=sha(self.experiment / 'audit/scientific-summary.json')))
|
||||
seal(self.experiment)
|
||||
with self.assertRaisesRegex(ValueError, 'cohort-mode blocks'):
|
||||
self.publish()
|
||||
|
||||
def test_false_reference_exactness_is_not_hidden_by_pass_all(self):
|
||||
write(self.experiment / 'regression/frozen-five.json', dict(pass_all=True,
|
||||
trials=[dict(exact={'step': False})] * 5))
|
||||
seal(self.experiment)
|
||||
with self.assertRaisesRegex(ValueError, 'five exact trial'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_unknown_or_unfinalized_status_is_not_promoted(self):
|
||||
for status in ('INCOMPLETE-NOT-SEALED', 'PASS-FROZEN-D18-INDEPENDENT-VALIDATION'):
|
||||
write(self.experiment / 'audit/final-status.json', dict(self.status, status=status))
|
||||
seal(self.experiment)
|
||||
with self.assertRaisesRegex(ValueError, 'Unrecognized Phase 4AM'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_manifest_path_escape_rejected_before_destination(self):
|
||||
path = self.experiment / 'manifest.json'
|
||||
manifest = json.loads(path.read_text(encoding='utf-8'))
|
||||
manifest['files'][0]['path'] = '../escape.json'
|
||||
write(path, manifest)
|
||||
controls = self.experiment / 'checksums.sha256'
|
||||
controls.write_text('\n'.join([r['sha256'] + ' ' + r['path'] for r in manifest['files']] +
|
||||
[sha(path) + ' manifest.json']) + '\n', encoding='utf-8')
|
||||
with self.assertRaisesRegex(ValueError, 'escaping relative path'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_completed_status_without_frozen_reference_is_not_publishable(self):
|
||||
(self.experiment / 'regression/frozen-five.json').unlink()
|
||||
seal(self.experiment)
|
||||
with self.assertRaisesRegex(ValueError, 'Required finalized evidence absent'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_existing_registry_never_overwritten_or_resumed(self):
|
||||
self.registry.mkdir(parents=True)
|
||||
sentinel = self.registry / 'keep.txt'
|
||||
sentinel.write_bytes(b'keep these bytes')
|
||||
with self.assertRaisesRegex(ValueError, 'Registry already exists'):
|
||||
self.publish()
|
||||
self.assertEqual(sentinel.read_bytes(), b'keep these bytes')
|
||||
self.assertEqual(list(self.registry.iterdir()), [sentinel])
|
||||
|
||||
def test_missing_integrity_report_blocks_before_destination(self):
|
||||
(self.experiment / 'regression/old-integrity-after.json').unlink()
|
||||
seal(self.experiment)
|
||||
with self.assertRaisesRegex(ValueError, 'Required finalized evidence absent'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_bulk_corruption_blocks_even_though_bulk_is_not_copied(self):
|
||||
path = self.experiment / 'bulk.npy'
|
||||
path.write_bytes(b'x' * path.stat().st_size)
|
||||
with self.assertRaisesRegex(ValueError, 'Payload SHA-256 mismatch'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_unlisted_payload_and_bad_manifest_self_hash_fail(self):
|
||||
extra = self.experiment / 'unlisted.csv'
|
||||
extra.write_text('unlisted\n')
|
||||
with self.assertRaisesRegex(ValueError, 'unlisted/missing'):
|
||||
self.publish()
|
||||
extra.unlink()
|
||||
controls = self.experiment / 'checksums.sha256'
|
||||
lines = controls.read_text().splitlines()
|
||||
lines[-1] = '0' * 64 + ' manifest.json'
|
||||
controls.write_text('\n'.join(lines) + '\n')
|
||||
with self.assertRaisesRegex(ValueError, 'manifest self-checksum'):
|
||||
self.publish()
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
def test_large_json_verified_but_not_copied(self):
|
||||
path = self.experiment / 'large.json'
|
||||
path.write_bytes(b' ' * (publication.MAX_EVIDENCE_BYTES + 1))
|
||||
seal(self.experiment)
|
||||
self.publish()
|
||||
self.assertFalse((self.registry / 'evidence/large.json').exists())
|
||||
receipt = json.loads((self.registry / 'publication-receipt.json').read_text(encoding='utf-8'))
|
||||
self.assertEqual(next(r for r in receipt['experiment_payload_not_copied'] if r['path'] == 'large.json')['reason'],
|
||||
'exceeds_4_MiB')
|
||||
|
||||
def test_interrupted_copy_retained_and_cannot_resume(self):
|
||||
with mock.patch.object(publication, '_copy', side_effect=OSError('synthetic interrupted copy')):
|
||||
with self.assertRaisesRegex(OSError, 'synthetic interrupted copy'):
|
||||
self.publish()
|
||||
self.assertTrue((self.registry / 'publication-failure.json').is_file())
|
||||
self.assertFalse((self.registry / 'publication-receipt.json').exists())
|
||||
with self.assertRaisesRegex(ValueError, 'Registry already exists'):
|
||||
self.publish()
|
||||
|
||||
def test_after_copy_full_verification_detects_bulk_change(self):
|
||||
original = publication._copy
|
||||
changed = False
|
||||
|
||||
def copy_then_change(helper, source, destination, size, checksum):
|
||||
nonlocal changed
|
||||
original(helper, source, destination, size, checksum)
|
||||
if not changed:
|
||||
changed = True
|
||||
(self.experiment / 'bulk.npy').write_bytes(b'changed after initial seal check')
|
||||
|
||||
with mock.patch.object(publication, '_copy', side_effect=copy_then_change):
|
||||
with self.assertRaises(ValueError):
|
||||
self.publish()
|
||||
self.assertTrue((self.registry / 'publication-failure.json').is_file())
|
||||
self.assertFalse((self.registry / 'publication-receipt.json').exists())
|
||||
|
||||
def test_helper_pin_checked_without_modifying_old_globals_or_cache(self):
|
||||
helper = publication.load_helpers(HELPER)
|
||||
self.assertEqual(helper.MODEL_RELATIVE, Path('models/male-cns-lif/pn-recruitment-audit-v1'))
|
||||
self.assertEqual(helper.FINAL_STATUS, 'PASS-CONDITIONAL-PN-APL-SOURCE-AUDIT')
|
||||
wrong = self.root / 'wrong-helper.py'
|
||||
wrong.write_bytes(b'raise RuntimeError("must never execute")\n')
|
||||
with self.assertRaisesRegex(ValueError, 'helper hash mismatch'):
|
||||
publication.publish(self.root, wrong)
|
||||
self.assertFalse(self.registry.exists())
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user