Digital Drosophila Archive
101 lines
4.6 KiB
Python
101 lines
4.6 KiB
Python
"""Authority, full-byte IPC and cleanup tests without original production runs."""
|
|
import base64
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
|
|
SCRIPT = Path(__file__).with_name('profile_original.py')
|
|
SPEC = importlib.util.spec_from_file_location('test_original_profile_module', SCRIPT)
|
|
P = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(P)
|
|
|
|
|
|
class BoundaryTests(unittest.TestCase):
|
|
def test_array_packet_uses_raw_signed_zero_bytes(self):
|
|
reference = np.asarray([0., 1.], dtype=np.float32)
|
|
P.compare_packet(P.pack_array(reference), reference)
|
|
changed = np.asarray([-0., 1.], dtype=np.float32)
|
|
with self.assertRaisesRegex(ValueError, 'full array bytes'):
|
|
P.compare_packet(P.pack_array(changed), reference)
|
|
|
|
def test_packet_dtype_and_shape_are_not_repaired(self):
|
|
reference = np.asarray([1.], dtype=np.float32)
|
|
for array in (reference.astype(np.float64), reference.reshape(1, 1)):
|
|
with self.assertRaisesRegex(ValueError, 'identity'):
|
|
P.compare_packet(P.pack_array(array), reference)
|
|
|
|
def test_response_chain_binds_all_output_bytes_and_role(self):
|
|
response = dict(role='producer', sequence=1, previous_response_sha256=None, complete=True,
|
|
arrays={'delta_after': P.pack_array(np.zeros(3, np.float32))})
|
|
response['response_sha256'] = P.payload_sha(response)
|
|
self.assertEqual(P.verify_response(response, None, 1, 'producer'), response['response_sha256'])
|
|
with self.assertRaisesRegex(ValueError, 'chain/role'):
|
|
P.verify_response(response, None, 1, 'verifier')
|
|
response['arrays']['delta_after']['data_base64'] = base64.b64encode(b'bad').decode()
|
|
with self.assertRaisesRegex(ValueError, 'payload hash'):
|
|
P.verify_response(response, None, 1, 'producer')
|
|
|
|
def test_response_chain_reordering_rejected(self):
|
|
response = dict(role='producer', sequence=2, previous_response_sha256='a' * 64, complete=True)
|
|
response['response_sha256'] = P.payload_sha(response)
|
|
with self.assertRaisesRegex(ValueError, 'chain/role'):
|
|
P.verify_response(response, 'b' * 64, 2, 'producer')
|
|
|
|
def test_manifest_paths_cannot_escape(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary).resolve()
|
|
for name in ('../other', '/absolute', 'foo\\bar', 'x/../bar', 'C:/windows'):
|
|
with self.assertRaises(ValueError):
|
|
P.contained(root, name)
|
|
self.assertEqual(P.contained(root, 'valid/input.npz'), root / 'valid/input.npz')
|
|
|
|
def test_output_writes_are_exclusive(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary).resolve() / 'result.json'
|
|
P.save(path, {'complete': True})
|
|
original = path.read_bytes()
|
|
with self.assertRaises(FileExistsError):
|
|
P.save(path, {'complete': False})
|
|
self.assertEqual(path.read_bytes(), original)
|
|
|
|
def test_mismatching_source_pin_rejected(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary).resolve() / 'source.py'
|
|
path.write_text('raise AssertionError("must not run")', encoding='utf-8')
|
|
with self.assertRaisesRegex(ValueError, 'identity'):
|
|
P.load_source('untrusted_original_profile_test', path, '0' * 64)
|
|
|
|
def test_profile_categories_do_not_use_cumulative_totals(self):
|
|
self.assertEqual(P.profile_category('/usr/lib/python3.10/ast.py', 'dump'), 'source_parse_compile_load')
|
|
self.assertEqual(P.profile_category('/frozen/phase4j.py::trace-expression-only', '<module>'),
|
|
'trace_expression_and_tensor_ops')
|
|
self.assertEqual(P.profile_category('/frozen/bridge.py', 'trace_from_events'), 'trace_apply_python_self')
|
|
|
|
|
|
class CleanupTests(unittest.TestCase):
|
|
def test_cleanup_attempts_every_owned_child_after_failure(self):
|
|
calls = []
|
|
class Good:
|
|
def abort(self):
|
|
calls.append('good')
|
|
class Bad:
|
|
def abort(self):
|
|
calls.append('bad')
|
|
raise RuntimeError('simulated abort error')
|
|
original = list(P.CHILDREN)
|
|
try:
|
|
P.CHILDREN[:] = [Good(), Bad()]
|
|
with self.assertRaisesRegex(ValueError, 'cleanup errors'):
|
|
P.cleanup_children()
|
|
self.assertEqual(calls, ['bad', 'good'])
|
|
finally:
|
|
P.CHILDREN[:] = original
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|