"""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()