feat: improve Korean neural voice quality for v0.1.1
This commit is contained in:
@@ -0,0 +1,256 @@
|
||||
"""외부 provider 대신 작은 PCM fixture로 실패·복구·원자 공개 계약을 검증한다.
|
||||
|
||||
실제 네트워크 생성은 별도 인수 단계에서 101개 전부 실행한다. 이 테스트의
|
||||
가짜 MP3는 adapter 경계용이며 public 자산이나 생성 evidence로 쓰지 않는다.
|
||||
"""
|
||||
import contextlib
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
import wave
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
import neural_audio as audio
|
||||
|
||||
ENTRIES = [{"file": "intro.wav", "text": "안내 문장"}] + [
|
||||
{"file": f"target-{i}.wav", "text": f"목표 문장 {i}"} for i in range(100)
|
||||
]
|
||||
VERSIONS = {"edge-tts": "7.3.1", "ffmpeg": "ffmpeg version fixture"}
|
||||
|
||||
|
||||
def wav(path, *, rate=24000, channels=1, width=2, duration=1, amplitude=4000):
|
||||
values = [int(amplitude * math.sin(2 * math.pi * 220 * i / rate)) for i in range(int(rate * duration))]
|
||||
with wave.open(str(path), "wb") as stream:
|
||||
stream.setparams((channels, width, rate, 0, "NONE", "not compressed"))
|
||||
stream.writeframes(b"".join(struct.pack("<h", x) for x in values))
|
||||
|
||||
|
||||
def snapshot(path):
|
||||
return {str(p.relative_to(path)): hashlib.sha256(p.read_bytes()).hexdigest() for p in path.rglob("*") if p.is_file()}
|
||||
|
||||
|
||||
class GeneratorContract(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
self.output = self.root / "audio"
|
||||
self.output.mkdir()
|
||||
(self.output / "intro.wav").write_bytes(b"old voice")
|
||||
(self.output / "manifest.json").write_text('{"old":true}')
|
||||
self.before = snapshot(self.output)
|
||||
self.calls = []
|
||||
self.delays = []
|
||||
self.fixture = self.root / "fixture.wav"
|
||||
wav(self.fixture)
|
||||
|
||||
def synthesize(self, text, path):
|
||||
self.assertEqual(snapshot(self.output), self.before, "생성 중 기존 공개 자산은 변하면 안 된다")
|
||||
self.assertFalse(path.exists(), "재시도는 이전 불완전 MP3를 제거해야 한다")
|
||||
self.calls.append(text)
|
||||
path.write_bytes(b"fake mp3 for adapter boundary")
|
||||
|
||||
def convert(self, source, destination):
|
||||
self.assertTrue(source.exists())
|
||||
destination.write_bytes(self.fixture.read_bytes())
|
||||
|
||||
def run_generate(self, synthesize=None, convert=None):
|
||||
with patch.object(audio, "tool_versions", return_value=VERSIONS, create=True):
|
||||
return audio.generate(ENTRIES, self.output, synthesize=synthesize or self.synthesize,
|
||||
convert=convert or self.convert, sleep=self.delays.append)
|
||||
|
||||
def assert_preserved(self):
|
||||
self.assertEqual(snapshot(self.output), self.before)
|
||||
self.assertEqual(sorted(p.name for p in self.root.iterdir()), ["audio", "fixture.wav"])
|
||||
|
||||
def test_all_101_are_validated_before_single_atomic_publication(self):
|
||||
with patch.object(audio, "atomic_publish", wraps=audio.atomic_publish) as publish:
|
||||
result = self.run_generate()
|
||||
self.assertEqual(publish.call_count, 1)
|
||||
self.assertEqual(len(self.calls), 101)
|
||||
self.assertEqual(len(result["files"]), 101)
|
||||
self.assertEqual(result, json.loads((self.output / "manifest.json").read_text()))
|
||||
self.assertEqual(result["voice"], "ko-KR-HyunsuMultilingualNeural")
|
||||
self.assertEqual(result["toolVersions"], VERSIONS)
|
||||
self.assertEqual(sorted(p.name for p in self.output.iterdir()), sorted([e["file"] for e in ENTRIES]+["manifest.json"]))
|
||||
self.assertEqual(sorted(p.name for p in self.root.iterdir()), ["audio", "fixture.wav"])
|
||||
|
||||
def test_transient_provider_failures_retry_cleanly_with_bounded_backoff(self):
|
||||
attempts = []
|
||||
def flaky(text, path):
|
||||
self.assertFalse(path.exists())
|
||||
attempts.append(text)
|
||||
if len(attempts) < 3:
|
||||
path.write_bytes(b"partial")
|
||||
raise subprocess.CalledProcessError(1, ["provider"])
|
||||
self.synthesize(text, path)
|
||||
self.run_generate(synthesize=flaky)
|
||||
self.assertEqual(len(attempts), 103)
|
||||
self.assertEqual(self.delays, [1, 2])
|
||||
|
||||
def test_exhausted_provider_failure_preserves_existing_files_and_redacts_secrets(self):
|
||||
attempts = []
|
||||
def broken(text, path):
|
||||
attempts.append(text)
|
||||
path.write_bytes(b"partial")
|
||||
if len(attempts) <= 2:
|
||||
return
|
||||
raise subprocess.CalledProcessError(1, ["provider", "credential-sentinel"], stderr="cookie=credential-sentinel")
|
||||
captured = io.StringIO()
|
||||
with contextlib.redirect_stdout(captured), contextlib.redirect_stderr(captured):
|
||||
with self.assertRaisesRegex(RuntimeError, "target-1.wav.*3") as failure:
|
||||
self.run_generate(synthesize=broken)
|
||||
self.assertNotIn("credential-sentinel", str(failure.exception) + captured.getvalue())
|
||||
self.assertEqual(len(attempts), 5)
|
||||
self.assertEqual(self.delays, [1, 2])
|
||||
self.assert_preserved()
|
||||
|
||||
def test_timeout_is_bounded_and_preserves_existing_files(self):
|
||||
attempts = []
|
||||
def timeout(text, path):
|
||||
attempts.append(text)
|
||||
raise subprocess.TimeoutExpired("provider", 60)
|
||||
with self.assertRaisesRegex(RuntimeError, "intro.wav.*3"):
|
||||
self.run_generate(synthesize=timeout)
|
||||
self.assertEqual(len(attempts), 3)
|
||||
self.assert_preserved()
|
||||
|
||||
def test_invalid_pcm_never_reaches_publication(self):
|
||||
def invalid(source, destination):
|
||||
wav(destination, amplitude=0)
|
||||
with self.assertRaisesRegex(ValueError, "무음|RMS"):
|
||||
self.run_generate(convert=invalid)
|
||||
self.assert_preserved()
|
||||
|
||||
def test_ffmpeg_failure_is_not_retried_and_preserves_existing_files(self):
|
||||
def broken(source, destination):
|
||||
raise subprocess.CalledProcessError(1, ["ffmpeg"])
|
||||
with self.assertRaisesRegex(RuntimeError, "intro.wav.*변환"):
|
||||
self.run_generate(convert=broken)
|
||||
self.assertEqual(len(self.calls), 1)
|
||||
self.assertEqual(self.delays, [])
|
||||
self.assert_preserved()
|
||||
|
||||
def test_interruption_preserves_existing_files_and_cleans_temporary_data(self):
|
||||
def interrupt(text, path):
|
||||
path.write_bytes(b"partial")
|
||||
raise KeyboardInterrupt()
|
||||
with self.assertRaises(KeyboardInterrupt):
|
||||
self.run_generate(synthesize=interrupt)
|
||||
self.assert_preserved()
|
||||
|
||||
def test_publication_failure_preserves_existing_files(self):
|
||||
with patch.object(audio, "atomic_publish", side_effect=OSError("exchange unavailable")):
|
||||
with self.assertRaisesRegex(OSError, "exchange unavailable"):
|
||||
self.run_generate()
|
||||
self.assert_preserved()
|
||||
|
||||
def test_full_validation_detects_changed_earlier_file_before_publication(self):
|
||||
def tamper(source, destination):
|
||||
self.convert(source, destination)
|
||||
if destination.name == "target-99.wav":
|
||||
(destination.parent / "intro.wav").write_bytes(b"damaged")
|
||||
with self.assertRaises((ValueError, EOFError, wave.Error)):
|
||||
self.run_generate(convert=tamper)
|
||||
self.assert_preserved()
|
||||
|
||||
def test_wrong_edge_version_fails_before_network_or_asset_changes(self):
|
||||
with patch("importlib.metadata.version", return_value="7.2.7"):
|
||||
with self.assertRaisesRegex(RuntimeError, "7.3.1"):
|
||||
audio.generate(ENTRIES, self.output, synthesize=self.synthesize)
|
||||
self.assertEqual(self.calls, [])
|
||||
self.assert_preserved()
|
||||
|
||||
def test_rejects_incomplete_or_duplicate_or_unsafe_inventory_before_requests(self):
|
||||
for entries in [ENTRIES[:-1], [ENTRIES[0]] * 101,
|
||||
[{"file": "../escape.wav", "text": "외부"}] + ENTRIES[1:]]:
|
||||
with self.subTest(entries=entries[0]):
|
||||
with self.assertRaises(ValueError):
|
||||
audio.generate(entries, self.output, synthesize=self.synthesize)
|
||||
self.assertEqual(self.calls, [])
|
||||
self.assert_preserved()
|
||||
|
||||
def test_cli_import_has_no_generation_or_filesystem_side_effects(self):
|
||||
with patch("subprocess.check_output", side_effect=AssertionError("import 실행 금지")), \
|
||||
patch("subprocess.run", side_effect=AssertionError("import 실행 금지")):
|
||||
spec = importlib.util.spec_from_file_location("generate_audio_import_test", ROOT / "scripts/generate_audio.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
def test_edge_adapter_uses_pinned_python_prosody_timeout_and_no_shell(self):
|
||||
path = self.root / "sample.mp3"
|
||||
with patch("subprocess.run") as run:
|
||||
audio.synthesize_edge("한국어 ; $(literal)", path)
|
||||
args, kwargs = run.call_args
|
||||
self.assertEqual(args[0], [sys.executable, "-m", "edge_tts", "--voice", "ko-KR-HyunsuMultilingualNeural",
|
||||
"--rate=-6%", "--pitch=-4Hz", "--text", "한국어 ; $(literal)", "--write-media", str(path)])
|
||||
self.assertLessEqual(kwargs["timeout"], 60)
|
||||
self.assertTrue(kwargs["check"])
|
||||
self.assertFalse(kwargs.get("shell", False))
|
||||
self.assertEqual(kwargs["stderr"], subprocess.DEVNULL)
|
||||
|
||||
def test_ffmpeg_adapter_actually_decodes_mp3_to_normalized_pcm24(self):
|
||||
mp3 = self.root / "test.mp3"
|
||||
subprocess.run(["ffmpeg", "-v", "error", "-i", str(self.fixture), str(mp3)], check=True)
|
||||
output = self.root / "normalized.wav"
|
||||
audio.convert_mp3(mp3, output)
|
||||
with wave.open(str(output)) as stream:
|
||||
self.assertEqual((stream.getframerate(), stream.getnchannels(), stream.getsampwidth()), (24000, 1, 2))
|
||||
samples = struct.unpack("<" + "h" * stream.getnframes(), stream.readframes(stream.getnframes()))
|
||||
rms = math.sqrt(sum(x*x for x in samples)/len(samples))/32768
|
||||
self.assertAlmostEqual(rms, 0.1, places=3)
|
||||
self.assertLess(max(abs(x) for x in samples), 32767)
|
||||
|
||||
def test_atomic_exchange_keeps_old_tree_at_staging_path(self):
|
||||
staged = self.root / "staged"
|
||||
staged.mkdir()
|
||||
(staged / "new.wav").write_bytes(b"complete")
|
||||
audio.atomic_publish(staged, self.output)
|
||||
self.assertEqual(snapshot(staged), self.before)
|
||||
self.assertEqual((self.output / "new.wav").read_bytes(), b"complete")
|
||||
|
||||
def test_atomic_first_publication(self):
|
||||
staged = self.root / "staged"
|
||||
staged.mkdir()
|
||||
(staged / "new.wav").write_bytes(b"complete")
|
||||
output = self.root / "first"
|
||||
audio.atomic_publish(staged, output)
|
||||
self.assertFalse(staged.exists())
|
||||
self.assertEqual((output / "new.wav").read_bytes(), b"complete")
|
||||
|
||||
|
||||
class PCMContract(unittest.TestCase):
|
||||
def test_rejects_silence_clipping_bad_format_and_abnormal_duration(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "invalid.wav"
|
||||
for options in [{"amplitude": 0}, {"amplitude": 32767}, {"rate": 16000},
|
||||
{"channels": 2}, {"width": 1}, {"duration": 0.2}, {"duration": 21}]:
|
||||
with self.subTest(options=options):
|
||||
wav(path, **options)
|
||||
with self.assertRaises(ValueError):
|
||||
audio.inspect_wav(path)
|
||||
|
||||
def test_valid_pcm_returns_measured_hash_bytes_duration_and_levels(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "valid.wav"
|
||||
wav(path)
|
||||
result = audio.inspect_wav(path)
|
||||
self.assertEqual(result["duration"], 1)
|
||||
self.assertEqual(result["bytes"], path.stat().st_size)
|
||||
self.assertEqual(result["sha256"], hashlib.sha256(path.read_bytes()).hexdigest())
|
||||
self.assertGreater(result["rms"], 0)
|
||||
self.assertLess(result["peak"], 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user