Digital Drosophila Archive
69 lines
4.1 KiB
Python
69 lines
4.1 KiB
Python
"""CPU-only chain/mutation/real-source loading tests; no production events."""
|
|
import copy
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
import numpy as np
|
|
import validate_update_cpu as V
|
|
|
|
|
|
class ValidationTests(unittest.TestCase):
|
|
def fixture(self):
|
|
parent=Path('/synthetic/parent');auth={'epoch_id':'one','command':{'fixture':True},
|
|
'diagnostic_mode':'reference','checkpoint_manifest_sha256':'2'*64}
|
|
roles=('state','spec','requests','overlay_before','overlay_manifest_before')
|
|
paths={r:Path('/synthetic/'+r) for r in roles};pins={p:str(i+3)*64 for i,p in enumerate(paths.values())}
|
|
hashes={name:format(i+8,'x')*64 for i,name in enumerate(V.FILES)}
|
|
chain=dict(schema='phase4an-epoch-postprocess-chain-v1',complete=True,
|
|
weights_installed=False,brain_executed=False,learning_pass_claimed=False,candidate_generated=False,
|
|
coordinator_source_sha256=V.E_SHA,authority_sha256='1'*64,epoch_id='one',command=auth['command'],
|
|
diagnostic_mode='reference',source_neural_checkpoint_dir=str(parent),
|
|
source_neural_checkpoint_manifest_sha256='2'*64,input_files_sha256={str(p):v for p,v in pins.items()},
|
|
output_files_sha256={n:s for n,s in hashes.items() if n!='chain-receipt.json'},
|
|
after_overlay_file_sha256=hashes['delta-after.npz'],after_overlay_manifest_sha256=hashes['overlay-manifest.json'])
|
|
for field,role in (('source_state_file_sha256','state'),('source_spec_sha256','spec'),
|
|
('source_requests_sha256','requests'),('before_overlay_file_sha256','overlay_before'),
|
|
('before_overlay_manifest_sha256','overlay_manifest_before')):chain[field]=pins[paths[role]]
|
|
return chain,auth,'1'*64,pins,paths,parent,hashes
|
|
|
|
def test_exact_chain(self):V.chain_binding(*self.fixture())
|
|
|
|
def test_wrong_parent_command_or_payload_rejected(self):
|
|
for field,value in (('source_neural_checkpoint_dir','elsewhere'),('source_neural_checkpoint_manifest_sha256','0'*64),
|
|
('authority_sha256','0'*64),('command',{}),('before_overlay_file_sha256','0'*64),('source_state_file_sha256','0'*64),
|
|
('after_overlay_file_sha256','0'*64),('output_files_sha256',{}),('input_files_sha256',{}),('complete',False),
|
|
('weights_installed',True),('learning_pass_claimed',True),('coordinator_source_sha256','0'*64)):
|
|
args=list(self.fixture());args[0]=copy.deepcopy(args[0]);args[0][field]=value
|
|
with self.subTest(field=field),self.assertRaises(ValueError):V.chain_binding(*args)
|
|
|
|
def test_bit_identity_includes_signed_zero(self):
|
|
self.assertTrue(V.same(np.zeros(2,np.float32),np.zeros(2,np.float32)))
|
|
self.assertFalse(V.same(np.zeros(2,np.float32),np.zeros(2,np.float64)))
|
|
self.assertFalse(V.same(np.array([0.],np.float32),np.array([-0.],np.float32)))
|
|
|
|
def test_source_substitution_rejected_before_exec(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p=Path(d)/'fake.py';p.write_text('raise RuntimeError("should not execute")')
|
|
with self.assertRaisesRegex(ValueError,'Unverified source'):V.load('_fixture_not_loaded',p,'0'*64)
|
|
self.assertNotIn('_fixture_not_loaded',sys.modules)
|
|
|
|
def test_actual_frozen_sources_load_without_torch_or_ordinary_R(self):
|
|
code="""
|
|
import sys
|
|
sys.path.insert(0,sys.argv[1]);import validate_update_cpu as V
|
|
e=V.load('_v_cpu_test_E',V.LAB/'conditioning_epoch.py',V.E_SHA)
|
|
V.load('engineering_compare',V.LAB/'engineering_compare.py',V.CMP_SHA)
|
|
q=V.load('_v_cpu_test_Q',V.LAB/'conditioning_preflight_compare.py',V.Q_SHA)
|
|
assert 'torch' not in sys.modules and 'runtime_light' not in sys.modules
|
|
assert callable(e._authority) and callable(q.audit_delta_binding)
|
|
print('ACTUAL_SOURCE_IMPORT_OK')
|
|
"""
|
|
result=subprocess.run([sys.executable,'-B','-c',code,str(Path(V.__file__).parent)],text=True,capture_output=True)
|
|
self.assertEqual(result.returncode,0,result.stderr);self.assertIn('ACTUAL_SOURCE_IMPORT_OK',result.stdout)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|