/
/
/
1"""Tests for MusicController.in_progress_items playlog queries."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import Mock, patch
7
8from music_assistant_models.enums import MediaType
9
10from music_assistant.constants import DB_TABLE_PLAYLOG, DB_TABLE_PROVIDER_MAPPINGS
11from music_assistant.helpers.datetime import utc_timestamp
12from music_assistant.mass import MusicAssistant
13
14if TYPE_CHECKING:
15 import pytest
16
17GET_CURRENT_USER = "music_assistant.controllers.music.controller.get_current_user"
18
19
20async def _add_in_progress_row(
21 mass: MusicAssistant,
22 item_id: str,
23 *,
24 provider: str = "library",
25 media_type: MediaType = MediaType.AUDIOBOOK,
26 userid: str = "user-a",
27) -> None:
28 """Insert a single in-progress (partially played) playlog row."""
29 await mass.music.database.insert(
30 DB_TABLE_PLAYLOG,
31 {
32 "item_id": item_id,
33 "provider": provider,
34 "media_type": media_type.value,
35 "name": f"Item {item_id}",
36 "timestamp": int(utc_timestamp()),
37 "fully_played": False,
38 "seconds_played": 60,
39 "userid": userid,
40 "user_initiated": True,
41 },
42 )
43
44
45async def _add_provider_mapping(
46 mass: MusicAssistant,
47 item_id: int,
48 provider_instance: str,
49 media_type: MediaType = MediaType.AUDIOBOOK,
50 *,
51 available: bool = True,
52) -> None:
53 """Insert a provider mapping row for a library item."""
54 await mass.music.database.insert(
55 DB_TABLE_PROVIDER_MAPPINGS,
56 {
57 "media_type": media_type.value,
58 "item_id": item_id,
59 "provider_domain": provider_instance.rstrip("0123456789_"),
60 "provider_instance": provider_instance,
61 "provider_item_id": f"{provider_instance}-{item_id}",
62 "available": available,
63 "in_library": True,
64 },
65 )
66
67
68async def test_in_progress_items_omitted_providers_unchanged(
69 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
70) -> None:
71 """Existing callers (no providers filter) keep seeing every in-progress entry."""
72 monkeypatch.setattr(mass.music, "get_active_provider_instances", lambda: ["local_1"])
73 await _add_provider_mapping(mass, item_id=1, provider_instance="local_1")
74 await _add_in_progress_row(mass, "1", provider="library")
75
76 result = await mass.music.in_progress_items(limit=10)
77
78 assert {item.item_id for item in result} == {"1"}
79
80
81async def test_in_progress_items_filters_direct_provider_entries(
82 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
83) -> None:
84 """Direct (non-library) playlog entries are matched against their own provider."""
85 monkeypatch.setattr(
86 mass.music, "get_active_provider_instances", lambda: ["audible_1", "local_1"]
87 )
88 await _add_in_progress_row(mass, "audible-book", provider="audible_1")
89 await _add_in_progress_row(mass, "local-book", provider="local_1")
90
91 result = await mass.music.in_progress_items(limit=10, providers=["local_1"])
92
93 assert {item.item_id for item in result} == {"local-book"}
94
95
96async def test_in_progress_items_library_item_or_matches_any_mapping(
97 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
98) -> None:
99 """A library item mapped to two providers remains when only one is requested."""
100 monkeypatch.setattr(
101 mass.music, "get_active_provider_instances", lambda: ["audible_1", "local_1"]
102 )
103 await _add_provider_mapping(mass, item_id=1, provider_instance="audible_1")
104 await _add_provider_mapping(mass, item_id=1, provider_instance="local_1")
105 await _add_in_progress_row(mass, "1", provider="library")
106
107 result = await mass.music.in_progress_items(limit=10, providers=["local_1"])
108
109 assert {item.item_id for item in result} == {"1"}
110
111
112async def test_in_progress_items_explicit_empty_providers_returns_no_items(
113 mass: MusicAssistant,
114) -> None:
115 """An explicit empty provider list returns no items, distinct from omitting the filter."""
116 await _add_in_progress_row(mass, "book-1")
117
118 result = await mass.music.in_progress_items(limit=10, providers=[])
119
120 assert result == []
121
122
123async def test_in_progress_items_combines_explicit_and_user_provider_filter(
124 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
125) -> None:
126 """A user's provider_filter narrows an explicit filter that would otherwise pass."""
127 monkeypatch.setattr(mass.music, "get_active_provider_instances", lambda: ["local_1"])
128 await _add_provider_mapping(mass, item_id=1, provider_instance="audible_1")
129 await _add_provider_mapping(mass, item_id=1, provider_instance="local_1")
130 await _add_in_progress_row(mass, "1", provider="library")
131
132 with patch(GET_CURRENT_USER, return_value=Mock(user_id="user-a", provider_filter=["local_1"])):
133 result = await mass.music.in_progress_items(limit=10, providers=["local_1"])
134 assert {item.item_id for item in result} == {"1"}
135
136 # requesting a provider the user isn't permitted to use must not leak the item back
137 # in, even though the item does have a (permission-restricted) mapping to it.
138 with patch(GET_CURRENT_USER, return_value=Mock(user_id="user-a", provider_filter=["local_1"])):
139 result = await mass.music.in_progress_items(limit=10, providers=["audible_1"])
140 assert result == []
141
142
143async def test_in_progress_items_excludes_unavailable_mapping(
144 mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
145) -> None:
146 """A mapping that is no longer available on the requested provider cannot match."""
147 monkeypatch.setattr(
148 mass.music, "get_active_provider_instances", lambda: ["local_1", "audible_1"]
149 )
150 await _add_provider_mapping(mass, item_id=1, provider_instance="local_1")
151 await _add_provider_mapping(mass, item_id=1, provider_instance="audible_1", available=False)
152 await _add_in_progress_row(mass, "1", provider="library")
153
154 result = await mass.music.in_progress_items(limit=10, providers=["audible_1"])
155
156 assert result == []
157