music-assistant-server

1"""Unit tests for AI Radio host normalization and persistence."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7import logging
8from typing import Any
9
10import pytest
11from music_assistant_models.errors import InvalidDataError
12
13from music_assistant.providers.ai_radio.constants import DEFAULT_LLM_INSTRUCTIONS
14from music_assistant.providers.ai_radio.hosts import AIRadioHostsMixin
15from music_assistant.providers.ai_radio.storage import AIRadioStorageMixin
16
17
18class DummyHosts(AIRadioHostsMixin, AIRadioStorageMixin):
19    """Minimal harness combining host and storage helpers."""
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(section_id: str, section_type: str = "ai_text") -> dict[str, Any]:
30    return {"id": section_id, "name": section_id, "type": section_type, "prompt": "Prompt"}
31
32
33def _host(section_ids: list[str]) -> dict[str, Any]:
34    return {
35        "id": "rick",
36        "name": "Rick",
37        "instructions": "Laid-back evening host.",
38        "tts_engine": "",
39        "section_ids": section_ids,
40        "section_order": [
41            {"when": "between_songs", "flow": [{"MUST": section_ids[0]}]},
42        ],
43    }
44
45
46def test_normalize_host_returns_known_schema() -> None:
47    """Normalize a valid host payload to the known schema."""
48    dummy = DummyHosts()
49    dummy._sections = {"Song_Transition": _section("Song_Transition")}
50    normalized = dummy._normalize_host(_host(["Song_Transition"]))
51    assert normalized["id"] == "rick"
52    assert normalized["name"] == "Rick"
53    assert normalized["instructions"] == "Laid-back evening host."
54    assert normalized["tts_engine"] == ""
55    assert normalized["section_ids"] == ["Song_Transition"]
56    assert normalized["merge_section_id"] == ""
57
58
59def test_normalize_host_defaults_empty_instructions() -> None:
60    """Fall back to the default instructions when the payload's are blank."""
61    dummy = DummyHosts()
62    dummy._sections = {"Song_Transition": _section("Song_Transition")}
63    payload = _host(["Song_Transition"])
64    payload["instructions"] = "   "
65    normalized = dummy._normalize_host(payload)
66    assert normalized["instructions"] == DEFAULT_LLM_INSTRUCTIONS
67
68
69def test_normalize_host_requires_name() -> None:
70    """Reject hosts with a blank name."""
71    dummy = DummyHosts()
72    dummy._sections = {"Song_Transition": _section("Song_Transition")}
73    payload = _host(["Song_Transition"])
74    payload["name"] = " "
75    with pytest.raises(InvalidDataError):
76        dummy._normalize_host(payload)
77
78
79def test_normalize_host_persists_language() -> None:
80    """Persist a host's language override through normalization."""
81    dummy = DummyHosts()
82    dummy._sections = {"Song_Transition": _section("Song_Transition")}
83    payload = _host(["Song_Transition"])
84    payload["language"] = "nl_NL"
85    normalized = dummy._normalize_host(payload)
86    assert normalized["language"] == "nl_NL"
87
88
89def test_normalize_host_defaults_missing_language_to_empty() -> None:
90    """Normalize a host with no 'language' key at all, for backwards compatibility."""
91    dummy = DummyHosts()
92    dummy._sections = {"Song_Transition": _section("Song_Transition")}
93    payload = _host(["Song_Transition"])
94    assert "language" not in payload
95    normalized = dummy._normalize_host(payload)
96    assert normalized["language"] == ""
97
98
99def test_normalize_host_strips_whitespace_only_language_to_empty() -> None:
100    """Treat a whitespace-only language as 'follow the server locale'."""
101    dummy = DummyHosts()
102    dummy._sections = {"Song_Transition": _section("Song_Transition")}
103    payload = _host(["Song_Transition"])
104    payload["language"] = "   "
105    normalized = dummy._normalize_host(payload)
106    assert normalized["language"] == ""
107
108
109def test_normalize_host_rejects_unknown_section_reference() -> None:
110    """Reject hosts that reference shared sections that do not exist."""
111    dummy = DummyHosts()
112    dummy._sections = {}
113    with pytest.raises(InvalidDataError):
114        dummy._normalize_host(_host(["Missing_Section"]))
115
116
117def test_normalize_host_rejects_non_meta_merge_section() -> None:
118    """Reject merge sections that do not point to an ai_meta section."""
119    dummy = DummyHosts()
120    dummy._sections = {"Song_Transition": _section("Song_Transition")}
121    payload = _host(["Song_Transition"])
122    payload["merge_section_id"] = "Song_Transition"
123    with pytest.raises(InvalidDataError):
124        dummy._normalize_host(payload)
125
126
127def test_normalize_host_rejects_non_numeric_optional_chance() -> None:
128    """Reject OPTIONAL flow entries with non-numeric chance values."""
129    dummy = DummyHosts()
130    dummy._sections = {"Song_Transition": _section("Song_Transition")}
131    payload = _host(["Song_Transition"])
132    payload["section_order"] = [
133        {
134            "when": "between_songs",
135            "flow": [
136                {
137                    "OPTIONAL": {
138                        "section": "Song_Transition",
139                        "chance": "invalid",
140                    }
141                }
142            ],
143        }
144    ]
145    with pytest.raises(InvalidDataError, match="OPTIONAL chance must be numeric"):
146        dummy._normalize_host(payload)
147
148
149def test_normalize_host_rejects_non_numeric_alternative_weight() -> None:
150    """Reject ALTERNATIVE choices with non-numeric weight values."""
151    dummy = DummyHosts()
152    dummy._sections = {"Song_Transition": _section("Song_Transition")}
153    payload = _host(["Song_Transition"])
154    payload["section_order"] = [
155        {
156            "when": "between_songs",
157            "flow": [
158                {
159                    "ALTERNATIVE": {
160                        "choices": [{"section": "Song_Transition", "weight": "not-a-number"}]
161                    }
162                }
163            ],
164        }
165    ]
166    with pytest.raises(InvalidDataError, match="weight"):
167        dummy._normalize_host(payload)
168
169
170def test_default_host_template_is_normalizable() -> None:
171    """Ensure the built-in host template passes normalization."""
172    dummy = DummyHosts()
173    defaults = dummy._default_sections_template()
174    dummy._sections = {item["id"]: item for item in defaults}
175    template = dummy._default_host_template()
176    normalized = dummy._normalize_host(template)
177    assert normalized["merge_section_id"] == "Between_Songs_Smoother"
178
179
180def test_write_and_load_hosts_round_trips_normalized_host(tmp_path: Any) -> None:
181    """Round trip a normalized host through write and load."""
182    sections = {"Song_Transition": _section("Song_Transition")}
183    hosts_file = tmp_path / "hosts.json"
184
185    writer = DummyHosts()
186    writer._sections = sections
187    writer._hosts_file = hosts_file
188    normalized = writer._normalize_host(_host(["Song_Transition"]))
189    writer._hosts = {normalized["id"]: normalized}
190
191    asyncio.run(writer._write_hosts())
192
193    reader = DummyHosts()
194    reader._sections = sections
195    reader._hosts_file = hosts_file
196
197    asyncio.run(reader._load_hosts())
198
199    assert reader._hosts == {normalized["id"]: normalized}
200
201
202def test_load_hosts_recovers_from_invalid_json(tmp_path: Any) -> None:
203    """Continue with no hosts when the hosts file is not valid JSON."""
204    hosts_file = tmp_path / "hosts.json"
205    hosts_file.write_text("{not valid json")
206
207    dummy = DummyHosts()
208    dummy._hosts_file = hosts_file
209
210    asyncio.run(dummy._load_hosts())
211
212    assert dummy._hosts == {}
213    assert hosts_file.read_text() == "{not valid json"
214
215
216def _v2_station(station_id: str, name: str, instructions: str) -> dict[str, Any]:
217    return {
218        "id": station_id,
219        "name": name,
220        "source_playlist_id": "playlist-1",
221        "source_playlist_provider": "library",
222        "general": {"instructions": instructions},
223        "section_ids": ["Song_Transition"],
224        "sections": [_section("Song_Transition")],
225        "section_order": [
226            {"when": "between_songs", "flow": [{"MUST": "Song_Transition"}]},
227        ],
228        "merge_section_id": "",
229    }
230
231
232def test_migrate_v2_extracts_host_and_slims_station() -> None:
233    """Extract a host profile out of a v2 station and slim the station in place."""
234    dummy = DummyHosts()
235    dummy._sections = {"Song_Transition": _section("Song_Transition")}
236    stations = [_v2_station("station_a", "Evening Chill", "Laid-back host.")]
237    dummy._migrate_stations_v2_to_v3(stations)
238    assert len(dummy._hosts) == 1
239    host = next(iter(dummy._hosts.values()))
240    assert host["instructions"] == "Laid-back host."
241    assert host["section_ids"] == ["Song_Transition"]
242    assert stations[0]["host_id"] == host["id"]
243    assert "section_order" not in stations[0]
244    assert "general" not in stations[0]
245
246
247def test_migrate_v2_dedupes_identical_hosts() -> None:
248    """Reuse the same host for stations that share the same persona fingerprint."""
249    dummy = DummyHosts()
250    dummy._sections = {"Song_Transition": _section("Song_Transition")}
251    stations = [
252        _v2_station("station_a", "Morning", "Same persona."),
253        _v2_station("station_b", "Evening", "Same persona."),
254    ]
255    dummy._migrate_stations_v2_to_v3(stations)
256    assert len(dummy._hosts) == 1
257    assert stations[0]["host_id"] == stations[1]["host_id"]
258
259
260def test_migrate_v2_keeps_distinct_hosts_apart() -> None:
261    """Create separate hosts for stations with distinct personas."""
262    dummy = DummyHosts()
263    dummy._sections = {"Song_Transition": _section("Song_Transition")}
264    stations = [
265        _v2_station("station_a", "Morning", "Persona A."),
266        _v2_station("station_b", "Evening", "Persona B."),
267    ]
268    dummy._migrate_stations_v2_to_v3(stations)
269    assert len(dummy._hosts) == 2
270    assert stations[0]["host_id"] != stations[1]["host_id"]
271
272
273def test_migrate_v2_renames_host_on_slug_collision() -> None:
274    """Rename the second host's slug when two distinct personas share the same name."""
275    dummy = DummyHosts()
276    dummy._sections = {"Song_Transition": _section("Song_Transition")}
277    stations = [
278        _v2_station("station_a", "Same Name", "Persona A."),
279        _v2_station("station_b", "Same Name", "Persona B."),
280    ]
281    dummy._migrate_stations_v2_to_v3(stations)
282    assert len(dummy._hosts) == 2
283    assert stations[0]["host_id"] != stations[1]["host_id"]
284    assert "same_name_host" in dummy._hosts
285    renamed_ids = set(dummy._hosts) - {"same_name_host"}
286    assert len(renamed_ids) == 1
287    assert next(iter(renamed_ids)).startswith("same_name_host_")
288
289
290def test_load_stations_skips_station_that_fails_host_migration(tmp_path: Any) -> None:
291    """Isolate one station's migration failure so the rest of the file still loads."""
292    good_station = _v2_station("station_a", "Morning", "Persona A.")
293    bad_station = _v2_station("station_b", "Evening", "Persona B.")
294    # non-numeric OPTIONAL.chance makes _normalize_host raise during migration
295    bad_station["section_order"] = [
296        {
297            "when": "between_songs",
298            "flow": [{"OPTIONAL": {"section": "Song_Transition", "chance": "invalid"}}],
299        }
300    ]
301    stations_file = tmp_path / "stations.json"
302    stations_file.write_text(json.dumps({"version": 2, "stations": [good_station, bad_station]}))
303
304    dummy = DummyHosts()
305    dummy._stations_file = stations_file
306    dummy._sections_file = tmp_path / "sections.json"
307    dummy._hosts_file = tmp_path / "hosts.json"
308    dummy._sections = {"Song_Transition": _section("Song_Transition")}
309
310    asyncio.run(dummy._load_stations())
311
312    assert len(dummy._hosts) == 1
313    assert len(dummy._stations) == 1
314    assert next(iter(dummy._stations.values()))["name"] == "Morning"
315
316
317def test_load_stations_migrates_v2_file_on_disk(tmp_path: Any) -> None:
318    """
319    Round-trip a v2 stations file through _load_stations.
320
321    Covers: hosts extracted from embedded personas, stations normalized to v3
322    in memory, hosts.json and stations.json written, embedded sections
323    without a matching section_ids list (fallback derivation) upserted into
324    _sections, and stations.json rewritten exactly once.
325    """
326    v2_station = {
327        "id": "station_a",
328        "name": "Evening Chill",
329        "source_playlist_id": "playlist-1",
330        "source_playlist_provider": "library",
331        "general": {"instructions": "Laid-back host."},
332        # no section_ids: forces the embedded-sections-only fallback
333        "sections": [_section("Song_Transition")],
334        "section_order": [
335            {"when": "between_songs", "flow": [{"MUST": "Song_Transition"}]},
336        ],
337        "merge_section_id": "",
338    }
339    stations_file = tmp_path / "stations.json"
340    stations_file.write_text(json.dumps({"version": 2, "stations": [v2_station]}))
341
342    dummy = DummyHosts()
343    dummy._stations_file = stations_file
344    dummy._sections_file = tmp_path / "sections.json"
345    dummy._hosts_file = tmp_path / "hosts.json"
346
347    asyncio.run(dummy._load_hosts())
348
349    write_calls = 0
350    original_write_stations = dummy._write_stations
351
352    async def _counting_write_stations() -> None:
353        nonlocal write_calls
354        write_calls += 1
355        await original_write_stations()
356
357    dummy._write_stations = _counting_write_stations  # type: ignore[method-assign]
358
359    asyncio.run(dummy._load_stations())
360
361    assert write_calls == 1
362    assert len(dummy._hosts) == 1
363    assert "Song_Transition" in dummy._sections
364
365    host = next(iter(dummy._hosts.values()))
366    assert host["instructions"] == "Laid-back host."
367    assert host["section_ids"] == ["Song_Transition"]
368
369    station = next(iter(dummy._stations.values()))
370    assert station["host_id"] == host["id"]
371    for legacy_key in ("general", "sections", "section_ids", "section_order", "merge_section_id"):
372        assert legacy_key not in station
373
374    assert dummy._hosts_file.exists()
375    hosts_payload = json.loads(dummy._hosts_file.read_text())
376    assert hosts_payload["hosts"][0]["instructions"] == "Laid-back host."
377
378    stations_payload = json.loads(stations_file.read_text())
379    assert stations_payload["version"] == 3
380
381
382def _preset_dummy(tmp_path: Any) -> DummyHosts:
383    """Build a host harness pointed at an empty storage dir."""
384    dummy = DummyHosts()
385    dummy._hosts_file = tmp_path / "hosts.json"
386    dummy._stations_file = tmp_path / "stations.json"
387    dummy._sections_file = tmp_path / "sections.json"
388    return dummy
389
390
391def test_seed_preset_hosts_seeds_four_hosts_on_fresh_install(tmp_path: Any) -> None:
392    """Seed the four preset hosts with host-scoped sections and persist both files."""
393    dummy = _preset_dummy(tmp_path)
394
395    asyncio.run(dummy._seed_preset_hosts())
396
397    assert list(dummy._hosts) == ["morning_show", "minimal_dj", "music_nerd", "party_host"]
398    host = dummy._hosts["morning_show"]
399    assert host["name"] == "Morning show"
400    assert host["section_ids"] == [
401        "morning_show_intro",
402        "morning_show_transition",
403        "morning_show_weather",
404        "morning_show_news",
405        "morning_show_sign_off",
406        "morning_show_smoother",
407    ]
408    assert host["merge_section_id"] == "morning_show_smoother"
409    assert all(section_id in dummy._sections for section_id in host["section_ids"])
410    # the same segment name on two hosts must not share a section
411    assert dummy._sections["minimal_dj_transition"]["constraints"] == {"max_chars": 300}
412    assert dummy._sections["morning_show_transition"]["constraints"] == {"max_chars": 650}
413
414    hosts_payload = json.loads(dummy._hosts_file.read_text())
415    assert [item["id"] for item in hosts_payload["hosts"]] == [
416        "minimal_dj",
417        "morning_show",
418        "music_nerd",
419        "party_host",
420    ]
421    section_ids = {item["id"] for item in json.loads(dummy._sections_file.read_text())["sections"]}
422    assert section_ids == set(dummy._sections)
423    assert not dummy._stations_file.exists()
424
425
426def test_seed_preset_hosts_skipped_for_migrated_install(tmp_path: Any) -> None:
427    """Leave a pre-v3 install alone: its hosts come from the station migration."""
428    dummy = _preset_dummy(tmp_path)
429    dummy._stations_file.write_text(json.dumps({"version": 2, "stations": []}))
430
431    asyncio.run(dummy._seed_preset_hosts())
432
433    assert dummy._hosts == {}
434    assert dummy._sections == {}
435    assert not dummy._hosts_file.exists()
436
437
438def test_seed_preset_hosts_skipped_when_hosts_file_exists(tmp_path: Any) -> None:
439    """Never re-seed once a hosts file exists, even when the user emptied it."""
440    dummy = _preset_dummy(tmp_path)
441    dummy._hosts_file.write_text(json.dumps({"version": 1, "hosts": []}))
442
443    asyncio.run(dummy._seed_preset_hosts())
444
445    assert dummy._hosts == {}
446    assert dummy._sections == {}
447
448
449def test_preset_hosts_and_sections_are_already_normalized() -> None:
450    """Ensure every seeded preset passes host and section validation untouched."""
451    dummy = DummyHosts()
452    for host, sections in dummy._default_preset_hosts():
453        for section in sections:
454            assert dummy._normalize_section(section) == section
455            dummy._sections[section["id"]] = section
456        assert dummy._normalize_host(host) == host
457
458
459def test_morning_show_preset_section_order_matches_frontend_compiler() -> None:
460    """Pin the compiled cadence of the morning show preset to the frontend's compiler output."""
461    dummy = DummyHosts()
462    hosts = {host["id"]: host for host, _ in dummy._default_preset_hosts()}
463
464    assert hosts["morning_show"]["section_order"] == [
465        {"when": "start_of_playlist", "flow": [{"MUST": "morning_show_intro"}]},
466        {
467            "when": "between_songs",
468            "flow": [
469                {
470                    "OPTIONAL": {
471                        "section": "morning_show_transition",
472                        "chance": 2 / 3,
473                        "guards": {
474                            "min_gap_songs": 3,
475                            "max_per_60min": 0,
476                            "require_placeholders_present": [],
477                        },
478                    }
479                },
480                {
481                    "OPTIONAL": {
482                        "section": "morning_show_weather",
483                        "chance": 1.0,
484                        "guards": {
485                            "min_gap_songs": 0,
486                            "max_per_60min": 1,
487                            "require_placeholders_present": ["<weather_hourly>", "<timestamp>"],
488                        },
489                    }
490                },
491                {
492                    "OPTIONAL": {
493                        "section": "morning_show_news",
494                        "chance": 1.0,
495                        "guards": {
496                            "min_gap_songs": 0,
497                            "max_per_60min": 1,
498                            "require_placeholders_present": ["<timestamp>"],
499                        },
500                    }
501                },
502            ],
503        },
504        {"when": "end_of_playlist", "flow": [{"MUST": "morning_show_sign_off"}]},
505    ]
506
507
508def test_music_nerd_preset_compiles_its_recurring_segments_as_optional() -> None:
509    """Compile an every-2-songs segment into a certain OPTIONAL and an occasional one by chance."""
510    dummy = DummyHosts()
511    hosts = {host["id"]: host for host, _ in dummy._default_preset_hosts()}
512
513    assert hosts["music_nerd"]["section_order"] == [
514        {"when": "start_of_playlist", "flow": [{"MUST": "music_nerd_intro"}]},
515        {
516            "when": "between_songs",
517            "flow": [
518                {
519                    "OPTIONAL": {
520                        "section": "music_nerd_artist_fact",
521                        "chance": 1.0,
522                        "guards": {
523                            "min_gap_songs": 2,
524                            "max_per_60min": 0,
525                            "require_placeholders_present": [],
526                        },
527                    }
528                },
529                {
530                    "OPTIONAL": {
531                        "section": "music_nerd_transition",
532                        "chance": 0.2,
533                        "guards": {
534                            "min_gap_songs": 1,
535                            "max_per_60min": 0,
536                            "require_placeholders_present": [],
537                        },
538                    }
539                },
540            ],
541        },
542    ]
543