"""Persistence invariants: complete data, compatible sets and recoverable write failures.""" from __future__ import annotations import json import os from pathlib import Path import stat import tempfile from typing import cast import unittest from unittest import mock import numpy as np from python_app.models.dataset_model import ( ComboKey, ResultBlock, ResultCollection, ResultPayload, SweepCollection, TraceData, ) from python_app.storage.npz.store import NpzStore from python_app.storage.npz.vna_history_json import build_vna_history_payload from python_app.storage.npz.vna_history_json import collection_directories, load_vna_history_snapshot from python_app.workflows.sequential_capture_workflow import combine_collections_via_median, combine_traces_via_median def _sweep(cid: int = 1, ns: int = 20, *, start: int = 10, end: int = 15) -> SweepCollection: return SweepCollection( collection_id=cid, monotonic_ns=ns, capture_start_ns=start, capture_end_ns=end, traces=[TraceData( combo=ComboKey(0, 0), frequency_hz=np.array([1e9, 2e9], dtype=np.float32), s21=np.array([5 - 6j, 7 + 8j], dtype=np.complex64), capture_start_ns=start, capture_end_ns=end, )], ) def _payloads() -> list[ResultPayload]: return [ ResultPayload('trace', 1, frequency_hz=np.array([1, 2], dtype=np.float32), trace=np.array([1 + 2j, 3 - 4j], dtype=np.complex64)), ResultPayload('scalar', 2, scalar_value=3.5), ResultPayload('image', 3, image_x_axis=np.array([1, 2], dtype=np.float32), image_y_axis=np.array([3], dtype=np.float32), image=np.array([[4, 5]], dtype=np.float32)), ResultPayload('table', 4, table=np.array([[1, 2, 3], [4, 5, 6]], dtype=np.float32)), ] class PersistenceIntegrityTest(unittest.TestCase): def setUp(self) -> None: temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.store = NpzStore(self.root / 'sets') self.output = self.root / 'output' def test_all_calibration_and_reference_kinds_preserve_arrays_and_timing(self) -> None: for kind in ('s21_calibration', 's21_reference'): with self.subTest(kind=kind): original = _sweep(7, 100) second = _sweep().traces[0] second.combo = ComboKey(2, 3) original.traces.append(second) self.store.save_set(kind, 'radar', 'full', original) loaded = self.store.load_set(kind, 'radar', 'full') self.assertEqual((loaded.collection_id, loaded.monotonic_ns), (7, 100)) self.assertEqual((loaded.capture_start_ns, loaded.capture_end_ns), (10, 15)) self.assertTrue(self.store.has_combo_coverage(kind, 'radar', 'full', [ComboKey(0, 0), ComboKey(2, 3)])) self.assertFalse(self.store.has_combo_coverage(kind, 'radar', 'full', [ComboKey(9, 9)])) for before, after in zip(original.traces, loaded.traces, strict=True): self.assertEqual(before.combo, after.combo) self.assertEqual((after.capture_start_ns, after.capture_end_ns), (10, 15)) for field in ('frequency_hz', 's21'): np.testing.assert_array_equal(getattr(before, field), getattr(after, field)) def test_failed_set_commit_preserves_previous_data_and_sidecar(self) -> None: self.store.save_set('s21_reference', 'radar', 'full', _sweep(1, 20)) original_meta = (self.root / 'sets/s21_reference/radar/full.json').read_bytes() replace = Path.replace def fail_archive_commit(path, destination): if Path(destination).suffix == '.npz': raise OSError('disk full during commit') return replace(path, destination) with mock.patch.object(Path, 'replace', fail_archive_commit), self.assertRaises(OSError): self.store.save_set('s21_reference', 'radar', 'full', _sweep(2, 200)) self.assertEqual(self.store.load_set('s21_reference', 'radar', 'full').collection_id, 1) self.assertEqual((self.root / 'sets/s21_reference/radar/full.json').read_bytes(), original_meta) self.assertEqual(list(self.root.rglob('*.tmp')), []) def test_failed_new_set_commit_does_not_publish_a_sidecar_only_set(self) -> None: replace = Path.replace def fail_archive_commit(path, destination): if Path(destination).suffix == '.npz': raise OSError('commit failed') return replace(path, destination) with mock.patch.object(Path, 'replace', fail_archive_commit), self.assertRaises(OSError): self.store.save_set('s21_reference', 'radar', 'full', _sweep()) self.assertEqual(self.store.list_sets('s21_reference', 'radar'), []) self.assertEqual(list(self.root.rglob('*.json')), []) self.assertEqual(list(self.root.rglob('*.tmp')), []) def test_failed_sidecar_commit_restores_previous_archive_without_rewriting_it(self) -> None: self.store.save_set('s21_reference', 'radar', 'full', _sweep(1, 20)) original = (self.root / 'sets/s21_reference/radar/full.npz').read_bytes() replace = Path.replace def fail_metadata_commit(path, destination): if Path(destination).suffix == '.json': raise OSError('metadata commit failed') return replace(path, destination) with mock.patch.object(Path, 'replace', fail_metadata_commit), self.assertRaises(OSError): self.store.save_set('s21_reference', 'radar', 'full', _sweep(2, 200)) self.assertEqual((self.root / 'sets/s21_reference/radar/full.npz').read_bytes(), original) self.assertEqual(self.store.load_set('s21_reference', 'radar', 'full').collection_id, 1) self.assertEqual(list(self.root.rglob('*.tmp')), []) def test_set_files_are_synced_before_commit_and_directory_afterwards(self) -> None: events = [] fsync, replace = os.fsync, Path.replace def sync_file(descriptor): events.append('directory' if stat.S_ISDIR(os.fstat(descriptor).st_mode) else 'file') fsync(descriptor) def commit(path, destination): events.append('replace') return replace(path, destination) with mock.patch('os.fsync', side_effect=sync_file), mock.patch.object(Path, 'replace', commit): self.store.save_set('s21_calibration', 'radar', 'full', _sweep()) first_file = events.index('file') self.assertEqual(events[:first_file], ['directory', 'directory']) self.assertEqual(events[first_file:], ['file', 'file', 'replace', 'replace', 'directory']) def test_set_sync_failure_does_not_replace_previous_measurement(self) -> None: self.store.save_set('s21_calibration', 'radar', 'full', _sweep()) archive = self.root / 'sets/s21_calibration/radar/full.npz' original = archive.read_bytes() for failed_sync in (1, 2): with self.subTest(failed_sync=failed_sync), mock.patch('python_app.storage.npz.store.logger.exception'): with mock.patch('os.fsync', side_effect=[None] * (failed_sync - 1) + [OSError('sync failed')]): with self.assertRaises(OSError): self.store.save_set('s21_calibration', 'radar', 'full', _sweep(2, 30)) self.assertEqual(archive.read_bytes(), original) self.assertEqual(list(archive.parent.glob('*.tmp')), []) def test_sets_and_snapshots_preserve_permissions_and_respect_umask(self) -> None: previous = os.umask(0o027) try: self.store.save_set('s21_calibration', 'radar', 'full', _sweep()) directory = self.store.save_runtime_snapshot_numpy(self.output, 'full', [_sweep()], [], []) finally: os.umask(previous) for path in (self.root / 'sets/s21_calibration/radar').iterdir(): self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o640) path.chmod(0o664) for path in [directory, *directory.rglob('*')]: self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o750 if path.is_dir() else 0o640) self.store.save_set('s21_calibration', 'radar', 'full', _sweep(2, 30)) for path in (self.root / 'sets/s21_calibration/radar').iterdir(): self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o664) def test_invalid_sets_are_rejected_before_overwriting_valid_data(self) -> None: self.store.save_set('s21_calibration', 'radar', 'full', _sweep()) invalid = [_sweep(), _sweep(), _sweep(), _sweep()] invalid[0].traces.append(invalid[0].traces[0]) invalid[1].traces[0].s21 = np.array([1], dtype=np.complex64) invalid[2].traces[0].frequency_hz = np.array([[1, 2]], dtype=np.float32) invalid[3].traces[0].s21[0] = complex(float('nan'), 0) for collection in invalid: with self.subTest(collection=collection), self.assertRaises(ValueError): self.store.save_set('s21_calibration', 'radar', 'full', collection) self.assertEqual(self.store.load_set('s21_calibration', 'radar', 'full').monotonic_ns, 20) def test_set_names_cannot_escape_store(self) -> None: for name in ('../escape', 'sub/name', '..', '.', '', r'sub\name'): with self.subTest(name=name), self.assertRaises(ValueError): self.store.save_set('s21_reference', 'radar', name, _sweep()) def test_invalid_channel_identifiers_are_rejected_without_truncation(self) -> None: for position in (0.5, True, -1, 2 ** 32): sweep = _sweep() sweep.traces[0].combo = ComboKey(cast(int, position), 0) with self.subTest(position=position), self.assertRaises(ValueError): self.store.save_set('s21_reference', 'radar', 'invalid', sweep) self.assertEqual(self.store.list_sets('s21_reference', 'radar'), []) def test_corrupted_metadata_types_are_rejected_without_truncation(self) -> None: self.store.save_set('s21_reference', 'radar', 'full', _sweep()) archive = self.root / 'sets/s21_reference/radar/full.npz' with np.load(archive, allow_pickle=False) as stored: arrays = {key: stored[key] for key in stored.files} original = json.loads(str(arrays['metadata_json'].item())) for malformed in ([], {**original, 'collection_id': 1.5}, {**original, 'monotonic_ns': True}, {**original, 'combos': [{**original['combos'][0], 'input': 0.5}]}): with self.subTest(metadata=malformed): np.savez(archive, **{**arrays, 'metadata_json': np.asarray(json.dumps(malformed))}) with self.assertRaisesRegex(RuntimeError, 'Corrupted preprocess set'): self.store.load_set('s21_reference', 'radar', 'full') def test_complete_snapshot_keeps_unmatched_stage_tails(self) -> None: raw = [_sweep(1, 10), _sweep(2, 30)] pre = [_sweep(1, 20), _sweep(3, 40)] result = [ResultCollection(1, 20)] directory = self.store.save_runtime_snapshot_numpy( self.output, 'full', raw, pre, result, ) for stage, ids in [('raw', [1, 2]), ('preprocessed', [1, 3]), ('results', [1])]: metadata = [json.loads(path.read_text()) for path in sorted((directory / stage).glob('*/meta.json'))] self.assertEqual([item['collection_id'] for item in metadata], ids) def test_snapshot_result_payloads_preserve_every_kind_at_both_scopes(self) -> None: payloads = _payloads() result = ResultCollection(1, 20, processing_duration_ns=12345, settings_revision=2**63 + 7, capture_start_ns=2**60 + 3, capture_end_ns=2**60 + 9, speed_m_s=1.25, collection_payloads=payloads, blocks=[ResultBlock(ComboKey(0, 0), payloads)]) directory = self.store.save_runtime_snapshot_numpy(self.output, 'full', [], [], [result]) collection_dir = next((directory / 'results').iterdir()) meta = json.loads((collection_dir / 'meta.json').read_text()) self.assertEqual(meta['processing_duration_ns'], 12345) self.assertEqual((meta['settings_revision'], meta['capture_start_ns'], meta['capture_end_ns'], meta['speed_m_s']), (2**63 + 7, 2**60 + 3, 2**60 + 9, 1.25)) for base, entries in [(collection_dir, meta['collection_payloads']), (collection_dir / meta['blocks'][0]['dir'], meta['blocks'][0]['payloads'])]: self.assertEqual([entry['kind'] for entry in entries], [1, 2, 3, 4]) np.testing.assert_array_equal(np.load(base / entries[0]['freq_file']), payloads[0].frequency_hz) np.testing.assert_array_equal(np.load(base / entries[0]['trace_file']), payloads[0].trace) np.testing.assert_array_equal(np.load(base / entries[1]['scalar_file']), np.array([3.5], dtype=np.float32)) np.testing.assert_array_equal(np.load(base / entries[2]['x_axis_file']), payloads[2].image_x_axis) np.testing.assert_array_equal(np.load(base / entries[2]['y_axis_file']), payloads[2].image_y_axis) np.testing.assert_array_equal(np.load(base / entries[2]['image_file']), payloads[2].image) np.testing.assert_array_equal(np.load(base / entries[3]['table_file']), payloads[3].table) def test_bad_result_payload_rolls_back_entire_snapshot(self) -> None: for payload in (ResultPayload('unknown', 9), ResultPayload('bad_table', 4, table=np.array([1, 2]))): with self.subTest(kind=payload.kind), self.assertRaises(ValueError): self.store.save_runtime_snapshot_numpy( self.output, 'failed', [_sweep()], [], [ResultCollection(1, 20, collection_payloads=[payload])], ) self.assertFalse((self.output / 'failed').exists()) self.assertEqual(list(self.output.iterdir()), []) def test_snapshot_write_failure_can_be_retried_with_same_name(self) -> None: with mock.patch('numpy.save', side_effect=OSError('disk full')), self.assertRaises(OSError): self.store.save_runtime_snapshot_numpy(self.output, 'retry', [_sweep()], [], []) directory = self.store.save_runtime_snapshot_numpy(self.output, 'retry', [_sweep()], [], []) self.assertTrue(directory.exists()) with self.assertRaises(FileExistsError): self.store.save_runtime_snapshot_numpy(self.output, 'retry', [_sweep()], [], []) def test_stream_failure_keeps_prior_chunks_and_retries_without_partial_data(self) -> None: writer = self.store.create_snapshot_stream(self.output, 'stream') writer.append([_sweep()], [], [ResultCollection(1, 20)]) with mock.patch('numpy.save', side_effect=OSError('disk full')), self.assertRaises(OSError): writer.append([_sweep(2, 30)], [], [ResultCollection(2, 40)]) self.assertEqual(writer.result_count, 1) self.assertEqual(len(list((writer.directory / 'raw').iterdir())), 1) self.assertEqual(list(writer.directory.glob('.chunk-*')), []) writer.append([_sweep(2, 30)], [], [ResultCollection(2, 40)]) self.assertEqual(writer.result_count, 2) self.assertEqual([path.name.split('_')[0] for path in sorted((writer.directory / 'raw').iterdir())], ['0000', '0001']) def test_stream_failure_preserves_collections_already_published_in_current_chunk(self) -> None: writer = self.store.create_snapshot_stream(self.output, 'partial') good = ResultCollection(1, 20) bad = ResultCollection(2, 40, collection_payloads=[ResultPayload('invalid', 9)]) with self.assertRaises(ValueError): writer.append([_sweep(1, 10), _sweep(2, 30)], [], [good, bad]) self.assertEqual(writer.result_count, 1) self.assertEqual(len(list((writer.directory / 'raw').iterdir())), 2) self.assertEqual(len(list((writer.directory / 'results').iterdir())), 1) writer.append([], [], [ResultCollection(2, 40)]) self.assertEqual(writer.result_count, 2) self.assertEqual([path.name.split('_')[0] for path in sorted((writer.directory / 'results').iterdir())], ['0000', '0001']) def test_processing_revisions_are_saved_and_cannot_be_reassigned(self) -> None: settings = {4: {'settings_revision': 4, 'gain': 2}} directory = self.store.save_runtime_snapshot_numpy( self.output, 'snapshot', [], [], [], processing_settings=settings, ) self.assertEqual(json.loads((directory / 'processing_settings.json').read_text()), {'4': settings[4]}) writer = self.store.create_snapshot_stream(self.output, 'stream') writer.append([], [], [ResultCollection(1, 10)], processing_settings=settings) later = {5: {'settings_revision': 5, 'gain': 3}} writer.append([], [], [ResultCollection(2, 20)], processing_settings=later) path = writer.directory / 'processing_settings.json' self.assertEqual(json.loads(path.read_text()), {'4': settings[4], '5': later[5]}) with self.assertRaisesRegex(ValueError, 'already has different settings'): writer.append([], [], [ResultCollection(3, 30)], processing_settings={4: {'gain': 9}}) self.assertEqual(writer.result_count, 2) self.assertEqual(json.loads(path.read_text()), {'4': settings[4], '5': later[5]}) def test_failed_processing_settings_write_does_not_publish_results(self) -> None: writer = self.store.create_snapshot_stream(self.output, 'stream') with mock.patch('python_app.storage.npz.snapshot_numpy.atomic_write_json', side_effect=OSError('disk full')): with self.assertRaises(OSError): writer.append([], [], [ResultCollection(1, 10)], processing_settings={1: {'gain': 2}}) self.assertEqual(writer.result_count, 0) self.assertEqual(list(writer.directory.iterdir()), []) def test_json_export_preserves_unmatched_raw_sweeps_and_matches_reused_ids_by_timing(self) -> None: raw_old = _sweep(1, 10, start=1, end=5) raw_new = _sweep(1, 100, start=90, end=95) raw_new.traces[0].s21[:] = 99 pre_new = _sweep(1, 110, start=90, end=95) payload = build_vna_history_payload([raw_old, raw_new], [pre_new], [], input_index=0, output_index=0) self.assertEqual(len(payload['sweep_history']), 2) self.assertEqual(payload['sweep_history'][-1]['sweep_points'], [[99.0, 0.0], [99.0, 0.0]]) self.assertEqual(payload['sweep_history'][-1]['calibrated_points'], [[5.0, -6.0], [7.0, 8.0]]) def test_json_batch_failure_does_not_publish_partial_export(self) -> None: sweep = _sweep() second = _sweep().traces[0] second.combo = ComboKey(0, 1) sweep.traces.append(second) dump = json.dump writes = 0 def fail_second_file(*args, **kwargs): nonlocal writes writes += 1 if writes == 2: raise OSError('disk full') return dump(*args, **kwargs) with mock.patch('python_app.storage.npz.store.json.dump', fail_second_file), self.assertRaises(OSError): self.store.save_runtime_vna_history_json_batch(self.output, 'export', [sweep], [], []) self.assertEqual(list(self.output.iterdir()), []) paths = self.store.save_runtime_vna_history_json_batch(self.output, 'export', [sweep], [], []) self.assertEqual(len(paths), 2) self.assertEqual(len(paths), 2) for path in paths: self.assertEqual(json.loads(path.read_text())['channel'], 's21') def test_json_export_can_follow_numpy_without_overwriting_either_format(self) -> None: directory = self.store.save_runtime_snapshot_numpy(self.output, 'both', [_sweep()], [], []) directory.chmod(0o775) before = {path: path.read_bytes() for path in directory.rglob('*') if path.is_file()} paths = self.store.save_runtime_vna_history_json_batch(self.output, 'both', [_sweep()], [], []) self.assertEqual(stat.S_IMODE(directory.stat().st_mode), 0o775) self.assertEqual({path: path.read_bytes() for path in before}, before) exported = paths[0].read_bytes() with self.assertRaises(FileExistsError): self.store.save_runtime_vna_history_json_batch(self.output, 'both', [_sweep(2, 30)], [], []) self.assertEqual(paths[0].read_bytes(), exported) self.assertEqual({path: path.read_bytes() for path in before}, before) def test_long_recording_converter_uses_numeric_directory_order(self) -> None: writer = self.store.create_snapshot_stream(self.output, 'long') writer.append([_sweep(1, 10), _sweep(2, 20), _sweep(3, 30)], [], []) folders = sorted((writer.directory / 'raw').iterdir()) for folder, index in zip(folders, (9998, 9999, 10000), strict=True): folder.rename(folder.with_name(f'{index}_{folder.name.split("_", 1)[1]}')) self.assertEqual([int(path.name.split('_')[0]) for path in collection_directories(writer.directory / 'raw')], [9998, 9999, 10000]) def test_offline_and_runtime_json_exports_share_matching_and_preserve_orphan_sweeps(self) -> None: raw = [_sweep(1, 10, start=1, end=5), _sweep(1, 100, start=90, end=95)] pre = [_sweep(1, 110, start=90, end=95)] directory = self.store.save_runtime_snapshot_numpy( self.output, 'convert', raw, pre, [], ) offline = load_vna_history_snapshot(directory, input_index=0, output_index=0)["sweep_history"] runtime = build_vna_history_payload(raw, pre, [], input_index=0, output_index=0) self.assertEqual(offline, runtime['sweep_history']) self.assertEqual(len(offline), 2) class CalibrationMedianIntegrityTest(unittest.TestCase): def test_rejects_frequency_axes_that_changed_between_captures(self) -> None: first, second = _sweep(), _sweep() second.traces[0].frequency_hz[1] += 1e8 with self.assertRaisesRegex(RuntimeError, 'frequency-axis'): combine_traces_via_median([first.traces[0], second.traces[0]]) def test_rejects_extra_combinations_instead_of_silently_discarding_them(self) -> None: first, second = _sweep(), _sweep() extra = _sweep().traces[0] extra.combo = ComboKey(0, 1) second.traces.append(extra) with self.assertRaisesRegex(RuntimeError, 'trace counts'): combine_collections_via_median([first, second]) def test_collection_and_trace_windows_span_all_median_captures(self) -> None: first, second = _sweep(start=10, end=15), _sweep(start=30, end=35) combined = combine_collections_via_median([first, second]) self.assertEqual((combined.capture_start_ns, combined.capture_end_ns), (10, 35)) self.assertEqual((combined.traces[0].capture_start_ns, combined.traces[0].capture_end_ns), (10, 35)) if __name__ == '__main__': unittest.main()