import os import sys import time import types import pytest from pathlib import Path from unittest.mock import MagicMock BACKEND_DIR = Path(__file__).resolve().parents[1] if str(BACKEND_DIR) not in sys.path: sys.path.insert(0, str(BACKEND_DIR)) def _make_torch_stub(cuda_avail=False, mps_avail=False): class DummyTensor: def __matmul__(self, other): return self def matmul(self, other): return self stub = types.SimpleNamespace() stub.float32 = "float32" stub.float16 = "float16" stub.randn = lambda *a, **k: DummyTensor() stub.mm = lambda a, b: DummyTensor() stub.from_numpy = lambda arr: DummyTensor() stub.nn = types.SimpleNamespace() stub.nn.Linear = MagicMock(return_value=MagicMock()) stub.nn.Module = type("Module", (), {}) stub.no_grad = MagicMock() stub.no_grad.return_value.__enter__ = MagicMock(return_value=None) stub.no_grad.return_value.__exit__ = MagicMock(return_value=False) stub.backends = types.SimpleNamespace() stub.backends.mps = types.SimpleNamespace() stub.backends.mps.is_available = lambda: mps_avail stub.backends.mps.is_built = lambda: mps_avail stub.cuda = types.SimpleNamespace() stub.cuda.is_available = lambda: cuda_avail stub.cuda.device_count = lambda: 1 if cuda_avail else 0 stub.cuda.get_device_properties = lambda n: types.SimpleNamespace(total_memory=8*1024*1024*1024) stub.cuda.empty_cache = lambda: None stub.mps = types.SimpleNamespace() stub.mps.is_available = lambda: mps_avail stub.mps.is_built = lambda: mps_avail stub.mps.empty_cache = lambda: None stub.device = lambda s: s stub.Tensor = MagicMock() return stub def _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device=None): for mod_name in list(sys.modules.keys()): if mod_name.startswith("tts_asr") or mod_name == "torch": del sys.modules[mod_name] torch_stub = _make_torch_stub(cuda_avail=cuda_avail, mps_avail=mps_avail) sys.modules["torch"] = torch_stub if env_device is not None: os.environ["TTS_ASR_DEVICE"] = env_device elif "TTS_ASR_DEVICE" in os.environ: del os.environ["TTS_ASR_DEVICE"] import tts_asr tts_asr._device_caps = None tts_asr._tts_pipeline = None tts_asr._asr_pipeline = None tts_asr._tts_last_used = 0 tts_asr._asr_last_used = 0 return tts_asr @pytest.fixture(autouse=True) def _clean_tts_env(): saved = {} for k in ["TTS_ASR_DEVICE", "TTS_ASR_IDLE_TIMEOUT", "TTS_ASR_MODEL_SIZE", "TTS_ASR_QUANTIZE", "TTS_ASR_OFFLINE_MODE", "TTS_ASR_WARMUP", "TTS_ASR_MPS_MEMORY_LIMIT_MB"]: saved[k] = os.environ.get(k) if k in os.environ: del os.environ[k] yield for k, v in saved.items(): if v is not None: os.environ[k] = v elif k in os.environ: del os.environ[k] # --- Cache clearing --- def test_clear_cuda_cache(): tts = _reload_tts_asr(cuda_avail=True, mps_avail=False, env_device="cpu") tts._clear_cuda_cache() def test_clear_mps_cache(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=True, env_device="cpu") tts._clear_mps_cache() # --- Model cache check --- def test_check_model_cached_non_offline(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) os.environ["TTS_ASR_OFFLINE_MODE"] = "false" import importlib importlib.reload(tts) assert tts._check_model_cached("openai/whisper-tiny") is True def test_check_model_cached_offline_mode(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) os.environ["TTS_ASR_OFFLINE_MODE"] = "true" import importlib importlib.reload(tts) assert tts._check_model_cached("openai/whisper-tiny") is False # --- Torch dtype --- def test_get_torch_dtype_cpu(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device="cpu") assert tts._get_torch_dtype() == "float32" def test_get_torch_dtype_mps(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=True, env_device="mps") assert tts._get_torch_dtype() == "float32" def test_get_torch_dtype_cuda(): tts = _reload_tts_asr(cuda_avail=True, mps_avail=False, env_device="cuda") assert tts._get_torch_dtype() == "float16" # --- Device detection --- def test_get_device_cpu_env(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device="cpu") assert tts._get_device() == "cpu" def test_get_device_mps_available(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=True, env_device="mps") assert tts._get_device() == "mps" def test_get_device_mps_not_available_falls_back(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device="mps") assert tts._get_device() == "cpu" def test_get_device_cuda_available(): tts = _reload_tts_asr(cuda_avail=True, mps_avail=False, env_device="cuda") assert tts._get_device() == "cuda" def test_get_device_cuda_not_available_falls_back(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device="cuda") assert tts._get_device() == "cpu" def test_get_device_auto_mps(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=True, env_device=None) assert tts._get_device() == "mps" def test_get_device_auto_cuda(): tts = _reload_tts_asr(cuda_avail=True, mps_avail=False, env_device=None) assert tts._get_device() == "cuda" def test_get_device_auto_cpu(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device=None) assert tts._get_device() == "cpu" def test_device_arg_cuda(): tts = _reload_tts_asr(cuda_avail=True, mps_avail=False, env_device="cuda") assert tts._device_arg() == "cuda:0" def test_device_arg_cpu(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device="cpu") assert tts._device_arg() == "cpu" def test_device_arg_mps(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=True, env_device="mps") assert tts._device_arg() == "mps" def test_test_device_capability_cpu(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) ok, err = tts._test_device_capability("cpu") assert ok is True assert err == "" def test_test_device_capability_mps_not_available(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) ok, err = tts._test_device_capability("mps") assert ok is False assert len(err) > 0 def test_test_device_capability_cuda_not_available(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) ok, err = tts._test_device_capability("cuda") assert ok is False assert len(err) > 0 def test_test_device_capability_unknown_device(): tts = _reload_tts_asr() ok, err = tts._test_device_capability("vulkan") assert ok is False assert len(err) > 0 # --- Idle model unload --- def test_check_and_unload_idle_models_timeout_zero(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) os.environ["TTS_ASR_IDLE_TIMEOUT"] = "0" tts._tts_pipeline = "pipeline" tts._asr_pipeline = "pipeline" tts._tts_last_used = time.time() tts._asr_last_used = time.time() tts._check_and_unload_idle_models() assert tts._tts_pipeline == "pipeline" def test_check_and_unload_idle_models_unloads_when_expired(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) os.environ["TTS_ASR_IDLE_TIMEOUT"] = "1" tts._tts_pipeline = "pipeline" tts._asr_pipeline = "pipeline" tts._tts_last_used = time.time() - 10 tts._asr_last_used = time.time() - 10 import importlib importlib.reload(tts) tts._check_and_unload_idle_models() assert True # Function executed without error def test_check_and_unload_idle_models_keeps_when_not_expired(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) os.environ["TTS_ASR_IDLE_TIMEOUT"] = "60" tts._tts_pipeline = "pipeline" tts._asr_pipeline = "pipeline" tts._tts_last_used = time.time() tts._asr_last_used = time.time() tts._check_and_unload_idle_models() assert tts._tts_pipeline == "pipeline" # --- API key --- def test_get_api_key_success(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) key = tts.get_api_key("your-secret-key-here") assert key == "your-secret-key-here" def test_get_api_key_wrong_key_raises(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) with pytest.raises(Exception): tts.get_api_key("wrong-key") def test_get_api_key_missing_key_raises(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) with pytest.raises(Exception): tts.get_api_key("") # --- Pydantic models --- def test_tts_request_model(): tts = _reload_tts_asr() req = tts.TTSRequest(text="hello") assert req.text == "hello" assert req.voice == "af_bella" assert req.rate == 1.0 assert req.format == "wav" def test_asr_request_model(): tts = _reload_tts_asr() req = tts.ASRRequest(audio_base64="base64data", language="zh") assert req.audio_base64 == "base64data" assert req.language == "zh" # --- Device capabilities --- def test_detect_device_capabilities_cpu(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) caps = tts._detect_device_capabilities() assert caps.device == "cpu" assert caps.mps_available is False assert caps.cuda_available is False def test_detect_device_capabilities_mps(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=True) caps = tts._detect_device_capabilities() assert caps.device == "mps" assert caps.mps_available is True def test_detect_device_capabilities_cuda(): tts = _reload_tts_asr(cuda_avail=True, mps_avail=False) caps = tts._detect_device_capabilities() assert caps.device == "cuda" assert caps.cuda_available is True # --- Apple Silicon check --- def test_is_apple_silicon_windows(): tts = _reload_tts_asr() assert tts._is_apple_silicon() is False # --- Model size --- def test_recommended_model_size_auto(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False, env_device="cpu") size = tts._get_recommended_model_size() assert size in tts.WHISPER_MODEL_SIZES or size == "auto" def test_recommended_model_size_explicit(): tts = _reload_tts_asr(cuda_avail=False, mps_avail=False) os.environ["TTS_ASR_MODEL_SIZE"] = "tiny" import importlib importlib.reload(tts) size = tts._get_recommended_model_size() assert size == "tiny"