import copy import importlib.util from pathlib import Path import tempfile import unittest from unittest import mock import numpy as np import soundfile as sf SPEC = importlib.util.spec_from_file_location('generate_voice', Path(__file__).with_name('generate_voice.py')) voice = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(voice) class VoiceHistoryTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.catalog = {'schemaVersion': 1, 'clips': [ {'id': 'count_01', 'text': '하나', 'usage': 'count', 'stageIds': [1], 'required': True}, {'id': 'guide_collect', 'text': '친구를 찾아 끌어와 볼까?', 'usage': 'guide', 'stageIds': [1], 'required': True}]} self.profile = {'schemaVersion': 1, 'modelId': voice.MODEL_ID, 'modelRevision': 'a' * 40, 'language': 'Korean', 'speaker': 'Sohee', 'instruct': '천천히 읽어 주세요.', 'device': 'cuda:0', 'dtype': 'bfloat16', 'attention': 'sdpa', 'seed': 42, 'generationParameters': {'max_new_tokens': 2048}} self.calls = [] def tearDown(self): self.temp.cleanup() def fake(self, clip): self.calls.append(clip['id']) return np.sin(np.linspace(0, 20, 2400)).astype(np.float32) * .2, 24000 def run_generation(self, catalog=None, profile=None, ids=None): return voice.generate(catalog or self.catalog, profile or self.profile, self.root, ids=ids, synthesize=self.fake, environment={'test': True}) def test_partial_changed_clip_regenerates_and_preserves_original(self): first = self.run_generation() old = (self.root / first['clips'][0]['file']).read_bytes() revised = copy.deepcopy(self.catalog) revised['clips'][1]['text'] = '친구를 끌어와 볼까?' second = self.run_generation(catalog=revised) self.assertEqual(['reused', 'generated'], [c['result'] for c in second['clips']]) self.assertEqual(['count_01', 'guide_collect', 'guide_collect'], self.calls) self.assertEqual(old, (self.root / first['clips'][0]['file']).read_bytes()) self.assertNotEqual(first['runId'], second['runId']) self.assertEqual([], voice.validate_manifest(second, revised, self.profile, self.root)) def test_empty_text_duplicate_ids_and_bad_stage_rejected(self): for edit in ('empty', 'duplicate', 'stage'): bad = copy.deepcopy(self.catalog) if edit == 'empty': bad['clips'][0]['text'] = ' ' if edit == 'duplicate': bad['clips'][1]['id'] = 'count_01' if edit == 'stage': bad['clips'][0]['stageIds'] = [31] with self.assertRaises(ValueError): self.run_generation(catalog=bad) self.assertEqual([], self.calls) def test_missing_corrupt_and_profile_mismatch_cannot_validate(self): manifest = self.run_generation() changed = copy.deepcopy(self.profile) changed['instruct'] = '밝게 읽어 주세요.' self.assertTrue(voice.validate_manifest(manifest, self.catalog, changed, self.root)) wav = self.root / manifest['clips'][0]['file'] wav.write_bytes(b'not wav') self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root)) wav.unlink() self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root)) def test_partial_scope_does_not_claim_all_required(self): manifest = self.run_generation(ids=['count_01']) self.assertEqual('selected', manifest['scope']) self.assertEqual([], voice.validate_manifest(manifest, self.catalog, self.profile, self.root, ['count_01'])) self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root)) def test_failed_or_invalid_waveform_not_success(self): def broken(clip): return np.array([np.nan]), 24000 manifest = voice.generate(self.catalog, self.profile, self.root, synthesize=broken) self.assertEqual(['failed', 'failed'], [c['result'] for c in manifest['clips']]) self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root)) def test_changed_profile_and_tampered_waveform_force_regeneration(self): first = self.run_generation() path = self.root / first['clips'][0]['file'] sf.write(path, np.zeros(2400), 24000) second = self.run_generation() self.assertEqual(['generated', 'reused'], [c['result'] for c in second['clips']]) revised = copy.deepcopy(self.profile) revised['seed'] = 43 third = self.run_generation(profile=revised) self.assertEqual(['generated', 'generated'], [c['result'] for c in third['clips']]) def test_unknown_requested_id_rejected(self): with self.assertRaises(ValueError): self.run_generation(ids=['missing']) def test_pcm_export_resamples_and_preserves_existing_runtime(self): import sys sys.path.insert(0, str(Path(__file__).parent)) from prepare_voice import prepare manifest = self.run_generation() target = self.root / 'runtime' selection = prepare(manifest, self.catalog, self.profile, self.root, target) info = sf.info(target / 'count_01.wav') self.assertEqual((44100, 1, 'PCM_16'), (info.samplerate, info.channels, info.subtype)) self.assertAlmostEqual(.1, info.duration) self.assertEqual('not_listened', selection['selections'][0]['review']) original = (target / 'count_01.wav').read_bytes() sf.write(target / 'count_01.wav', np.ones(4410) * .1, 44100) changed = (target / 'count_01.wav').read_bytes() with self.assertRaises(ValueError): prepare(manifest, self.catalog, self.profile, self.root, target) self.assertEqual(changed, (target / 'count_01.wav').read_bytes()) self.assertNotEqual(original, changed) def test_long_count_is_blocked_before_export(self): import sys sys.path.insert(0, str(Path(__file__).parent)) from prepare_voice import prepare def long_take(clip): return np.ones(72000, dtype=np.float32) * .1, 24000 manifest = voice.generate(self.catalog, self.profile, self.root, synthesize=long_take) target = self.root / 'runtime' with self.assertRaisesRegex(ValueError, 'exceeds 2 seconds'): prepare(manifest, self.catalog, self.profile, self.root, target) self.assertFalse(target.exists()) def test_new_take_preserves_original_and_other_clips(self): first = self.run_generation() original = (self.root / first['clips'][0]['file']).read_bytes() second = voice.generate(self.catalog, self.profile, self.root, ids=['count_01'], synthesize=self.fake, regenerate=True) self.assertEqual('generated', second['clips'][0]['result']) self.assertTrue(second['regenerateRequested']) self.assertEqual(1, len(second['clips'])) self.assertEqual(original, (self.root / first['clips'][0]['file']).read_bytes()) self.assertEqual(['count_01', 'guide_collect', 'count_01'], self.calls) def test_reuse_environment_records_packages_device_and_original_run_without_model_load(self): first = self.run_generation() entries = voice.reusable_entries(self.root, self.catalog, self.profile) with mock.patch.object(voice.importlib.metadata, 'version', return_value='test-version'), \ mock.patch.object(voice.subprocess, 'run', return_value=mock.Mock(stdout='Test GPU\n')): environment = voice.reuse_environment(self.profile, entries, self.root) second = voice.generate(self.catalog, self.profile, self.root, environment=environment) self.assertEqual(['reused', 'reused'], [c['result'] for c in second['clips']]) self.assertEqual(['count_01', 'guide_collect'], self.calls) self.assertEqual('not_needed', second['environment']['modelLoad']) self.assertEqual('Test GPU', second['environment']['device']) self.assertEqual('test-version', second['environment']['packages']['qwen-tts']) source = second['environment']['sourceGenerations'][0] self.assertEqual(first['runId'], source['runId']) 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()) def test_reuse_rejects_origin_manifest_device_changed_after_sha_recording(self): first = self.run_generation() entries = voice.reusable_entries(self.root, self.catalog, self.profile) with mock.patch.object(voice.importlib.metadata, 'version', return_value='test-version'), \ mock.patch.object(voice.subprocess, 'run', return_value=mock.Mock(stdout='Test GPU\n')): environment = voice.reuse_environment(self.profile, entries, self.root) second = voice.generate(self.catalog, self.profile, self.root, environment=environment) self.assertEqual([], voice.validate_manifest(second, self.catalog, self.profile, self.root)) changed_reference = copy.deepcopy(second) changed_reference['environment']['sourceGenerations'][0]['manifestSha256'] = '0' * 64 self.assertTrue(voice.validate_manifest(changed_reference, self.catalog, self.profile, self.root)) changed_environment = copy.deepcopy(second) changed_environment['environment']['sourceGenerations'][0]['environment']['device'] = 'CORRUPTED_DEVICE_RECORD' self.assertTrue(voice.validate_manifest(changed_environment, self.catalog, self.profile, self.root)) origin = self.root / 'generated' / first['runId'] / 'manifest.json' changed = voice.load_json(origin) changed['environment']['device'] = 'CORRUPTED_DEVICE_RECORD' voice.write_json(origin, changed) self.assertTrue(voice.validate_manifest(second, self.catalog, self.profile, self.root)) def test_legacy_migration_sha_seals_manifest_without_freezing_mutable_inputs(self): manifest = self.run_generation() path = self.root / 'generated' / manifest['runId'] / 'manifest.json' voice.write_json(self.root / 'input-snapshot-migration.json', { 'schemaVersion': 1, 'runs': [{'runId': manifest['runId'], 'originalManifestSha256': voice.file_hash(path)}], 'preservedArtifactSha256': {'qwen-profile.json': 'old mutable file hash'}}) voice.write_json(self.root / 'qwen-profile.json', {'seed': 43}) self.assertEqual([], voice.validate_manifest(manifest, self.catalog, self.profile, self.root)) changed = voice.load_json(path) changed['environment']['device'] = 'CORRUPTED_DEVICE_RECORD' voice.write_json(path, changed) self.assertTrue(voice.validate_manifest(changed, self.catalog, self.profile, self.root)) if __name__ == '__main__': unittest.main()