"""CPU-only synthetic safety/AST tests; not scientific performance evidence.""" import ast import copy import hashlib import importlib.util import json from pathlib import Path import sys import tempfile import time import unittest from unittest.mock import patch import numpy as np import perf_probe as P def fresh_validator(): path = Path(P.__file__).parent/'resident_validator.py' spec = importlib.util.spec_from_file_location('_test_resident_fixture', path) module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module) return module class Tests(unittest.TestCase): def test_safe_cache_reuses_same_verified_module(self): v = fresh_validator() with tempfile.TemporaryDirectory() as d: p = Path(d)/'m.py'; p.write_text('VALUE = 7\n', encoding='utf-8'); sha = P.digest(p) try: a = v.load('_perf_unit_one', p, sha) self.assertIs(a, v.load('_perf_unit_one', p, sha)) p.write_text('VALUE = 8\n', encoding='utf-8') with self.assertRaisesRegex(ValueError, 'bytes changed'): v.load('_perf_unit_one', p, sha) finally: sys.modules.pop('_perf_unit_one', None) def test_module_identity_substitution_rejected(self): v = fresh_validator() with tempfile.TemporaryDirectory() as d: p = Path(d)/'m.py'; p.write_text('VALUE=1', encoding='utf-8'); sha = P.digest(p) try: v.load('_perf_unit_two', p, sha); sys.modules['_perf_unit_two'] = object() with self.assertRaisesRegex(ValueError, 'identity differs'): v.load('_perf_unit_two', p, sha) finally: sys.modules.pop('_perf_unit_two', None) def test_inherited_torch_fails_and_poisoned_worker_stays_failed(self): v = fresh_validator() with patch.dict(sys.modules, {'torch': object()}): with self.assertRaisesRegex(ValueError, 'must not inherit'): v.validate('x','x','x','x','x') self.assertEqual(v._RESIDENT_STATE, 'failed') with self.assertRaisesRegex(ValueError, 'active or failed'): v.validate('x','x','x','x','x') def test_ordinary_runtime_rejected(self): v = fresh_validator() with patch.dict(sys.modules, {'runtime_light': object()}): with self.assertRaisesRegex(ValueError, 'Dedicated CPU-only'): v.validate('x','x','x','x','x') self.assertEqual(v._RESIDENT_STATE, 'failed') def test_bitwise_signed_zero(self): v = fresh_validator() self.assertFalse(v.same(np.array([0.],np.float32), np.array([-0.],np.float32))) self.assertFalse(v.same(np.array([0.],np.float32), np.array([0.],np.float64))) def test_copied_validation_body_AST_preserved(self): original = P.VALIDATOR if not original.exists(): original = Path(P.__file__).parent.parent/'conditioning-execution-v1/validate_update_cpu.py' old = ast.parse(original.read_text(encoding='utf-8')) new = ast.parse((Path(P.__file__).parent/'resident_validator.py').read_text(encoding='utf-8')) of = {n.name:n for n in old.body if isinstance(n,ast.FunctionDef)} nf = {n.name:n for n in new.body if isinstance(n,ast.FunctionDef)} for name in ('require','digest','same','unlinked','chain_binding'): self.assertEqual(ast.dump(of[name]), ast.dump(nf[name])) a,b=copy.deepcopy(of['validate']),copy.deepcopy(nf['_validate_checked']) a.body=a.body[2:];b.name='validate' for node in ast.walk(b): if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id=='receipt' for t in node.targets): node.value.keywords=[k for k in node.value.keywords if k.arg not in ('resident_prototype','original_validator_source','resident_cache_scope')] next(k for k in node.value.keywords if k.arg=='schema').value=ast.Constant('conditioning-update-cpu-validation-v1') self.assertEqual(ast.dump(a),ast.dump(b)) def fixture_receipts(self, root): original={'path':str(P.VALIDATOR),'sha256':P.V_SHA} candidate=Path(P.__file__).resolve().parent/'resident_validator.py' a={'schema':'conditioning-update-cpu-validation-v1','validator_source':original, 'complete':True,'test_scientific_value':{'raw_hash':'a'*64}} b=copy.deepcopy(a);b.update(schema='conditioning-resident-cpu-validation-prototype-v1', validator_source={'path':str(candidate),'sha256':P.digest(candidate)}, original_validator_source=original,resident_prototype=True, resident_cache_scope='SHA-verified imported modules only; no input/hash/state/delta/checkpoint caching') p,q=Path(root)/'original.json',Path(root)/'prototype.json' P.save(p,a);P.save(q,b) return p,q,b def test_only_explicit_provenance_differences_allowed(self): with tempfile.TemporaryDirectory() as d: p,q,b=self.fixture_receipts(d) self.assertTrue(P.compare_validation(p,q)['all_scientific_fields_identical']) b['test_scientific_value']['raw_hash']='b'*64 q.write_text(json.dumps(b),encoding='utf-8') with self.assertRaisesRegex(ValueError,'Scientific verifier'): P.compare_validation(p,q) def test_provenance_substitution_rejected(self): with tempfile.TemporaryDirectory() as d: p,q,b=self.fixture_receipts(d);b['original_validator_source']['sha256']='0'*64 q.write_text(json.dumps(b),encoding='utf-8') with self.assertRaisesRegex(ValueError,'original-source'): P.compare_validation(p,q) def test_exclusive_save_no_overwrite(self): with tempfile.TemporaryDirectory() as d: p=Path(d)/'record.json';P.save(p,{'one':1}) with self.assertRaises(FileExistsError): P.save(p,{'two':2}) self.assertEqual(P.read(p),{'one':1}) def test_parent_traversal_rejected(self): with self.assertRaisesRegex(ValueError, 'Parent traversal'): P.safe('/tmp/test/../escape') def test_absolute_deadline_enforced(self): process = object.__new__(P.Process); process.deadline = time.perf_counter()-1 with self.assertRaisesRegex(ValueError, 'absolute wall'): process.remaining() def test_bad_ready_child_is_reaped_before_constructor_raises(self): children=[]; original=P.subprocess.Popen def create_test_child(argv, **kwargs): child=original([sys.executable,'-B','-c','import time; time.sleep(30)'],**kwargs) children.append(child);return child with tempfile.TemporaryDirectory() as d: root=Path(d) for name in ('logs','processes'): (root/name).mkdir() plan=root/'plan.json';P.save(plan,{}) with patch.object(P.subprocess,'Popen',side_effect=create_test_child), patch.object(P.Process,'line',side_effect=ValueError('bad ready fixture')): with self.assertRaisesRegex(ValueError,'bad ready fixture'): P.Process('update','server',plan,root,'bad-start',time.perf_counter()+10) self.assertEqual(len(children),1);self.assertIsNotNone(children[0].poll()) self.assertIsNotNone(P.read(root/'processes/bad-start.aborted.json')['actual_parent_observed_exit']) if __name__=='__main__': unittest.main()