/
/
/
1"""
2End-to-end tests for the bookkeeping that credits an enqueued album as played.
3
4Boots a hermetic MusicAssistant (fake `test` provider + demo players) and drives the real
5``play_media`` path, so the wiring that arms and re-arms an album's credit is exercised
6rather than the decision helper alone.
7"""
8
9from __future__ import annotations
10
11from typing import TYPE_CHECKING, cast
12
13import pytest
14from music_assistant_models.enums import QueueOption
15
16from music_assistant.mass import MusicAssistant
17from music_assistant.models.music_provider import MusicProvider
18
19from .conftest import demo_players, wait_for
20
21if TYPE_CHECKING:
22 from music_assistant_models.media_items import Album, ItemMapping, MediaItemType, Track
23
24
25async def _album_with_track(mass: MusicAssistant, album_id: str) -> tuple[Album, Track]:
26 """Return a fake-provider album together with one of its tracks."""
27 test_prov = cast("MusicProvider", mass.get_provider("test"))
28 assert test_prov is not None
29 album = await test_prov.get_album(album_id)
30 tracks = await test_prov.get_album_tracks(album_id)
31 return album, tracks[0]
32
33
34async def _play(mass: MusicAssistant, queue_id: str, album: Album, **kwargs: object) -> None:
35 """Play (or enqueue) an album and wait until the queue has picked it up."""
36 await mass.player_queues.play_media(
37 queue_id,
38 cast("MediaItemType | ItemMapping | str", album),
39 **kwargs, # type: ignore[arg-type]
40 )
41 assert await wait_for(
42 lambda: (q := mass.player_queues.get(queue_id)) is not None and q.items > 0
43 ), "album never reached the queue"
44
45
46@pytest.mark.asyncio
47async def test_album_credit_is_armed_and_rearmed_through_play_media(
48 e2e_mass: MusicAssistant,
49) -> None:
50 """Playing an album arms its credit once; queueing it again arms it for another play."""
51 queue_id = demo_players(e2e_mass)[0].player_id
52 album, track = await _album_with_track(e2e_mass, "0_0")
53 tracker = e2e_mass.player_queues
54
55 await _play(e2e_mass, queue_id, album)
56 queue_data = tracker.queue_data(queue_id)
57 assert queue_data.credited_albums == set()
58
59 # the first of the album's tracks to complete claims the credit
60 assert tracker._claim_enqueued_album_credit(queue_data, track) is not None
61 assert queue_data.credited_albums == {album}
62 # every later track of the same album finds it already claimed
63 assert tracker._claim_enqueued_album_credit(queue_data, track) is None
64
65 # queueing the same album again is a new play of it
66 await _play(e2e_mass, queue_id, album, option=QueueOption.ADD)
67 assert queue_data.credited_albums == set()
68 assert tracker._claim_enqueued_album_credit(queue_data, track) is not None
69
70
71@pytest.mark.asyncio
72async def test_album_credit_is_cleared_when_a_new_play_replaces_the_queue(
73 e2e_mass: MusicAssistant,
74) -> None:
75 """A play that replaces the queue drops the credits recorded for the previous one."""
76 queue_id = demo_players(e2e_mass)[1].player_id
77 album, track = await _album_with_track(e2e_mass, "0_0")
78 other_album, _ = await _album_with_track(e2e_mass, "1_0")
79 tracker = e2e_mass.player_queues
80
81 await _play(e2e_mass, queue_id, album)
82 queue_data = tracker.queue_data(queue_id)
83 assert tracker._claim_enqueued_album_credit(queue_data, track) is not None
84 assert queue_data.credited_albums == {album}
85
86 await _play(e2e_mass, queue_id, other_album)
87 assert queue_data.credited_albums == set()
88
89
90@pytest.mark.asyncio
91async def test_album_credit_is_dropped_when_the_album_leaves_the_enqueued_list(
92 e2e_mass: MusicAssistant,
93) -> None:
94 """A credit is forgotten once its album falls off the (capped) enqueued list."""
95 queue_id = demo_players(e2e_mass)[2].player_id
96 album, track = await _album_with_track(e2e_mass, "0_0")
97 tracker = e2e_mass.player_queues
98
99 await _play(e2e_mass, queue_id, album)
100 queue_data = tracker.queue_data(queue_id)
101 assert tracker._claim_enqueued_album_credit(queue_data, track) is not None
102 assert queue_data.credited_albums == {album}
103
104 # push the album out of the enqueued list, which holds the last 10 items
105 for index in range(1, 12):
106 other, _ = await _album_with_track(e2e_mass, f"{index}_0")
107 await _play(e2e_mass, queue_id, other, option=QueueOption.ADD)
108
109 assert album not in queue_data.enqueued_media_items
110 assert queue_data.credited_albums == set()
111