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