Digital Drosophila Archive
162 lines
8.5 KiB
Python
162 lines
8.5 KiB
Python
"""Host-only authority, exact-comparison, ordering and process receipt tests."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import numpy as np
|
|
|
|
import brain_replay as replay
|
|
|
|
|
|
class ReplayTests(unittest.TestCase):
|
|
def test_fixed_forward_reverse_order(self):
|
|
result = replay.fixture_order(14)
|
|
self.assertEqual(len(result), 28)
|
|
self.assertEqual(result[:14], [('forward', i) for i in range(14)])
|
|
self.assertEqual(result[14:], [('reverse', i) for i in range(13, -1, -1)])
|
|
for count in (0, 13, 15):
|
|
with self.assertRaises(ValueError):
|
|
replay.fixture_order(count)
|
|
|
|
def test_exact_comparison_detects_signed_zero_dtype_shape_and_bit_changes(self):
|
|
expected = {'v': np.array([0., 1.], np.float64)}
|
|
self.assertTrue(replay.compare_arrays(expected, expected, ('v',))['v']['bitwise_exact'])
|
|
for array in (np.array([-0., 1.], np.float64), np.array([0., 1.], np.float32),
|
|
np.array([[0., 1.]], np.float64),
|
|
np.array([0., np.nextafter(1., 2.)], np.float64)):
|
|
self.assertFalse(replay.compare_arrays({'v': array}, expected, ('v',))['v']['bitwise_exact'])
|
|
with self.assertRaisesRegex(ValueError, 'schema'):
|
|
replay.compare_arrays({}, expected, ('v',))
|
|
|
|
def test_checkpoint_comparison_includes_all_arrays_requests_and_environment(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
actual_dir, old_dir = root / 'actual', root / 'old'
|
|
actual_dir.mkdir()
|
|
old_dir.mkdir()
|
|
for name in ('requests.csv', 'finite-audit.json', 'apl-pending-tail.float32.npy'):
|
|
(actual_dir / name).write_bytes(b'fixed exact reference')
|
|
(old_dir / name).write_bytes(b'fixed exact reference')
|
|
def record():
|
|
return dict(state={name: np.zeros(2, np.float64 if name in ('v', 'g') else np.int32)
|
|
for name in replay.STATE_KEYS},
|
|
terminal={'spike_ring': np.zeros((19, 2), np.uint8),
|
|
'apl_state': np.zeros(2, np.float32),
|
|
'apl_pending': np.zeros((18, 2), np.float32),
|
|
'next_tick': np.asarray(1200, np.int64)},
|
|
summary={**{key: 'same' for key in replay.SUMMARY_KEYS},
|
|
'apl_component': {'environment': {'torch': 'fixed-original'}}},
|
|
fresh=True, global_tick_offset=0)
|
|
actual, original = record(), record()
|
|
self.assertTrue(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact'])
|
|
actual['terminal']['spike_ring'][0, 1] = 1
|
|
self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact'])
|
|
actual = record()
|
|
actual['summary']['apl_component']['environment']['torch'] = 'different'
|
|
self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact'])
|
|
actual = record()
|
|
actual['summary']['effective_weights_raw_sha256'] = 'different'
|
|
self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact'])
|
|
actual = record()
|
|
(actual_dir / 'requests.csv').write_bytes(b'different acceptance')
|
|
self.assertFalse(replay.compare_checkpoints(actual, original, actual_dir, old_dir)['exact'])
|
|
|
|
def test_json_and_pin_authority_rejects_ambiguity_and_mutation(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
value = root / 'value.json'
|
|
value.write_text('{"key":1,"key":2}', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate'):
|
|
replay.read(value)
|
|
value.write_text('{"key":NaN}', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'Nonfinite'):
|
|
replay.read(value)
|
|
value.write_text('{"key":1}', encoding='utf-8')
|
|
pins = {}
|
|
replay.pin_add(pins, value, replay.sha(value))
|
|
replay.verify(pins)
|
|
with self.assertRaisesRegex(ValueError, 'Conflicting'):
|
|
replay.pin_add(pins, value, 'f' * 64)
|
|
value.write_text('{"key":2}', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'changed'):
|
|
replay.verify(pins)
|
|
with self.assertRaisesRegex(ValueError, 'leaves'):
|
|
replay.contained(root, str(root.parent / 'outside'))
|
|
|
|
def test_loader_executes_only_verified_bytes_and_cannot_rebind_module(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = Path(directory) / 'source.py'
|
|
source.write_text('value = 42\n', encoding='utf-8')
|
|
name = '_brain_replay_host_test_unique_module'
|
|
try:
|
|
with self.assertRaisesRegex(ValueError, 'Unverified'):
|
|
replay.load_source(name, source, '0' * 64)
|
|
self.assertNotIn(name, sys.modules)
|
|
module = replay.load_source(name, source, replay.sha(source))
|
|
self.assertEqual(module.value, 42)
|
|
with self.assertRaisesRegex(ValueError, 'repeated'):
|
|
replay.load_source(name, source, replay.sha(source))
|
|
finally:
|
|
sys.modules.pop(name, None)
|
|
|
|
def test_parent_environment_is_original_only(self):
|
|
with mock.patch.object(replay.sys, 'dont_write_bytecode', False):
|
|
with self.assertRaisesRegex(ValueError, 'original registered'):
|
|
replay.check_environment()
|
|
|
|
|
|
class ProcessReceiptTests(unittest.TestCase):
|
|
def paths(self, root):
|
|
for name in ('jobs', 'processes'):
|
|
(root / name).mkdir()
|
|
report_path = root / 'report.json'
|
|
replay.save(report_path, {'exact': True})
|
|
result_path = root / 'result.json'
|
|
replay.save(result_path, dict(complete=True, all_bitwise_exact=True, lane='fresh', count=1,
|
|
records=[dict(report=str(report_path), sha256=replay.sha(report_path))]))
|
|
return dict(lane='fresh', runs=[['fresh', 0]], result_file=str(result_path)), root / 'jobs/fresh.json', root / 'processes'
|
|
|
|
def test_child_receipt_requires_actual_return_and_keeps_outer_group(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
job, job_path, process_dir = self.paths(root)
|
|
process = mock.Mock(pid=4321, returncode=0)
|
|
process.wait.return_value = 0
|
|
with mock.patch('brain_replay.subprocess.Popen', return_value=process) as popen:
|
|
receipt, result = replay.launch(job, job_path, process_dir, replay.time.time() + 10)
|
|
self.assertTrue(receipt['exit_independently_observed'])
|
|
self.assertEqual(receipt['actual_returncode'], 0)
|
|
self.assertEqual(receipt['child_pid'], 4321)
|
|
self.assertNotIn('start_new_session', popen.call_args.kwargs)
|
|
self.assertEqual(popen.call_args.kwargs['env']['CUDA_VISIBLE_DEVICES'], replay.GPU)
|
|
self.assertEqual(popen.call_args.args[0][0], str(replay.ORIGINAL_PYTHON))
|
|
self.assertEqual(replay.read(process_dir / 'fresh.json')['actual_returncode'], 0)
|
|
self.assertTrue(result['all_bitwise_exact'])
|
|
|
|
def test_timeout_stops_child_retains_real_failure_receipt_and_never_retries(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
job, job_path, process_dir = self.paths(root)
|
|
process = mock.Mock(pid=4321, returncode=-15)
|
|
process.wait.side_effect = [subprocess.TimeoutExpired('mock child', 1), -15, -15]
|
|
process.poll.return_value = None
|
|
with mock.patch('brain_replay.subprocess.Popen', return_value=process) as popen:
|
|
with self.assertRaisesRegex(ValueError, 'child failed'):
|
|
replay.launch(job, job_path, process_dir, replay.time.time() + 10)
|
|
self.assertEqual(popen.call_count, 1)
|
|
process.terminate.assert_called_once()
|
|
receipt = replay.read(process_dir / 'fresh.json')
|
|
self.assertTrue(receipt['timed_out'])
|
|
self.assertTrue(receipt['child_terminated_on_failure'])
|
|
self.assertEqual(receipt['actual_returncode'], -15)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|