feat(transcription): add shared voice input support (#4232)
* feat(webui): add voice transcription input * feat(webui): render ANSI output in code blocks * refactor(webui): isolate voice recorder logic * refactor(transcription): keep websocket ingress thin * refactor(transcription): resolve channel audio settings on demand * style(webui): neutralize voice waveform color * feat(webui): add voice input tooltip * feat(webui): add voice input keyboard shortcut * fix(webui): distinguish voice shortcut platforms * fix(webui): place voice button after model selector * refactor(webui): share voice hold recording helpers * fix(desktop): allow microphone voice input * fix(webui): stabilize token usage month labels * feat(webui): show voice input on settings overview * fix(webui): label voice capability as recognition * fix(webui): align capability overview status * refactor(webui): isolate transcription socket handling * fix(webui): soften silent voice waveform * refactor(audio): clarify transcription service location * docs(transcription): clarify audio and provider boundaries * fix(exec): reduce session output polling flake
This commit is contained in:
@@ -12,7 +12,8 @@ from nanobot.bus.events import OutboundMessage
|
||||
from nanobot.bus.queue import MessageBus
|
||||
from nanobot.channels.base import BaseChannel
|
||||
from nanobot.channels.manager import ChannelManager
|
||||
from nanobot.config.schema import ChannelsConfig
|
||||
from nanobot.config.loader import save_config
|
||||
from nanobot.config.schema import ChannelsConfig, Config
|
||||
from nanobot.providers.transcription import GroqTranscriptionProvider as _GroqProvider
|
||||
from nanobot.providers.transcription import OpenAITranscriptionProvider as _OpenAIProvider
|
||||
from nanobot.utils.restart import RestartNotice
|
||||
@@ -238,102 +239,103 @@ async def test_manager_loads_plugin_from_dict_config():
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_manager_propagates_groq_transcription_api_base_to_channels():
|
||||
from nanobot.channels.manager import ChannelManager
|
||||
|
||||
fake_config = SimpleNamespace(
|
||||
channels=ChannelsConfig.model_validate({
|
||||
"fakeplugin": {"enabled": True, "allowFrom": ["*"]},
|
||||
"transcriptionLanguage": "en",
|
||||
}),
|
||||
providers=SimpleNamespace(
|
||||
groq=SimpleNamespace(api_key="groq-key", api_base="http://proxy.local/v1/audio/transcriptions"),
|
||||
openai=SimpleNamespace(api_key="openai-key", api_base="https://api.openai.com/v1/audio/transcriptions"),
|
||||
),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"nanobot.channels.registry.discover_enabled",
|
||||
return_value={"fakeplugin": _FakePlugin},
|
||||
):
|
||||
mgr = ChannelManager.__new__(ChannelManager)
|
||||
mgr.config = fake_config
|
||||
mgr.bus = MessageBus()
|
||||
mgr.channels = {}
|
||||
mgr._dispatch_task = None
|
||||
mgr._init_channels()
|
||||
|
||||
channel = mgr.channels["fakeplugin"]
|
||||
assert channel.transcription_provider == "groq"
|
||||
assert channel.transcription_api_key == "groq-key"
|
||||
assert channel.transcription_api_base == "http://proxy.local/v1/audio/transcriptions"
|
||||
assert channel.transcription_language == "en"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_manager_propagates_openai_transcription_api_base_to_channels():
|
||||
from nanobot.channels.manager import ChannelManager
|
||||
|
||||
fake_config = SimpleNamespace(
|
||||
channels=ChannelsConfig.model_validate({
|
||||
"fakeplugin": {"enabled": True, "allowFrom": ["*"]},
|
||||
"transcriptionProvider": "openai",
|
||||
}),
|
||||
providers=SimpleNamespace(
|
||||
openai=SimpleNamespace(
|
||||
api_key="openai-key",
|
||||
api_base="http://proxy.local/v1/audio/transcriptions",
|
||||
),
|
||||
groq=SimpleNamespace(api_key="groq-key", api_base=""),
|
||||
),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"nanobot.channels.registry.discover_enabled",
|
||||
return_value={"fakeplugin": _FakePlugin},
|
||||
):
|
||||
mgr = ChannelManager.__new__(ChannelManager)
|
||||
mgr.config = fake_config
|
||||
mgr.bus = MessageBus()
|
||||
mgr.channels = {}
|
||||
mgr._dispatch_task = None
|
||||
mgr._init_channels()
|
||||
|
||||
channel = mgr.channels["fakeplugin"]
|
||||
assert channel.transcription_provider == "openai"
|
||||
assert channel.transcription_api_key == "openai-key"
|
||||
assert channel.transcription_api_base == "http://proxy.local/v1/audio/transcriptions"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_base_channel_passes_api_base_to_openai_transcription_provider():
|
||||
"""BaseChannel.transcribe_audio must forward transcription_api_base to OpenAI."""
|
||||
async def test_base_channel_reads_current_transcription_config_each_call(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
"""BaseChannel.transcribe_audio resolves config at call time, not manager init time."""
|
||||
from nanobot.providers import transcription as transcription_mod
|
||||
|
||||
channel = _FakePlugin({"enabled": True, "allowFrom": ["*"]}, MessageBus())
|
||||
channel.transcription_provider = "openai"
|
||||
channel.transcription_api_key = "k"
|
||||
channel.transcription_api_base = "http://override/v1/audio/transcriptions"
|
||||
channel.transcription_language = "en"
|
||||
config_path = tmp_path / "config.json"
|
||||
config = Config()
|
||||
config.transcription.provider = "openai"
|
||||
config.transcription.model = "whisper-custom"
|
||||
config.transcription.language = "en"
|
||||
config.providers.openai.api_key = "openai-key"
|
||||
config.providers.openai.api_base = "http://openai.local/v1/audio/transcriptions"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
|
||||
captured: dict[str, object] = {}
|
||||
channel = _FakePlugin({"enabled": True, "allowFrom": ["*"]}, MessageBus())
|
||||
|
||||
calls: list[dict[str, object]] = []
|
||||
|
||||
class _StubOpenAI:
|
||||
def __init__(self, api_key=None, api_base=None, language=None):
|
||||
captured["api_key"] = api_key
|
||||
captured["api_base"] = api_base
|
||||
captured["language"] = language
|
||||
def __init__(self, api_key=None, api_base=None, language=None, model=None):
|
||||
calls.append({
|
||||
"provider": "openai",
|
||||
"api_key": api_key,
|
||||
"api_base": api_base,
|
||||
"language": language,
|
||||
"model": model,
|
||||
})
|
||||
|
||||
async def transcribe(self, file_path):
|
||||
return "ok"
|
||||
return "openai-ok"
|
||||
|
||||
with patch.object(transcription_mod, "OpenAITranscriptionProvider", _StubOpenAI):
|
||||
result = await channel.transcribe_audio("/tmp/does-not-matter.wav")
|
||||
class _StubGroq:
|
||||
def __init__(self, api_key=None, api_base=None, language=None, model=None):
|
||||
calls.append({
|
||||
"provider": "groq",
|
||||
"api_key": api_key,
|
||||
"api_base": api_base,
|
||||
"language": language,
|
||||
"model": model,
|
||||
})
|
||||
|
||||
assert result == "ok"
|
||||
assert captured["api_key"] == "k"
|
||||
assert captured["api_base"] == "http://override/v1/audio/transcriptions"
|
||||
assert captured["language"] == "en"
|
||||
async def transcribe(self, file_path):
|
||||
return "groq-ok"
|
||||
|
||||
with (
|
||||
patch.object(transcription_mod, "OpenAITranscriptionProvider", _StubOpenAI),
|
||||
patch.object(transcription_mod, "GroqTranscriptionProvider", _StubGroq),
|
||||
):
|
||||
assert await channel.transcribe_audio("/tmp/does-not-matter.wav") == "openai-ok"
|
||||
|
||||
config.transcription.provider = "groq"
|
||||
config.transcription.model = "whisper-large-v3-turbo"
|
||||
config.transcription.language = "ko"
|
||||
config.providers.groq.api_key = "groq-key"
|
||||
config.providers.groq.api_base = "http://groq.local/v1/audio/transcriptions"
|
||||
save_config(config, config_path)
|
||||
|
||||
assert await channel.transcribe_audio("/tmp/does-not-matter.wav") == "groq-ok"
|
||||
|
||||
assert calls == [
|
||||
{
|
||||
"provider": "openai",
|
||||
"api_key": "openai-key",
|
||||
"api_base": "http://openai.local/v1/audio/transcriptions",
|
||||
"language": "en",
|
||||
"model": "whisper-custom",
|
||||
},
|
||||
{
|
||||
"provider": "groq",
|
||||
"api_key": "groq-key",
|
||||
"api_base": "http://groq.local/v1/audio/transcriptions",
|
||||
"language": "ko",
|
||||
"model": "whisper-large-v3-turbo",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_base_channel_respects_disabled_transcription_config(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
config_path = tmp_path / "config.json"
|
||||
config = Config()
|
||||
config.transcription.enabled = False
|
||||
config.providers.groq.api_key = "groq-key"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
|
||||
channel = _FakePlugin({"enabled": True, "allowFrom": ["*"]}, MessageBus())
|
||||
|
||||
with patch("nanobot.providers.transcription.GroqTranscriptionProvider") as provider:
|
||||
assert await channel.transcribe_audio("/tmp/does-not-matter.wav") == ""
|
||||
provider.assert_not_called()
|
||||
|
||||
|
||||
def test_openai_transcription_provider_honors_api_base_argument():
|
||||
@@ -348,37 +350,6 @@ def test_openai_transcription_provider_honors_api_base_argument():
|
||||
assert custom.api_url == "http://override/v1/audio/transcriptions"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_base_channel_passes_language_to_groq_transcription_provider():
|
||||
"""BaseChannel.transcribe_audio must forward transcription_language to Groq."""
|
||||
from nanobot.providers import transcription as transcription_mod
|
||||
|
||||
channel = _FakePlugin({"enabled": True, "allowFrom": ["*"]}, MessageBus())
|
||||
channel.transcription_provider = "groq"
|
||||
channel.transcription_api_key = "k"
|
||||
channel.transcription_api_base = "http://override/v1/audio/transcriptions"
|
||||
channel.transcription_language = "ko"
|
||||
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
class _StubGroq:
|
||||
def __init__(self, api_key=None, api_base=None, language=None):
|
||||
captured["api_key"] = api_key
|
||||
captured["api_base"] = api_base
|
||||
captured["language"] = language
|
||||
|
||||
async def transcribe(self, file_path):
|
||||
return "ok"
|
||||
|
||||
with patch.object(transcription_mod, "GroqTranscriptionProvider", _StubGroq):
|
||||
result = await channel.transcribe_audio("/tmp/does-not-matter.wav")
|
||||
|
||||
assert result == "ok"
|
||||
assert captured["api_key"] == "k"
|
||||
assert captured["api_base"] == "http://override/v1/audio/transcriptions"
|
||||
assert captured["language"] == "ko"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Transcription provider HTTP tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -69,6 +69,7 @@ def _make_channel() -> WebSocketChannel:
|
||||
[
|
||||
("data:image/png;base64,AAAA", "image/png"),
|
||||
("data:image/jpeg;base64,AAAA", "image/jpeg"),
|
||||
("data:audio/webm;codecs=opus;base64,AAAA", "audio/webm"),
|
||||
("data:IMAGE/PNG;base64,AAAA", "image/png"),
|
||||
("data:image/svg+xml;base64,AAAA", "image/svg+xml"),
|
||||
("data:text/plain;base64,AAAA", "text/plain"),
|
||||
|
||||
@@ -271,8 +271,6 @@ async def test_lid_to_phone_cache_resolves_lid_only_messages():
|
||||
async def test_voice_message_transcription_uses_media_path():
|
||||
"""Voice messages are transcribed when media path is available."""
|
||||
ch = WhatsAppChannel({"enabled": True, "allowFrom": ["*"]}, MagicMock())
|
||||
ch.transcription_provider = "openai"
|
||||
ch.transcription_api_key = "sk-test"
|
||||
ch._handle_message = AsyncMock()
|
||||
ch.transcribe_audio = AsyncMock(return_value="Hello world")
|
||||
|
||||
|
||||
@@ -8,6 +8,8 @@ from unittest.mock import AsyncMock, patch
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from nanobot.audio.transcription import resolve_transcription_config
|
||||
from nanobot.config.schema import Config
|
||||
from nanobot.providers.transcription import (
|
||||
GroqTranscriptionProvider,
|
||||
OpenAITranscriptionProvider,
|
||||
@@ -33,6 +35,65 @@ def _raw_response(status: int, content: bytes) -> httpx.Response:
|
||||
return httpx.Response(status_code=status, content=content, request=request)
|
||||
|
||||
|
||||
def test_resolver_uses_legacy_channel_provider_when_top_level_is_unset() -> None:
|
||||
config = Config()
|
||||
config.channels.transcription_provider = "openai"
|
||||
config.channels.transcription_language = "en"
|
||||
config.providers.openai.api_key = "sk-test"
|
||||
config.providers.openai.api_base = "https://proxy.example/v1"
|
||||
|
||||
resolved = resolve_transcription_config(config)
|
||||
|
||||
assert resolved.provider == "openai"
|
||||
assert resolved.model == "whisper-1"
|
||||
assert resolved.language == "en"
|
||||
assert resolved.api_key == "sk-test"
|
||||
assert resolved.api_base == "https://proxy.example/v1"
|
||||
assert resolved.configured is True
|
||||
|
||||
|
||||
def test_resolver_prefers_top_level_transcription_over_legacy_channels() -> None:
|
||||
config = Config()
|
||||
config.channels.transcription_provider = "openai"
|
||||
config.channels.transcription_language = "en"
|
||||
config.transcription.provider = "groq"
|
||||
config.transcription.model = "whisper-large-v3-turbo"
|
||||
config.transcription.language = "ko"
|
||||
config.providers.groq.api_key = "gsk-test"
|
||||
config.providers.groq.api_base = "https://groq.example/openai/v1"
|
||||
|
||||
resolved = resolve_transcription_config(config)
|
||||
|
||||
assert resolved.provider == "groq"
|
||||
assert resolved.model == "whisper-large-v3-turbo"
|
||||
assert resolved.language == "ko"
|
||||
assert resolved.api_key == "gsk-test"
|
||||
assert resolved.api_base == "https://groq.example/openai/v1"
|
||||
|
||||
|
||||
def test_resolved_transcription_repr_hides_api_key() -> None:
|
||||
config = Config()
|
||||
config.providers.groq.api_key = "gsk-secret"
|
||||
|
||||
resolved = resolve_transcription_config(config)
|
||||
|
||||
assert "gsk-secret" not in repr(resolved)
|
||||
assert "api_key" not in repr(resolved)
|
||||
|
||||
|
||||
def test_resolver_keeps_enabled_and_limits_on_effective_config() -> None:
|
||||
config = Config()
|
||||
config.transcription.enabled = False
|
||||
config.transcription.max_duration_sec = 45
|
||||
config.transcription.max_upload_mb = 12
|
||||
|
||||
resolved = resolve_transcription_config(config)
|
||||
|
||||
assert resolved.enabled is False
|
||||
assert resolved.max_duration_sec == 45
|
||||
assert resolved.max_upload_mb == 12
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# OpenAI provider — retry on transient HTTP + network errors
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -215,6 +276,32 @@ async def test_provider_omits_language_when_unset(
|
||||
assert "language" not in files
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provider_forwards_custom_model_in_multipart(audio_file: Path) -> None:
|
||||
provider = GroqTranscriptionProvider(api_key="k", model="whisper-large-v3-turbo")
|
||||
post = AsyncMock(return_value=_response(200, {"text": "ok"}))
|
||||
with patch("httpx.AsyncClient.post", post), patch("asyncio.sleep", AsyncMock()):
|
||||
result = await provider.transcribe(audio_file)
|
||||
|
||||
assert result == "ok"
|
||||
files = post.await_args_list[0].kwargs["files"]
|
||||
assert files["model"] == (None, "whisper-large-v3-turbo")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provider_forwards_file_mime_type(tmp_path: Path) -> None:
|
||||
audio = tmp_path / "voice.webm"
|
||||
audio.write_bytes(b"audio")
|
||||
provider = GroqTranscriptionProvider(api_key="k")
|
||||
post = AsyncMock(return_value=_response(200, {"text": "ok"}))
|
||||
with patch("httpx.AsyncClient.post", post), patch("asyncio.sleep", AsyncMock()):
|
||||
result = await provider.transcribe(audio)
|
||||
|
||||
assert result == "ok"
|
||||
files = post.await_args_list[0].kwargs["files"]
|
||||
assert files["file"] == ("voice.webm", b"audio", "audio/webm")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_language_survives_retry(audio_file: Path) -> None:
|
||||
"""Regression: language must be present on every retry attempt, not just the first."""
|
||||
|
||||
@@ -6,8 +6,12 @@ import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from nanobot.agent.tools.exec_session import (
|
||||
ExecSessionManager,
|
||||
ListExecSessionsTool,
|
||||
WriteStdinTool,
|
||||
)
|
||||
from nanobot.agent.tools.shell import ExecTool
|
||||
from nanobot.agent.tools.exec_session import ExecSessionManager, ListExecSessionsTool, WriteStdinTool
|
||||
|
||||
|
||||
def _python_command(code: str) -> str:
|
||||
@@ -141,7 +145,7 @@ def test_exec_can_continue_with_stdin(tmp_path):
|
||||
return initial, result
|
||||
|
||||
initial, result = asyncio.run(run())
|
||||
assert "ready" in initial
|
||||
assert "ready" in initial + result
|
||||
assert "Process running" in initial
|
||||
assert "Elapsed:" in initial
|
||||
assert "got:ping" in result
|
||||
@@ -170,7 +174,7 @@ def test_write_stdin_can_close_stdin(tmp_path):
|
||||
return initial, result
|
||||
|
||||
initial, result = asyncio.run(run())
|
||||
assert "ready" in initial
|
||||
assert "ready" in initial + result
|
||||
assert "got:payload" in result
|
||||
assert "Stdin closed." in result
|
||||
assert "Exit code: 0" in result
|
||||
@@ -185,14 +189,20 @@ def test_write_stdin_can_terminate_session(tmp_path):
|
||||
"import time; print('ready', flush=True); time.sleep(30)"
|
||||
)
|
||||
|
||||
initial = await exec_tool.execute(command=command, yield_time_ms=500)
|
||||
initial = await exec_tool.execute(command=command, yield_time_ms=100)
|
||||
sid = _session_id(initial)
|
||||
waited = await stdin_tool.execute(
|
||||
session_id=sid,
|
||||
wait_for="ready",
|
||||
wait_timeout_ms=3000,
|
||||
yield_time_ms=0,
|
||||
)
|
||||
result = await stdin_tool.execute(
|
||||
session_id=sid,
|
||||
terminate=True,
|
||||
yield_time_ms=0,
|
||||
)
|
||||
return initial, result
|
||||
return initial + waited, result
|
||||
|
||||
initial, result = asyncio.run(run())
|
||||
assert "ready" in initial
|
||||
@@ -243,7 +253,7 @@ def test_write_stdin_preserves_completed_session_output_until_polled(tmp_path):
|
||||
|
||||
initial, final = asyncio.run(run())
|
||||
|
||||
assert "ready" in initial
|
||||
assert "ready" in initial + final
|
||||
assert "done" in final
|
||||
assert "Exit code: 0" in final
|
||||
|
||||
|
||||
@@ -8,8 +8,8 @@ import pytest
|
||||
|
||||
from nanobot.utils.media_decode import (
|
||||
DEFAULT_MAX_BYTES,
|
||||
FileSizeExceeded,
|
||||
MAX_FILE_SIZE,
|
||||
FileSizeExceeded,
|
||||
save_base64_data_url,
|
||||
)
|
||||
|
||||
@@ -25,6 +25,31 @@ def test_saves_png_with_correct_extension(tmp_path) -> None:
|
||||
assert (tmp_path / result.split("/")[-1]).read_bytes() == b"fake png"
|
||||
|
||||
|
||||
def test_saves_data_url_with_mime_parameters(tmp_path) -> None:
|
||||
result = save_base64_data_url(_data_url(b"voice", mime="audio/webm;codecs=opus"), tmp_path)
|
||||
assert result is not None
|
||||
assert result.endswith(".webm")
|
||||
assert (tmp_path / result.split("/")[-1]).read_bytes() == b"voice"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mime", "suffix"),
|
||||
[
|
||||
("audio/webm", ".webm"),
|
||||
("video/webm", ".webm"),
|
||||
("audio/ogg", ".ogg"),
|
||||
("audio/wav", ".wav"),
|
||||
("audio/mpga", ".mpga"),
|
||||
],
|
||||
)
|
||||
def test_saves_common_audio_with_api_friendly_extension(
|
||||
tmp_path, mime: str, suffix: str
|
||||
) -> None:
|
||||
result = save_base64_data_url(_data_url(b"voice", mime=mime), tmp_path)
|
||||
assert result is not None
|
||||
assert result.endswith(suffix)
|
||||
|
||||
|
||||
def test_returns_none_for_malformed_data_url(tmp_path) -> None:
|
||||
assert save_base64_data_url("not-a-data-url", tmp_path) is None
|
||||
|
||||
|
||||
@@ -18,6 +18,7 @@ from nanobot.webui.settings_api import (
|
||||
update_agent_settings,
|
||||
update_model_configuration,
|
||||
update_network_safety_settings,
|
||||
update_transcription_settings,
|
||||
)
|
||||
|
||||
|
||||
@@ -243,6 +244,75 @@ def test_settings_payload_includes_network_safety_fields(
|
||||
assert payload["advanced"]["ssrf_whitelist_count"] == 1
|
||||
|
||||
|
||||
def test_settings_payload_includes_effective_transcription_config(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "config.json"
|
||||
config = Config()
|
||||
config.channels.transcription_provider = "openai"
|
||||
config.channels.transcription_language = "en"
|
||||
config.providers.openai.api_key = "sk-test"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
|
||||
payload = settings_payload()
|
||||
|
||||
assert payload["transcription"]["enabled"] is True
|
||||
assert payload["transcription"]["provider"] == "openai"
|
||||
assert payload["transcription"]["provider_configured"] is True
|
||||
assert payload["transcription"]["model"] == "whisper-1"
|
||||
assert payload["transcription"]["language"] == "en"
|
||||
|
||||
|
||||
def test_update_transcription_settings_writes_top_level_only(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "config.json"
|
||||
config = Config()
|
||||
config.channels.transcription_provider = "openai"
|
||||
config.channels.transcription_language = "en"
|
||||
config.providers.groq.api_key = "gsk-test"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
|
||||
payload = update_transcription_settings(
|
||||
{
|
||||
"enabled": ["true"],
|
||||
"provider": ["groq"],
|
||||
"model": ["whisper-large-v3-turbo"],
|
||||
"language": ["ko"],
|
||||
"maxDurationSec": ["90"],
|
||||
"maxUploadMb": ["20"],
|
||||
}
|
||||
)
|
||||
|
||||
saved = load_config(config_path)
|
||||
assert saved.channels.transcription_provider == "openai"
|
||||
assert saved.channels.transcription_language == "en"
|
||||
assert saved.transcription.enabled is True
|
||||
assert saved.transcription.provider == "groq"
|
||||
assert saved.transcription.model == "whisper-large-v3-turbo"
|
||||
assert saved.transcription.language == "ko"
|
||||
assert saved.transcription.max_duration_sec == 90
|
||||
assert saved.transcription.max_upload_mb == 20
|
||||
assert payload["transcription"]["provider"] == "groq"
|
||||
assert payload["transcription"]["provider_configured"] is True
|
||||
|
||||
|
||||
def test_update_transcription_settings_validates_language(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "config.json"
|
||||
save_config(Config(), config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
|
||||
with pytest.raises(WebUISettingsError, match="transcription language"):
|
||||
update_transcription_settings({"language": ["en-US"]})
|
||||
|
||||
|
||||
def test_settings_payload_includes_token_usage_summary(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
"""Tests for WebUI transcription envelopes carried over the gateway socket."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from nanobot.config.loader import save_config
|
||||
from nanobot.config.schema import Config
|
||||
from nanobot.webui.transcription_ws import webui_transcription_event
|
||||
|
||||
|
||||
def _audio_data_url(payload: bytes = b"voice", mime: str = "audio/webm") -> str:
|
||||
return f"data:{mime};base64,{base64.b64encode(payload).decode('ascii')}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webui_transcribe_audio_rejects_unconfigured_provider(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "config.json"
|
||||
config = Config()
|
||||
config.transcription.provider = "groq"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
|
||||
event, payload = await webui_transcription_event({
|
||||
"request_id": "voice-1",
|
||||
"data_url": _audio_data_url(),
|
||||
})
|
||||
|
||||
assert event == "transcription_error"
|
||||
assert payload == {
|
||||
"request_id": "voice-1",
|
||||
"detail": "not_configured",
|
||||
"provider": "groq",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webui_transcribe_audio_rejects_unsupported_mime(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "config.json"
|
||||
config = Config()
|
||||
config.transcription.provider = "groq"
|
||||
config.providers.groq.api_key = "gsk-test"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
|
||||
event, payload = await webui_transcription_event({
|
||||
"request_id": "voice-1",
|
||||
"data_url": _audio_data_url(mime="text/plain"),
|
||||
})
|
||||
|
||||
assert event == "transcription_error"
|
||||
assert payload["request_id"] == "voice-1"
|
||||
assert payload["detail"] == "mime"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webui_transcribe_audio_rejects_oversized_audio(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "config.json"
|
||||
config = Config()
|
||||
config.transcription.provider = "groq"
|
||||
config.transcription.max_upload_mb = 1
|
||||
config.providers.groq.api_key = "gsk-test"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
monkeypatch.setattr("nanobot.audio.transcription.get_media_dir", lambda _channel=None: tmp_path)
|
||||
|
||||
event, payload = await webui_transcription_event({
|
||||
"request_id": "voice-1",
|
||||
"data_url": _audio_data_url(payload=b"x" * (1024 * 1024 + 1)),
|
||||
})
|
||||
|
||||
assert event == "transcription_error"
|
||||
assert payload["request_id"] == "voice-1"
|
||||
assert payload["detail"] == "size"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webui_transcribe_audio_returns_text_and_removes_temp_file(
|
||||
tmp_path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "config.json"
|
||||
media_dir = tmp_path / "media"
|
||||
media_dir.mkdir()
|
||||
config = Config()
|
||||
config.transcription.provider = "groq"
|
||||
config.providers.groq.api_key = "gsk-test"
|
||||
save_config(config, config_path)
|
||||
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
|
||||
monkeypatch.setattr(
|
||||
"nanobot.audio.transcription.get_media_dir",
|
||||
lambda _channel=None: media_dir,
|
||||
)
|
||||
captured_paths: list[Path] = []
|
||||
|
||||
async def fake_transcribe_audio_file(path: str | Path, _resolved: Any) -> str:
|
||||
p = Path(path)
|
||||
assert p.exists()
|
||||
captured_paths.append(p)
|
||||
return "hello voice"
|
||||
|
||||
monkeypatch.setattr(
|
||||
"nanobot.audio.transcription.transcribe_audio_file",
|
||||
fake_transcribe_audio_file,
|
||||
)
|
||||
|
||||
event, payload = await webui_transcription_event({
|
||||
"request_id": "voice-1",
|
||||
"data_url": _audio_data_url(payload=b"webm voice", mime="audio/webm;codecs=opus"),
|
||||
"duration_ms": 1200,
|
||||
})
|
||||
|
||||
assert event == "transcription_result"
|
||||
assert payload == {"request_id": "voice-1", "text": "hello voice"}
|
||||
assert captured_paths
|
||||
assert not captured_paths[0].exists()
|
||||
Reference in New Issue
Block a user