music-assistant-server

1"""Tests for skipping DRM protected Soundcloud tracks."""
2
3from __future__ import annotations
4
5import logging
6from collections.abc import AsyncGenerator
7from typing import Any
8
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import InvalidDataError, MediaNotFoundError
12
13from music_assistant.models.music_provider import SyncRunState
14from music_assistant.providers.soundcloud import (
15    DrmProtectedTrackError,
16    SoundcloudMusicProvider,
17    _is_drm_protected,
18)
19
20# transcodings as returned by the Soundcloud API for DRM protected content: the encrypted
21# HLS variants are the only playable ones, while the mp3 entries are advertised but 404
22DRM_TRANSCODINGS = [
23    {
24        "preset": "aac_160k",
25        "url": "https://api/aac_cbc",
26        "format": {"protocol": "cbc-encrypted-hls"},
27    },
28    {
29        "preset": "aac_160k",
30        "url": "https://api/aac_ctr",
31        "format": {"protocol": "ctr-encrypted-hls"},
32    },
33    {"preset": "mp3_1_0", "url": "https://api/mp3_hls", "format": {"protocol": "hls"}},
34    {"preset": "mp3_1_0", "url": "https://api/mp3_prog", "format": {"protocol": "progressive"}},
35]
36PLAIN_TRANSCODINGS = [
37    {"preset": "mp3_0_1", "url": "https://api/mp3_hls", "format": {"protocol": "hls"}},
38    {"preset": "mp3_0_1", "url": "https://api/mp3_prog", "format": {"protocol": "progressive"}},
39]
40
41
42def _track_obj(track_id: int, title: str, transcodings: list[dict[str, Any]]) -> dict[str, Any]:
43    """Build a Soundcloud API track object with the given transcodings."""
44    return {
45        "id": track_id,
46        "title": title,
47        "duration": 235818,
48        "full_duration": 235771,
49        "kind": "track",
50        "permalink_url": f"https://soundcloud.com/artist/{track_id}",
51        "policy": "MONETIZE",
52        "monetization_model": "AD_SUPPORTED",
53        "track_authorization": "auth-token",
54        "user": {"id": 1, "username": "Some Artist", "permalink": "some-artist"},
55        "media": {"transcodings": transcodings},
56    }
57
58
59def test_is_drm_protected_detects_encrypted_transcodings() -> None:
60    """Encrypted HLS transcodings mark a track as DRM protected."""
61    assert _is_drm_protected(_track_obj(1, "Danceteria", DRM_TRANSCODINGS)) is True
62
63
64def test_is_drm_protected_allows_plain_transcodings() -> None:
65    """A track with only plain transcodings is not DRM protected."""
66    assert _is_drm_protected(_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)) is False
67
68
69def test_is_drm_protected_allows_partial_track_object() -> None:
70    """A partial track object without media details is not treated as DRM protected."""
71    assert _is_drm_protected({"id": 3, "kind": "track"}) is False
72
73
74async def test_parse_track_rejects_drm_track(provider: SoundcloudMusicProvider) -> None:
75    """Parsing a DRM protected track raises an error the listing paths already skip."""
76    with pytest.raises(DrmProtectedTrackError):
77        await provider._parse_track(_track_obj(1, "Danceteria", DRM_TRANSCODINGS))
78    assert issubclass(DrmProtectedTrackError, InvalidDataError)
79
80
81async def test_parse_track_accepts_plain_track(provider: SoundcloudMusicProvider) -> None:
82    """Parsing a track without DRM returns a Track."""
83    track = await provider._parse_track(_track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS))
84    assert track.item_id == "2"
85    assert track.name == "Lofi Beat"
86
87
88async def test_library_tracks_skip_drm_and_log_count(
89    provider: SoundcloudMusicProvider, caplog: pytest.LogCaptureFixture
90) -> None:
91    """DRM protected tracks are not imported and the number skipped is logged."""
92
93    async def _liked_tracks(_user_id: str) -> AsyncGenerator[dict[str, Any]]:
94        yield _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
95        yield _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)
96        yield _track_obj(3, "The Fate of Ophelia", DRM_TRANSCODINGS)
97
98    provider._soundcloud.get_track_details_liked = _liked_tracks
99
100    with caplog.at_level(logging.INFO):
101        tracks = [track async for track in provider.get_library_tracks()]
102
103    assert [track.item_id for track in tracks] == ["2"]
104    assert "2" in caplog.text
105    assert "DRM" in caplog.text
106
107
108async def test_library_drm_tracks_are_not_reported_as_sync_failures(
109    provider: SoundcloudMusicProvider, sync_run: SyncRunState
110) -> None:
111    """
112    A DRM protected track is left out without counting as a failed sync item.
113
114    Soundcloud never allows those to be imported, so they are a permanent and expected
115    category rather than something that went wrong, and the count is logged in one go.
116    """
117
118    async def _liked_tracks(_user_id: str) -> AsyncGenerator[dict[str, Any]]:
119        yield _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
120        yield _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS)
121
122    provider._soundcloud.get_track_details_liked = _liked_tracks
123
124    tracks = [track async for track in provider.get_library_tracks()]
125
126    assert [track.item_id for track in tracks] == ["2"]
127    assert sync_run.skipped_item_ids == {}
128    assert sync_run.failures == 0
129    assert not sync_run.incomplete_media_types
130
131
132async def test_playlist_tracks_skip_drm(provider: SoundcloudMusicProvider) -> None:
133    """DRM protected tracks are left out of playlist listings."""
134    provider._soundcloud.get_playlist_details.return_value = {
135        "id": 10,
136        "tracks": [
137            _track_obj(1, "Danceteria", DRM_TRANSCODINGS),
138            _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS),
139        ],
140    }
141
142    get_playlist_tracks: Any = SoundcloudMusicProvider.get_playlist_tracks.__wrapped__  # type: ignore[attr-defined]
143    tracks = await get_playlist_tracks(provider, "10")
144
145    assert [track.item_id for track in tracks] == ["2"]
146
147
148async def test_search_skips_drm_track_and_keeps_others(
149    provider: SoundcloudMusicProvider,
150) -> None:
151    """A DRM protected search hit is skipped without dropping the other results."""
152    provider._soundcloud.search.return_value = {
153        "collection": [
154            _track_obj(1, "Danceteria", DRM_TRANSCODINGS),
155            _track_obj(2, "Lofi Beat", PLAIN_TRANSCODINGS),
156        ]
157    }
158
159    search: Any = SoundcloudMusicProvider.search.__wrapped__  # type: ignore[attr-defined]
160    result = await search(provider, "madonna", [MediaType.TRACK], 10)
161
162    assert [track.item_id for track in result.tracks] == ["2"]
163
164
165async def test_get_stream_details_reports_drm(provider: SoundcloudMusicProvider) -> None:
166    """Streaming an already imported DRM track fails with a message naming the cause."""
167    provider._soundcloud.get_track_details.return_value = [
168        _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
169    ]
170
171    with pytest.raises(MediaNotFoundError, match="DRM"):
172        await provider.get_stream_details("1", MediaType.TRACK)
173
174
175async def test_get_track_raises_media_not_found_when_unparsable(
176    provider: SoundcloudMusicProvider,
177) -> None:
178    """An unparsable track results in MediaNotFoundError instead of an internal error."""
179    provider._soundcloud.get_track_details.return_value = [
180        _track_obj(1, "Danceteria", DRM_TRANSCODINGS)
181    ]
182
183    get_track: Any = SoundcloudMusicProvider.get_track.__wrapped__  # type: ignore[attr-defined]
184    with pytest.raises(MediaNotFoundError):
185        await get_track(provider, "1")
186