wehub-resource-sync
541 lines
18 KiB
Python
541 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from opensquilla.gateway.config import AudioConfig
|
|
from opensquilla.provider.audio import (
|
|
AudioGenerationResult,
|
|
DubbingResult,
|
|
ElevenLabsSharedVoicesResult,
|
|
ElevenLabsSubscriptionResult,
|
|
ElevenLabsVoicesListResult,
|
|
MusicGenerationResult,
|
|
VoiceCloneResult,
|
|
VoiceConversionResult,
|
|
)
|
|
from opensquilla.tools.types import CallerKind, ToolContext, current_tool_context
|
|
|
|
|
|
def _audio_config(*, enabled: bool = True, api_key: str = "el-test") -> AudioConfig:
|
|
config = AudioConfig(enabled=enabled)
|
|
config.providers.elevenlabs.api_key = api_key
|
|
config.providers.elevenlabs.api_key_env = "ELEVENLABS_API_KEY"
|
|
return config
|
|
|
|
|
|
def _tool_context(tmp_path: Path) -> ToolContext:
|
|
return ToolContext(
|
|
caller_kind=CallerKind.AGENT,
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_tts_uses_elevenlabs_and_writes_audio(monkeypatch, tmp_path: Path) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def text_to_speech(self, request):
|
|
captured["request"] = request
|
|
return AudioGenerationResult(
|
|
audio_bytes=b"spoken-audio",
|
|
provider="elevenlabs",
|
|
model=request.model_id,
|
|
voice=request.voice,
|
|
response_format=request.output_format,
|
|
mime_type="audio/mpeg",
|
|
generation_id="gen_tts",
|
|
)
|
|
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.tts(
|
|
text="Speak this",
|
|
voice="voice_123",
|
|
output_path="speech.mp3",
|
|
language_code="en-GB",
|
|
speed=0.92,
|
|
stability=0.65,
|
|
similarity_boost=0.8,
|
|
style=0.0,
|
|
use_speaker_boost=True,
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
target = tmp_path / "workspace" / "speech.mp3"
|
|
result = json.loads(payload)
|
|
assert target.read_bytes() == b"spoken-audio"
|
|
assert result["status"] == "ok"
|
|
assert result["provider"] == "elevenlabs"
|
|
assert result["path"] == str(target)
|
|
request = captured["request"]
|
|
assert getattr(request, "text") == "Speak this"
|
|
assert getattr(request, "voice") == "voice_123"
|
|
assert getattr(request, "language_code") == "en-GB"
|
|
assert getattr(request, "voice_settings") == {
|
|
"stability": 0.65,
|
|
"similarity_boost": 0.8,
|
|
"style": 0.0,
|
|
"use_speaker_boost": True,
|
|
"speed": 0.92,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_tts_uses_configured_elevenlabs_voice_when_voice_is_omitted(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def text_to_speech(self, request):
|
|
captured["voice"] = request.voice
|
|
return AudioGenerationResult(
|
|
audio_bytes=b"default-voice-audio",
|
|
provider="elevenlabs",
|
|
model=request.model_id,
|
|
voice=request.voice,
|
|
response_format=request.output_format,
|
|
mime_type="audio/mpeg",
|
|
)
|
|
|
|
config = _audio_config()
|
|
config.tts.voice = "configured_voice"
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(config)
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.tts(text="Default voice", output_path="default.mp3")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
result = json.loads(payload)
|
|
assert result["status"] == "ok"
|
|
assert result["voice"] == "configured_voice"
|
|
assert captured["voice"] == "configured_voice"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_voice_clone_requires_consent_metadata(tmp_path: Path) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
sample = tmp_path / "workspace" / "sample.mp3"
|
|
sample.parent.mkdir(parents=True)
|
|
sample.write_bytes(b"ID3sample")
|
|
media.configure_audio(_audio_config())
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.voice_clone(sample_audio="sample.mp3", name="Demo")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
result = json.loads(payload)
|
|
assert result["status"] == "consent_required"
|
|
assert result["tool"] == "voice_clone"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_voice_clone_calls_elevenlabs_with_consent(monkeypatch, tmp_path: Path) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def clone_voice(self, request):
|
|
captured["request"] = request
|
|
return VoiceCloneResult(
|
|
provider="elevenlabs",
|
|
voice_id="voice_new",
|
|
name=request.name,
|
|
)
|
|
|
|
sample = tmp_path / "workspace" / "sample.mp3"
|
|
sample.parent.mkdir(parents=True)
|
|
sample.write_bytes(b"ID3sample")
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.voice_clone(
|
|
sample_audio="sample.mp3",
|
|
name="Demo",
|
|
consent_metadata={"speaker": "me", "consent": True},
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
result = json.loads(payload)
|
|
assert result["status"] == "ok"
|
|
assert result["voice_id"] == "voice_new"
|
|
request = captured["request"]
|
|
assert getattr(request, "sample_audio_bytes") == b"ID3sample"
|
|
assert getattr(request, "name") == "Demo"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_voice_convert_calls_elevenlabs_and_writes_audio(monkeypatch, tmp_path: Path) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def convert_voice(self, request):
|
|
return VoiceConversionResult(
|
|
audio_bytes=b"converted-audio",
|
|
provider="elevenlabs",
|
|
model=request.model_id,
|
|
voice=request.target_voice,
|
|
response_format=request.output_format,
|
|
mime_type="audio/mpeg",
|
|
)
|
|
|
|
source = tmp_path / "workspace" / "source.mp3"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_bytes(b"ID3source")
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.voice_convert(
|
|
source_audio="source.mp3",
|
|
target_voice="voice_123",
|
|
output_path="converted.mp3",
|
|
consent_metadata={"speaker": "me", "consent": True},
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
target = tmp_path / "workspace" / "converted.mp3"
|
|
result = json.loads(payload)
|
|
assert target.read_bytes() == b"converted-audio"
|
|
assert result["status"] == "ok"
|
|
assert result["voice"] == "voice_123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_dubbing_generate_submits_elevenlabs_job(monkeypatch, tmp_path: Path) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def create_dubbing(self, request):
|
|
captured["request"] = request
|
|
return DubbingResult(
|
|
provider="elevenlabs",
|
|
dubbing_id="dub_123",
|
|
status="submitted",
|
|
target_language=request.target_language,
|
|
source_language=request.source_language,
|
|
)
|
|
|
|
source = tmp_path / "workspace" / "source.mp3"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_bytes(b"ID3source")
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.dubbing_generate(
|
|
source_media="source.mp3",
|
|
target_language="zh",
|
|
source_language="en",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
result = json.loads(payload)
|
|
assert result["status"] == "ok"
|
|
assert result["dubbing_id"] == "dub_123"
|
|
assert result["target_language"] == "zh"
|
|
assert getattr(captured["request"], "watermark") is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_music_generate_calls_elevenlabs_and_writes_audio(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def generate_music(self, request):
|
|
captured["request"] = request
|
|
return MusicGenerationResult(
|
|
audio_bytes=b"music-audio",
|
|
provider="elevenlabs",
|
|
model=request.model_id,
|
|
response_format=request.output_format,
|
|
mime_type="audio/mpeg",
|
|
generation_id="music_123",
|
|
)
|
|
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.music_generate(
|
|
prompt="upbeat product intro",
|
|
style="electronic",
|
|
duration_seconds=12,
|
|
output_path="intro.mp3",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
target = tmp_path / "workspace" / "intro.mp3"
|
|
result = json.loads(payload)
|
|
assert target.read_bytes() == b"music-audio"
|
|
assert result["status"] == "ok"
|
|
assert result["path"] == str(target)
|
|
request = captured["request"]
|
|
assert getattr(request, "prompt") == "upbeat product intro\nStyle: electronic"
|
|
assert getattr(request, "duration_seconds") == 12
|
|
assert getattr(request, "force_instrumental") is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_song_generate_calls_elevenlabs_music_provider_with_lyrics(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def generate_music(self, request):
|
|
captured["request"] = request
|
|
return MusicGenerationResult(
|
|
audio_bytes=b"song-audio",
|
|
provider="elevenlabs",
|
|
model=request.model_id,
|
|
response_format=request.output_format,
|
|
mime_type="audio/mpeg",
|
|
)
|
|
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.song_generate(
|
|
lyrics="original line one\noriginal line two",
|
|
vocal_style="warm alto",
|
|
backing_style="acoustic",
|
|
duration_seconds=30,
|
|
output_path="song.mp3",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
target = tmp_path / "workspace" / "song.mp3"
|
|
result = json.loads(payload)
|
|
assert target.read_bytes() == b"song-audio"
|
|
assert result["status"] == "ok"
|
|
request = captured["request"]
|
|
assert getattr(request, "lyrics") == "original line one\noriginal line two"
|
|
assert getattr(request, "force_instrumental") is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_song_generate_retries_short_preview_when_quota_exceeded(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
requests: list[object] = []
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def generate_music(self, request):
|
|
requests.append(request)
|
|
if len(requests) == 1:
|
|
raise RuntimeError(
|
|
"Music generation provider returned HTTP 401 Unauthorized: "
|
|
'{"detail":{"code":"quota_exceeded","message":"This request '
|
|
'exceeds your API key quota. You have 2344 credits remaining, '
|
|
'while 2888 credits are required for this request."}}'
|
|
)
|
|
return MusicGenerationResult(
|
|
audio_bytes=b"short-song-audio",
|
|
provider="elevenlabs",
|
|
model=request.model_id,
|
|
response_format=request.output_format,
|
|
mime_type="audio/mpeg",
|
|
)
|
|
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
|
|
long_lyrics = "\n".join(
|
|
[
|
|
"[副歌]",
|
|
"夏天的风吹过你的头发",
|
|
"那是我们一起走过的年华",
|
|
"阳光洒在石板路上",
|
|
"你的笑容像花儿一样绽放",
|
|
"[主歌]",
|
|
"还记得那年放学的傍晚",
|
|
"你背着书包站在校门口等",
|
|
"我说今天作业又留了一大堆",
|
|
"你说没关系慢慢做别太认真",
|
|
]
|
|
)
|
|
token = current_tool_context.set(_tool_context(tmp_path))
|
|
try:
|
|
payload = await media.song_generate(
|
|
lyrics=long_lyrics,
|
|
vocal_style="温暖男声",
|
|
backing_style="木吉他民谣",
|
|
output_path="song.mp3",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
media.configure_audio(None)
|
|
|
|
target = tmp_path / "workspace" / "song.mp3"
|
|
result = json.loads(payload)
|
|
assert target.read_bytes() == b"short-song-audio"
|
|
assert result["status"] == "ok"
|
|
assert result["quota_retry"]["strategy"] == "short_preview"
|
|
assert len(requests) == 2
|
|
retry_request = requests[1]
|
|
assert getattr(retry_request, "duration_seconds") == 8.0
|
|
assert len(getattr(retry_request, "lyrics")) < len(long_lyrics)
|
|
assert getattr(retry_request, "force_instrumental") is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_audio_provider_capabilities_reports_live_subscription(
|
|
monkeypatch,
|
|
) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def get_subscription(self, _request):
|
|
return ElevenLabsSubscriptionResult(
|
|
provider="elevenlabs",
|
|
tier="creator",
|
|
status="active",
|
|
raw={"tier": "creator"},
|
|
)
|
|
|
|
async def list_voices(self, _request):
|
|
return ElevenLabsVoicesListResult(
|
|
provider="elevenlabs",
|
|
voices=[{"voice_id": "voice_123"}],
|
|
raw={"voices": [{"voice_id": "voice_123"}]},
|
|
)
|
|
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
try:
|
|
payload = await media.audio_provider_capabilities(probe_live=True)
|
|
finally:
|
|
media.configure_audio(None)
|
|
|
|
result = json.loads(payload)
|
|
assert result["status"] == "ok"
|
|
assert result["provider"] == "elevenlabs"
|
|
assert result["configured"] is True
|
|
assert result["subscription"]["tier"] == "creator"
|
|
assert result["voice_count"] == 1
|
|
assert result["capabilities"]["voice_cloning"]["status"] == "available"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_voice_search_returns_shared_voices_for_locale_matching(monkeypatch) -> None:
|
|
from opensquilla.tools.builtin import media
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProvider:
|
|
def __init__(self, **_kwargs):
|
|
return None
|
|
|
|
async def search_shared_voices(self, request):
|
|
captured["request"] = request
|
|
return ElevenLabsSharedVoicesResult(
|
|
provider="elevenlabs",
|
|
voices=[
|
|
{
|
|
"name": "Mandarin Narrator",
|
|
"voice_id": "voice_zh",
|
|
"public_owner_id": "owner_123",
|
|
"language": "zh",
|
|
"accent": "beijing mandarin",
|
|
"gender": "female",
|
|
"category": "professional",
|
|
}
|
|
],
|
|
raw={"has_more": False},
|
|
)
|
|
|
|
monkeypatch.setattr(media, "ElevenLabsAudioProductionProvider", FakeProvider)
|
|
media.configure_audio(_audio_config())
|
|
try:
|
|
payload = await media.voice_search(
|
|
language="zh",
|
|
accent="beijing mandarin",
|
|
page_size=3,
|
|
)
|
|
finally:
|
|
media.configure_audio(None)
|
|
|
|
result = json.loads(payload)
|
|
assert result["status"] == "ok"
|
|
assert result["voices"][0]["voice_id"] == "voice_zh"
|
|
assert getattr(captured["request"], "language") == "zh"
|
|
assert getattr(captured["request"], "accent") == "beijing mandarin"
|
|
assert getattr(captured["request"], "page_size") == 3
|