NINI
268 lines
16 KiB
Python
268 lines
16 KiB
Python
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))
|
|
duplicated = copy.deepcopy(second)
|
|
duplicated['environment']['sourceGenerations'].append(copy.deepcopy(duplicated['environment']['sourceGenerations'][0]))
|
|
self.assertTrue(voice.validate_manifest(duplicated, self.catalog, self.profile, self.root))
|
|
changed_path = copy.deepcopy(second)
|
|
changed_path['environment']['sourceGenerations'][0]['manifest'] = '../outside/manifest.json'
|
|
self.assertTrue(voice.validate_manifest(changed_path, 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()
|