Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 98 additions & 1 deletion tests/tools/test_transcription_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import subprocess
import types
import wave
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock, mock_open, patch

import pytest

Expand Down Expand Up @@ -1580,3 +1580,100 @@ def test_no_env_var_uses_list_mode(self, monkeypatch):
monkeypatch.delenv(LOCAL_STT_COMMAND_ENV, raising=False)
use_shell = bool(os.getenv(LOCAL_STT_COMMAND_ENV, "").strip())
assert use_shell is False


# ============================================================================
# Groq STT config — null-subsection guard
# ============================================================================
class TestGroqSttLanguageConfig:
"""_transcribe_groq language lookup must survive stt.groq: null in config.

``_load_stt_config().get("groq", {}).get("language")`` crashes when
``stt.groq`` is explicitly ``null`` because ``.get("groq", {})`` returns
the null value (the default only applies when the key is *missing*).
All provider-subsection reads must use ``.get(key) or {}`` instead.
"""

GROQ_API_KEY = "sk-test-groq-key-placeholder"

def _make_groq_config(self, groq_value, local_value=None):
"""Build a fake _load_stt_config return dict."""
cfg = {}
if groq_value is not None:
cfg["groq"] = groq_value
if local_value is not None:
cfg["local"] = local_value
return cfg

def test_groq_override_present(self, monkeypatch):
"""stt.groq.language is set -> used."""
cfg = self._make_groq_config({"language": "fr"})
monkeypatch.setattr("tools.transcription_tools._load_stt_config", lambda: cfg)
monkeypatch.setattr("tools.transcription_tools.get_env_value", lambda k: self.GROQ_API_KEY if k == "GROQ_API_KEY" else None)
monkeypatch.setattr("tools.transcription_tools._HAS_OPENAI", True)

from tools.transcription_tools import _transcribe_groq
with patch("openai.OpenAI") as mock_openai, \
patch("builtins.open", mock_open(read_data=b"fake audio data")):
mock_client = mock_openai.return_value
mock_client.audio.transcriptions.create.return_value = "bonjour"
result = _transcribe_groq("/tmp/test.wav", "whisper-1")

assert result.get("success") is True
# The language should be passed as 'fr'
create_kwargs = mock_openai.return_value.audio.transcriptions.create.call_args[1]
assert create_kwargs.get("language") == "fr"

def test_local_fallback(self, monkeypatch):
"""stt.groq is empty but stt.local.language is set -> fallback used."""
cfg = self._make_groq_config({}, {"language": "de"})
monkeypatch.setattr("tools.transcription_tools._load_stt_config", lambda: cfg)
monkeypatch.setattr("tools.transcription_tools.get_env_value", lambda k: self.GROQ_API_KEY if k == "GROQ_API_KEY" else None)
monkeypatch.setattr("tools.transcription_tools._HAS_OPENAI", True)

from tools.transcription_tools import _transcribe_groq
with patch("openai.OpenAI") as mock_openai, \
patch("builtins.open", mock_open(read_data=b"fake audio data")):
mock_client = mock_openai.return_value
mock_client.audio.transcriptions.create.return_value = "guten tag"
result = _transcribe_groq("/tmp/test.wav", "whisper-1")

assert result.get("success") is True
create_kwargs = mock_openai.return_value.audio.transcriptions.create.call_args[1]
assert create_kwargs.get("language") == "de"

def test_no_language_configured(self, monkeypatch):
"""Neither groq nor local has language -> None (auto-detect)."""
cfg = self._make_groq_config({}, {})
monkeypatch.setattr("tools.transcription_tools._load_stt_config", lambda: cfg)
monkeypatch.setattr("tools.transcription_tools.get_env_value", lambda k: self.GROQ_API_KEY if k == "GROQ_API_KEY" else None)
monkeypatch.setattr("tools.transcription_tools._HAS_OPENAI", True)

from tools.transcription_tools import _transcribe_groq
with patch("openai.OpenAI") as mock_openai, \
patch("builtins.open", mock_open(read_data=b"fake audio data")):
mock_client = mock_openai.return_value
mock_client.audio.transcriptions.create.return_value = "hello"
result = _transcribe_groq("/tmp/test.wav", "whisper-1")

assert result.get("success") is True
create_kwargs = mock_openai.return_value.audio.transcriptions.create.call_args[1]
assert create_kwargs.get("language") is None

def test_stt_groq_null(self, monkeypatch):
"""stt.groq is explicitly null (YAML null) — must not crash."""
cfg = self._make_groq_config(None, {})
monkeypatch.setattr("tools.transcription_tools._load_stt_config", lambda: cfg)
monkeypatch.setattr("tools.transcription_tools.get_env_value", lambda k: self.GROQ_API_KEY if k == "GROQ_API_KEY" else None)
monkeypatch.setattr("tools.transcription_tools._HAS_OPENAI", True)

from tools.transcription_tools import _transcribe_groq
with patch("openai.OpenAI") as mock_openai, \
patch("builtins.open", mock_open(read_data=b"fake audio data")):
mock_client = mock_openai.return_value
mock_client.audio.transcriptions.create.return_value = "hello from null-guard"
result = _transcribe_groq("/tmp/test.wav", "whisper-1")

assert result.get("success") is True
create_kwargs = mock_openai.return_value.audio.transcriptions.create.call_args[1]
assert create_kwargs.get("language") is None # null subsection -> no override
25 changes: 18 additions & 7 deletions tools/transcription_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -1315,20 +1315,31 @@ def _transcribe_groq(file_path: str, model_name: str) -> Dict[str, Any]:
logger.info("Model %s not available on Groq, using %s", model_name, DEFAULT_GROQ_STT_MODEL)
model_name = DEFAULT_GROQ_STT_MODEL

# Language: config.yaml (stt.groq.language > stt.local.language) > auto-detect.
stt_config = _load_stt_config()
language = (
(stt_config.get("groq") or {}).get("language")
or (stt_config.get("local") or {}).get("language")
)

try:
from openai import OpenAI, APIError, APIConnectionError, APITimeoutError
client = OpenAI(api_key=api_key, base_url=GROQ_BASE_URL, timeout=30, max_retries=0)
try:
with open(file_path, "rb") as audio_file:
transcription = client.audio.transcriptions.create(
model=model_name,
file=audio_file,
response_format="text",
)
create_kwargs = {
"model": model_name,
"file": audio_file,
"response_format": "text",
}
if language:
create_kwargs["language"] = language
transcription = client.audio.transcriptions.create(**create_kwargs)

transcript_text = str(transcription).strip()
logger.info("Transcribed %s via Groq API (%s, %d chars)",
Path(file_path).name, model_name, len(transcript_text))
logger.info("Transcribed %s via Groq API (%s, %d chars, lang=%s)",
Path(file_path).name, model_name, len(transcript_text),
language or "auto")

return {"success": True, "transcript": transcript_text, "provider": "groq"}
finally:
Expand Down
Loading