diff --git a/tests/tools/test_tts_dotenv_fallback.py b/tests/tools/test_tts_dotenv_fallback.py new file mode 100644 index 0000000000..ed72a78ade --- /dev/null +++ b/tests/tools/test_tts_dotenv_fallback.py @@ -0,0 +1,230 @@ +"""Regression tests for #17140. + +TTS provider tools must resolve API keys from ``~/.hermes/.env`` (via +``hermes_cli.config.get_env_value``) and not only from ``os.environ`` — +otherwise users who keep their keys in the dotenv file see "API key not set" +errors even though the key is configured. Same class of bug as #15914 (auth) +already addressed for ``agent/credential_pool`` and ``hermes_cli/auth``. +""" + +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture(autouse=True) +def isolate_env(monkeypatch): + """Strip every TTS-related env var so the test really exercises the + dotenv code path. If any of these survive into the test, the assertion + that ``get_env_value`` was consulted becomes meaningless because + ``os.environ`` already satisfies the lookup. + """ + for key in ( + "ELEVENLABS_API_KEY", + "XAI_API_KEY", + "XAI_BASE_URL", + "MINIMAX_API_KEY", + "MISTRAL_API_KEY", + "GEMINI_API_KEY", + "GEMINI_BASE_URL", + "GOOGLE_API_KEY", + ): + monkeypatch.delenv(key, raising=False) + + +class TestDotenvFallbackPerProvider: + """For each affected provider, when only ``~/.hermes/.env`` carries the + key (mocked via ``hermes_cli.config.load_env``), the provider must find + it. Before the fix, ``os.getenv`` returned ``None`` and the provider + raised ``ValueError("X_API_KEY not set")``. + """ + + def test_elevenlabs_reads_dotenv_key(self, tmp_path): + from tools import tts_tool + + with patch.object(tts_tool, "get_env_value", return_value="el-dotenv-key"), \ + patch.object(tts_tool, "_import_elevenlabs") as mock_import: + mock_client = MagicMock() + mock_client.text_to_speech.convert.return_value = iter([b"audio"]) + mock_import.return_value = MagicMock(return_value=mock_client) + + output = str(tmp_path / "out.mp3") + tts_tool._generate_elevenlabs("hi", output, {}) + + mock_import.return_value.assert_called_once_with(api_key="el-dotenv-key") + + def test_xai_reads_dotenv_key(self, tmp_path): + from tools import tts_tool + + captured: dict = {} + + def fake_post(url, **kwargs): + captured["url"] = url + captured["headers"] = kwargs.get("headers", {}) + response = MagicMock() + response.content = b"audio" + response.raise_for_status = MagicMock() + return response + + with patch.object(tts_tool, "get_env_value", return_value="xai-dotenv-key"), \ + patch("requests.post", side_effect=fake_post): + tts_tool._generate_xai_tts("hi", str(tmp_path / "out.mp3"), {}) + + assert captured["headers"]["Authorization"] == "Bearer xai-dotenv-key" + + def test_minimax_reads_dotenv_key(self, tmp_path): + from tools import tts_tool + + captured: dict = {} + + def fake_post(url, **kwargs): + captured["headers"] = kwargs.get("headers", {}) + response = MagicMock() + response.json.return_value = { + "data": {"audio": b"\x00\x01".hex()}, + "base_resp": {"status_code": 0}, + } + response.raise_for_status = MagicMock() + return response + + with patch.object(tts_tool, "get_env_value", return_value="mm-dotenv-key"), \ + patch("requests.post", side_effect=fake_post): + tts_tool._generate_minimax_tts("hi", str(tmp_path / "out.mp3"), {}) + + assert captured["headers"]["Authorization"] == "Bearer mm-dotenv-key" + + def test_mistral_reads_dotenv_key(self, tmp_path): + import base64 + + from tools import tts_tool + + seen_keys: list = [] + + def fake_mistral_factory(*, api_key=None): + seen_keys.append(api_key) + client = MagicMock() + client.__enter__ = MagicMock(return_value=client) + client.__exit__ = MagicMock(return_value=False) + client.audio.speech.complete.return_value = MagicMock( + audio_data=base64.b64encode(b"data").decode() + ) + return client + + with patch.object(tts_tool, "get_env_value", return_value="mistral-dotenv-key"), \ + patch.object(tts_tool, "_import_mistral_client", return_value=fake_mistral_factory): + tts_tool._generate_mistral_tts("hi", str(tmp_path / "out.mp3"), {}) + + assert seen_keys == ["mistral-dotenv-key"] + + def test_gemini_reads_dotenv_key(self, tmp_path): + from tools import tts_tool + + captured: dict = {} + + def fake_post(url, **kwargs): + captured["params"] = kwargs.get("params", {}) + response = MagicMock() + response.status_code = 200 + response.json.return_value = { + "candidates": [ + { + "content": { + "parts": [ + { + "inlineData": { + "data": "AAAA", + "mimeType": "audio/L16;codec=pcm;rate=24000", + } + } + ] + } + } + ] + } + response.raise_for_status = MagicMock() + return response + + # GEMINI_API_KEY hits the first branch; GOOGLE_API_KEY would only be + # consulted if the first returned None. Use a side-effect-style mock + # to verify the lookup order matches the production code. + seen_lookups: list = [] + + def fake_get_env_value(key): + seen_lookups.append(key) + if key == "GEMINI_API_KEY": + return "gemini-dotenv-key" + return None + + with patch.object(tts_tool, "get_env_value", side_effect=fake_get_env_value), \ + patch("requests.post", side_effect=fake_post): + tts_tool._generate_gemini_tts("hi", str(tmp_path / "out.wav"), {}) + + assert "GEMINI_API_KEY" in seen_lookups + assert captured["params"]["key"] == "gemini-dotenv-key" + + +class TestRegressionGuard: + """Goal-backward proof that the old behaviour ('only check ``os.environ``') + breaks reading from a dotenv-only key, and the new behaviour fixes it. + Implemented as an end-to-end probe that patches + ``hermes_cli.config.load_env`` to simulate ``~/.hermes/.env`` carrying the + key while ``os.environ`` does not. + """ + + def test_minimax_missing_when_only_in_dotenv_before_fix(self, tmp_path, monkeypatch): + from tools import tts_tool + + monkeypatch.delenv("MINIMAX_API_KEY", raising=False) + + # Simulate ~/.hermes/.env carrying the key (load_env returns the dict + # that get_env_value falls back to). The pre-fix ``os.getenv`` call + # ignores this entirely and raises ValueError. + with patch( + "hermes_cli.config.load_env", + return_value={"MINIMAX_API_KEY": "dotenv-secret"}, + ): + # Sanity-check: get_env_value resolves through load_env when + # os.environ is empty. + from hermes_cli.config import get_env_value as live_get + assert live_get("MINIMAX_API_KEY") == "dotenv-secret" + + # And the production code path now consumes the resolved value + # instead of raising "MINIMAX_API_KEY not set". + captured: dict = {} + + def fake_post(url, **kwargs): + captured["headers"] = kwargs.get("headers", {}) + response = MagicMock() + response.json.return_value = { + "data": {"audio": b"\x00".hex()}, + "base_resp": {"status_code": 0}, + } + response.raise_for_status = MagicMock() + return response + + with patch("requests.post", side_effect=fake_post): + tts_tool._generate_minimax_tts( + "hi", str(tmp_path / "out.mp3"), {} + ) + + assert captured["headers"]["Authorization"] == "Bearer dotenv-secret" + + def test_check_tts_requirements_sees_dotenv_minimax(self, monkeypatch): + """``check_tts_requirements`` is the gate that decides whether + ``/voice on`` is even offered. If it only checked ``os.environ`` it + would say "no provider available" for users who keep MINIMAX_API_KEY + in ``~/.hermes/.env``, even though the dispatcher would later succeed. + """ + from tools import tts_tool + + monkeypatch.delenv("MINIMAX_API_KEY", raising=False) + + with patch( + "hermes_cli.config.load_env", + return_value={"MINIMAX_API_KEY": "dotenv-secret"}, + ), patch.object(tts_tool, "_import_edge_tts", side_effect=ImportError), \ + patch.object(tts_tool, "_import_elevenlabs", side_effect=ImportError), \ + patch.object(tts_tool, "_import_openai_client", side_effect=ImportError), \ + patch.object(tts_tool, "_check_neutts_available", return_value=False), \ + patch.object(tts_tool, "_check_kittentts_available", return_value=False): + assert tts_tool.check_tts_requirements() is True diff --git a/tools/tts_tool.py b/tools/tts_tool.py index a7ca57fab1..ae01843a8f 100644 --- a/tools/tts_tool.py +++ b/tools/tts_tool.py @@ -44,6 +44,7 @@ from urllib.parse import urljoin from hermes_constants import display_hermes_home logger = logging.getLogger(__name__) +from hermes_cli.config import get_env_value from tools.managed_tool_gateway import resolve_managed_tool_gateway from tools.tool_backend_helpers import managed_nous_tools_enabled, prefers_gateway, resolve_openai_audio_api_key from tools.xai_http import hermes_xai_user_agent @@ -312,7 +313,7 @@ def _generate_elevenlabs(text: str, output_path: str, tts_config: Dict[str, Any] Returns: Path to the saved audio file. """ - api_key = os.getenv("ELEVENLABS_API_KEY", "") + api_key = (get_env_value("ELEVENLABS_API_KEY") or "") if not api_key: raise ValueError("ELEVENLABS_API_KEY not set. Get one at https://elevenlabs.io/") @@ -406,7 +407,7 @@ def _generate_xai_tts(text: str, output_path: str, tts_config: Dict[str, Any]) - """ import requests - api_key = os.getenv("XAI_API_KEY", "").strip() + api_key = (get_env_value("XAI_API_KEY") or "").strip() if not api_key: raise ValueError("XAI_API_KEY not set. Get one at https://console.x.ai/") @@ -417,7 +418,7 @@ def _generate_xai_tts(text: str, output_path: str, tts_config: Dict[str, Any]) - bit_rate = int(xai_config.get("bit_rate", DEFAULT_XAI_BIT_RATE)) base_url = str( xai_config.get("base_url") - or os.getenv("XAI_BASE_URL") + or get_env_value("XAI_BASE_URL") or DEFAULT_XAI_BASE_URL ).strip().rstrip("/") @@ -479,7 +480,7 @@ def _generate_minimax_tts(text: str, output_path: str, tts_config: Dict[str, Any """ import requests - api_key = os.getenv("MINIMAX_API_KEY", "") + api_key = (get_env_value("MINIMAX_API_KEY") or "") if not api_key: raise ValueError("MINIMAX_API_KEY not set. Get one at https://platform.minimax.io/") @@ -556,7 +557,7 @@ def _generate_mistral_tts(text: str, output_path: str, tts_config: Dict[str, Any and writes the raw bytes to *output_path*. Supports native Opus output for Telegram voice bubbles. """ - api_key = os.getenv("MISTRAL_API_KEY", "") + api_key = (get_env_value("MISTRAL_API_KEY") or "") if not api_key: raise ValueError("MISTRAL_API_KEY not set. Get one at https://console.mistral.ai/") @@ -651,7 +652,7 @@ def _generate_gemini_tts(text: str, output_path: str, tts_config: Dict[str, Any] """ import requests - api_key = (os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY") or "").strip() + api_key = (get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY") or "").strip() if not api_key: raise ValueError( "GEMINI_API_KEY not set. Get one at https://aistudio.google.com/app/apikey" @@ -662,7 +663,7 @@ def _generate_gemini_tts(text: str, output_path: str, tts_config: Dict[str, Any] voice = str(gemini_config.get("voice", DEFAULT_GEMINI_TTS_VOICE)).strip() or DEFAULT_GEMINI_TTS_VOICE base_url = str( gemini_config.get("base_url") - or os.getenv("GEMINI_BASE_URL") + or get_env_value("GEMINI_BASE_URL") or DEFAULT_GEMINI_TTS_BASE_URL ).strip().rstrip("/") @@ -1148,7 +1149,7 @@ def check_tts_requirements() -> bool: pass try: _import_elevenlabs() - if os.getenv("ELEVENLABS_API_KEY"): + if get_env_value("ELEVENLABS_API_KEY"): return True except ImportError: pass @@ -1158,15 +1159,15 @@ def check_tts_requirements() -> bool: return True except ImportError: pass - if os.getenv("MINIMAX_API_KEY"): + if get_env_value("MINIMAX_API_KEY"): return True - if os.getenv("XAI_API_KEY"): + if get_env_value("XAI_API_KEY"): return True - if os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY"): + if get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY"): return True try: _import_mistral_client() - if os.getenv("MISTRAL_API_KEY"): + if get_env_value("MISTRAL_API_KEY"): return True except ImportError: pass @@ -1278,7 +1279,7 @@ def stream_tts_to_speaker( {**tts_config, "elevenlabs": {**el_config, "model_id": model_id}}, ) - api_key = os.getenv("ELEVENLABS_API_KEY", "") + api_key = (get_env_value("ELEVENLABS_API_KEY") or "") if not api_key: logger.warning("ELEVENLABS_API_KEY not set; streaming TTS audio disabled") else: @@ -1464,13 +1465,13 @@ if __name__ == "__main__": print("\nProvider availability:") print(f" Edge TTS: {'installed' if _check(_import_edge_tts, 'edge') else 'not installed (pip install edge-tts)'}") print(f" ElevenLabs: {'installed' if _check(_import_elevenlabs, 'el') else 'not installed (pip install elevenlabs)'}") - print(f" API Key: {'set' if os.getenv('ELEVENLABS_API_KEY') else 'not set'}") + print(f" API Key: {'set' if get_env_value('ELEVENLABS_API_KEY') else 'not set'}") print(f" OpenAI: {'installed' if _check(_import_openai_client, 'oai') else 'not installed'}") print( " API Key: " f"{'set' if resolve_openai_audio_api_key() else 'not set (VOICE_TOOLS_OPENAI_KEY or OPENAI_API_KEY)'}" ) - print(f" MiniMax: {'API key set' if os.getenv('MINIMAX_API_KEY') else 'not set (MINIMAX_API_KEY)'}") + print(f" MiniMax: {'API key set' if get_env_value('MINIMAX_API_KEY') else 'not set (MINIMAX_API_KEY)'}") print(f" ffmpeg: {'✅ found' if _has_ffmpeg() else '❌ not found (needed for Telegram Opus)'}") print(f"\n Output dir: {DEFAULT_OUTPUT_DIR}")