From 4a6e76b771ccfd869bd9861ca1b202d4e03d0d3d Mon Sep 17 00:00:00 2001 From: rUv Date: Tue, 13 Oct 2026 02:03:13 -0700 Subject: [PATCH] Bind confirmation plans to dataset records --- ...427-enforced-confirmation-training-mode.md | 18 ++- scripts/csi_training/confirmation.py | 128 ++++++++++++++---- scripts/csi_training/espfi.py | 5 +- scripts/csi_training/espfi_train.py | 4 +- .../tests/test_csi_training_confirmation.py | 93 ++++++++++++- scripts/tests/test_espfi.py | 12 +- scripts/tests/test_espfi_train.py | 46 +++++-- 7 files changed, 261 insertions(+), 45 deletions(-) diff --git a/docs/adr/ADR-427-enforced-confirmation-training-mode.md b/docs/adr/ADR-427-enforced-confirmation-training-mode.md index 282023f42..9538f6b7d 100644 --- a/docs/adr/ADR-427-enforced-confirmation-training-mode.md +++ b/docs/adr/ADR-427-enforced-confirmation-training-mode.md @@ -17,7 +17,16 @@ trainer. The mode requires a complete metadata plan, exposure ledger, frozen protocol SHA-256 and frozen dataset-manifest SHA-256. Before importing the dataset loader, decoding data, creating output, or training, the gate validates bounded regular files, leakage declarations, exposure history, coverage, and -both frozen hashes. Invalid input exits 1; valid but blocked evidence exits 2. +both frozen hashes. It also requires the plan and dataset manifest to have the +same retained signal-digest set, canonical dataset/revision, partition mapping +(`train`, `validation`, `test` to `train`, `development`, `confirmation`) and +numeric activity IDs. Room and participant declarations bind to the loader's +numeric source identities using canonical `room:N` and `participant:N` tokens. +Session and, when claimed, device identifiers must also exist in the dataset +manifest and match; missing provenance blocks the gate. Trial number is not +treated as a physical session. This binds supplied declarations to the exact +records the supervised loader will consume without assigning activity semantics. Invalid +input exits 1; valid but blocked evidence exits 2. The dataset loader's returned manifest digest is checked again before training. Confirmation mode cannot optimize an existing bundle and every receipt remains `promotionAuthorized: false`. @@ -30,10 +39,15 @@ the legacy training path. ## Consequences This closes accidental gate bypass within the selected mode and binds the -declared review to exact manifest bytes. It does not authenticate captures, +declared review to exact manifest bytes and retained record identities. It does not authenticate captures, consent, licenses, aliases, stable physical identities, or exposure-ledger completeness. The inputs are unsigned. The manifest hash binds declarations and referenced payload hashes, not provenance truth. A concurrent manifest change may be decoded before the loader's digest mismatch aborts, although it cannot reach training. Passing software tests makes no sensing-accuracy claim. +The current ESP-Fi adapter pins the canonical dataset name and source revision, +but its upstream records do not establish physical session or device identity. +Together with its prior exposure, this makes it correctly ineligible for an +independent-confirmation run even when its file hashes are intact. + diff --git a/scripts/csi_training/confirmation.py b/scripts/csi_training/confirmation.py index 2fc75eec9..f288d4dcd 100644 --- a/scripts/csi_training/confirmation.py +++ b/scripts/csi_training/confirmation.py @@ -113,45 +113,111 @@ def _pairs(pairs): return result -def read_json(path): - if not hasattr(os, 'O_NOFOLLOW') or not hasattr(os, 'O_NONBLOCK'): +def _read_regular(path, maximum): + if (type(maximum) is not int or not 1 <= maximum <= 16 * 1024 * 1024 + or not hasattr(os, 'O_NOFOLLOW') or not hasattr(os, 'O_NONBLOCK')): raise ValueError('Safe file intake unavailable on this platform') fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) with os.fdopen(fd, 'rb') as handle: info = os.fstat(handle.fileno()) - if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_BYTES: + if not stat.S_ISREG(info.st_mode) or info.st_size > maximum: raise ValueError('Expected bounded regular file') - raw = handle.read(MAX_BYTES + 1) - if len(raw) > MAX_BYTES: raise ValueError('Input too large') + raw = handle.read(info.st_size + 1) + if len(raw) != info.st_size: + raise ValueError('File changed during read') + return raw + + +def read_json_and_sha256(path, maximum=MAX_BYTES): + raw = _read_regular(path, maximum) def invalid(value): raise ValueError('Nonfinite JSON constant') - return json.loads(raw, object_pairs_hook=_pairs, parse_constant=invalid) + return (json.loads(raw, object_pairs_hook=_pairs, parse_constant=invalid), + hashlib.sha256(raw).hexdigest()) + + +def read_json(path): + return read_json_and_sha256(path)[0] def regular_sha256(path, maximum=MAX_DATASET_MANIFEST_BYTES): """Hash one bounded regular file through the descriptor that was checked.""" - if (type(maximum) is not int or not 1 <= maximum <= 16 * 1024 * 1024 - or not hasattr(os, 'O_NOFOLLOW') or not hasattr(os, 'O_NONBLOCK')): - raise ValueError('Safe file hashing unavailable') - fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) - digest_value = hashlib.sha256() - with os.fdopen(fd, 'rb') as handle: - info = os.fstat(handle.fileno()) - if not stat.S_ISREG(info.st_mode) or info.st_size > maximum: - raise ValueError('Expected bounded regular file') - remaining = info.st_size - while remaining: - block = handle.read(min(1024 * 1024, remaining)) - if not block: - raise ValueError('File changed during hashing') - digest_value.update(block) - remaining -= len(block) - if handle.read(1): - raise ValueError('File changed during hashing') - return digest_value.hexdigest() + return hashlib.sha256(_read_regular(path, maximum)).hexdigest() + + +def _dataset_binding(manifest, dataset, required_source=None): + """Bind declared confirmation records to a supervised ESP-Fi manifest.""" + if type(dataset) is not dict: + raise ValueError('Dataset manifest must be an object') + if type(dataset.get('schema_version')) is not int or dataset['schema_version'] != 1: + raise ValueError('Unsupported dataset manifest version') + dataset_name = dataset.get('dataset') + revision = dataset.get('source_revision') + if type(dataset_name) is not str or not 1 <= len(dataset_name) <= 128: + raise ValueError('Invalid dataset identity') + token(revision) + dataset_source = (dataset_name, revision) + records = dataset.get('records') + if type(records) is not list or not 1 <= len(records) <= MAX_RECORDS: + raise ValueError('Invalid dataset record count') + split_map = {'train': 'train', 'validation': 'development', 'test': 'confirmation'} + retained, duplicates = {}, False + for row in records: + if type(row) is not dict or row.get('split') not in (*split_map, None): + raise ValueError('Invalid dataset record') + signal = digest(row.get('signal_sha256')) + identity = row.get('identity') + if (type(identity) is not list or len(identity) != 4 + or any(type(value) is not int for value in identity) + or not 1 <= identity[2] <= 7): + raise ValueError('Invalid dataset identity tuple') + if row['split'] is not None: + if signal in retained: + duplicates = True + retained[signal] = row + planned = {row['signal_sha256']: row for row in manifest['records']} + common = set(retained) & set(planned) + reasons = set() + if required_source is not None: + if (type(required_source) is not tuple or len(required_source) != 2 + or type(required_source[0]) is not str or type(required_source[1]) is not str): + raise ValueError('Invalid required source') + if dataset_source != required_source: + reasons.add('canonical_dataset_source_mismatch') + if duplicates: + reasons.add('dataset_duplicate_signal_digest') + if set(retained) != set(planned): + reasons.add('dataset_record_set_mismatch') + expected_source = dataset_source + for signal in common: + plan, row = planned[signal], retained[signal] + if source(plan['source']) != expected_source: + reasons.add('dataset_source_mismatch') + if plan['split'] != split_map[row['split']]: + reasons.add('dataset_split_mismatch') + if plan['activity'] != row['identity'][2] - 1: + reasons.add('dataset_activity_mismatch') + if plan['room'] != 'room:' + str(row['identity'][0]): + reasons.add('dataset_room_mismatch') + if plan['participant'] != 'participant:' + str(row['identity'][1]): + reasons.add('dataset_participant_mismatch') + session = row.get('session_id') + if session is None: + reasons.add('dataset_session_identity_unbound') + elif plan['session'] != token(session): + reasons.add('dataset_session_mismatch') + if manifest['claim'].endswith('_device'): + device = row.get('device_id') + if device is None: + reasons.add('dataset_device_identity_unbound') + elif plan['device'] != token(device): + reasons.add('dataset_device_mismatch') + return reasons, {'retained_records': len(retained), 'planned_records': len(planned), + 'matched_records': len(common), 'exact_record_set': set(retained) == set(planned)} def training_gate(manifest_path, ledger_path, dataset_manifest_path, - expected_protocol_sha256, expected_dataset_manifest_sha256): + expected_protocol_sha256, expected_dataset_manifest_sha256, + required_source=None): """Evaluate declared metadata and bind it to frozen protocol/dataset bytes. This remains a metadata gate. It does not authenticate captures, consent, @@ -159,19 +225,23 @@ def training_gate(manifest_path, ledger_path, dataset_manifest_path, """ digest(expected_protocol_sha256) digest(expected_dataset_manifest_sha256) - report = evaluate(read_json(manifest_path), read_json(ledger_path)) + manifest = read_json(manifest_path) + report = evaluate(manifest, read_json(ledger_path)) reasons = set(report['reasons']) if report['protocol_sha256'] != expected_protocol_sha256: reasons.add('protocol_digest_mismatch') - actual = regular_sha256(dataset_manifest_path) + dataset, actual = read_json_and_sha256(dataset_manifest_path, MAX_DATASET_MANIFEST_BYTES) if actual != expected_dataset_manifest_sha256: reasons.add('dataset_manifest_digest_mismatch') + binding_reasons, binding = _dataset_binding(manifest, dataset, required_source) + reasons.update(binding_reasons) result = dict(report) result.update({'preflight_passed': not reasons, 'reasons': sorted(reasons), 'dataset_manifest_sha256': actual, 'expected_dataset_manifest_sha256': expected_dataset_manifest_sha256, - 'scope': report['scope'] + ' Dataset bytes are hash-pinned; provenance claims remain unauthenticated.'}) + 'dataset_binding': binding, + 'scope': report['scope'] + ' Dataset records and bytes are bound to declarations; provenance claims remain unauthenticated.'}) return result diff --git a/scripts/csi_training/espfi.py b/scripts/csi_training/espfi.py index ba04e899b..cb1b2ed25 100644 --- a/scripts/csi_training/espfi.py +++ b/scripts/csi_training/espfi.py @@ -14,6 +14,7 @@ import zipfile import numpy as np REVISION = 'c8bf6aa0d680fa02695c85a82b05e79b3ebb10c4' +DATASET = 'ESP-Fi HAR' ARCHIVE_SHA256 = '85f3df633c5b1a48ca4b1423467fce22821b21eeb3eed085a8d9f983dc156660' ARCHIVE_BYTES = 644939981 SOURCE_URL = f'https://media.githubusercontent.com/media/AutoSmartGroup/ESP-Fi-HAR/{REVISION}/ESP-Fi_dataset.rar' @@ -126,7 +127,7 @@ def prepare(archive, output, pool=5): output.mkdir(parents=True) target = output / 'amplitudes.npz' np.savez_compressed(target, **arrays) - manifest = {'schema_version': 1, 'dataset': 'ESP-Fi HAR', 'source_url': SOURCE_URL, + manifest = {'schema_version': 1, 'dataset': DATASET, 'source_url': SOURCE_URL, 'source_revision': REVISION, 'archive_sha256': ARCHIVE_SHA256, 'license': 'CC-BY-4.0', 'license_url': f'https://github.com/AutoSmartGroup/ESP-Fi-HAR/blob/{REVISION}/Model%20Code/DATA_LICENSE.txt', 'attribution': 'Wen et al. ESP-Fi HAR. Ad Hoc Networks 186 (2026) 104192. doi:10.1016/j.adhoc.2026.104192', @@ -164,7 +165,7 @@ def _load_partitions(manifest, splits): meta = json.loads(path.read_text()) if meta.get('schema_version') != 1 or meta.get('classes') != CLASSES: raise ValueError('unsupported dataset contract') - if (meta.get('source_revision') != REVISION or meta.get('archive_sha256') != ARCHIVE_SHA256 + if (meta.get('dataset') != DATASET or meta.get('source_revision') != REVISION or meta.get('archive_sha256') != ARCHIVE_SHA256 or meta.get('source_url') != SOURCE_URL or meta.get('license') != 'CC-BY-4.0' or meta.get('split_policy') != 'joint_room_subject_intersection_v1' or meta.get('temporal_mean_pool') not in (1, 2, 5, 10, 19, 25, 50, 95) diff --git a/scripts/csi_training/espfi_train.py b/scripts/csi_training/espfi_train.py index 9f5286d46..524846ce6 100644 --- a/scripts/csi_training/espfi_train.py +++ b/scripts/csi_training/espfi_train.py @@ -378,10 +378,12 @@ def main(argv=None): preflight = None if gate_enabled: from .confirmation import training_gate + from .espfi import DATASET, REVISION try: preflight = training_gate(args.confirmation_manifest, args.exposure_ledger, args.manifest, args.expected_protocol_sha256, - args.expected_dataset_manifest_sha256) + args.expected_dataset_manifest_sha256, + required_source=(DATASET, REVISION)) except (ValueError, OSError, TypeError, RecursionError, OverflowError): print(json.dumps({'status': 'invalid_input', 'promotionAuthorized': False})) raise SystemExit(1) diff --git a/scripts/tests/test_csi_training_confirmation.py b/scripts/tests/test_csi_training_confirmation.py index 66d6e6132..2372efdbf 100644 --- a/scripts/tests/test_csi_training_confirmation.py +++ b/scripts/tests/test_csi_training_confirmation.py @@ -16,14 +16,27 @@ def fixture(): for activity in range(7): for trial in range(10 if split == 'confirmation' else 1): name = f'{split}:{activity}:{trial}' + group = {'train': 1, 'development': 2}.get(split, 3 + trial % 2) records.append(dict(id=name, signal_sha256=hashlib.sha256(name.encode()).hexdigest(), source={'dataset':'synthetic', 'revision':'1'}, split=split, - room=f'{split}:room:{trial%2}', participant=f'{split}:person:{trial%2}', + room=f'room:{group}', participant=f'participant:{group}', session=f'{split}:session:{trial%2}', device=f'{split}:device', activity=activity)) return {'schema_version':1, 'protocol_sha256':'a'*64, 'claim':'new_room_person_device', 'records':records}, {'schema_version':1,'exposed_sources':[]} +def dataset_fixture(manifest): + split_map = {'train':'train', 'development':'validation', 'confirmation':'test'} + return {'schema_version':1, 'dataset':'synthetic', 'source_revision':'1', + 'records':[{'signal_sha256':row['signal_sha256'], + 'split':split_map[row['split']], + 'identity':[int(row['room'].split(':')[1]), + int(row['participant'].split(':')[1]), + row['activity'] + 1, index + 1], + 'session_id':row['session'], 'device_id':row['device']} + for index,row in enumerate(manifest['records'])]} + + class ConfirmationTests(unittest.TestCase): def test_success_is_only_metadata(self): m,l=fixture(); r=evaluate(m,l) @@ -96,7 +109,7 @@ class ConfirmationTests(unittest.TestCase): with tempfile.TemporaryDirectory() as d: m,l=fixture();root=Path(d);dataset=root/'dataset.json' p=root/'m.json';q=root/'l.json' - dataset.write_text('{"schema_version":1}') + dataset.write_text(json.dumps(dataset_fixture(m))) p.write_text(json.dumps(m));q.write_text(json.dumps(l)) frozen=regular_sha256(dataset) report=training_gate(p,q,dataset,'a'*64,frozen) @@ -106,6 +119,82 @@ class ConfirmationTests(unittest.TestCase): self.assertIn('protocol_digest_mismatch',training_gate(p,q,dataset,'b'*64,frozen)['reasons']) self.assertIn('dataset_manifest_digest_mismatch',training_gate(p,q,dataset,'a'*64,'0'*64)['reasons']) + def test_training_gate_binds_each_dataset_record_field(self): + changes = [('signal_sha256','dataset_record_set_mismatch'), + ('source','dataset_source_mismatch'), + ('split','dataset_split_mismatch'), + ('activity','dataset_activity_mismatch'), + ('room','dataset_room_mismatch'), + ('participant','dataset_participant_mismatch'), + ('session','dataset_session_mismatch'), + ('device','dataset_device_mismatch')] + with tempfile.TemporaryDirectory() as d: + root=Path(d) + for field,reason in changes: + with self.subTest(field=field): + m,l=fixture();dataset=dataset_fixture(m) + if field == 'signal_sha256': + dataset['records'][0]['signal_sha256']='f'*64 + elif field == 'source': + m['records'][0]['source']={'dataset':'other','revision':'1'} + elif field == 'split': + dataset['records'][0]['split']='validation' + elif field == 'activity': + dataset['records'][0]['identity'][2]=2 if m['records'][0]['activity'] != 1 else 3 + elif field == 'room': + dataset['records'][0]['identity'][0]=8 + elif field == 'participant': + dataset['records'][0]['identity'][1]=8 + elif field == 'session': + dataset['records'][0]['session_id']='other:session' + else: + dataset['records'][0]['device_id']='other:device' + p=root/'m.json';q=root/'l.json';data=root/'dataset.json' + p.write_text(json.dumps(m));q.write_text(json.dumps(l));data.write_text(json.dumps(dataset)) + report=training_gate(p,q,data,'a'*64,regular_sha256(data)) + self.assertIn(reason,report['reasons']) + self.assertFalse(report['preflight_passed']) + + def test_training_gate_rejects_duplicate_retained_signal(self): + with tempfile.TemporaryDirectory() as d: + m,l=fixture();dataset=dataset_fixture(m) + dataset['records'][1]['signal_sha256']=dataset['records'][0]['signal_sha256'] + root=Path(d);p=root/'m';q=root/'l';data=root/'data' + p.write_text(json.dumps(m));q.write_text(json.dumps(l));data.write_text(json.dumps(dataset)) + reasons=training_gate(p,q,data,'a'*64,regular_sha256(data))['reasons'] + self.assertIn('dataset_duplicate_signal_digest',reasons) + self.assertIn('dataset_record_set_mismatch',reasons) + + def test_training_gate_ignores_excluded_dataset_record(self): + with tempfile.TemporaryDirectory() as d: + m,l=fixture();dataset=dataset_fixture(m) + dataset['records'].append({'signal_sha256':'f'*64, 'split':None, + 'identity':[9,9,1,1]}) + root=Path(d);p=root/'m';q=root/'l';data=root/'data' + p.write_text(json.dumps(m));q.write_text(json.dumps(l));data.write_text(json.dumps(dataset)) + report=training_gate(p,q,data,'a'*64,regular_sha256(data)) + self.assertTrue(report['preflight_passed'],report['reasons']) + self.assertTrue(report['dataset_binding']['exact_record_set']) + + def test_training_gate_rejects_malformed_dataset_binding(self): + with tempfile.TemporaryDirectory() as d: + m,l=fixture();dataset=dataset_fixture(m);dataset['records'][0]['identity'][2]=True + root=Path(d);p=root/'m';q=root/'l';data=root/'data' + p.write_text(json.dumps(m));q.write_text(json.dumps(l));data.write_text(json.dumps(dataset)) + with self.assertRaises(ValueError): + training_gate(p,q,data,'a'*64,regular_sha256(data)) + + def test_training_gate_fails_closed_without_session_or_device_identity(self): + with tempfile.TemporaryDirectory() as d: + m,l=fixture();dataset=dataset_fixture(m) + for row in dataset['records']: + row.pop('session_id');row.pop('device_id') + root=Path(d);p=root/'m';q=root/'l';data=root/'data' + p.write_text(json.dumps(m));q.write_text(json.dumps(l));data.write_text(json.dumps(dataset)) + reasons=training_gate(p,q,data,'a'*64,regular_sha256(data))['reasons'] + self.assertIn('dataset_session_identity_unbound',reasons) + self.assertIn('dataset_device_identity_unbound',reasons) + def test_regular_sha256_rejects_unsafe_files(self): with tempfile.TemporaryDirectory() as d: root=Path(d);target=root/'target';target.write_bytes(b'ok') diff --git a/scripts/tests/test_espfi.py b/scripts/tests/test_espfi.py index aba693d29..dfacb7a30 100644 --- a/scripts/tests/test_espfi.py +++ b/scripts/tests/test_espfi.py @@ -83,7 +83,7 @@ class EspfiTests(unittest.TestCase): 'signal_sha256': selected.get(i, hashlib.sha256(str(i).encode()).hexdigest())}) npz = Path(tmp) / 'amplitudes.npz' np.savez_compressed(npz, **arrays) - meta = dict(schema_version=1, classes=espfi.CLASSES, source_revision=espfi.REVISION, + meta = dict(schema_version=1, dataset=espfi.DATASET, classes=espfi.CLASSES, source_revision=espfi.REVISION, source_url=espfi.SOURCE_URL, archive_sha256=espfi.ARCHIVE_SHA256, license='CC-BY-4.0', split_policy='joint_room_subject_intersection_v1', temporal_mean_pool=95, window_samples=10, records=records, @@ -148,6 +148,16 @@ class EspfiTests(unittest.TestCase): with self.assertRaisesRegex(ValueError, 'provenance'): espfi.load_dataset(p) + def test_loader_rejects_source_alias(self): + with tempfile.TemporaryDirectory() as tmp: + manifest, _, _, meta = self.fixture(tmp) + for value in (None, 'renamed-source'): + with self.subTest(value=value): + meta['dataset'] = value + manifest.write_text(json.dumps(meta)) + with self.assertRaisesRegex(ValueError, 'provenance'): + espfi.load_dataset(manifest) + if __name__ == '__main__': unittest.main() diff --git a/scripts/tests/test_espfi_train.py b/scripts/tests/test_espfi_train.py index 93b11d88e..3dfc86053 100644 --- a/scripts/tests/test_espfi_train.py +++ b/scripts/tests/test_espfi_train.py @@ -13,7 +13,7 @@ import torch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from csi_training.espfi_train import AmplitudeBundle, AmplitudeClassifier, main, optimize, optimize_bundle, scores from csi_training.confirmation import regular_sha256 -from scripts.tests.test_csi_training_confirmation import fixture as confirmation_fixture +from scripts.tests.test_csi_training_confirmation import dataset_fixture, fixture as confirmation_fixture def fixture(): @@ -116,7 +116,7 @@ class ConfirmationGateIntegrationTests(unittest.TestCase): plan, ledger = confirmation_fixture() dataset = root / 'dataset.json' plan_path, ledger_path = root / 'plan.json', root / 'ledger.json' - dataset.write_text('{"schema_version":1}') + dataset.write_text(json.dumps(dataset_fixture(plan))) plan_path.write_text(json.dumps(plan)) ledger_path.write_text(json.dumps(ledger)) return dataset, plan_path, ledger_path @@ -132,7 +132,8 @@ class ConfirmationGateIntegrationTests(unittest.TestCase): root=Path(d);dataset,plan,ledger=self.gate_files(root) value=json.loads(plan.read_text());value['records'][-1]['room']=value['records'][0]['room'] plan.write_text(json.dumps(value)) - with patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with patch.multiple('csi_training.espfi',DATASET='synthetic',REVISION='1'), \ + patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: with self.assertRaisesRegex(SystemExit, '2'): main(self.argv(root,dataset,plan,ledger)) load.assert_not_called();train.assert_not_called() @@ -144,12 +145,16 @@ class ConfirmationGateIntegrationTests(unittest.TestCase): cases=[(self.argv(root,dataset,plan,ledger)[:-2],1), (self.argv(root,dataset,plan,ledger)[:-1]+['0'*64],2)] for args,code in cases: - with self.subTest(code=code), patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with self.subTest(code=code), \ + patch.multiple('csi_training.espfi',DATASET='synthetic',REVISION='1'), \ + patch('csi_training.espfi.load_dataset') as load, \ + patch('csi_training.espfi_train.optimize') as train: with self.assertRaisesRegex(SystemExit,str(code)):main(args) load.assert_not_called();train.assert_not_called() self.assertFalse((root/'out').exists()) plan.write_text('{bad json') - with patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with patch.multiple('csi_training.espfi',DATASET='synthetic',REVISION='1'), \ + patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: with self.assertRaisesRegex(SystemExit,'1'): main(self.argv(root,dataset,plan,ledger)) load.assert_not_called();train.assert_not_called() @@ -181,15 +186,39 @@ class ConfirmationGateIntegrationTests(unittest.TestCase): root=Path(d);dataset,plan,ledger=self.gate_files(root) args=self.argv(root,dataset,plan,ledger) args[args.index('--expected-protocol-sha256')+1]='b'*64 - with patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with patch.multiple('csi_training.espfi',DATASET='synthetic',REVISION='1'), \ + patch('csi_training.espfi.load_dataset') as load, \ + patch('csi_training.espfi_train.optimize') as train: with self.assertRaisesRegex(SystemExit,'2'):main(args) load.assert_not_called();train.assert_not_called() + def test_record_binding_mismatch_never_loads(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + value=json.loads(dataset.read_text());value['records'][0]['split']='validation' + dataset.write_text(json.dumps(value)) + with patch.multiple('csi_training.espfi',DATASET='synthetic',REVISION='1'), \ + patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with self.assertRaisesRegex(SystemExit,'2'): + main(self.argv(root,dataset,plan,ledger)) + load.assert_not_called();train.assert_not_called() + self.assertFalse((root/'out').exists()) + + def test_canonical_source_mismatch_never_loads(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + with patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with self.assertRaisesRegex(SystemExit,'2'): + main(self.argv(root,dataset,plan,ledger)) + load.assert_not_called();train.assert_not_called() + self.assertFalse((root/'out').exists()) + def test_manifest_race_stops_before_training(self): with tempfile.TemporaryDirectory() as d: root=Path(d);dataset,plan,ledger=self.gate_files(root) parts,provenance=fixture();provenance['manifest_sha256']='f'*64 - with patch('csi_training.espfi.load_dataset',return_value=(parts,provenance)) as load, \ + with patch.multiple('csi_training.espfi',DATASET='synthetic',REVISION='1'), \ + patch('csi_training.espfi.load_dataset',return_value=(parts,provenance)) as load, \ patch('csi_training.espfi_train.optimize') as train: with self.assertRaisesRegex(ValueError,'changed'): main(self.argv(root,dataset,plan,ledger)) @@ -209,7 +238,8 @@ class ConfirmationGateIntegrationTests(unittest.TestCase): root=Path(d);dataset,plan,ledger=self.gate_files(root) parts,provenance=fixture();provenance['manifest_sha256']=regular_sha256(dataset) report={'variant':'statistics','test':{},'latency':{},'promotionAuthorized':False} - with patch('csi_training.espfi.load_dataset',return_value=(parts,provenance)) as load, \ + with patch.multiple('csi_training.espfi',DATASET='synthetic',REVISION='1'), \ + patch('csi_training.espfi.load_dataset',return_value=(parts,provenance)) as load, \ patch('csi_training.espfi_train.optimize',return_value=report) as train: main(self.argv(root,dataset,plan,ledger)) load.assert_called_once();train.assert_called_once()