music-assistant-server

1"""
2Tests for applying per-user resume info to a provider's podcast episode listing.
3
4Resume info for a listing is read from the playlog table in one batched query, no matter
5how many episodes the podcast has. The integration tests use the ``mass`` fixture from
6``tests/conftest.py``, which creates a full MusicAssistant instance with a real SQLite
7database in a temporary directory.
8"""
9
10from __future__ import annotations
11
12from collections.abc import AsyncGenerator, Callable
13from typing import TYPE_CHECKING, Any
14from unittest.mock import AsyncMock, MagicMock, patch
15
16import pytest
17from music_assistant_models.enums import MediaType
18from music_assistant_models.media_items import Podcast, PodcastEpisode, ProviderMapping
19
20from music_assistant.constants import DB_TABLE_PLAYLOG
21from music_assistant.mass import MusicAssistant
22from music_assistant.models.music_provider import MusicProvider
23
24if TYPE_CHECKING:
25    from music_assistant_models.auth import User
26
27PROVIDER_ID = "test_podcast_prov"
28PODCAST_ID = "show-001"
29
30
31class _StubPodcastProvider(MusicProvider):
32    """Minimal music provider yielding a prepared list of podcast episodes."""
33
34    def __init__(self, episodes: list[PodcastEpisode]) -> None:
35        """
36        Initialize the stub provider.
37
38        :param episodes: The episodes to yield from get_podcast_episodes.
39        """
40        self.episodes = episodes
41        self.config = MagicMock()
42        self.config.instance_id = PROVIDER_ID
43        self.manifest = MagicMock()
44        self.manifest.domain = PROVIDER_ID
45        self.logger = MagicMock()
46
47    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
48        """Yield the prepared episodes for the given podcast id."""
49        for episode in self.episodes:
50            yield episode
51
52
53def _episode(index: int, **kwargs: Any) -> PodcastEpisode:
54    """
55    Build a provider podcast episode.
56
57    :param index: Episode number, also used to derive its item id.
58    :param kwargs: Extra PodcastEpisode attributes (e.g. resume info).
59    """
60    item_id = f"ep-{index:03d}"
61    return PodcastEpisode(
62        item_id=item_id,
63        provider=PROVIDER_ID,
64        name=f"Episode {index}",
65        provider_mappings={
66            ProviderMapping(
67                item_id=item_id,
68                provider_domain=PROVIDER_ID,
69                provider_instance=PROVIDER_ID,
70            )
71        },
72        position=index,
73        podcast=Podcast(
74            item_id=PODCAST_ID,
75            provider=PROVIDER_ID,
76            name="My Podcast",
77            provider_mappings={
78                ProviderMapping(
79                    item_id=PODCAST_ID,
80                    provider_domain=PROVIDER_ID,
81                    provider_instance=PROVIDER_ID,
82                )
83            },
84        ),
85        **kwargs,
86    )
87
88
89async def _add_playlog_row(
90    mass: MusicAssistant,
91    item_id: str,
92    userid: str,
93    seconds_played: int,
94    fully_played: bool,
95    timestamp: int = 1000,
96) -> None:
97    """Seed one playlog row for a podcast episode."""
98    await mass.music.database.insert(
99        DB_TABLE_PLAYLOG,
100        {
101            "item_id": item_id,
102            "provider": PROVIDER_ID,
103            "media_type": MediaType.PODCAST_EPISODE.value,
104            "name": item_id,
105            "userid": userid,
106            "seconds_played": seconds_played,
107            "fully_played": fully_played,
108            "timestamp": timestamp,
109        },
110        allow_replace=True,
111    )
112
113
114@pytest.fixture(name="count_playlog_queries")
115def count_playlog_queries_fixture(
116    mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
117) -> Callable[[], int]:
118    """Count every database read of the playlog table, single row or batched."""
119    calls = 0
120
121    def _counted(name: str) -> Callable[..., Any]:
122        original = getattr(mass.music.database, name)
123
124        async def _wrapper(table: str, *args: Any, **kwargs: Any) -> Any:
125            nonlocal calls
126            if table == DB_TABLE_PLAYLOG:
127                calls += 1
128            return await original(table, *args, **kwargs)
129
130        return _wrapper
131
132    for name in ("get_row", "get_rows"):
133        monkeypatch.setattr(mass.music.database, name, _counted(name))
134    return lambda: calls
135
136
137async def _list_episodes(
138    mass: MusicAssistant,
139    episodes: list[PodcastEpisode],
140    user: User | None = None,
141) -> list[PodcastEpisode]:
142    """
143    Run the controller's provider listing for a stub provider.
144
145    :param mass: The MusicAssistant instance to run against.
146    :param episodes: The episodes the stub provider yields.
147    :param user: The session user the request is made for, if any.
148    """
149    provider = _StubPodcastProvider(episodes)
150    with (
151        patch.object(mass, "get_provider", return_value=provider),
152        patch(
153            "music_assistant.controllers.music.media.podcasts.get_current_user",
154            return_value=user,
155        ),
156    ):
157        return [
158            x
159            async for x in mass.music.podcasts._get_provider_podcast_episodes(
160                PODCAST_ID, PROVIDER_ID
161            )
162        ]
163
164
165async def _explain_resume_query(mass: MusicAssistant, user: User | None) -> list[str]:
166    """
167    Return the query plan rows for the batched resume query as the controller emits it.
168
169    :param mass: The MusicAssistant instance to run against.
170    :param user: The session user the request is made for, if any.
171    """
172    database = mass.music.database
173    # drop the planner statistics so SQLite assumes its default (large) table size instead of
174    # the handful of rows seeded here, where scanning the table really is the cheaper plan
175    await database.execute("ANALYZE")
176    await database.execute("DELETE FROM sqlite_stat1")
177    await database.execute("ANALYZE sqlite_master")
178
179    captured: dict[str, Any] = {}
180    original = database._db.execute_fetchall
181
182    async def _spy(sql: str, params: Any = None) -> Any:
183        if DB_TABLE_PLAYLOG in sql and sql.startswith("SELECT"):
184            captured["sql"], captured["params"] = sql, params
185        return await original(sql, params)
186
187    with patch.object(database._db, "execute_fetchall", _spy):
188        await _list_episodes(mass, [_episode(1)], user=user)
189
190    # the emitted query is explained rather than a copy of it, so the plan cannot drift
191    # away from the query this test claims to cover
192    assert "sql" in captured, "no playlog select was emitted"
193    plan = await database.get_rows_from_query(
194        f"EXPLAIN QUERY PLAN {captured['sql']}", captured["params"], limit=0
195    )
196    return [row["detail"] for row in plan]
197
198
199async def test_resume_info_is_scoped_to_the_requesting_user(mass: MusicAssistant) -> None:
200    """The requesting user's progress is applied; another user's progress is ignored."""
201    user = await mass.webserver.auth.create_user("podcastresume")
202    other_user = await mass.webserver.auth.create_user("podcastresumeother")
203    await _add_playlog_row(mass, "ep-001", user.user_id, seconds_played=90, fully_played=False)
204    await _add_playlog_row(mass, "ep-002", other_user.user_id, seconds_played=30, fully_played=True)
205
206    result = await _list_episodes(mass, [_episode(1), _episode(2)], user=user)
207
208    assert result[0].resume_position_ms == 90000
209    assert result[0].fully_played is False
210    assert result[1].resume_position_ms is None
211    assert result[1].fully_played is None
212
213
214async def test_native_resume_info_is_not_overwritten(mass: MusicAssistant) -> None:
215    """An episode that arrives with resume info from its provider is left untouched."""
216    user = await mass.webserver.auth.create_user("podcastnative")
217    await _add_playlog_row(mass, "ep-001", user.user_id, seconds_played=90, fully_played=True)
218
219    episode = _episode(1, fully_played=False, resume_position_ms=5000)
220    result = await _list_episodes(mass, [episode], user=user)
221
222    assert result[0].fully_played is False
223    assert result[0].resume_position_ms == 5000
224
225
226async def test_listing_uses_a_single_playlog_query(
227    mass: MusicAssistant, count_playlog_queries: Callable[[], int]
228) -> None:
229    """Resume info for an entire listing costs one query, not one query per episode."""
230    user = await mass.webserver.auth.create_user("podcastbatched")
231    await _add_playlog_row(mass, "ep-005", user.user_id, seconds_played=120, fully_played=False)
232
233    episodes = [_episode(index) for index in range(1, 26)]
234    result = await _list_episodes(mass, episodes, user=user)
235
236    assert count_playlog_queries() == 1
237    assert len(result) == 25
238    assert result[4].resume_position_ms == 120000
239
240
241async def test_no_playlog_query_when_provider_supplies_resume_info(
242    mass: MusicAssistant, count_playlog_queries: Callable[[], int]
243) -> None:
244    """A provider that reports resume info natively triggers no playlog query at all."""
245    user = await mass.webserver.auth.create_user("podcastnativeonly")
246    episodes = [_episode(index, fully_played=False, resume_position_ms=1000) for index in (1, 2)]
247
248    result = await _list_episodes(mass, episodes, user=user)
249
250    assert count_playlog_queries() == 0
251    assert all(x.resume_position_ms == 1000 for x in result)
252
253
254async def test_resume_info_is_not_capped_by_the_default_row_limit(mass: MusicAssistant) -> None:
255    """
256    Resume info survives a playlog holding more rows than the default query limit.
257
258    get_rows caps at 500 rows unless limit=0 is passed, and the batched query sorts by
259    ascending timestamp - so the cap keeps the *oldest* rows and drops the newest, which are
260    exactly the episodes a listener is part way through. The playlog keeps 90 days of history
261    across every podcast of a provider, so busy households cross 500 rows.
262    """
263    user = await mass.webserver.auth.create_user("podcastrowlimit")
264    row_count = 501
265    for index in range(1, row_count + 1):
266        # ascending timestamps: the newest row sorts last and is the first one a cap drops
267        await _add_playlog_row(
268            mass,
269            f"ep-{index:03d}",
270            user.user_id,
271            seconds_played=index,
272            fully_played=False,
273            timestamp=1000 + index,
274        )
275
276    # the oldest row doubles as a control: it survives either way, so a failure below
277    # points at the cap rather than at resume lookup being broken altogether
278    result = await _list_episodes(mass, [_episode(1), _episode(row_count)], user=user)
279
280    assert result[0].resume_position_ms == 1000
281    assert result[1].resume_position_ms == row_count * 1000
282
283
284async def test_resume_query_uses_the_provider_media_type_index(mass: MusicAssistant) -> None:
285    """
286    The batched resume query is served by an index instead of scanning the playlog.
287
288    It filters on provider/media_type/userid, which the item_id-first unique index cannot
289    serve, so without a dedicated index SQLite falls back to the userid index and reads every
290    row that user ever played, of every media type. The playlog holds 90 days of history for
291    the whole household, so that cost grows with listening activity rather than podcast size.
292    """
293    user = await mass.webserver.auth.create_user("podcastindexplan")
294    await _add_playlog_row(mass, "ep-001", user.user_id, seconds_played=10, fully_played=False)
295
296    details = await _explain_resume_query(mass, user)
297
298    assert any(f"USING INDEX {DB_TABLE_PLAYLOG}_provider_media_type_idx" in x for x in details), (
299        details
300    )
301    # with userid in the filter the equality prefix reaches timestamp, so the index satisfies
302    # the ORDER BY on its own. That part does not survive userid dropping out of the filter -
303    # the test below covers what still holds there
304    assert not any("TEMP B-TREE" in x for x in details), details
305
306
307async def test_resume_query_uses_the_index_without_a_session_user(mass: MusicAssistant) -> None:
308    """
309    The batched resume query still avoids a playlog scan when no user can be resolved.
310
311    Without a userid to filter on, the usable equality prefix stops at provider/media_type and
312    SQLite sorts the matched rows rather than reading them already ordered. The lookup itself
313    still has to go through the index: that sort covers one provider's episodes, where a table
314    scan would cover the whole household's 90 days of history.
315    """
316    await _add_playlog_row(mass, "ep-001", "user-a", seconds_played=10, fully_played=False)
317
318    with patch.object(
319        mass.music, "_get_user_for_provider", new_callable=AsyncMock, return_value=None
320    ):
321        details = await _explain_resume_query(mass, None)
322
323    assert any(f"USING INDEX {DB_TABLE_PLAYLOG}_provider_media_type_idx" in x for x in details), (
324        details
325    )
326
327
328async def test_without_a_session_user_the_newest_progress_wins(mass: MusicAssistant) -> None:
329    """With no user to scope to, the most recently played row is applied."""
330    # the newest row is deliberately both inserted first and owned by the alphabetically
331    # first userid, so neither insertion order nor the index's (userid, timestamp) key order
332    # can leave it last. The map the controller builds keeps whichever row it sees last, so
333    # only an explicit sort on timestamp lands on the 240s one
334    await _add_playlog_row(mass, "ep-001", "user-a", 240, fully_played=False, timestamp=2000)
335    await _add_playlog_row(mass, "ep-001", "user-b", 30, fully_played=False, timestamp=1000)
336
337    with patch.object(
338        mass.music, "_get_user_for_provider", new_callable=AsyncMock, return_value=None
339    ):
340        result = await _list_episodes(mass, [_episode(1)], user=None)
341
342    assert result[0].resume_position_ms == 240000
343