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.constants import DEFAULT_LLM_INSTRUCTIONS
16from music_assistant.providers.ai_radio.storage import AIRadioStorageMixin
17
18
19class DummyStorage(AIRadioStorageMixin):
20    """Minimal storage harness for unit testing helper methods."""
21
22    def __init__(self) -> None:
23        """Initialize dummy mixin state."""
24        self.logger = logging.getLogger(__name__)
25        self._sections: dict[str, dict[str, Any]] = {}
26        self._stations: 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_normalize_general_uses_known_schema_only() -> None:
63    """Normalize general config to known keys and converted value types."""
64    storage = DummyStorage()
65    raw_general = {
66        "instructions": "Keep it concise and conversational.",
67        "weather_provider": "open_meteo",
68        "weather_timeout_seconds": "30",
69        # timezone/location moved to provider config; a legacy station carrying them
70        # must not resurrect them in the normalized output
71        "timezone": "Europe/Berlin",
72        "location": {"city": "Berlin", "country": "DE"},
73        "foo": "bar",
74    }
75
76    normalized = storage._normalize_general(raw_general)
77
78    assert normalized["instructions"] == "Keep it concise and conversational."
79    assert normalized["weather_timeout_seconds"] == 30
80    assert "foo" not in normalized
81    assert "timezone" not in normalized
82    assert "location" not in normalized
83
84    default_keys = set(storage._default_station_template()["general"].keys())
85    assert set(normalized.keys()) == default_keys
86
87
88def test_materialize_sections_reports_missing_and_returns_copies() -> None:
89    """Resolve sections and report unknown ids without mutating source map."""
90    storage = DummyStorage()
91    storage._sections = {
92        "Song_Transition": {
93            "id": "Song_Transition",
94            "name": "Song Transition",
95            "type": "ai_text",
96            "prompt": "Transition",
97        }
98    }
99
100    sections, missing = storage._materialize_sections(["Song_Transition", "Unknown_Section"])
101
102    assert missing == ["Unknown_Section"]
103    assert sections[0]["id"] == "Song_Transition"
104    sections[0]["name"] = "Changed"
105    assert storage._sections["Song_Transition"]["name"] == "Song Transition"
106
107
108def test_refresh_station_sections_rebuilds_embedded_sections() -> None:
109    """Rebuild embedded station sections from selected shared ids."""
110    storage = DummyStorage()
111    storage._sections = {
112        "Song_Transition": {
113            "id": "Song_Transition",
114            "name": "Song Transition",
115            "type": "ai_text",
116            "prompt": "Transition",
117        }
118    }
119    storage._stations = {
120        "station_a": {
121            "id": "station_a",
122            "section_ids": ["Song_Transition", "Missing_Section"],
123            "sections": [],
124        }
125    }
126
127    storage._refresh_station_sections()
128
129    sections = storage._stations["station_a"]["sections"]
130    assert len(sections) == 1
131    assert sections[0]["id"] == "Song_Transition"
132
133
134def test_normalize_section_rejects_invalid_type() -> None:
135    """Reject shared sections with unsupported types."""
136    storage = DummyStorage()
137
138    with pytest.raises(InvalidDataError, match="invalid type"):
139        storage._normalize_section(
140            {
141                "id": "Bad_Section",
142                "name": "Bad Section",
143                "type": "unsupported",
144                "prompt": "Prompt",
145            }
146        )
147
148
149def test_normalize_section_normalizes_invalid_web_search_and_constraints() -> None:
150    """Clamp section fields to the supported ai_text schema."""
151    storage = DummyStorage()
152
153    normalized = storage._normalize_section(
154        {
155            "id": "Global_News",
156            "name": "Global News",
157            "type": "ai_text",
158            "prompt": "News prompt",
159            "web_search": "sometimes",
160            "constraints": {"max_chars": "450"},
161        }
162    )
163
164    assert normalized["web_search"] == "disabled"
165    assert normalized["constraints"] == {"max_chars": 450}
166
167
168def test_normalize_station_rejects_missing_source_playlist_id() -> None:
169    """Reject stations without a source playlist reference."""
170    storage = DummyStorage()
171    storage._sections = {"Song_Transition": _section("Song_Transition")}
172    station = _station(["Song_Transition"])
173    station["source_playlist_id"] = ""
174
175    with pytest.raises(InvalidDataError, match="source_playlist_id is required"):
176        storage._normalize_station(station)
177
178
179def test_normalize_station_rejects_unknown_section_reference() -> None:
180    """Reject stations that reference shared sections that do not exist."""
181    storage = DummyStorage()
182
183    with pytest.raises(InvalidDataError, match="unknown sections: Song_Transition"):
184        storage._normalize_station(_station(["Song_Transition"]))
185
186
187def test_normalize_station_rejects_merge_section_outside_selected_ids() -> None:
188    """Reject merge sections that are not part of the station section list."""
189    storage = DummyStorage()
190    storage._sections = {
191        "Song_Transition": _section("Song_Transition"),
192        "Between_Songs_Mix": _section(
193            "Between_Songs_Mix",
194            name="Between Songs Mix",
195            section_type="ai_meta",
196        ),
197    }
198    station = _station(["Song_Transition"])
199    station["merge_section_id"] = "Between_Songs_Mix"
200
201    with pytest.raises(InvalidDataError, match="must be selected in station section_ids"):
202        storage._normalize_station(station)
203
204
205def test_normalize_station_rejects_non_meta_merge_section() -> None:
206    """Reject merge sections that do not point to an ai_meta section."""
207    storage = DummyStorage()
208    storage._sections = {
209        "Song_Transition": _section("Song_Transition"),
210        "Weather_Short": _section("Weather_Short", name="Weather Short"),
211    }
212    station = _station(["Song_Transition", "Weather_Short"])
213    station["merge_section_id"] = "Weather_Short"
214
215    with pytest.raises(InvalidDataError, match="must reference an ai_meta section"):
216        storage._normalize_station(station)
217
218
219def test_normalize_station_rejects_non_numeric_optional_chance() -> None:
220    """Reject OPTIONAL flow entries with non-numeric chance values."""
221    storage = DummyStorage()
222    storage._sections = {"Song_Transition": _section("Song_Transition")}
223    station = _station(["Song_Transition"])
224    station["section_order"] = [
225        {
226            "when": "between_songs",
227            "flow": [
228                {
229                    "OPTIONAL": {
230                        "section": "Song_Transition",
231                        "chance": "invalid",
232                    }
233                }
234            ],
235        }
236    ]
237
238    with pytest.raises(InvalidDataError, match="OPTIONAL chance must be numeric"):
239        storage._normalize_station(station)
240
241
242def test_load_sections_persists_defaults_when_file_corrupt(tmp_path: Any) -> None:
243    """Persist default sections to disk when on-disk payload is empty."""
244    sections_file = tmp_path / "sections.json"
245    sections_file.write_text(json.dumps({"version": 1, "sections": []}))
246
247    storage = DummyStorage()
248    storage._sections_file = sections_file
249
250    asyncio.run(storage._load_sections())
251
252    parsed = json.loads(sections_file.read_text())
253    assert len(parsed["sections"]) > 0
254
255
256def test_load_sections_recovers_from_invalid_json(tmp_path: Any) -> None:
257    """Fall back to default sections when the sections file is not valid JSON."""
258    sections_file = tmp_path / "sections.json"
259    sections_file.write_text("{not valid json")
260
261    storage = DummyStorage()
262    storage._sections_file = sections_file
263
264    asyncio.run(storage._load_sections())
265
266    assert len(storage._sections) > 0
267
268
269def test_load_stations_recovers_from_invalid_json(tmp_path: Any) -> None:
270    """Continue with no stations when the stations file is not valid JSON."""
271    stations_file = tmp_path / "stations.json"
272    stations_file.write_text("{not valid json")
273
274    storage = DummyStorage()
275    storage._stations_file = stations_file
276    storage._sections_file = tmp_path / "sections.json"
277
278    asyncio.run(storage._load_stations())
279
280    assert storage._stations == {}
281
282
283def test_write_sections_keeps_existing_file_when_write_fails(
284    tmp_path: Any, monkeypatch: pytest.MonkeyPatch
285) -> None:
286    """Preserve the previous sections file when writing the new content fails."""
287    sections_file = tmp_path / "sections.json"
288    original_content = json.dumps({"version": 1, "sections": []})
289    sections_file.write_text(original_content)
290
291    storage = DummyStorage()
292    storage._sections_file = sections_file
293    storage._sections = {"s1": {"id": "s1", "name": "S1", "type": "ai_text", "prompt": "P"}}
294
295    def failing_fsync(_fd: int) -> None:
296        """Raise to simulate a failed disk write."""
297        raise OSError("disk full")
298
299    monkeypatch.setattr(cast("Any", storage_module).os, "fsync", failing_fsync)
300
301    with pytest.raises(OSError, match="disk full"):
302        asyncio.run(storage._write_sections())
303
304    assert sections_file.read_text() == original_content
305
306
307def test_normalize_section_handles_non_numeric_max_chars_cleanly() -> None:
308    """Handle non-numeric constraint values without leaking a raw ValueError."""
309    storage = DummyStorage()
310
311    with pytest.raises(InvalidDataError):
312        storage._normalize_section(
313            {
314                "id": "s1",
315                "name": "s1",
316                "type": "ai_text",
317                "prompt": "p",
318                "constraints": {"max_chars": "abc"},
319            }
320        )
321
322
323def test_normalize_station_rejects_non_numeric_alternative_weight() -> None:
324    """Reject ALTERNATIVE choices with non-numeric weight values."""
325    storage = DummyStorage()
326    storage._sections = {"Song_Transition": _section("Song_Transition")}
327    station = _station(["Song_Transition"])
328    station["section_order"] = [
329        {
330            "when": "between_songs",
331            "flow": [
332                {
333                    "ALTERNATIVE": {
334                        "choices": [{"section": "Song_Transition", "weight": "not-a-number"}]
335                    }
336                }
337            ],
338        }
339    ]
340
341    with pytest.raises(InvalidDataError, match="weight"):
342        storage._normalize_station(station)
343
344
345def test_load_stations_does_not_persist_invalid_default_station(tmp_path: Any) -> None:
346    """Do not persist a default station that fails normalization to disk."""
347    storage = DummyStorage()
348    storage._stations_file = tmp_path / "stations.json"
349    storage._sections_file = tmp_path / "sections.json"
350    storage._sections = {item["id"]: item for item in storage._default_sections_template()}
351
352    asyncio.run(storage._load_stations())
353
354    if storage._stations_file.exists():
355        parsed = json.loads(storage._stations_file.read_text())
356        for station in parsed.get("stations", []):
357            assert station.get("source_playlist_id") != ""
358
359
360def test_normalize_general_replaces_null_instructions_with_default() -> None:
361    """Replace null instructions with the default LLM instructions string."""
362    storage = DummyStorage()
363
364    result = storage._normalize_general(
365        {
366            "instructions": None,
367            "weather_provider": "open_meteo",
368            "weather_timeout_seconds": 20,
369        }
370    )
371
372    assert result["instructions"] == DEFAULT_LLM_INSTRUCTIONS
373
374
375@pytest.mark.parametrize("field", ["max_duration_minutes"])
376def test_normalize_station_rejects_non_numeric_numeric_field(field: str) -> None:
377    """Reject station numeric fields containing non-numeric values."""
378    storage = DummyStorage()
379    storage._sections = {"Song_Transition": _section("Song_Transition")}
380    station = _station(["Song_Transition"])
381    station[field] = "not-a-number"
382
383    with pytest.raises(InvalidDataError):
384        storage._normalize_station(station)
385
386
387def test_write_json_file_round_trips_non_ascii_content_as_utf8(
388    tmp_path: Any, monkeypatch: pytest.MonkeyPatch
389) -> None:
390    """Write JSON content with an explicit utf-8 encoding and round-trip non-ASCII text."""
391    storage = DummyStorage()
392    target = tmp_path / "sections.json"
393    payload = {"name": "Café Wörld 日本語", "emoji": "🎧"}
394
395    seen_encodings: list[str | None] = []
396    real_open = Path.open
397
398    def spy_open(self: Path, *args: Any, **kwargs: Any) -> Any:
399        seen_encodings.append(kwargs.get("encoding"))
400        return real_open(self, *args, **kwargs)
401
402    monkeypatch.setattr(Path, "open", spy_open)
403
404    asyncio.run(storage._write_json_file(target, payload))
405
406    assert seen_encodings == ["utf-8"]
407    assert json.loads(target.read_bytes().decode("utf-8")) == payload
408