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