music-assistant-server

1"""Unit tests for AI Radio storage normalization helpers."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7import logging
8from pathlib import Path
9from typing import Any, cast
10
11import pytest
12from music_assistant_models.errors import InvalidDataError
13
14from music_assistant.providers.ai_radio import storage as storage_module
15from music_assistant.providers.ai_radio.storage import AIRadioStorageMixin
16
17
18class DummyStorage(AIRadioStorageMixin):
19    """Minimal storage harness for unit testing helper methods."""
20
21    def __init__(self) -> None:
22        """Initialize dummy mixin state."""
23        self.logger = logging.getLogger(__name__)
24        self._sections: dict[str, dict[str, Any]] = {}
25        self._stations: dict[str, dict[str, Any]] = {}
26        self._hosts: dict[str, dict[str, Any]] = {}
27
28
29def _section(
30    section_id: str,
31    *,
32    name: str | None = None,
33    section_type: str = "ai_text",
34    prompt: str = "Prompt",
35) -> dict[str, Any]:
36    """Build a minimal shared section payload."""
37    return {
38        "id": section_id,
39        "name": name or section_id,
40        "type": section_type,
41        "prompt": prompt,
42    }
43
44
45def _station(section_ids: list[str]) -> dict[str, Any]:
46    """Build a minimal valid station payload."""
47    return {
48        "id": "station_a",
49        "name": "Station A",
50        "source_playlist_id": "playlist-1",
51        "source_playlist_provider": "library",
52        "section_ids": section_ids,
53        "section_order": [
54            {
55                "when": "between_songs",
56                "flow": [{"MUST": section_ids[0]}],
57            }
58        ],
59    }
60
61
62def test_materialize_sections_reports_missing_and_returns_copies() -> None:
63    """Resolve sections and report unknown ids without mutating source map."""
64    storage = DummyStorage()
65    storage._sections = {
66        "Song_Transition": {
67            "id": "Song_Transition",
68            "name": "Song Transition",
69            "type": "ai_text",
70            "prompt": "Transition",
71        }
72    }
73
74    sections, missing = storage._materialize_sections(["Song_Transition", "Unknown_Section"])
75
76    assert missing == ["Unknown_Section"]
77    assert sections[0]["id"] == "Song_Transition"
78    sections[0]["name"] = "Changed"
79    assert storage._sections["Song_Transition"]["name"] == "Song Transition"
80
81
82def test_normalize_section_rejects_invalid_type() -> None:
83    """Reject shared sections with unsupported types."""
84    storage = DummyStorage()
85
86    with pytest.raises(InvalidDataError, match="invalid type"):
87        storage._normalize_section(
88            {
89                "id": "Bad_Section",
90                "name": "Bad Section",
91                "type": "unsupported",
92                "prompt": "Prompt",
93            }
94        )
95
96
97def test_normalize_section_normalizes_invalid_web_search_and_constraints() -> None:
98    """Clamp section fields to the supported ai_text schema."""
99    storage = DummyStorage()
100
101    normalized = storage._normalize_section(
102        {
103            "id": "Global_News",
104            "name": "Global News",
105            "type": "ai_text",
106            "prompt": "News prompt",
107            "web_search": "sometimes",
108            "constraints": {"max_chars": "450"},
109        }
110    )
111
112    assert normalized["web_search"] == "disabled"
113    assert normalized["constraints"] == {"max_chars": 450}
114
115
116def test_normalize_station_rejects_missing_source_playlist_id() -> None:
117    """Reject stations without a source playlist reference."""
118    storage = DummyStorage()
119    storage._sections = {"Song_Transition": _section("Song_Transition")}
120    station = _station(["Song_Transition"])
121    station["source_playlist_id"] = ""
122
123    with pytest.raises(InvalidDataError, match="source_playlist_id is required"):
124        storage._normalize_station(station)
125
126
127def test_normalize_station_rejects_unknown_host_reference() -> None:
128    """Reject stations that reference a host id that does not exist."""
129    storage = DummyStorage()
130
131    with pytest.raises(InvalidDataError, match="unknown host"):
132        storage._normalize_station(
133            {
134                "id": "station_a",
135                "name": "Station A",
136                "source_playlist_id": "playlist-1",
137                "host_id": "missing_host",
138            }
139        )
140
141
142def test_load_sections_persists_defaults_when_file_corrupt(tmp_path: Any) -> None:
143    """Persist default sections to disk when on-disk payload is empty."""
144    sections_file = tmp_path / "sections.json"
145    sections_file.write_text(json.dumps({"version": 1, "sections": []}))
146
147    storage = DummyStorage()
148    storage._sections_file = sections_file
149
150    asyncio.run(storage._load_sections())
151
152    parsed = json.loads(sections_file.read_text())
153    assert len(parsed["sections"]) > 0
154
155
156def test_load_sections_recovers_from_invalid_json(tmp_path: Any) -> None:
157    """Fall back to default sections when the sections file is not valid JSON."""
158    sections_file = tmp_path / "sections.json"
159    sections_file.write_text("{not valid json")
160
161    storage = DummyStorage()
162    storage._sections_file = sections_file
163
164    asyncio.run(storage._load_sections())
165
166    assert len(storage._sections) > 0
167
168
169def test_load_stations_recovers_from_invalid_json(tmp_path: Any) -> None:
170    """Continue with no stations when the stations file is not valid JSON."""
171    stations_file = tmp_path / "stations.json"
172    stations_file.write_text("{not valid json")
173
174    storage = DummyStorage()
175    storage._stations_file = stations_file
176    storage._sections_file = tmp_path / "sections.json"
177
178    asyncio.run(storage._load_stations())
179
180    assert storage._stations == {}
181
182
183def test_write_sections_keeps_existing_file_when_write_fails(
184    tmp_path: Any, monkeypatch: pytest.MonkeyPatch
185) -> None:
186    """Preserve the previous sections file when writing the new content fails."""
187    sections_file = tmp_path / "sections.json"
188    original_content = json.dumps({"version": 1, "sections": []})
189    sections_file.write_text(original_content)
190
191    storage = DummyStorage()
192    storage._sections_file = sections_file
193    storage._sections = {"s1": {"id": "s1", "name": "S1", "type": "ai_text", "prompt": "P"}}
194
195    def failing_fsync(_fd: int) -> None:
196        """Raise to simulate a failed disk write."""
197        raise OSError("disk full")
198
199    monkeypatch.setattr(cast("Any", storage_module).os, "fsync", failing_fsync)
200
201    with pytest.raises(OSError, match="disk full"):
202        asyncio.run(storage._write_sections())
203
204    assert sections_file.read_text() == original_content
205
206
207def test_normalize_section_handles_non_numeric_max_chars_cleanly() -> None:
208    """Handle non-numeric constraint values without leaking a raw ValueError."""
209    storage = DummyStorage()
210
211    with pytest.raises(InvalidDataError):
212        storage._normalize_section(
213            {
214                "id": "s1",
215                "name": "s1",
216                "type": "ai_text",
217                "prompt": "p",
218                "constraints": {"max_chars": "abc"},
219            }
220        )
221
222
223def test_load_stations_does_not_persist_invalid_default_station(tmp_path: Any) -> None:
224    """Do not persist a default station that fails normalization to disk."""
225    storage = DummyStorage()
226    storage._stations_file = tmp_path / "stations.json"
227    storage._sections_file = tmp_path / "sections.json"
228    storage._sections = {item["id"]: item for item in storage._default_sections_template()}
229
230    asyncio.run(storage._load_stations())
231
232    if storage._stations_file.exists():
233        parsed = json.loads(storage._stations_file.read_text())
234        for station in parsed.get("stations", []):
235            assert station.get("source_playlist_id") != ""
236
237
238@pytest.mark.parametrize("field", ["max_duration_minutes"])
239def test_normalize_station_rejects_non_numeric_numeric_field(field: str) -> None:
240    """Reject station numeric fields containing non-numeric values."""
241    storage = DummyStorage()
242    storage._sections = {"Song_Transition": _section("Song_Transition")}
243    storage._hosts = {"host_a": {"id": "host_a", "name": "Host A"}}
244    station = _station(["Song_Transition"])
245    station["host_id"] = "host_a"
246    station[field] = "not-a-number"
247
248    with pytest.raises(InvalidDataError, match="must be numeric"):
249        storage._normalize_station(station)
250
251
252def test_write_json_file_round_trips_non_ascii_content_as_utf8(
253    tmp_path: Any, monkeypatch: pytest.MonkeyPatch
254) -> None:
255    """Write JSON content with an explicit utf-8 encoding and round-trip non-ASCII text."""
256    storage = DummyStorage()
257    target = tmp_path / "sections.json"
258    payload = {"name": "Café Wörld 日本語", "emoji": "🎧"}
259
260    seen_encodings: list[str | None] = []
261    real_open = Path.open
262
263    def spy_open(self: Path, *args: Any, **kwargs: Any) -> Any:
264        seen_encodings.append(kwargs.get("encoding"))
265        return real_open(self, *args, **kwargs)
266
267    monkeypatch.setattr(Path, "open", spy_open)
268
269    asyncio.run(storage._write_json_file(target, payload))
270
271    assert seen_encodings == ["utf-8"]
272    assert json.loads(target.read_bytes().decode("utf-8")) == payload
273
274
275def _record_write(writes: list[str], name: str) -> Any:
276    """Return a persistence stub that records that it was called."""
277
278    async def _write() -> None:
279        writes.append(name)
280
281    return _write
282
283
284def test_load_stations_v3_skips_the_migration_and_rewrites_nothing(tmp_path: Any) -> None:
285    """A file already at v3 loads as-is, inventing no hosts and rewriting no file."""
286    stations_file = tmp_path / "stations.json"
287    payload = {
288        "version": 3,
289        "stations": [
290            {
291                "id": "station_a",
292                "name": "Station A",
293                "source_playlist_id": "playlist-1",
294                "host_id": "rick",
295            }
296        ],
297    }
298    stations_file.write_text(json.dumps(payload))
299
300    storage = DummyStorage()
301    storage._stations_file = stations_file
302    storage._sections_file = tmp_path / "sections.json"
303    storage._hosts = {"rick": {"id": "rick", "name": "Rick"}}
304    writes: list[str] = []
305    for name in ("_write_stations", "_write_sections", "_write_hosts"):
306        setattr(storage, name, _record_write(writes, name))
307
308    asyncio.run(storage._load_stations())
309
310    assert list(storage._stations) == ["station_a"]
311    assert storage._stations["station_a"]["host_id"] == "rick"
312    assert list(storage._hosts) == ["rick"]
313    assert writes == []
314    assert json.loads(stations_file.read_text()) == payload
315
316
317def test_normalize_station_v3_requires_known_host() -> None:
318    """Reject stations that reference a host that does not exist."""
319    dummy = DummyStorage()
320    dummy._hosts = {}
321    station = {
322        "id": "station_a",
323        "name": "Station A",
324        "source_playlist_id": "playlist-1",
325        "host_id": "rick",
326    }
327    with pytest.raises(InvalidDataError):
328        dummy._normalize_station(station)
329
330
331def test_normalize_station_v3_returns_slim_schema() -> None:
332    """Normalize a v3 station payload without any legacy embedded fields."""
333    dummy = DummyStorage()
334    dummy._hosts = {"rick": {"id": "rick", "name": "Rick"}}
335    station = {
336        "id": "station_a",
337        "name": "Station A",
338        "source_playlist_id": "playlist-1",
339        "host_id": "rick",
340    }
341    normalized = dummy._normalize_station(station)
342    assert normalized["host_id"] == "rick"
343    for legacy_key in ("general", "sections", "section_ids", "section_order", "merge_section_id"):
344        assert legacy_key not in normalized
345