Digital Drosophila Archive
400 lines
23 KiB
Python
400 lines
23 KiB
Python
"""CPU synthetic checkpoint/clock tests; not a brain or GPU equivalence test."""
|
|
import ast
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
import runtime_continuation as c
|
|
|
|
|
|
R = c.R
|
|
AJ = Path(__file__).resolve().parent.parent / 'apl-timing-lab-v1/gpu_timing.py'
|
|
|
|
|
|
class RingTests(unittest.TestCase):
|
|
def execute(self, start, steps, raw, phase, apl_pending, events, *, admit_global=True):
|
|
ring = c.rotate_spike_ring(raw, phase)
|
|
apl = apl_pending.copy()
|
|
due, applied = [], []
|
|
for tick in range(steps):
|
|
absolute = start + tick
|
|
ring[tick % 19] = events[absolute]
|
|
admitted = absolute >= 18 if admit_global else tick >= 18
|
|
due.append(ring[(tick - 18) % 19].copy() if admitted else np.zeros(ring.shape[1], np.uint8))
|
|
slot = tick % 18
|
|
applied.append(apl[slot].copy())
|
|
apl[slot] = np.float32(absolute + 1)
|
|
head = steps % 18
|
|
return ring, steps, np.concatenate((apl[head:], apl[:head])), np.asarray(due), np.asarray(applied)
|
|
|
|
def test_multi_segments_preserve_all_future_arrivals_and_apl_slots(self):
|
|
rng = np.random.default_rng(193)
|
|
events = rng.integers(0, 2, (80, 4), dtype=np.uint8)
|
|
initial, pending = np.zeros((19, 4), np.uint8), np.zeros((18, 2), np.float32)
|
|
full = self.execute(0, 80, initial, 0, pending, events)
|
|
for lengths in ((5, 13, 1, 22, 39), (17, 1, 1, 18, 43), (19, 19, 19, 23), (1,) * 80):
|
|
ring, phase, tail, offset, due, applied = initial, 0, pending, 0, [], []
|
|
for steps in lengths:
|
|
ring, phase, tail, d, a = self.execute(offset, steps, ring, phase, tail, events)
|
|
due.append(d); applied.append(a); offset += steps
|
|
np.testing.assert_array_equal(np.concatenate(due), full[3])
|
|
np.testing.assert_array_equal(np.concatenate(applied), full[4])
|
|
np.testing.assert_array_equal(c.rotate_spike_ring(ring, phase), c.rotate_spike_ring(full[0], 80))
|
|
np.testing.assert_array_equal(tail, full[2])
|
|
|
|
def test_parent_shorter_than_delay_needs_global_admission(self):
|
|
events = np.zeros((24, 2), np.uint8); events[0, 1] = 1
|
|
parent = self.execute(0, 5, np.zeros((19, 2), np.uint8), 0, np.zeros((18, 2), np.float32), events)
|
|
child = self.execute(5, 19, *parent[:3], events)
|
|
wrong = self.execute(5, 19, *parent[:3], events, admit_global=False)
|
|
self.assertEqual(child[3][13, 1], 1) # absolute tick 18, local 13
|
|
self.assertEqual(wrong[3][13, 1], 0)
|
|
|
|
def test_local_tick_zero_request_uses_global_clock_and_ring_phase(self):
|
|
events = np.zeros((50, 3), np.uint8)
|
|
events[19, 2] = 1 # Synthetic US request at child's local tick zero.
|
|
first = self.execute(0, 19, np.zeros((19, 3), np.uint8), 0, np.zeros((18, 2), np.float32), events)
|
|
child = self.execute(19, 31, *first[:3], events)
|
|
self.assertEqual(child[3][18, 2], 1)
|
|
self.assertEqual(np.count_nonzero(child[3][:18, 2]), 0)
|
|
|
|
def test_rotation_is_copy_and_rejects_invalid_ring(self):
|
|
raw = np.zeros((19, 2), np.uint8); raw[18, 1] = 1
|
|
before = raw.tobytes()
|
|
result = c.rotate_spike_ring(raw, 19)
|
|
result[:] = 1
|
|
self.assertEqual(raw.tobytes(), before)
|
|
for value, phase in ((raw.astype(np.int32), 0), (raw[:18], 0), (raw, -1), (raw, True)):
|
|
with self.assertRaises(ValueError):
|
|
c.rotate_spike_ring(value, phase)
|
|
|
|
|
|
class ASTTests(unittest.TestCase):
|
|
def test_only_restore_and_global_admission_edits_in_both_modes(self):
|
|
for mode in ('light', 'reference'):
|
|
tree, audit = c.derive_continued(AJ.read_bytes(), mode)
|
|
self.assertTrue(audit['full_AST_rollback_identical'])
|
|
self.assertTrue(audit['kernels_sorting_and_local_ring_math_unchanged'])
|
|
self.assertEqual(sum(isinstance(node, ast.If) and ast.unparse(node.test) == c.CLOCK_NEW for node in ast.walk(tree)), 1)
|
|
self.assertEqual(sum(isinstance(node, ast.Expr) and ast.unparse(node) == c.RESTORE for node in ast.walk(tree)), 1)
|
|
compile(tree, '<pure continuation AST>', 'exec')
|
|
|
|
def test_reference_loop_diff_is_exactly_admission_condition(self):
|
|
base, _ = R.derive_ast(AJ.read_bytes(), 'reference')
|
|
actual, _ = c.derive_continued(AJ.read_bytes(), 'reference')
|
|
loops = lambda tree: next(node for node in ast.walk(tree) if isinstance(node, ast.For) and
|
|
isinstance(node.target, ast.Name) and node.target.id == 'tick')
|
|
modified = copy.deepcopy(loops(actual))
|
|
changed = [node for node in ast.walk(modified) if isinstance(node, ast.If) and ast.unparse(node.test) == c.CLOCK_NEW]
|
|
changed[0].test = ast.parse(c.CLOCK_OLD, mode='eval').body
|
|
self.assertEqual(R._dump(modified), R._dump(loops(base)))
|
|
|
|
def test_compile_never_mutates_sealed_module_namespace(self):
|
|
module = types.ModuleType('fake_aj'); module.__file__ = str(AJ)
|
|
original = dict(module.__dict__)
|
|
runtime = types.SimpleNamespace(write_json=lambda *_: None, audit_sequence=lambda *_: None)
|
|
function, _ = c.compile_continued(module, runtime, 'light')
|
|
self.assertEqual(module.__dict__, original)
|
|
self.assertEqual(function.__globals__['__file__'], str(Path(c.__file__).resolve()))
|
|
self.assertIs(function.__globals__['_continuation'], runtime)
|
|
|
|
|
|
class ArrayTarget:
|
|
def __init__(self, value):
|
|
self.value = np.array(value, copy=True)
|
|
|
|
def copy_(self, value):
|
|
self.value[...] = value
|
|
|
|
def any(self):
|
|
return np.asarray(self.value.any())
|
|
|
|
|
|
class RestoreTests(unittest.TestCase):
|
|
def test_reference_fifo_audit_preserves_carried_bits_for_short_segment(self):
|
|
initial = np.arange(36, dtype=np.float32).reshape(18, 2)
|
|
initial[0, 0] = np.float32(-0.)
|
|
generated = np.arange(10, dtype=np.float32).reshape(5, 2)
|
|
joined = np.concatenate((initial, generated))
|
|
applied, pending = joined[:5], joined[5:]
|
|
result = c.audit_carried_sequence(generated, applied, pending, initial)
|
|
self.assertTrue(result['sample_bits_preserved'])
|
|
bad = applied.copy(); bad[0, 0] = np.float32(0.)
|
|
with self.assertRaisesRegex(ValueError, 'bitwise'):
|
|
c.audit_carried_sequence(generated, bad, pending, initial)
|
|
|
|
def test_timing_counts_and_ranges_use_absolute_clock_not_segment_length(self):
|
|
obj = object.__new__(c.ContinuedRuntime)
|
|
obj.global_offset, obj.spec, obj.mode = 41, {'steps': 5}, 'reference'
|
|
obj.parent = {'terminal': {'apl_pending': np.zeros((18, 4064), np.float32)}}
|
|
result = obj._timing_metadata({'pending_generated_count': 5, 'pending_initial_zero_prefix': 13,
|
|
'initial_padding_positive_zero': True})
|
|
self.assertEqual(result['pending_generated_count'], 18)
|
|
self.assertEqual(result['pending_initial_zero_prefix'], 0)
|
|
self.assertEqual(result['pending_application_tick_start'], 46)
|
|
self.assertEqual(result['pending_generated_tick_start'], 28)
|
|
self.assertEqual(result['pending_segment_generated_rows'], 5)
|
|
self.assertEqual(result['pending_parent_generated_rows'], 13)
|
|
self.assertEqual(result['applied_parent_generated_rows'], 5)
|
|
self.assertNotIn('initial_padding_positive_zero', result)
|
|
|
|
def test_restore_carries_exact_state_and_queues_but_not_segment_counts(self):
|
|
parent = {'global_next_tick': 41, 'state': {'v': np.array([-.05, -.04], np.float64),
|
|
'g': np.array([-0., -.002], np.float64), 'ref': np.array([22, 3], np.int32)},
|
|
'terminal': {'spike_ring': np.zeros((19, 2), np.uint8), 'next_tick': np.asarray(5, np.int64),
|
|
'apl_pending': np.arange(36, dtype=np.float32).reshape(18, 2), 'apl_state': np.array([2., 3.], np.float32)}}
|
|
parent['terminal']['spike_ring'][4, 1] = 1
|
|
obj = object.__new__(c.ContinuedRuntime)
|
|
obj.parent, obj.installations, obj.restorations = parent, 1, 0
|
|
obj.torch = types.SimpleNamespace(float64=np.float64, float32=np.float32, int32=np.int32, uint8=np.uint8)
|
|
v, g, ref, counts = (ArrayTarget(np.zeros(2, dtype)) for dtype in (np.float64, np.float64, np.int32, np.int32))
|
|
ring, apl_ring = ArrayTarget(np.zeros((19, 2), np.uint8)), ArrayTarget(np.zeros((18, 2), np.float32))
|
|
apl = types.SimpleNamespace(state=ArrayTarget(np.zeros(2, np.float32)))
|
|
upload = lambda value, dtype: np.array(value, dtype=dtype, copy=True)
|
|
obj.restore(v, g, ref, counts, ring, apl_ring, apl, upload)
|
|
self.assertEqual(g.value.tobytes(), parent['state']['g'].tobytes())
|
|
np.testing.assert_array_equal(ref.value, parent['state']['ref'])
|
|
np.testing.assert_array_equal(counts.value, [0, 0])
|
|
np.testing.assert_array_equal(ring.value, c.rotate_spike_ring(parent['terminal']['spike_ring'], 5))
|
|
np.testing.assert_array_equal(apl_ring.value, parent['terminal']['apl_pending'])
|
|
np.testing.assert_array_equal(apl.state.value, parent['terminal']['apl_state'])
|
|
with self.assertRaises(ValueError):
|
|
obj.restore(v, g, ref, counts, ring, apl_ring, apl, upload)
|
|
|
|
|
|
class CheckpointTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tmp.name).resolve()
|
|
self.n = patch.object(c, 'N', 4); self.e = patch.object(c, 'E', 6)
|
|
self.n.start(); self.e.start()
|
|
|
|
def tearDown(self):
|
|
self.e.stop(); self.n.stop(); self.tmp.cleanup()
|
|
|
|
def save(self, path, value):
|
|
R._save(path, value)
|
|
|
|
def make(self, name, steps=5, parent=None, corrupt_carry=None):
|
|
directory = self.root / name; directory.mkdir()
|
|
inputs = self.root / (name + '-inputs'); inputs.mkdir()
|
|
spec = {'steps': steps}
|
|
if parent is not None:
|
|
spec['parent_checkpoint_manifest_sha256'] = parent['manifest_sha256']
|
|
self.save(inputs / 'spec.json', spec)
|
|
overlay_bytes = b'synthetic-bound-overlay-not-a-brain'
|
|
(inputs / 'overlay.npz').write_bytes(overlay_bytes)
|
|
meta = {'overlay_sha256': R.digest(inputs / 'overlay.npz'), 'mask_sha256': R.MASK_SHA,
|
|
'base_candidate_sha256': 'b' * 64}
|
|
self.save(inputs / 'overlay-manifest.json', meta)
|
|
(directory / 'overlay-input.npz').write_bytes(overlay_bytes)
|
|
self.save(directory / 'overlay-manifest.json', meta)
|
|
event_tick, event_neuron = np.array([0], np.int32), np.array([1], np.int32)
|
|
counts = np.array([0, 1, 0, 0], np.int32)
|
|
state = {'step': event_tick, 'neuron_idx': event_neuron, 'counts': counts,
|
|
'v': np.full(4, -.052, np.float64), 'g': np.array([-0., -.003, 0., 1.], np.float64),
|
|
'ref': np.array([0, 22, 3, 0], np.int32)}
|
|
ring = np.zeros((19, 4), np.uint8) if parent is None else c.rotate_spike_ring(parent['terminal']['spike_ring'], parent['steps'])
|
|
for tick in range(steps):
|
|
ring[tick % 19] = 0
|
|
if tick == 0:
|
|
ring[0, 1] = 1
|
|
pending = np.zeros((18, 4064), np.float32)
|
|
if corrupt_carry == 'spike':
|
|
ring[18, 0] ^= np.uint8(1)
|
|
elif corrupt_carry == 'apl':
|
|
pending[0, 0] = np.float32(1.)
|
|
terminal = {'spike_ring': ring, 'apl_pending': pending, 'apl_state': np.array([1., 2.], np.float32),
|
|
'next_tick': np.asarray(steps, np.int64)}
|
|
np.savez_compressed(directory / 'state.npz', **state)
|
|
np.savez_compressed(directory / 'terminal-queues.npz', **terminal)
|
|
np.save(directory / 'apl-pending-tail.float32.npy', pending)
|
|
self.save(directory / 'apl-timing-audit.json', {'apl_delay_steps': 18, 'brain_tail_advanced': False,
|
|
'pending_raw_sha256': R._raw(pending), 'final_apl_state_u32_bits': terminal['apl_state'].view(np.uint32).tolist()})
|
|
offset = 0 if parent is None else parent['global_next_tick']
|
|
summary = {'complete': True, 'pass_all': True, 'all_finite': True, 'steps': steps, 'completed_ticks': steps,
|
|
'nodes': 4, 'edges': 6, 'precision': 'float64', 'apl_precision': 'float32',
|
|
'within_trial_weight_updates': False, 'checkpoint_continuation_used': parent is not None,
|
|
'effective_weights_raw_sha256': 'e'*64, 'global_tick_offset': offset, 'global_next_tick': offset + steps,
|
|
'input_sha256': {'spec.json': R.digest(inputs / 'spec.json')}}
|
|
if parent is not None:
|
|
summary['parent_checkpoint_manifest_sha256'] = parent['manifest_sha256']
|
|
self.save(directory / 'summary.json', summary)
|
|
self.save(directory / 'finite-audit.json', {'all_finite': True, 'first_failure_tick': -1, 'finite_observed_ticks': steps})
|
|
self.save(directory / 'overlay-audit.json', {'complete': True, 'weights_updated_during_run': False,
|
|
'effective_weights_raw_sha256': 'e'*64, 'mask_sha256': R.MASK_SHA, 'am_helper_sha256': R.AM_OVERLAY_SHA,
|
|
'base_candidate_sha256': 'b'*64})
|
|
provenance = {'schema': 'phase4an-reporting-adapter-v1' if parent is None else 'phase4an-continuation-runtime-v1',
|
|
'complete': True, 'weights_updated_during_run': False, 'old_globals_patched': False,
|
|
'source_file': R.__file__ if parent is None else c.__file__,
|
|
'source_sha256': c.LIGHT_SHA if parent is None else R.digest(c.__file__),
|
|
'input_sha256': {str(path): R.digest(path) for path in inputs.iterdir()}}
|
|
if parent is not None:
|
|
provenance.update(parent_checkpoint_manifest_sha256=parent['manifest_sha256'],
|
|
global_tick_offset=offset, global_next_tick=offset + steps)
|
|
for key, file in (('summary_sha256', 'summary.json'), ('finite_audit_sha256', 'finite-audit.json'),
|
|
('terminal_queues_sha256', 'terminal-queues.npz'), ('overlay_audit_sha256', 'overlay-audit.json'),
|
|
('state_sha256', 'state.npz')):
|
|
provenance[key] = R.digest(directory / file)
|
|
self.save(directory / 'runtime-provenance.json', provenance)
|
|
if parent is not None:
|
|
self.save(directory / c.RECEIPT, {'schema': 'phase4an-continuation-receipt-v1', 'complete': True,
|
|
'parent_checkpoint_dir': str(parent['directory']), 'parent_checkpoint_manifest_sha256': parent['manifest_sha256'],
|
|
'segment_steps': steps, 'counts_scope': 'segment_only_not_lifetime', 'weights_updated_during_run': False,
|
|
'global_tick_offset': offset, 'global_next_tick': offset + steps, 'payload_sha256': c._tree_hashes(directory)})
|
|
return directory
|
|
|
|
def test_fresh_seal_is_exclusive_and_binds_state_bytes(self):
|
|
directory = self.make('fresh')
|
|
sealed = c.seal_checkpoint(directory)
|
|
loaded = c.load_checkpoint(directory, sealed['sha256'])
|
|
self.assertEqual(loaded['global_next_tick'], 5)
|
|
self.assertEqual(loaded['state']['g'].tobytes(), np.array([-0., -.003, 0., 1.], np.float64).tobytes())
|
|
self.assertIn('state.npz', sealed['files_sha256'])
|
|
self.assertFalse(sealed['pre_seal_state_integrity_independently_proven'])
|
|
with self.assertRaisesRegex(ValueError, 'already sealed'):
|
|
c.seal_checkpoint(directory)
|
|
with (directory / 'state.npz').open('ab') as stream:
|
|
stream.write(b'tampered')
|
|
with self.assertRaisesRegex(ValueError, 'inventory/hash'):
|
|
c.load_checkpoint(directory, sealed['sha256'])
|
|
|
|
def test_multisegment_receipts_bind_absolute_offset_and_local_phase(self):
|
|
parent = None
|
|
total = 0
|
|
for i, steps in enumerate((5, 13, 1, 19)):
|
|
directory = self.make('segment-' + str(i), steps, parent)
|
|
sealed = c.seal_checkpoint(directory)
|
|
parent = c.load_checkpoint(directory, sealed['sha256'])
|
|
total += steps
|
|
self.assertEqual(parent['global_next_tick'], total)
|
|
self.assertEqual(parent['terminal']['next_tick'].item(), steps)
|
|
self.assertEqual(parent['ancestry_directories'], [self.root / ('segment-' + str(j)) for j in range(i, -1, -1)])
|
|
self.assertEqual(total, 38)
|
|
|
|
def test_run_rejects_output_and_build_in_every_ancestor_before_overlay_or_gpu(self):
|
|
parent = None
|
|
for i in range(3):
|
|
directory = self.make('ancestor-' + str(i), 20, parent)
|
|
seal = c.seal_checkpoint(directory)
|
|
parent = c.load_checkpoint(directory, seal['sha256'])
|
|
case = self.root / 'next-case'; case.mkdir()
|
|
self.save(case / 'spec.json', {'max_steps': 5, 'steps': 5, 'dt_ms': .1, 'duration_ms': .5,
|
|
'apl_delay_steps': 18, 'record_history': False,
|
|
'parent_checkpoint_manifest_sha256': parent['manifest_sha256']})
|
|
before = {path: c._tree_hashes(path) for path in parent['ancestry_directories']}
|
|
with patch.object(R, 'load_overlay', side_effect=AssertionError('Overlay/GPU preparation must not begin')) as overlay:
|
|
for ancestor in parent['ancestry_directories']:
|
|
for forbidden in (ancestor, ancestor / 'new-child'):
|
|
for output, build in ((forbidden, self.root / 'safe-build'), (self.root / 'safe-output', forbidden)):
|
|
with self.subTest(ancestor=ancestor.name, output=output, build=build), self.assertRaises(ValueError):
|
|
c.run_continued(case, output, build, self.root / 'unused-overlay.npz',
|
|
self.root / 'unused-manifest.json', parent['directory'])
|
|
overlay.assert_not_called()
|
|
self.assertEqual(before, {path: c._tree_hashes(path) for path in parent['ancestry_directories']})
|
|
self.assertFalse((self.root / 'safe-output').exists())
|
|
self.assertFalse((self.root / 'safe-build').exists())
|
|
|
|
def test_reject_bad_external_pin_extra_file_and_error_marker(self):
|
|
directory = self.make('fresh')
|
|
sealed = c.seal_checkpoint(directory)
|
|
with self.assertRaisesRegex(ValueError, 'external pin'):
|
|
c.load_checkpoint(directory, '0' * 64)
|
|
(directory / 'unexpected.bin').write_bytes(b'extra')
|
|
with self.assertRaisesRegex(ValueError, 'inventory/hash'):
|
|
c.load_checkpoint(directory, sealed['sha256'])
|
|
failure = self.make('failure')
|
|
self.save(failure / 'runtime-error.json', {'complete': False})
|
|
with self.assertRaisesRegex(ValueError, 'failure/error'):
|
|
c.seal_checkpoint(failure)
|
|
|
|
def test_self_consistent_receipts_do_not_hide_unwritten_carry_slot_corruption(self):
|
|
directory = self.make('parent', 20)
|
|
sealed = c.seal_checkpoint(directory)
|
|
parent = c.load_checkpoint(directory, sealed['sha256'])
|
|
for kind, message in (('spike', 'carried spike'), ('apl', 'carried APL')):
|
|
child = self.make('bad-' + kind, 5, parent, corrupt_carry=kind)
|
|
# The synthetic writer hashes its incorrect output consistently;
|
|
# the independent parent-to-child carry invariant must still reject.
|
|
with self.subTest(kind=kind), self.assertRaisesRegex(ValueError, message):
|
|
c.seal_checkpoint(child)
|
|
|
|
def test_reject_partial_receipt_and_modified_original_input(self):
|
|
directory = self.make('bad-receipt')
|
|
(directory / 'summary.json').write_text('{}', encoding='utf-8')
|
|
with self.assertRaises(ValueError):
|
|
c.seal_checkpoint(directory)
|
|
directory = self.make('modified-input')
|
|
(self.root / 'modified-input-inputs' / 'spec.json').write_text('{}', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'input/source'):
|
|
c.seal_checkpoint(directory)
|
|
|
|
def test_descendant_rejects_mutated_ancestor_even_if_own_files_are_unchanged(self):
|
|
directory = self.make('ancestor', 20)
|
|
sealed = c.seal_checkpoint(directory)
|
|
parent = c.load_checkpoint(directory, sealed['sha256'])
|
|
child = self.make('descendant', 5, parent)
|
|
child_seal = c.seal_checkpoint(child)
|
|
with (directory / 'state.npz').open('ab') as stream:
|
|
stream.write(b'ancestor changed after child seal')
|
|
with self.assertRaisesRegex(ValueError, 'inventory/hash'):
|
|
c.load_checkpoint(child, child_seal['sha256'])
|
|
|
|
def test_self_consistent_hashes_cannot_hide_parent_pin_or_weight_disagreement(self):
|
|
directory = self.make('bound-parent', 20)
|
|
sealed = c.seal_checkpoint(directory)
|
|
parent = c.load_checkpoint(directory, sealed['sha256'])
|
|
def rewrite(path, value):
|
|
path.write_text(json.dumps(value, sort_keys=True), encoding='utf-8')
|
|
for kind in ('case', 'summary', 'provenance', 'weights'):
|
|
child = self.make('bound-child-' + kind, 5, parent)
|
|
spec_path = self.root / ('bound-child-' + kind + '-inputs') / 'spec.json'
|
|
summary = R._read_json(child / 'summary.json')
|
|
provenance = R._read_json(child / 'runtime-provenance.json')
|
|
if kind == 'case':
|
|
spec = R._read_json(spec_path); spec['parent_checkpoint_manifest_sha256'] = '0' * 64
|
|
rewrite(spec_path, spec)
|
|
summary['input_sha256']['spec.json'] = R.digest(spec_path)
|
|
provenance['input_sha256'][str(spec_path)] = R.digest(spec_path)
|
|
elif kind in ('summary', 'provenance'):
|
|
target = summary if kind == 'summary' else provenance
|
|
target['parent_checkpoint_manifest_sha256'] = '0' * 64
|
|
else:
|
|
summary['effective_weights_raw_sha256'] = 'f' * 64
|
|
audit = R._read_json(child / 'overlay-audit.json'); audit['effective_weights_raw_sha256'] = 'f' * 64
|
|
rewrite(child / 'overlay-audit.json', audit)
|
|
provenance['overlay_audit_sha256'] = R.digest(child / 'overlay-audit.json')
|
|
rewrite(child / 'summary.json', summary)
|
|
provenance['summary_sha256'] = R.digest(child / 'summary.json')
|
|
rewrite(child / 'runtime-provenance.json', provenance)
|
|
receipt = R._read_json(child / c.RECEIPT)
|
|
receipt['payload_sha256'] = c._tree_hashes(child); receipt['payload_sha256'].pop(c.RECEIPT)
|
|
rewrite(child / c.RECEIPT, receipt)
|
|
# All individual byte hashes are now consistent. Only semantic
|
|
# cross-receipt binding or parent/child weight equality rejects it.
|
|
with self.subTest(kind=kind), self.assertRaisesRegex(ValueError, 'weights changed|pin disagrees'):
|
|
c.seal_checkpoint(child)
|
|
|
|
def test_reject_invalid_state_dtype_events_ref_and_ring(self):
|
|
directory = self.make('arrays')
|
|
state = c._npz(directory / 'state.npz', ('step', 'neuron_idx', 'counts', 'v', 'g', 'ref'))
|
|
terminal = c._npz(directory / 'terminal-queues.npz', ('spike_ring', 'apl_state', 'apl_pending', 'next_tick'))
|
|
for key, value in (('v', state['v'].astype(np.float32)), ('g', np.full(4, np.nan)),
|
|
('ref', np.full(4, 23, np.int32)), ('step', np.array([5], np.int32)),
|
|
('counts', np.zeros(4, np.int32))):
|
|
with self.subTest(key=key), self.assertRaises(ValueError):
|
|
c.validate_state(dict(state, **{key: value}), terminal, 5, fresh=True)
|
|
bad = dict(terminal, spike_ring=terminal['spike_ring'].copy()); bad['spike_ring'][0, 1] = 0
|
|
with self.assertRaisesRegex(ValueError, 'emitted'):
|
|
c.validate_state(state, bad, 5, fresh=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|