mirror of
https://github.com/denizsafak/abogen.git
synced 2026-09-20 11:40:57 +02:00
Artem Akymenko
2b70b9ca45
Language enum expanded from 9 to 33 languages:
- Added 24 new ISO 639-1 languages: AR, BG, CS, DA, DE, EL, ET, FI,
HR, HU, ID, KO, LT, LV, NL, PL, RO, RU, SK, SL, SV, TR, UK, VI
- Updated display_name, is_cjk (added KO)
Supertonic language mapping (32 languages, no ZH):
- engine.py: _SUPERTONIC_LANG_MAP, engine_language(), supported_languages()
- __init__.py: create_engine() passes config.language to pipeline
- pipeline.py: __init__() accepts language, resolves to ISO code;
__call__() passes lang= to TTS.synthesize()
total_steps propagation:
- tts_segments(): +total_steps param, conditionally passed to backend
- synthesize_text(): +total_steps param
- run_tts_segment_loop(): +total_steps param
- executor: all 5 synthesize_text() calls pass total_steps
Integration:
- pipeline_factory: create_pipeline_for_job() passes language to supertonic
- preview path: create_pipeline('supertonic', language=language)
Tests updated to accept total_steps in FakeBackend.__call__
185 lines
8.1 KiB
Python
185 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from abogen.domain.enums import Language
|
|
from abogen.domain.pipeline_factory import (
|
|
PipelinePool,
|
|
create_pipeline_for_job,
|
|
dispose_pipelines,
|
|
resolve_device,
|
|
)
|
|
|
|
|
|
class TestResolveDevice:
|
|
@patch("abogen.utils.load_config", return_value={"use_gpu": True})
|
|
@patch("abogen.domain.pipeline_factory.select_device", return_value="cuda:0")
|
|
def test_gpu_enabled(self, _sel, _cfg):
|
|
assert resolve_device(use_gpu=True) == "cuda:0"
|
|
|
|
@patch("abogen.utils.load_config", return_value={"use_gpu": True})
|
|
def test_gpu_disabled_by_job(self, _cfg):
|
|
assert resolve_device(use_gpu=False) == "cpu"
|
|
|
|
@patch("abogen.utils.load_config", return_value={"use_gpu": False})
|
|
@patch("abogen.domain.pipeline_factory.select_device", return_value="cuda:0")
|
|
def test_gpu_disabled_by_config(self, _sel, _cfg):
|
|
assert resolve_device(use_gpu=True) == "cpu"
|
|
|
|
|
|
class TestCreatePipelineForJob:
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline")
|
|
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
|
|
def test_supertonic_provider(self, _reg, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
result = create_pipeline_for_job("supertonic", Language.EN_US, use_gpu=True)
|
|
mock_create.assert_called_once_with("supertonic", language=Language.EN_US)
|
|
assert result is mock_create.return_value
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline")
|
|
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
|
|
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
|
|
def test_kokoro_provider(self, _dev, _reg, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
result = create_pipeline_for_job("kokoro", Language.EN_US, use_gpu=False)
|
|
mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
|
|
assert result is mock_create.return_value
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline")
|
|
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
|
|
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
|
|
def test_kokoro_provider_en_gb(self, _dev, _reg, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
result = create_pipeline_for_job("kokoro", Language.EN_GB, use_gpu=False)
|
|
mock_create.assert_called_once_with("kokoro", language=Language.EN_GB, device="cpu")
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline")
|
|
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=False)
|
|
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
|
|
def test_unknown_provider_falls_back_to_kokoro(self, _dev, _reg, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
result = create_pipeline_for_job("unknown_provider", Language.EN_US, use_gpu=False)
|
|
mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline")
|
|
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
|
|
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
|
|
def test_empty_provider_defaults_to_kokoro(self, _dev, _reg, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
result = create_pipeline_for_job("", Language.EN_US, use_gpu=False)
|
|
mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline")
|
|
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
|
|
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
|
|
def test_none_provider_defaults_to_kokoro(self, _dev, _reg, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
result = create_pipeline_for_job(None, Language.EN_US, use_gpu=False)
|
|
mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
|
|
|
|
|
|
class TestDisposePipelines:
|
|
def test_disposes_all_and_clears(self):
|
|
p1 = MagicMock()
|
|
p2 = MagicMock()
|
|
pipelines = {"kokoro": p1, "supertonic": p2}
|
|
dispose_pipelines(pipelines)
|
|
p1.dispose.assert_called_once()
|
|
p2.dispose.assert_called_once()
|
|
assert pipelines == {}
|
|
|
|
def test_handles_dispose_error(self):
|
|
p1 = MagicMock()
|
|
p1.dispose.side_effect = RuntimeError("boom")
|
|
pipelines = {"kokoro": p1}
|
|
dispose_pipelines(pipelines)
|
|
assert pipelines == {}
|
|
|
|
def test_empty_dict(self):
|
|
pipelines = {}
|
|
dispose_pipelines(pipelines)
|
|
assert pipelines == {}
|
|
|
|
|
|
class TestPipelinePool:
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
|
|
@patch("abogen.domain.pipeline_factory.initialize_voice_cache")
|
|
def test_get_creates_and_caches(self, _cache, mock_create):
|
|
mock_pipeline = MagicMock()
|
|
mock_create.return_value = mock_pipeline
|
|
pool = PipelinePool()
|
|
|
|
result = pool.get("kokoro", Language.EN_US, use_gpu=True)
|
|
assert result is mock_pipeline
|
|
mock_create.assert_called_once()
|
|
|
|
result2 = pool.get("kokoro", Language.EN_US, use_gpu=True)
|
|
assert result2 is mock_pipeline
|
|
assert mock_create.call_count == 1
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
|
|
@patch("abogen.domain.pipeline_factory.initialize_voice_cache")
|
|
def test_get_initializes_voice_cache_once(self, mock_cache, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
pool = PipelinePool()
|
|
|
|
request = MagicMock()
|
|
pool.get("kokoro", Language.EN_US, use_gpu=True, request=request)
|
|
assert mock_cache.call_count == 1
|
|
|
|
pool.get("kokoro", Language.EN_US, use_gpu=True, request=request)
|
|
assert mock_cache.call_count == 1
|
|
|
|
@patch("abogen.domain.pipeline_factory.initialize_voice_cache")
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
|
|
def test_get_no_job_skips_voice_cache(self, mock_create, mock_cache):
|
|
mock_create.return_value = MagicMock()
|
|
pool = PipelinePool()
|
|
pool.get("kokoro", Language.EN_US, use_gpu=True)
|
|
mock_cache.assert_not_called()
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
|
|
def test_get_separately_per_provider(self, mock_create):
|
|
p1 = MagicMock(name="kokoro")
|
|
p2 = MagicMock(name="supertonic")
|
|
mock_create.side_effect = [p1, p2]
|
|
pool = PipelinePool()
|
|
|
|
r1 = pool.get("kokoro", Language.EN_US, use_gpu=True)
|
|
r2 = pool.get("supertonic", Language.EN_US, use_gpu=True)
|
|
assert r1 is p1
|
|
assert r2 is p2
|
|
assert mock_create.call_count == 2
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
|
|
@patch("abogen.domain.pipeline_factory.initialize_voice_cache")
|
|
def test_dispose_all(self, mock_cache, mock_create):
|
|
p1 = MagicMock(name="kokoro")
|
|
p2 = MagicMock(name="supertonic")
|
|
mock_create.side_effect = [p1, p2]
|
|
pool = PipelinePool()
|
|
|
|
pool.get("kokoro", Language.EN_US, use_gpu=True)
|
|
pool.get("supertonic", Language.EN_US, use_gpu=True)
|
|
pool.dispose_all()
|
|
|
|
p1.dispose.assert_called_once()
|
|
p2.dispose.assert_called_once()
|
|
assert pool._pipelines == {}
|
|
assert pool._voice_cache_initialized is False
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
|
|
def test_dispose_empty_pool(self, mock_create):
|
|
pool = PipelinePool()
|
|
pool.dispose_all()
|
|
mock_create.assert_not_called()
|
|
|
|
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
|
|
@patch("abogen.domain.pipeline_factory.initialize_voice_cache")
|
|
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=False)
|
|
def test_unknown_provider_falls_back(self, _reg, _cache, mock_create):
|
|
mock_create.return_value = MagicMock()
|
|
pool = PipelinePool()
|
|
pool.get("bogus_provider", Language.EN_US, use_gpu=True)
|
|
mock_create.assert_called_once_with("kokoro", Language.EN_US, True)
|