/
/
/
1"""Tests for de-duplicating the doubled end-of-queue play report."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, cast
6from unittest.mock import Mock
7
8from music_assistant.controllers.player_queues import PlayerQueuesController
9from music_assistant.controllers.player_queues.state import PlayerQueueData
10
11if TYPE_CHECKING:
12 from music_assistant_models.player_queue import PlayerQueue
13
14
15def _tracker() -> PlayerQueuesController:
16 """Build a bare controller with the two reported queues registered."""
17 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
18 ctrl._queue_data = {
19 queue_id: PlayerQueueData(queue=cast("PlayerQueue", Mock(queue_id=queue_id)))
20 for queue_id in ("q1", "q2")
21 }
22 return ctrl
23
24
25def test_completed_play_counts_once() -> None:
26 """The doubled end-of-queue completion report is only counted once."""
27 tracker = _tracker()
28 # first completion report -> counted
29 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is True
30 # second (duplicate) completion report for the same item -> skipped
31 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is False
32
33
34def test_repeat_counts_again_after_restart() -> None:
35 """A looped track is counted again once it restarts."""
36 tracker = _tracker()
37 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is True
38 # the track restarts (a not-fully-played report) -> guard resets
39 assert tracker._should_mark_played("q1", "item1", fully_played=False, is_playing=True) is True
40 # the next completion is counted again
41 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is True
42
43
44def test_now_playing_reports_are_never_deduplicated() -> None:
45 """Now-playing / partial reports are always forwarded."""
46 tracker = _tracker()
47 assert tracker._should_mark_played("q1", "item1", fully_played=False, is_playing=True) is True
48 assert tracker._should_mark_played("q1", "item1", fully_played=False, is_playing=True) is True
49
50
51def test_distinct_items_each_count() -> None:
52 """Two different queue items are counted independently."""
53 tracker = _tracker()
54 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is True
55 assert tracker._should_mark_played("q1", "item2", fully_played=True, is_playing=False) is True
56
57
58def test_partial_report_for_other_item_does_not_reset_guard() -> None:
59 """A not-fully-played report for a different item must not re-arm another item's guard."""
60 tracker = _tracker()
61 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is True
62 # a partial report for a *different* item must not reset item1's guard
63 assert tracker._should_mark_played("q1", "item2", fully_played=False, is_playing=True) is True
64 # so the duplicate completion of item1 is still skipped
65 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is False
66
67
68def test_queues_are_independent() -> None:
69 """Each queue tracks its own last-counted play."""
70 tracker = _tracker()
71 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is True
72 # the same item id on a different queue is counted independently
73 assert tracker._should_mark_played("q2", "item1", fully_played=True, is_playing=False) is True
74 # and each queue's own duplicate is skipped
75 assert tracker._should_mark_played("q1", "item1", fully_played=True, is_playing=False) is False
76 assert tracker._should_mark_played("q2", "item1", fully_played=True, is_playing=False) is False
77
78
79def test_not_fully_played_first_report_is_forwarded() -> None:
80 """A not-fully-played report with nothing counted yet is forwarded without error."""
81 tracker = _tracker()
82 assert tracker._should_mark_played("q1", "item1", fully_played=False, is_playing=False) is True
83