mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
addb8d0be8
When the active model can't ingest a kind natively, the wire resolver converts it client-side instead of sending a part the model can't read. Per-kind ownership, no shared machinery: PDF text-extraction is a pure-local PDF concern; audio transcription is an STT concern memoized in the audio domain. - core/pdf.py: extract_pdf_text via pypdfium2 (pure-local, no network, no cache — re-run per build; page-capped) - core/audio.py: transcribe_cached — non-raising, memoized by (alias, content-hash); backend failures not cached - session._wire_content_part: per-kind dispatch — native where the model supports the kind (supports_pdf / supports_audio_input), else fallback; display/export resolve natively so no conversion fires on a render - image left ungated (pre-existing behavior unchanged) - pyproject: pypdfium2 dependency + mypy untyped-import override - tests: pdf extraction, transcript memoization, per-kind gate dispatch
269 lines
11 KiB
Python
269 lines
11 KiB
Python
"""Unit tests for the STT/TTS audio helper (model-role resolution + backends).
|
|
|
|
``transcribe`` / ``synthesize`` are exercised through the registry boundary
|
|
with a mocked OpenAI-SDK client (mocking ``client.audio.*``), so the real
|
|
helper code runs end-to-end without a network call.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from turnstone.core import audio
|
|
|
|
|
|
class _Cfg:
|
|
"""Stand-in for ModelConfig — only the fields audio.py reads."""
|
|
|
|
def __init__(self, model: str, capabilities: dict | None = None) -> None:
|
|
self.model = model
|
|
self.capabilities = capabilities or {}
|
|
|
|
|
|
class _FakeConfigStore:
|
|
def __init__(self, **values: str) -> None:
|
|
self._values = values
|
|
|
|
def get(self, key: str, default: str = "") -> str:
|
|
return self._values.get(key, default)
|
|
|
|
|
|
class _FakeRegistry:
|
|
"""Minimal registry exposing the surface audio.py uses."""
|
|
|
|
def __init__(self, alias: str, cfg: _Cfg, client: object) -> None:
|
|
self._alias = alias
|
|
self._cfg = cfg
|
|
self._client = client
|
|
|
|
def has_alias(self, alias: str) -> bool:
|
|
return alias == self._alias
|
|
|
|
def get_config(self, alias: str) -> _Cfg:
|
|
if alias != self._alias:
|
|
raise ValueError(alias)
|
|
return self._cfg
|
|
|
|
def resolve(self, alias: str | None = None):
|
|
if alias not in (None, self._alias):
|
|
raise ValueError(alias)
|
|
return self._client, self._cfg.model, self._cfg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capability gating
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestModelSupportsRole:
|
|
def test_explicit_flag_wins(self):
|
|
assert audio.model_supports_role(_Cfg("anything", {"supports_transcription": True}), "stt")
|
|
# Explicit False overrides the would-be inference from the model name.
|
|
assert not audio.model_supports_role(
|
|
_Cfg("gpt-4o-mini-tts", {"supports_speech_synthesis": False}), "tts"
|
|
)
|
|
|
|
def test_infers_known_openai_audio_models(self):
|
|
assert audio.model_supports_role(_Cfg("gpt-4o-mini-transcribe"), "stt")
|
|
assert audio.model_supports_role(_Cfg("whisper-1"), "stt")
|
|
assert audio.model_supports_role(_Cfg("gpt-4o-mini-tts"), "tts")
|
|
assert audio.model_supports_role(_Cfg("tts-1"), "tts")
|
|
|
|
def test_chat_model_not_eligible(self):
|
|
assert not audio.model_supports_role(_Cfg("gpt-5"), "stt")
|
|
# Anthropic has no audio API — gated out of every audio role.
|
|
assert not audio.model_supports_role(_Cfg("claude-opus-4-8"), "tts")
|
|
assert not audio.model_supports_role(_Cfg("claude-opus-4-8"), "stt")
|
|
|
|
def test_unknown_role(self):
|
|
assert not audio.model_supports_role(_Cfg("whisper-1"), "vision_eval")
|
|
|
|
def test_hint_seed_lists_are_pinned(self):
|
|
# Mirrored verbatim in admin.js AUDIO_MODEL_HINTS — if these change,
|
|
# update the JS dropdown gate too (this pin makes the change deliberate).
|
|
assert audio._AUDIO_MODEL_HINTS == {
|
|
"stt": ("transcribe", "whisper", "-asr"),
|
|
"tts": ("tts-", "-tts"),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Role resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResolveRoleAlias:
|
|
def test_resolves_configured_capable_alias(self):
|
|
reg = _FakeRegistry("voice", _Cfg("gpt-4o-mini-transcribe"), MagicMock())
|
|
cs = _FakeConfigStore(**{"audio.stt_model_alias": "voice"})
|
|
assert audio.resolve_role_alias(config_store=cs, registry=reg, role="stt") == "voice"
|
|
|
|
def test_none_when_unset(self):
|
|
reg = _FakeRegistry("voice", _Cfg("gpt-4o-mini-transcribe"), MagicMock())
|
|
assert (
|
|
audio.resolve_role_alias(config_store=_FakeConfigStore(), registry=reg, role="stt")
|
|
is None
|
|
)
|
|
|
|
def test_none_when_alias_missing_from_registry(self):
|
|
reg = _FakeRegistry("voice", _Cfg("gpt-4o-mini-transcribe"), MagicMock())
|
|
cs = _FakeConfigStore(**{"audio.stt_model_alias": "ghost"})
|
|
assert audio.resolve_role_alias(config_store=cs, registry=reg, role="stt") is None
|
|
|
|
def test_none_when_alias_not_capability_eligible(self):
|
|
# Alias exists but its model can't do TTS -> gated out (Anthropic case).
|
|
reg = _FakeRegistry("brain", _Cfg("claude-opus-4-8"), MagicMock())
|
|
cs = _FakeConfigStore(**{"audio.tts_model_alias": "brain"})
|
|
assert audio.resolve_role_alias(config_store=cs, registry=reg, role="tts") is None
|
|
|
|
def test_none_when_no_registry_or_store(self):
|
|
assert audio.resolve_role_alias(config_store=None, registry=None, role="stt") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# transcribe / synthesize — boundary: mocked OpenAI-SDK client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTranscribe:
|
|
def test_calls_audio_transcriptions_and_returns_text(self):
|
|
client = MagicMock()
|
|
client.audio.transcriptions.create.return_value = MagicMock(text=" hello world ")
|
|
reg = _FakeRegistry("voice", _Cfg("gpt-4o-mini-transcribe"), client)
|
|
res = audio.transcribe(
|
|
registry=reg, alias="voice", data=b"RIFFfake", filename="speech.webm"
|
|
)
|
|
assert res.transcript == "hello world"
|
|
assert res.model_alias == "voice"
|
|
assert res.model == "gpt-4o-mini-transcribe"
|
|
kwargs = client.audio.transcriptions.create.call_args.kwargs
|
|
assert kwargs["model"] == "gpt-4o-mini-transcribe"
|
|
assert kwargs["file"] == ("speech.webm", b"RIFFfake")
|
|
|
|
def test_prompt_forwarded_when_set(self):
|
|
client = MagicMock()
|
|
client.audio.transcriptions.create.return_value = MagicMock(text="ok")
|
|
reg = _FakeRegistry("voice", _Cfg("whisper-1"), client)
|
|
audio.transcribe(
|
|
registry=reg, alias="voice", data=b"x", filename="a.wav", prompt="ACME jargon"
|
|
)
|
|
assert client.audio.transcriptions.create.call_args.kwargs["prompt"] == "ACME jargon"
|
|
|
|
def test_prompt_omitted_when_blank(self):
|
|
client = MagicMock()
|
|
client.audio.transcriptions.create.return_value = MagicMock(text="ok")
|
|
reg = _FakeRegistry("voice", _Cfg("whisper-1"), client)
|
|
audio.transcribe(registry=reg, alias="voice", data=b"x", filename="a.wav")
|
|
assert "prompt" not in client.audio.transcriptions.create.call_args.kwargs
|
|
|
|
def test_backend_failure_raises_backend_error(self):
|
|
client = MagicMock()
|
|
client.audio.transcriptions.create.side_effect = RuntimeError("boom")
|
|
reg = _FakeRegistry("voice", _Cfg("whisper-1"), client)
|
|
with pytest.raises(audio.AudioBackendError):
|
|
audio.transcribe(registry=reg, alias="voice", data=b"x", filename="a.wav")
|
|
|
|
|
|
class TestSynthesize:
|
|
def test_calls_audio_speech_and_returns_bytes(self):
|
|
client = MagicMock()
|
|
speech = MagicMock()
|
|
speech.read.return_value = b"RIFF...wavbytes"
|
|
client.audio.speech.create.return_value = speech
|
|
reg = _FakeRegistry("voice", _Cfg("gpt-4o-mini-tts"), client)
|
|
res = audio.synthesize(registry=reg, alias="voice", text="hi", voice="nova")
|
|
assert res.audio_bytes == b"RIFF...wavbytes"
|
|
assert res.media_type == "audio/mpeg"
|
|
assert res.model_alias == "voice"
|
|
kwargs = client.audio.speech.create.call_args.kwargs
|
|
assert kwargs["voice"] == "nova"
|
|
assert kwargs["input"] == "hi"
|
|
|
|
def test_default_voice_when_empty(self):
|
|
client = MagicMock()
|
|
client.audio.speech.create.return_value = MagicMock(read=lambda: b"a")
|
|
reg = _FakeRegistry("voice", _Cfg("gpt-4o-mini-tts"), client)
|
|
audio.synthesize(registry=reg, alias="voice", text="hi", voice="")
|
|
assert client.audio.speech.create.call_args.kwargs["voice"] == "alloy"
|
|
|
|
def test_backend_failure_raises_backend_error(self):
|
|
client = MagicMock()
|
|
client.audio.speech.create.side_effect = RuntimeError("down")
|
|
reg = _FakeRegistry("voice", _Cfg("gpt-4o-mini-tts"), client)
|
|
with pytest.raises(audio.AudioBackendError):
|
|
audio.synthesize(registry=reg, alias="voice", text="hi", voice="nova")
|
|
|
|
|
|
class TestOpenAIAudioModelsKnown:
|
|
"""The current OpenAI STT/TTS lineup is registered in the static capability
|
|
table, so the admin 'suggested capabilities' recognizes them and they show
|
|
in the known-models list. (Role gating also works via name inference for
|
|
openai-compatible/local backends that aren't in the static table.)"""
|
|
|
|
def test_stt_models_flagged(self):
|
|
from turnstone.core.providers import lookup_model_capabilities
|
|
|
|
for m in (
|
|
"whisper-1",
|
|
"gpt-4o-transcribe",
|
|
"gpt-4o-mini-transcribe",
|
|
"gpt-4o-transcribe-diarize", # prefix variant
|
|
):
|
|
caps = lookup_model_capabilities("openai", m) or {}
|
|
assert caps.get("supports_transcription") is True, m
|
|
assert caps.get("supports_speech_synthesis") is False, m
|
|
|
|
def test_tts_models_flagged(self):
|
|
from turnstone.core.providers import lookup_model_capabilities
|
|
|
|
for m in ("tts-1", "tts-1-hd", "gpt-4o-mini-tts"): # tts-1-hd is a prefix variant
|
|
caps = lookup_model_capabilities("openai", m) or {}
|
|
assert caps.get("supports_speech_synthesis") is True, m
|
|
assert caps.get("supports_transcription") is False, m
|
|
|
|
def test_chat_model_has_no_audio_flags(self):
|
|
from turnstone.core.providers import lookup_model_capabilities
|
|
|
|
caps = lookup_model_capabilities("openai", "gpt-5") or {}
|
|
assert not caps.get("supports_transcription")
|
|
assert not caps.get("supports_speech_synthesis")
|
|
|
|
|
|
class TestTranscribeCached:
|
|
"""The memoized, non-raising transcribe used by the no-native-audio wire
|
|
fallback. Caching an STT result is an audio-domain concern, so it lives here
|
|
next to ``transcribe`` rather than bundled with PDF text extraction."""
|
|
|
|
def _result(self, text: str):
|
|
return audio.TranscriptionResult(transcript=text, model_alias="w", model="m")
|
|
|
|
def test_memoizes_by_alias_and_hash(self, monkeypatch):
|
|
audio._clear_transcript_cache_for_test()
|
|
calls = []
|
|
|
|
def fake(*, registry, alias, data, filename):
|
|
calls.append(1)
|
|
return self._result("hello world")
|
|
|
|
monkeypatch.setattr(audio, "transcribe", fake)
|
|
kw = dict(registry=object(), alias="w", content_hash="h1", data=b"x", filename="a.wav")
|
|
assert audio.transcribe_cached(**kw) == "hello world"
|
|
assert audio.transcribe_cached(**kw) == "hello world"
|
|
assert len(calls) == 1 # second served from cache
|
|
|
|
def test_backend_failure_returns_empty_and_is_not_cached(self, monkeypatch):
|
|
audio._clear_transcript_cache_for_test()
|
|
calls = []
|
|
|
|
def boom(*, registry, alias, data, filename):
|
|
calls.append(1)
|
|
raise audio.AudioBackendError("down")
|
|
|
|
monkeypatch.setattr(audio, "transcribe", boom)
|
|
kw = dict(registry=object(), alias="w", content_hash="h2", data=b"x", filename="a.wav")
|
|
assert audio.transcribe_cached(**kw) == ""
|
|
audio.transcribe_cached(**kw)
|
|
assert len(calls) == 2 # failure not cached -> retried
|