/
/
/
1"""Tests for the OpenAI Text-to-speech provider."""
2
3from __future__ import annotations
4
5from pathlib import Path
6from typing import TYPE_CHECKING
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from aiohttp import ClientError, web
11from music_assistant_models.enums import ContentType, MediaType, ProviderType, StreamType
12
13from music_assistant.providers.openai_tts import (
14 CONF_VOICES,
15 DEFAULT_VOICES,
16 SUPPORTED_FEATURES,
17 OpenAITTSProvider,
18 fetch_backend_voices,
19)
20
21if TYPE_CHECKING:
22 from music_assistant_models.config_entries import ConfigValueType
23
24INSTANCE_ID = "openai_tts_instance"
25
26
27def create_provider(**config_values: ConfigValueType) -> OpenAITTSProvider:
28 """Construct an openai_tts provider with stubbed mass/manifest/config."""
29 mass = MagicMock()
30 mass.streams.base_url = "http://mass.local:8095"
31 # no setup data, so every read resolves through the config values below
32 mass.config.get = MagicMock(return_value={})
33 # no voice listing endpoint available in these tests
34 mass.http_session.get = MagicMock(side_effect=ClientError("no connection"))
35 manifest = MagicMock()
36 manifest.type = ProviderType.PLUGIN
37 manifest.domain = "openai_tts"
38 config = MagicMock()
39 config.name = "OpenAI Text-to-speech"
40 config.instance_id = INSTANCE_ID
41 config.values = {}
42 config.get_value = MagicMock(
43 side_effect=lambda key, default=None: config_values.get(key, default)
44 )
45 return OpenAITTSProvider(mass, manifest, config, SUPPORTED_FEATURES)
46
47
48async def test_resolve_voices_honours_config_override() -> None:
49 """The config override wins, stripped and de-duplicated with empty entries dropped."""
50 provider = create_provider(**{CONF_VOICES: [" nova ", "alloy", "", " nova", "echo "]})
51 assert await provider._resolve_voices() == ["nova", "alloy", "echo"]
52
53
54async def test_resolve_voices_splits_values_holding_commas() -> None:
55 """A whole list pasted into one value is split into the separate voices."""
56 provider = create_provider(**{CONF_VOICES: [" nova ,alloy", "echo", " nova"]})
57 assert await provider._resolve_voices() == ["nova", "alloy", "echo"]
58
59
60async def test_resolve_voices_accepts_a_hand_edited_string() -> None:
61 """A raw string in the stored config is read as a list of voices."""
62 provider = create_provider(**{CONF_VOICES: " nova ,alloy,, nova,echo "})
63 assert await provider._resolve_voices() == ["nova", "alloy", "echo"]
64
65
66async def test_resolve_voices_ignores_an_empty_override() -> None:
67 """An empty override is the shape stored when nothing is configured."""
68 provider = create_provider(**{CONF_VOICES: []})
69 assert await provider._resolve_voices() == list(DEFAULT_VOICES)
70
71
72async def test_voices_config_entry_accepts_a_cleared_field() -> None:
73 """Clearing the field submits no value, which must still parse as a list."""
74 provider = create_provider()
75 entry = next(e for e in await provider.get_config_entries() if e.key == CONF_VOICES)
76 assert entry.parse_value(None) == []
77
78
79async def test_resolve_voices_falls_back_to_defaults() -> None:
80 """Without an override and with failing discovery, the default voices are used."""
81 provider = create_provider()
82 assert await provider._resolve_voices() == list(DEFAULT_VOICES)
83
84
85async def test_get_tts_engines_yields_engine_per_voice() -> None:
86 """Every voice is exposed as an engine, using the voice identifier verbatim."""
87 provider = create_provider()
88 provider._voices = ["alloy", "nova"]
89 engines = await provider.get_tts_engines()
90 assert [(engine.id, engine.name) for engine in engines] == [
91 ("alloy", "alloy"),
92 ("nova", "nova"),
93 ]
94 assert all(engine.provider is provider for engine in engines)
95
96
97async def test_get_tts_message_returns_http_streamdetails() -> None:
98 """The rendered clip is served as MP3 over the instance's own stream route."""
99 provider = create_provider()
100 provider._voices = ["alloy"]
101 file_id = "a" * 64
102 with patch.object(provider, "_render_speech", AsyncMock(return_value=file_id)):
103 streamdetails = await provider.get_tts_message("hello there")
104 assert streamdetails.stream_type == StreamType.HTTP
105 assert streamdetails.audio_format.content_type == ContentType.MP3
106 assert streamdetails.media_type == MediaType.SOUND_EFFECT
107 assert streamdetails.item_id == file_id
108 assert streamdetails.path == f"http://mass.local:8095/{INSTANCE_ID}_speech?id={file_id}"
109
110
111def create_voices_session(payload: object) -> MagicMock:
112 """Return an http session whose voices endpoint responds with the given payload."""
113 response = MagicMock()
114 response.raise_for_status = MagicMock()
115 response.json = AsyncMock(return_value=payload)
116 session = MagicMock()
117 session.get = MagicMock(
118 return_value=MagicMock(
119 __aenter__=AsyncMock(return_value=response), __aexit__=AsyncMock(return_value=False)
120 )
121 )
122 return session
123
124
125async def test_fetch_backend_voices_accepts_known_payload_shapes() -> None:
126 """A voices listing may be wrapped or bare, holding plain names or objects."""
127 for payload, expected in (
128 ({"voices": ["af_bella", "am_adam"]}, ["af_bella", "am_adam"]),
129 (["af_bella"], ["af_bella"]),
130 ({"voices": [{"id": "af_bella"}, {"name": "am_adam"}]}, ["af_bella", "am_adam"]),
131 ({"voices": [{"unexpected": "shape"}, "af_bella"]}, ["af_bella"]),
132 ({"voices": []}, []),
133 # an unusable entry must not discard the voices around it
134 ({"voices": [None, 5, "af_bella", {"id": "am_adam"}]}, ["af_bella", "am_adam"]),
135 # a bare json string is not a voice list, and must not yield one-letter ids
136 ("af_bella", []),
137 ({"voices": "af_bella"}, []),
138 ):
139 session = create_voices_session(payload)
140 assert await fetch_backend_voices(session, "http://localhost:8880/v1") == expected
141
142
143async def test_fetch_backend_voices_never_raises() -> None:
144 """A backend without a voice listing yields no voices instead of an error."""
145 session = MagicMock()
146 session.get = MagicMock(side_effect=ClientError("no connection"))
147 assert await fetch_backend_voices(session, "https://api.openai.com/v1") == []
148
149
150async def test_index_cache_adopts_only_rendered_clips(tmp_path: Path) -> None:
151 """Anything this provider did not write itself stays unreachable through the route."""
152 provider = create_provider()
153 provider._cache_dir = str(tmp_path)
154 clip = tmp_path / f"{'b' * 64}.mp3"
155 clip.write_bytes(b"clip")
156 (tmp_path / "not-a-hash.mp3").write_bytes(b"clip")
157 (tmp_path / f"{'b' * 64}.txt").write_bytes(b"clip")
158 (tmp_path / f"{'c' * 64}.mp3").symlink_to(clip)
159
160 assert await provider._index_cache() == {"b" * 64: str(clip)}
161
162
163async def test_handle_speech_request_rejects_missing_id() -> None:
164 """A request without an id is rejected."""
165 provider = create_provider()
166 provider._clips = {}
167 request = MagicMock(spec=web.Request)
168 request.query = {}
169 response = await provider._handle_speech_request(request)
170 assert response.status == 400
171
172
173async def test_handle_speech_request_serves_indexed_clips_only() -> None:
174 """An id that this instance did not render never reaches the filesystem."""
175 provider = create_provider()
176 provider._clips = {}
177 for query in ({"id": "../../../etc/passwd"}, {"id": "NOTAHASH"}, {"id": "a" * 64}):
178 request = MagicMock(spec=web.Request)
179 request.query = query
180 with pytest.raises(web.HTTPNotFound):
181 await provider._handle_speech_request(request)
182