music-assistant-server

1"""Tests for response Brief dataclasses + ``_common`` adapters."""
2
3from __future__ import annotations
4
5import dataclasses
6from types import SimpleNamespace
7from typing import Any, cast
8
9import pytest
10
11from music_assistant.providers.fastmcp_server import models
12from music_assistant.providers.fastmcp_server.models import (
13    AlbumBrief,
14    ArtistBrief,
15    PlayerBrief,
16    PlaylistBrief,
17    QueueBrief,
18    RadioBrief,
19    TrackBrief,
20)
21from music_assistant.providers.fastmcp_server.tools._common import (
22    _external_now_playing,
23    page_args,
24    to_brief_album,
25    to_brief_artist,
26    to_brief_player,
27    to_brief_playlist,
28    to_brief_queue,
29    to_brief_radio,
30    to_brief_track,
31)
32
33
34def test_track_brief_defaults() -> None:
35    """TrackBrief fills sensible defaults."""
36    t = TrackBrief(uri="library://track/1", name="X")
37    assert t.artists == []
38    assert t.album is None
39    assert t.duration is None
40
41
42def test_to_brief_track_extracts_artists_and_album() -> None:
43    """``to_brief_track`` reads names from artists/album attributes."""
44    track = SimpleNamespace(
45        uri="library://track/42",
46        name="Sample",
47        artists=[SimpleNamespace(name="A1"), SimpleNamespace(name="A2")],
48        album=SimpleNamespace(name="Album"),
49        duration=180,
50    )
51    brief = to_brief_track(track)
52    assert brief == TrackBrief(
53        uri="library://track/42",
54        name="Sample",
55        artists=["A1", "A2"],
56        album="Album",
57        duration=180,
58    )
59
60
61def test_to_brief_album_falls_back_to_artists_list() -> None:
62    """``to_brief_album`` uses ``artists[0]`` when there's no scalar artist."""
63    album = SimpleNamespace(
64        uri="library://album/1",
65        name="Album",
66        artist=None,
67        artists=[SimpleNamespace(name="A1")],
68        year=2020,
69    )
70    assert to_brief_album(album) == AlbumBrief(
71        uri="library://album/1", name="Album", artist="A1", year=2020
72    )
73
74
75def test_to_brief_artist() -> None:
76    """``to_brief_artist`` extracts uri and name."""
77    artist = SimpleNamespace(uri="library://artist/x", name="X")
78    assert to_brief_artist(artist) == ArtistBrief(uri="library://artist/x", name="X")
79
80
81def test_to_brief_playlist() -> None:
82    """``to_brief_playlist`` includes track_count and owner when available."""
83    playlist = SimpleNamespace(
84        uri="library://playlist/1",
85        name="Mix",
86        track_count=12,
87        owner=SimpleNamespace(name="me"),
88    )
89    assert to_brief_playlist(playlist) == PlaylistBrief(
90        uri="library://playlist/1", name="Mix", track_count=12, owner="me"
91    )
92
93
94def test_to_brief_radio() -> None:
95    """``to_brief_radio`` maps name + description."""
96    radio = SimpleNamespace(uri="library://radio/1", name="R", description="d")
97    assert to_brief_radio(radio) == RadioBrief(uri="library://radio/1", name="R", description="d")
98
99
100def test_to_brief_player_reads_playback_state() -> None:
101    """
102    ``to_brief_player`` reads the canonical ``Player.playback_state`` enum.
103
104    The expected ``PlayerBrief`` pins every defaulted field explicitly —
105    leaving them implicit means the test passes only as long as the
106    dataclass defaults match what ``to_brief_player`` falls back to for
107    legacy stubs. Pinning them here keeps a future default flip from
108    silently breaking the playback-state-read contract this test is
109    actually about.
110    """
111    player = SimpleNamespace(
112        player_id="kitchen",
113        name="Kitchen",
114        playback_state=SimpleNamespace(value="playing"),
115        volume_level=42,
116        powered=True,
117        current_media=None,
118    )
119    brief = to_brief_player(player)
120    assert brief == PlayerBrief(
121        player_id="kitchen",
122        name="Kitchen",
123        state="playing",
124        volume_level=42,
125        powered=True,
126        current_item=None,
127        available=True,
128        enabled=True,
129        needs_setup=False,
130        active_group=None,
131        synced_to=None,
132    )
133
134
135def test_to_brief_player_falls_back_to_legacy_state_attr() -> None:
136    """
137    When only the legacy ``state`` attr exists, ``to_brief_player`` still resolves it.
138
139    Kept for back-compat with older shims / hand-built test stubs.
140    """
141    player = SimpleNamespace(
142        player_id="kitchen",
143        name="Kitchen",
144        state=SimpleNamespace(value="paused"),
145        volume_level=10,
146        powered=True,
147        current_media=None,
148    )
149    assert to_brief_player(player).state == "paused"
150
151
152def test_to_brief_player_current_item_prefers_title() -> None:
153    """``current_item`` uses :class:`PlayerMedia.title` when available."""
154    player = SimpleNamespace(
155        player_id="p1",
156        name="P1",
157        playback_state=SimpleNamespace(value="playing"),
158        volume_level=50,
159        powered=True,
160        current_media=SimpleNamespace(uri="spotify://track/x", title="Song Name"),
161    )
162    assert to_brief_player(player).current_item == "Song Name"
163
164
165def test_to_brief_player_current_item_falls_back_to_uri() -> None:
166    """No title → ``current_item`` falls back to URI (always present on PlayerMedia)."""
167    player = SimpleNamespace(
168        player_id="p1",
169        name="P1",
170        playback_state=SimpleNamespace(value="playing"),
171        volume_level=50,
172        powered=True,
173        current_media=SimpleNamespace(uri="spotify://track/x", title=None),
174    )
175    assert to_brief_player(player).current_item == "spotify://track/x"
176
177
178def test_to_brief_player_no_current_media() -> None:
179    """``current_item`` is ``None`` when the player is idle (no current media)."""
180    player = SimpleNamespace(
181        player_id="p1",
182        name="P1",
183        playback_state=SimpleNamespace(value="idle"),
184        volume_level=0,
185        powered=False,
186        current_media=None,
187    )
188    assert to_brief_player(player).current_item is None
189
190
191@pytest.mark.parametrize(
192    ("player_powered", "state_powered", "expected", "case"),
193    [
194        # 1. ``state`` present, value differs from raw ``powered`` →
195        #    canonical state wins.
196        (False, True, True, "state.powered=True overrides raw .powered=False"),
197        # 2. Contradictory direction — the other way — also wins via state.
198        #    Without this case, a test that always read raw .powered would
199        #    still pass case #1 (because it happens to match expected=True
200        #    when state.powered=True).
201        (True, False, False, "state.powered=False overrides raw .powered=True"),
202    ],
203)
204def test_to_brief_player_powered_prefers_state(
205    player_powered: bool, state_powered: bool, expected: bool, case: str
206) -> None:
207    """
208    When ``Player.state.powered`` is present, it wins over raw ``Player.powered``.
209
210    MA core builds ``_state.powered`` from ``__final_power_state`` and
211    serialises it in the REST API; the raw ``Player.powered`` property
212    returns ``_attr_powered`` which lags behind (and stays ``False`` for
213    some virtual player types). The brief must match what
214    ``Player.state.to_dict()`` would emit — i.e. the canonical ``state.powered``
215    value, not the raw attribute.
216    """
217    player = SimpleNamespace(
218        player_id="p1",
219        name="P1",
220        playback_state=SimpleNamespace(value="playing"),
221        volume_level=100,
222        powered=player_powered,
223        current_media=None,
224        state=SimpleNamespace(powered=state_powered, current_media=None),
225    )
226    assert to_brief_player(player).powered is expected, case
227
228
229def test_to_brief_player_powered_falls_back_to_raw_when_no_state() -> None:
230    """
231    Without a ``state`` attribute, the raw ``Player.powered`` is the only signal.
232
233    Pairs with the parametrized test above to pin both branches of the
234    canonical-vs-raw selection: state present (state wins) and state absent
235    (raw wins).
236    """
237    player = SimpleNamespace(
238        player_id="p1",
239        name="P1",
240        playback_state=SimpleNamespace(value="playing"),
241        volume_level=100,
242        powered=False,
243        current_media=None,
244        # NO `state` attribute at all.
245    )
246    assert to_brief_player(player).powered is False
247
248
249def test_to_brief_player_current_item_uses_state_current_media() -> None:
250    """
251    ``current_item`` is cleared when ``Player.state.current_media`` is None.
252
253    After ``stop`` MA core clears ``_state.current_media``, but the raw
254    ``_attr_current_media`` may persist until the next playback. The brief
255    must reflect the canonical state so the LLM doesn't think a track is
256    still playing.
257    """
258    stale = SimpleNamespace(uri="library://track/48", title="07")
259    player = SimpleNamespace(
260        player_id="p1",
261        name="P1",
262        playback_state=SimpleNamespace(value="idle"),
263        volume_level=0,
264        powered=True,
265        current_media=stale,
266        state=SimpleNamespace(powered=True, current_media=None),
267    )
268    assert to_brief_player(player).current_item is None
269
270
271def test_to_brief_player_exposes_available_and_enabled() -> None:
272    """
273    ``available`` / ``enabled`` flow through, and the state ladder fires.
274
275    Combined assert: a regression that breaks the state override only
276    when both blocker fields are set would otherwise slip through —
277    the dedicated state tests above use single-axis stubs.
278    """
279    player = SimpleNamespace(
280        player_id="p1",
281        name="P1",
282        playback_state=SimpleNamespace(value="idle"),
283        volume_level=0,
284        powered=True,
285        current_media=None,
286        available=False,
287        enabled=False,
288    )
289    brief = to_brief_player(player)
290    assert brief.available is False
291    assert brief.enabled is False
292    # ``unavailable`` wins because it's higher in the priority ladder
293    # than ``disabled``; pinning both confirms the ladder ordering on
294    # the same stub that exercises the field exposure.
295    assert brief.state == "unavailable"
296
297
298def test_to_brief_player_available_enabled_default_true_when_attrs_missing() -> None:
299    """
300    Legacy stubs without ``available`` / ``enabled`` keep working (defaults to True).
301
302    Pins back-compat: tests built before this feature use bare
303    ``SimpleNamespace`` players, and they must still produce a usable brief.
304    """
305    player = SimpleNamespace(
306        player_id="p1",
307        name="P1",
308        playback_state=SimpleNamespace(value="playing"),
309        volume_level=50,
310        powered=True,
311        current_media=None,
312    )
313    brief = to_brief_player(player)
314    assert brief.available is True
315    assert brief.enabled is True
316
317
318def test_to_brief_player_unavailable_overrides_state() -> None:
319    """
320    ``state`` becomes ``"unavailable"`` when the player is offline.
321
322    Without the override the brief reports the cached ``playback_state``
323    (typically ``"idle"``) and an LLM cannot distinguish a quiet speaker
324    from one that fell off the network.
325    """
326    player = SimpleNamespace(
327        player_id="p1",
328        name="P1",
329        playback_state=SimpleNamespace(value="idle"),
330        volume_level=None,
331        powered=True,
332        current_media=None,
333        available=False,
334    )
335    assert to_brief_player(player).state == "unavailable"
336
337
338def _blocker_stub(**overrides: Any) -> SimpleNamespace:
339    """
340    Build a minimal player stub for state-ladder tests.
341
342    Every blocker field defaults to its "not blocked" value; tests pass
343    ``overrides`` for the axis they're exercising.
344    """
345    base: dict[str, Any] = {
346        "player_id": "p1",
347        "name": "P1",
348        "playback_state": SimpleNamespace(value="playing"),
349        "volume_level": None,
350        "powered": True,
351        "current_media": None,
352        "available": True,
353        "enabled": True,
354        "needs_setup": False,
355        "active_group": None,
356        "synced_to": None,
357    }
358    base.update(overrides)
359    return SimpleNamespace(**base)
360
361
362@pytest.mark.parametrize(
363    ("blocker", "expected_state"),
364    [
365        ({"available": False}, "unavailable"),
366        ({"enabled": False}, "disabled"),
367        ({"needs_setup": True}, "needs_setup"),
368        ({"synced_to": "leader-id"}, "synced"),
369        ({"active_group": "group-id"}, "synced"),
370    ],
371)
372def test_to_brief_player_state_override_per_blocker(
373    blocker: dict[str, object], expected_state: str
374) -> None:
375    """
376    Each blocker in isolation produces its dedicated ``state`` value.
377
378    Pins the per-rung behaviour of the state ladder. Without these
379    overrides the LLM would see ``state="playing"`` (the cached
380    playback_state on the stub) for every unusable device and pick the
381    wrong target.
382    """
383    assert to_brief_player(_blocker_stub(**blocker)).state == expected_state
384
385
386@pytest.mark.parametrize(
387    ("blockers", "expected_state", "case"),
388    [
389        (
390            {"available": False, "synced_to": "leader"},
391            "unavailable",
392            "unavailable beats synced",
393        ),
394        (
395            {"available": False, "enabled": False},
396            "unavailable",
397            "unavailable beats disabled",
398        ),
399        (
400            {"enabled": False, "needs_setup": True},
401            "disabled",
402            "disabled beats needs_setup",
403        ),
404        (
405            {"needs_setup": True, "active_group": "g"},
406            "needs_setup",
407            "needs_setup beats synced",
408        ),
409    ],
410)
411def test_to_brief_player_state_priority_chain(
412    blockers: dict[str, object], expected_state: str, case: str
413) -> None:
414    """
415    When multiple blockers are set, the most-blocking value wins.
416
417    The single ``state`` field has to summarise usability; an LLM that
418    only reads ``state`` (skipping the explicit booleans) must make the
419    safe call. Priority: unavailable > disabled > needs_setup > synced.
420    """
421    assert to_brief_player(_blocker_stub(**blockers)).state == expected_state, case
422
423
424def test_to_brief_player_exposes_new_blocker_fields() -> None:
425    """``needs_setup`` / ``active_group`` / ``synced_to`` flow through from MA."""
426    player = _blocker_stub(
427        needs_setup=True,
428        active_group="group-x",
429        synced_to="leader-y",
430    )
431    brief = to_brief_player(player)
432    assert brief.needs_setup is True
433    assert brief.active_group == "group-x"
434    assert brief.synced_to == "leader-y"
435
436
437def test_to_brief_player_new_fields_default_safely_when_attrs_missing() -> None:
438    """
439    Legacy stubs without the new attributes still produce a usable brief.
440
441    Mirrors the back-compat pattern already pinned for
442    ``available`` / ``enabled`` — tests built before this feature use
443    bare ``SimpleNamespace`` players, and they must keep working.
444    """
445    player = SimpleNamespace(
446        player_id="p1",
447        name="P1",
448        playback_state=SimpleNamespace(value="playing"),
449        volume_level=50,
450        powered=True,
451        current_media=None,
452    )
453    brief = to_brief_player(player)
454    assert brief.needs_setup is False
455    assert brief.active_group is None
456    assert brief.synced_to is None
457
458
459def test_to_brief_player_prefers_state_active_group_over_raw_attr() -> None:
460    """
461    ``Player.state.active_group`` is the canonical sync-membership signal.
462
463    MA's ``__final_active_group`` walks every GROUP player and resolves
464    membership / protocol translation; the raw ``Player.active_group``
465    dataclass attr lags and stays ``None`` for SyncGroupPlayer
466    followers. The brief must read the canonical value so a follower
467    captured by an active group surfaces as ``state="synced"``.
468    """
469    player = SimpleNamespace(
470        player_id="follower",
471        name="Lenco",
472        playback_state=SimpleNamespace(value="idle"),
473        volume_level=None,
474        powered=True,
475        current_media=None,
476        # Raw attribute stays None — the case that broke live verification.
477        active_group=None,
478        synced_to=None,
479        state=SimpleNamespace(
480            powered=True,
481            current_media=None,
482            active_group="syncgroup_x",
483            synced_to=None,
484        ),
485    )
486    brief = to_brief_player(player)
487    assert brief.active_group == "syncgroup_x"
488    assert brief.state == "synced"
489
490
491def test_to_brief_player_prefers_state_synced_to_over_raw_attr() -> None:
492    """``Player.state.synced_to`` translates protocol-player ids; the brief must use it."""
493    player = SimpleNamespace(
494        player_id="follower",
495        name="Speaker",
496        playback_state=SimpleNamespace(value="idle"),
497        volume_level=None,
498        powered=True,
499        current_media=None,
500        synced_to=None,
501        active_group=None,
502        state=SimpleNamespace(
503            powered=True,
504            current_media=None,
505            active_group=None,
506            synced_to="visible-leader-id",
507        ),
508    )
509    brief = to_brief_player(player)
510    assert brief.synced_to == "visible-leader-id"
511    assert brief.state == "synced"
512
513
514def test_to_brief_player_falls_back_to_raw_when_state_lacks_group_fields() -> None:
515    """
516    Back-compat: legacy stubs whose ``state`` lacks the new group fields fall through.
517
518    The existing ``test_to_brief_player_powered_prefers_state`` already
519    exercises a stub whose ``state`` has ``powered`` + ``current_media``
520    but no ``active_group`` / ``synced_to``. After this change those
521    older stubs must keep producing valid briefs — the canonical-read
522    branch must guard with ``hasattr(state, "active_group")``.
523    """
524    player = SimpleNamespace(
525        player_id="p1",
526        name="P1",
527        playback_state=SimpleNamespace(value="playing"),
528        volume_level=None,
529        powered=True,
530        current_media=None,
531        active_group="raw-group",
532        synced_to=None,
533        # state lacks the new fields entirely — only the older attrs.
534        state=SimpleNamespace(powered=True, current_media=None),
535    )
536    brief = to_brief_player(player)
537    assert brief.active_group == "raw-group"
538    assert brief.state == "synced"
539
540
541def test_to_brief_player_prefers_state_volume_muted_over_raw() -> None:
542    """``volume_muted`` flows through from the canonical state object."""
543    player = SimpleNamespace(
544        player_id="p1",
545        name="P1",
546        playback_state=SimpleNamespace(value="playing"),
547        volume_level=50,
548        volume_muted=False,
549        powered=True,
550        current_media=None,
551        state=SimpleNamespace(
552            powered=True,
553            current_media=None,
554            volume_muted=True,
555        ),
556    )
557    assert to_brief_player(player).volume_muted is True
558
559
560def test_to_brief_player_prefers_state_group_volume_over_raw() -> None:
561    """
562    ``group_volume`` is read from state — SyncGroupPlayer holds it there.
563
564    The raw ``Player.group_volume`` dataclass attr can lag; the canonical
565    property is exposed on ``Player.state`` (line 1497 of MA's
566    ``models/player.py``). A SyncGroupPlayer's brief must surface the
567    real group volume, not the cached ``None`` on the raw attribute.
568    """
569    player = SimpleNamespace(
570        player_id="syncgroup_x",
571        name="Group",
572        playback_state=SimpleNamespace(value="playing"),
573        volume_level=None,
574        group_volume=None,
575        powered=True,
576        current_media=None,
577        state=SimpleNamespace(
578            powered=True,
579            current_media=None,
580            group_volume=75,
581            group_volume_muted=False,
582        ),
583    )
584    brief = to_brief_player(player)
585    assert brief.group_volume == 75
586    assert brief.group_volume_muted is False
587
588
589def test_to_brief_player_new_volume_fields_default_to_none_when_attrs_missing() -> None:
590    """
591    Legacy stubs without volume_muted / group_volume / group_volume_muted attrs work.
592
593    Mirrors the back-compat pattern already pinned for the
594    ``active_group`` / ``synced_to`` additions.
595    """
596    player = SimpleNamespace(
597        player_id="p1",
598        name="P1",
599        playback_state=SimpleNamespace(value="playing"),
600        volume_level=50,
601        powered=True,
602        current_media=None,
603    )
604    brief = to_brief_player(player)
605    assert brief.volume_muted is None
606    assert brief.group_volume is None
607    assert brief.group_volume_muted is None
608
609
610def test_to_brief_player_volume_fields_fall_back_to_raw_when_state_lacks_them() -> None:
611    """
612    Stubs whose ``state`` lacks the volume fields fall through to raw attrs.
613
614    Back-compat with stubs that carry ``state`` for ``powered`` /
615    ``current_media`` but predate the new volume fields.
616    """
617    player = SimpleNamespace(
618        player_id="p1",
619        name="P1",
620        playback_state=SimpleNamespace(value="playing"),
621        volume_level=50,
622        volume_muted=True,
623        group_volume=80,
624        group_volume_muted=True,
625        powered=True,
626        current_media=None,
627        state=SimpleNamespace(powered=True, current_media=None),
628    )
629    brief = to_brief_player(player)
630    assert brief.volume_muted is True
631    assert brief.group_volume == 80
632    assert brief.group_volume_muted is True
633
634
635def test_to_brief_queue_with_items() -> None:
636    """``to_brief_queue`` builds a ``QueueBrief`` with item summaries."""
637    queue = SimpleNamespace(
638        queue_id="kitchen",
639        current_index=2,
640        items=10,
641        shuffle_enabled=True,
642        repeat_mode=SimpleNamespace(value="off"),
643    )
644    items = [
645        SimpleNamespace(
646            queue_item_id="i1",
647            name="One",
648            duration=120,
649            media_item=SimpleNamespace(artists=[SimpleNamespace(name="A1")]),
650        ),
651        SimpleNamespace(queue_item_id="i2", name="Two", duration=240, media_item=None),
652    ]
653    brief = to_brief_queue(queue, items=items)
654    assert isinstance(brief, QueueBrief)
655    assert brief.queue_id == "kitchen"
656    assert brief.shuffle is True
657    assert brief.repeat == "off"
658    assert len(brief.items) == 2
659    assert brief.items[0].index == 0
660    assert brief.items[1].index == 1
661    assert brief.items[0].artists == ["A1"]
662
663
664def test_to_brief_queue_exposes_insert_index_fields() -> None:
665    """``to_brief_queue`` sets index metadata for agent insert planning."""
666    queue = SimpleNamespace(
667        queue_id="kitchen",
668        current_index=2,
669        index_in_buffer=4,
670        items=10,
671        shuffle_enabled=True,
672        repeat_mode=SimpleNamespace(value="off"),
673    )
674    items = [
675        SimpleNamespace(
676            queue_item_id="i1",
677            name="One",
678            duration=120,
679            media_item=SimpleNamespace(artists=[SimpleNamespace(name="A1")]),
680        ),
681    ]
682    brief = to_brief_queue(queue, items=items, items_offset=5)
683    assert brief.index_in_buffer == 4
684    assert brief.next_insertable_index == 5
685    assert brief.items_start_index == 5
686    assert brief.items[0].index == 5
687
688
689def test_to_brief_queue_uses_canonical_items_int_for_count() -> None:
690    """
691    ``items`` (int) on the canonical PlayerQueue is the **total** length.
692
693    Earlier code mis-fell back to len(brief_items) (the truncated lookahead),
694    under-reporting real queue depth. ``items_count`` from the truncated
695    lookahead must not win over the explicit total.
696    """
697    queue = SimpleNamespace(
698        queue_id="q",
699        current_index=0,
700        items=42,  # canonical MA: total length, not a list
701        shuffle_enabled=False,
702        repeat_mode=None,
703    )
704    # Pass only 5 items as the truncated lookahead.
705    truncated = [
706        SimpleNamespace(queue_item_id=str(i), name=f"t{i}", duration=60, media_item=None)
707        for i in range(5)
708    ]
709    brief = to_brief_queue(queue, items=truncated)
710    assert brief.item_count == 42  # not 5
711    assert len(brief.items) == 5
712
713
714def test_page_args_clamps() -> None:
715    """``page_args`` clamps negatives and oversized limits."""
716    assert page_args(-5, 5000) == (0, 200)
717    assert page_args(0, 0) == (0, 1)
718    assert page_args(10, 25) == (10, 25)
719
720
721def test_to_brief_queue_returns_none_count_when_unknown() -> None:
722    """
723    When the queue exposes no canonical count, report ``item_count=None``.
724
725    A silent ``0`` (formerly returned via ``len(brief_items)`` when the
726    truncated lookahead was empty) would tell the LLM the queue is empty
727    when in fact it just doesn't know. ``None`` is the honest answer and
728    lets clients prompt the user instead of acting on false data.
729    """
730    queue = SimpleNamespace(
731        queue_id="q",
732        current_index=0,
733        # No `items` / `items_count` / `items_total` exposed at all.
734        shuffle_enabled=False,
735        repeat_mode=None,
736    )
737    brief = to_brief_queue(queue, items=None)
738    assert brief.item_count is None
739    assert brief.items == []
740
741
742def test_to_brief_queue_exposes_available() -> None:
743    """
744    ``available`` flows through from ``PlayerQueue`` so callers see the offline case.
745
746    Mirrors the parallel fix on the player side: a queue belonging to
747    an offline player is still returned by ``get_active_queue`` but
748    now carries an explicit ``available=False`` signal.
749    """
750    queue = SimpleNamespace(
751        queue_id="q",
752        current_index=0,
753        items=0,
754        shuffle_enabled=False,
755        repeat_mode=None,
756        available=False,
757    )
758    assert to_brief_queue(queue).available is False
759
760
761def test_to_brief_queue_available_defaults_true_when_attr_missing() -> None:
762    """Legacy queue stubs without ``available`` keep working (defaults to True)."""
763    queue = SimpleNamespace(
764        queue_id="q",
765        current_index=0,
766        items=0,
767        shuffle_enabled=False,
768        repeat_mode=None,
769    )
770    assert to_brief_queue(queue).available is True
771
772
773def test_player_brief_external_source_defaults_none() -> None:
774    """A self-driven player exposes ``external_source = None`` by default."""
775    player = SimpleNamespace(
776        player_id="p1",
777        name="Speaker",
778        playback_state=SimpleNamespace(value="idle"),
779        volume_level=None,
780        current_media=None,
781    )
782    assert to_brief_player(player).external_source is None
783
784
785def _audio_source_item(
786    *, provider: str, title: str | None, name: str = "Wrapper"
787) -> SimpleNamespace:
788    """Build a queue item stub whose stream is a plugin AUDIO_SOURCE."""
789    return SimpleNamespace(
790        name=name,
791        streamdetails=SimpleNamespace(
792            media_type=SimpleNamespace(value="audio_source"),
793            provider=provider,
794            stream_metadata=SimpleNamespace(title=title),
795        ),
796    )
797
798
799def test_external_now_playing_returns_provider_and_title() -> None:
800    """``_external_now_playing`` returns (provider, title) for an AUDIO_SOURCE item."""
801    item = _audio_source_item(provider="yandex_ynison--PL8BnL7a", title="Behind Your Walls")
802    assert _external_now_playing(item) == ("yandex_ynison--PL8BnL7a", "Behind Your Walls")
803
804
805def test_external_now_playing_none_for_normal_track() -> None:
806    """``_external_now_playing`` returns ``None`` for a normal track item."""
807    item = SimpleNamespace(
808        name="Real Track",
809        streamdetails=SimpleNamespace(
810            media_type=SimpleNamespace(value="track"),
811            provider="yandex_music--abc",
812            stream_metadata=None,
813        ),
814    )
815    assert _external_now_playing(item) is None
816
817
818def test_external_now_playing_none_when_no_streamdetails() -> None:
819    """``_external_now_playing`` returns ``None`` when there are no stream details."""
820    assert _external_now_playing(SimpleNamespace(name="x", streamdetails=None)) is None
821    assert _external_now_playing(None) is None
822
823
824def test_external_now_playing_title_may_be_none() -> None:
825    """``_external_now_playing`` returns ``(provider, None)`` when the title is absent."""
826    item = _audio_source_item(provider="airplay--1", title=None)
827    assert _external_now_playing(item) == ("airplay--1", None)
828
829
830def test_external_now_playing_accepts_legacy_plugin_source() -> None:
831    """The deprecated ``plugin_source`` media type is still treated as external."""
832    item = SimpleNamespace(
833        name="Wrapper",
834        streamdetails=SimpleNamespace(
835            media_type=SimpleNamespace(value="plugin_source"),
836            provider="spotify--1",
837            stream_metadata=SimpleNamespace(title="Some Song"),
838        ),
839    )
840    assert _external_now_playing(item) == ("spotify--1", "Some Song")
841
842
843def _queue(*, state: str, current_item: SimpleNamespace | None) -> SimpleNamespace:
844    return SimpleNamespace(state=SimpleNamespace(value=state), current_item=current_item)
845
846
847def test_to_brief_player_external_source_playing() -> None:
848    """Idle player + active queue playing an AUDIO_SOURCE reports the real state."""
849    player = SimpleNamespace(
850        player_id="lenco",
851        name="Lenco LS-500",
852        playback_state=SimpleNamespace(value="idle"),
853        volume_level=None,
854        current_media=None,
855    )
856    queue = _queue(
857        state="playing",
858        current_item=_audio_source_item(
859            provider="yandex_ynison--PL8BnL7a", title="Behind Your Walls"
860        ),
861    )
862    brief = to_brief_player(player, active_queue=queue)
863    assert brief.state == "playing"
864    assert brief.external_source == "yandex_ynison--PL8BnL7a"
865    assert brief.current_item == "Behind Your Walls"
866
867
868def test_to_brief_player_normal_active_queue_unchanged() -> None:
869    """A normal track in the active queue leaves external_source None."""
870    player = SimpleNamespace(
871        player_id="p",
872        name="Speaker",
873        playback_state=SimpleNamespace(value="playing"),
874        volume_level=None,
875        current_media=SimpleNamespace(uri="ym://track/1", title="Song"),
876    )
877    normal_item = SimpleNamespace(
878        name="Song",
879        streamdetails=SimpleNamespace(
880            media_type=SimpleNamespace(value="track"),
881            provider="yandex_music--x",
882            stream_metadata=None,
883        ),
884    )
885    brief = to_brief_player(player, active_queue=_queue(state="playing", current_item=normal_item))
886    assert brief.external_source is None
887    assert brief.state == "playing"
888    assert brief.current_item == "Song"
889
890
891def test_to_brief_player_blocking_ladder_wins_over_queue() -> None:
892    """An unavailable player keeps state=unavailable even with a playing queue."""
893    player = SimpleNamespace(
894        player_id="p",
895        name="Offline",
896        playback_state=SimpleNamespace(value="idle"),
897        volume_level=None,
898        current_media=None,
899        available=False,
900    )
901    queue = _queue(
902        state="playing",
903        current_item=_audio_source_item(provider="airplay--1", title="X"),
904    )
905    assert to_brief_player(player, active_queue=queue).state == "unavailable"
906
907
908def test_to_brief_player_synced_wins_over_queue() -> None:
909    """A sync follower keeps state=synced even though its leader's queue plays."""
910    player = SimpleNamespace(
911        player_id="p",
912        name="Follower",
913        playback_state=SimpleNamespace(value="idle"),
914        volume_level=None,
915        current_media=None,
916        synced_to="leader",
917    )
918    queue = _queue(
919        state="playing",
920        current_item=_audio_source_item(provider="airplay--1", title="X"),
921    )
922    assert to_brief_player(player, active_queue=queue).state == "synced"
923
924
925def test_to_brief_player_disabled_wins_over_queue() -> None:
926    """An admin-disabled player keeps state=disabled even with a playing queue."""
927    player = SimpleNamespace(
928        player_id="p",
929        name="Disabled",
930        playback_state=SimpleNamespace(value="idle"),
931        volume_level=None,
932        current_media=None,
933        enabled=False,
934    )
935    queue = _queue(
936        state="playing",
937        current_item=_audio_source_item(provider="airplay--1", title="X"),
938    )
939    assert to_brief_player(player, active_queue=queue).state == "disabled"
940
941
942def test_to_brief_player_needs_setup_wins_over_queue() -> None:
943    """A not-yet-configured player keeps state=needs_setup even with a playing queue."""
944    player = SimpleNamespace(
945        player_id="p",
946        name="Unconfigured",
947        playback_state=SimpleNamespace(value="idle"),
948        volume_level=None,
949        current_media=None,
950        needs_setup=True,
951    )
952    queue = _queue(
953        state="playing",
954        current_item=_audio_source_item(provider="airplay--1", title="X"),
955    )
956    assert to_brief_player(player, active_queue=queue).state == "needs_setup"
957
958
959def test_to_brief_player_external_source_without_title_keeps_current_media() -> None:
960    """A titleless external source still sets external_source but does not blank current_item."""
961    player = SimpleNamespace(
962        player_id="p",
963        name="Speaker",
964        playback_state=SimpleNamespace(value="idle"),
965        volume_level=None,
966        current_media=SimpleNamespace(uri="airplay://x", title="Fallback"),
967    )
968    queue = _queue(
969        state="playing",
970        current_item=_audio_source_item(provider="airplay--1", title=None),
971    )
972    brief = to_brief_player(player, active_queue=queue)
973    assert brief.state == "playing"
974    assert brief.external_source == "airplay--1"
975    assert brief.current_item == "Fallback"
976
977
978def test_to_brief_player_no_active_queue_legacy_behaviour() -> None:
979    """With active_queue omitted, state comes from player.playback_state."""
980    player = SimpleNamespace(
981        player_id="p",
982        name="Speaker",
983        playback_state=SimpleNamespace(value="idle"),
984        volume_level=None,
985        current_media=None,
986    )
987    brief = to_brief_player(player)
988    assert brief.state == "idle"
989    assert brief.external_source is None
990
991
992def test_to_brief_queue_relabels_external_item() -> None:
993    """An AUDIO_SOURCE item shows the real track title; normal items keep theirs."""
994    external = _audio_source_item(
995        provider="yandex_ynison--PL8BnL7a",
996        title="Behind Your Walls",
997        name="Yandex Music Connect (Ynison)",
998    )
999    external.queue_item_id = "ext"
1000    external.duration = None
1001    external.media_item = None
1002    normal = SimpleNamespace(
1003        queue_item_id="n1",
1004        name="Ordinary Song",
1005        duration=120,
1006        media_item=SimpleNamespace(artists=[SimpleNamespace(name="A")]),
1007        streamdetails=None,
1008    )
1009    queue = SimpleNamespace(
1010        queue_id="q",
1011        current_index=0,
1012        items=2,
1013        shuffle_enabled=False,
1014        repeat_mode=SimpleNamespace(value="off"),
1015        available=True,
1016    )
1017    brief = to_brief_queue(queue, items=[external, normal])
1018    names = [it.name for it in brief.items]
1019    assert names == ["Behind Your Walls", "Ordinary Song"]
1020
1021
1022def test_to_brief_queue_external_item_without_title_keeps_wrapper_name() -> None:
1023    """A titleless AUDIO_SOURCE item falls back to its wrapper name, not an empty string."""
1024    external = _audio_source_item(
1025        provider="airplay--1",
1026        title=None,
1027        name="AirPlay",
1028    )
1029    external.queue_item_id = "ext"
1030    external.duration = None
1031    external.media_item = None
1032    queue = SimpleNamespace(
1033        queue_id="q",
1034        current_index=0,
1035        items=1,
1036        shuffle_enabled=False,
1037        repeat_mode=SimpleNamespace(value="off"),
1038        available=True,
1039    )
1040    brief = to_brief_queue(queue, items=[external])
1041    assert brief.items[0].name == "AirPlay"
1042
1043
1044_DEBUG_CLASSES = [
1045    ("PlayerInspect", {"player_id", "raw", "state", "truncated"}),
1046    ("QueueInspect", {"queue_id", "raw", "current_item", "truncated"}),
1047    ("ProviderInspect", {"instance_id", "raw", "manifest", "truncated"}),
1048    ("LogLine", {"timestamp", "level", "component", "message"}),
1049    (
1050        "LogTailResult",
1051        {
1052            "log_path",
1053            "lines",
1054            "bytes_scanned",
1055            "truncated",
1056            "has_more",
1057            "response_truncated",
1058            "next_call_hint",
1059        },
1060    ),
1061    ("ComponentCount", {"component", "count"}),
1062    (
1063        "LogStatsResult",
1064        {
1065            "log_path",
1066            "window_seconds",
1067            "total_records",
1068            "level_counts",
1069            "top_components",
1070            "first_timestamp",
1071            "last_timestamp",
1072            "bytes_scanned",
1073            "truncated",
1074        },
1075    ),
1076    ("EventRecord", {"timestamp", "event_type", "object_id", "data"}),
1077    ("EventSnapshot", {"events", "buffer_capacity", "total_seen"}),
1078    (
1079        "EventBufferStats",
1080        {"capacity", "current_size", "total_seen", "dropped", "subscribed_since", "by_type"},
1081    ),
1082    ("ProviderSummary", {"instance_id", "domain", "type", "name", "available", "last_error"}),
1083    ("ProviderList", {"providers"}),
1084    ("ConfigValueDump", {"key", "type", "value"}),
1085    ("ProviderConfigDump", {"instance_id", "domain", "values", "truncated"}),
1086    ("RouteEntry", {"method", "path", "registered_by"}),
1087    ("RouteList", {"routes"}),
1088    ("PackageVersions", {"packages"}),
1089    ("ReloadResult", {"instance_id", "duration_ms", "new_available", "last_error"}),
1090    (
1091        "HealthSummary",
1092        {
1093            "providers_loaded",
1094            "providers_disabled",
1095            "providers_error",
1096            "providers_error_details",
1097            "queues_total",
1098            "queues_with_active_playback",
1099            "queues_with_errors",
1100            "events_per_min_by_type",
1101            "log_errors_last_5min",
1102            "disabled_capabilities",
1103        },
1104    ),
1105]
1106
1107
1108@pytest.mark.parametrize(("name", "fields"), _DEBUG_CLASSES)
1109def test_debug_dataclass_shape(name: str, fields: set[str]) -> None:
1110    """Debug dataclasses are frozen, kw_only, and have the expected fields."""
1111    cls = cast("type", getattr(models, name))
1112    assert dataclasses.is_dataclass(cls), f"{name} is not a dataclass"
1113    assert cls.__dataclass_params__.frozen, f"{name} must be frozen"  # type: ignore[attr-defined]
1114    assert cls.__dataclass_params__.kw_only, f"{name} must be kw_only"  # type: ignore[attr-defined]
1115    actual = {f.name for f in dataclasses.fields(cls)}
1116    assert actual == fields, f"{name} fields drift: {actual - fields=} {fields - actual=}"
1117
1118
1119_CONFIG_CLASSES = [
1120    ("ConfigTarget", {"target_type", "target_id", "domain", "name", "enabled"}),
1121    ("ConfigTargetList", {"providers", "core", "players"}),
1122    ("CoreConfigDump", {"domain", "values", "truncated"}),
1123    ("PlayerConfigDump", {"player_id", "provider", "values", "truncated"}),
1124    (
1125        "ConfigEntryDump",
1126        {
1127            "key",
1128            "type",
1129            "label",
1130            "default_value",
1131            "required",
1132            "description",
1133            "options",
1134            "range",
1135            "advanced",
1136            "hidden",
1137            "requires_reload",
1138            "depends_on",
1139            "action",
1140            "current_value",
1141        },
1142    ),
1143    ("ConfigEntryList", {"target_type", "target_id", "entries", "truncated"}),
1144    ("DSPConfigDump", {"player_id", "enabled", "input_gain", "output_gain", "filters"}),
1145    ("ValueChange", {"key", "before", "after", "secret"}),
1146    ("DiffResult", {"target_type", "target_id", "changes"}),
1147    (
1148        "SetValueResult",
1149        {"target_type", "target_id", "key", "applied", "requires_reload", "audit_log_id", "diff"},
1150    ),
1151    (
1152        "SaveResult",
1153        {
1154            "target_type",
1155            "target_id",
1156            "applied",
1157            "changes",
1158            "requires_reload",
1159            "audit_log_id",
1160            "diff",
1161        },
1162    ),
1163    ("ActionResult", {"instance_id", "action_key", "new_entries", "extra_data", "audit_log_id"}),
1164]
1165
1166
1167@pytest.mark.parametrize(("name", "fields"), _CONFIG_CLASSES)
1168def test_config_dataclass_shape(name: str, fields: set[str]) -> None:
1169    """Config dataclasses are frozen, kw_only, and have the expected fields."""
1170    cls = cast("type", getattr(models, name))
1171    assert dataclasses.is_dataclass(cls), f"{name} is not a dataclass"
1172    assert cls.__dataclass_params__.frozen, f"{name} must be frozen"  # type: ignore[attr-defined]
1173    assert cls.__dataclass_params__.kw_only, f"{name} must be kw_only"  # type: ignore[attr-defined]
1174    actual = {f.name for f in dataclasses.fields(cls)}
1175    assert actual == fields, f"{name} fields drift: {actual - fields=} {fields - actual=}"
1176