"""Offline verification of the public replay runner; no inference or downloads.""" import copy import hashlib import importlib.util import json from pathlib import Path import tempfile import types import unittest from unittest.mock import patch spec = importlib.util.spec_from_file_location('runner', Path(__file__).with_name('reproduce_public.py')) r = importlib.util.module_from_spec(spec) spec.loader.exec_module(r) def h(raw): return hashlib.sha256(raw).hexdigest() def manifest(table): return json.dumps({'files_sha256': table}).encode() class IntegrityTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.data = {name: ('# fixture ' + name + '\n').encode() for name in r.LOCAL_PINS} self.pins = {name: h(raw) for name, raw in self.data.items()} for name, raw in self.data.items(): (self.root / name).write_bytes(raw) self.pin_patch = patch.object(r, 'LOCAL_PINS', self.pins) self.pin_patch.start() self.addCleanup(self.pin_patch.stop) self.addCleanup(self.temp.cleanup) def test_capture_exact_local_bytes(self): self.assertEqual(r.read_local(self.root), self.data) def test_changed_implementation_refused(self): (self.root / 'specialist_presentation.py').write_text('# modified\n') with self.assertRaisesRegex(ValueError, 'DIGEST_MISMATCH'): r.read_local(self.root) def test_changed_ui_refused(self): (self.root / 'specialist_review_ui.py').write_text('# modified\n') with self.assertRaisesRegex(ValueError, 'DIGEST_MISMATCH'): r.read_local(self.root) def test_zero_test_substitution_refused(self): (self.root / 'test_specialist_presentation.py').write_text('# contains no tests\n') with self.assertRaisesRegex(ValueError, 'DIGEST_MISMATCH'): r.read_local(self.root) def test_refusal_before_network_or_process(self): (self.root / 'test_specialist_presentation.py').write_text('# replacement\n') with patch.object(r, 'fetch') as fetch, patch.object(r.subprocess, 'run') as run: with self.assertRaises(ValueError): r.reproduce(self.root) fetch.assert_not_called() run.assert_not_called() def test_missing_local_source(self): (self.root / 'specialist_review_ui.py').unlink() with self.assertRaises(FileNotFoundError): r.read_local(self.root) def test_symlink_source_refused(self): p = self.root / 'specialist_review_ui.py' p.unlink() q = self.root / 'same-bytes';q.write_bytes(self.data[p.name]);p.symlink_to(q) with self.assertRaisesRegex(ValueError, 'REGULAR_SOURCE'): r.read_local(self.root) def test_directory_refused(self): p = self.root / 'specialist_review_ui.py';p.unlink();p.mkdir() with self.assertRaisesRegex(ValueError, 'REGULAR_SOURCE'): r.read_local(self.root) def test_size_bound(self): (self.root / 'specialist_review_ui.py').write_bytes(b'x'*(r.MAX_BYTES+1)) with self.assertRaisesRegex(ValueError, 'SIZE_BOUND'): r.read_local(self.root) def test_absolute_path_refused(self): with self.assertRaisesRegex(ValueError, 'PATH_REFUSED'):r.source_path('/tmp/x') def test_parent_path_refused(self): with self.assertRaisesRegex(ValueError, 'PATH_REFUSED'):r.source_path('join_assets/../x') def test_backslash_path_refused(self): with self.assertRaisesRegex(ValueError, 'PATH_REFUSED'):r.source_path('join_assets\\x') def test_empty_segment_refused(self): with self.assertRaisesRegex(ValueError, 'PATH_REFUSED'):r.source_path('join_assets//x') def test_correct_manifest_selects_fixed_source_scope(self): raw=manifest({'specialist_review.py':'a'*64,'join_assets/event.py':'b'*64,'unrelated.py':'c'*64}) with patch.object(r,'MANIFEST_SHA',h(raw)): self.assertEqual(set(r.parse_manifest(raw)),{'specialist_review.py','join_assets/event.py'}) def test_tampered_manifest_refused(self): raw=manifest({'specialist_review.py':'a'*64}) with patch.object(r,'MANIFEST_SHA',h(raw)): with self.assertRaisesRegex(ValueError,'DIGEST_MISMATCH'):r.parse_manifest(raw+b' ') def test_duplicate_manifest_key_refused(self): raw=b'{"files_sha256":{},"files_sha256":{}}' with patch.object(r,'MANIFEST_SHA',h(raw)): with self.assertRaisesRegex(ValueError,'DUPLICATE'):r.parse_manifest(raw) def test_manifest_missing_adapter_refused(self): raw=manifest({'join_assets/event.py':'a'*64}) with patch.object(r,'MANIFEST_SHA',h(raw)): with self.assertRaisesRegex(ValueError,'MISSING_ADAPTER'):r.parse_manifest(raw) def test_zero_tests_cannot_be_pass(self): with self.assertRaisesRegex(ValueError,'TEST_RUN_NOT_QUALIFIED'): r.require_test_result(0,'Ran 0 tests in 0.000s\n\nOK\n') def test_expected_18_tests_accepted(self): r.require_test_result(0,'..................\nRan 18 tests in 0.02s\n\nOK\n') def test_skipped_tests_refused(self): with self.assertRaises(ValueError):r.require_test_result(0,'Ran 18 tests in 0.1s\nOK (skipped=1)\n') def test_nonzero_returncode_refused(self): with self.assertRaises(ValueError):r.require_test_result(1,'Ran 18 tests in 0.1s\nOK\n') def test_no_summary_refused(self): with self.assertRaises(ValueError):r.require_test_result(0,'OK\n') def test_multiple_summaries_refused(self): with self.assertRaises(ValueError):r.require_test_result(0,'Ran 18 tests in 0.1s\nOK\nRan 0 tests in 0s\nOK\n') def test_expected_failure_refused(self): with self.assertRaises(ValueError):r.require_test_result(0,'Ran 18 tests in 0.1s\nOK (expected failures=1)\n') def test_source_capture_not_reread_after_validation(self): baseline=copy.deepcopy(self.data) remote={'specialist_review.py':b'# adapter','join_assets/event.py':b'# event'} m=manifest({name:h(raw) for name,raw in remote.items()}) def fetch(name): for local in self.data:(self.root/local).write_bytes(b'# changed during remote read') return m if name=='SOURCE_FILES.json' else remote[name] def run(*args,**kwargs): for name,raw in baseline.items():self.assertEqual((kwargs['cwd']/name).read_bytes(),raw) return types.SimpleNamespace(returncode=0,stdout='',stderr='Ran 18 tests in 0.1s\n\nOK\n') with patch.object(r,'MANIFEST_SHA',h(m)),patch.object(r,'fetch',side_effect=fetch),patch.object(r.subprocess,'run',side_effect=run): result=r.reproduce(self.root) self.assertEqual(result['tests'],18) self.assertEqual(result['local_sources_verified'],3) def test_manifest_count_bounded(self): raw=manifest({str(i):'a'*64 for i in range(65)}) with patch.object(r,'MANIFEST_SHA',h(raw)): with self.assertRaisesRegex(ValueError,'MANIFEST_SHAPE'):r.parse_manifest(raw) if __name__=='__main__':unittest.main(verbosity=2)