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']) if __name__ == '__main__': unittest.main()