Preserve immutable per-run voice inputs and validate source history
This commit is contained in:
@@ -152,6 +152,77 @@ class VoiceHistoryTests(unittest.TestCase):
|
||||
self.assertEqual(voice.file_hash(self.root / source['manifest']), source['manifestSha256'])
|
||||
self.assertEqual({'test': True}, source['environment'])
|
||||
|
||||
def test_run_input_snapshots_preserve_text_and_settings_after_shared_files_change(self):
|
||||
old_catalog, old_profile = copy.deepcopy(self.catalog), copy.deepcopy(self.profile)
|
||||
first = self.run_generation()
|
||||
run_dir = self.root / 'generated' / first['runId']
|
||||
original_wav = (self.root / first['clips'][0]['file']).read_bytes()
|
||||
self.catalog['clips'][0]['text'] = '하나예요.'
|
||||
self.profile.update(seed=43, instruct='밝게 읽어 주세요.', dtype='float16', attention='eager')
|
||||
voice.write_json(self.root / 'voice-catalog.json', self.catalog)
|
||||
voice.write_json(self.root / 'qwen-profile.json', self.profile)
|
||||
self.run_generation()
|
||||
self.assertEqual(old_catalog, voice.load_json(run_dir / 'catalog.json'))
|
||||
self.assertEqual(old_profile, voice.load_json(run_dir / 'profile.json'))
|
||||
self.assertEqual(first['catalogHash'], voice.digest(voice.load_json(run_dir / 'catalog.json')))
|
||||
self.assertEqual(first['profileHash'], voice.digest(voice.load_json(run_dir / 'profile.json')))
|
||||
self.assertEqual(original_wav, (self.root / first['clips'][0]['file']).read_bytes())
|
||||
self.assertTrue(voice.validate_manifest(first, self.catalog, self.profile, self.root))
|
||||
|
||||
def test_run_input_snapshots_missing_corrupt_or_hash_mismatch_fail_validation(self):
|
||||
manifest = self.run_generation()
|
||||
run_dir = self.root / 'generated' / manifest['runId']
|
||||
for name, data in [('catalog.json', self.catalog), ('profile.json', self.profile)]:
|
||||
path = run_dir / name
|
||||
for contents in (None, 'not json', '{}'):
|
||||
if contents is None:
|
||||
path.unlink(missing_ok=True)
|
||||
else:
|
||||
path.write_text(contents, encoding='utf-8')
|
||||
self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root))
|
||||
voice.write_json(path, data)
|
||||
changed = copy.deepcopy(manifest)
|
||||
changed['profileHash'] = 'f' * 64
|
||||
self.assertTrue(voice.validate_manifest(changed, self.catalog, self.profile, self.root))
|
||||
|
||||
def test_reused_run_rejects_missing_original_input_snapshot(self):
|
||||
first = self.run_generation()
|
||||
second = self.run_generation()
|
||||
self.assertEqual('reused', second['clips'][0]['result'])
|
||||
(self.root / 'generated' / first['runId'] / 'profile.json').unlink(missing_ok=True)
|
||||
self.assertTrue(voice.validate_manifest(second, self.catalog, self.profile, self.root))
|
||||
|
||||
def test_snapshot_validation_cli_does_not_load_model_or_use_network(self):
|
||||
import sys
|
||||
manifest = self.run_generation()
|
||||
catalog = self.root / 'voice-catalog.json'
|
||||
profile = self.root / 'qwen-profile.json'
|
||||
voice.write_json(catalog, self.catalog)
|
||||
voice.write_json(profile, self.profile)
|
||||
args = ['generate_voice.py', '--catalog', str(catalog), '--profile', str(profile),
|
||||
'--validate-only', '--manifest', str(self.root / 'generated' / manifest['runId'] / 'manifest.json')]
|
||||
with mock.patch.object(sys, 'argv', args), \
|
||||
mock.patch.object(voice, 'load_synthesizer', side_effect=AssertionError('model loaded')) as loader, \
|
||||
mock.patch.object(voice.subprocess, 'run', side_effect=AssertionError('device command ran')), \
|
||||
mock.patch.dict(sys.modules, {'torch': None, 'qwen_tts': None, 'huggingface_hub': None}):
|
||||
self.assertEqual(0, voice.main())
|
||||
snapshot = self.root / 'generated' / manifest['runId'] / 'catalog.json'
|
||||
for contents in (None, 'not json', '{}'):
|
||||
if contents is None:
|
||||
snapshot.unlink(missing_ok=True)
|
||||
else:
|
||||
snapshot.write_text(contents, encoding='utf-8')
|
||||
self.assertEqual(1, voice.main())
|
||||
loader.assert_not_called()
|
||||
|
||||
def test_legacy_manifest_uses_verified_sidecar_without_manifest_rewrite(self):
|
||||
manifest = self.run_generation()
|
||||
legacy = copy.deepcopy(manifest)
|
||||
legacy.pop('inputSnapshots')
|
||||
saved_manifest = (self.root / 'generated' / manifest['runId'] / 'manifest.json').read_bytes()
|
||||
self.assertEqual([], voice.validate_manifest(legacy, self.catalog, self.profile, self.root))
|
||||
self.assertEqual(saved_manifest, (self.root / 'generated' / manifest['runId'] / 'manifest.json').read_bytes())
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user