/
/
/
1"""Test MusicMe provider streaming methods."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, patch
6
7import pytest
8from music_assistant_models.enums import ContentType, MediaType, StreamType
9from music_assistant_models.errors import MediaNotFoundError
10
11from music_assistant.providers.musicme.constants import PARTNER_ID, STREAM_BASE
12from music_assistant.providers.musicme.provider import MusicMeProvider
13
14# ---------------------------------------------------------------------------
15# _get_stream_url
16# ---------------------------------------------------------------------------
17
18
19class TestGetStreamUrl:
20 """Tests for _get_stream_url."""
21
22 @pytest.mark.asyncio
23 async def test_returns_stream_url(self, provider: MusicMeProvider) -> None:
24 """Test that a valid ticket is decrypted and a stream URL is built."""
25 # Build a fake encrypted ticket that decrypt() can handle.
26 # We mock _api_get to return a response with a "ticket" field,
27 # then mock decrypt to return a known token.
28 fake_ticket = "FAKETOKEN123"
29 provider._api_get = AsyncMock(return_value={"ticket": "dummy_encrypted"}) # type: ignore[method-assign]
30
31 with patch(
32 "music_assistant.providers.musicme.provider.decrypt",
33 return_value=fake_ticket,
34 ):
35 url = await provider._get_stream_url("5034644330297-01_01")
36
37 assert url == f"{STREAM_BASE}/{PARTNER_ID}/{fake_ticket}.mp4"
38 provider._api_get.assert_called_once()
39 call_args = provider._api_get.call_args[0][0]
40 assert "/getstream/" in call_args
41 assert "ref=5034644330297-01_01" in call_args
42
43 @pytest.mark.asyncio
44 async def test_raises_on_no_ticket(self, provider: MusicMeProvider) -> None:
45 """Test that missing ticket raises MediaNotFoundError."""
46 provider._api_get = AsyncMock(return_value={"no_ticket_here": True}) # type: ignore[method-assign]
47
48 with pytest.raises(MediaNotFoundError, match="No streaming ticket"):
49 await provider._get_stream_url("5034644330297-01_01")
50
51 @pytest.mark.asyncio
52 async def test_raises_on_none_response(self, provider: MusicMeProvider) -> None:
53 """Test that None API response raises MediaNotFoundError."""
54 provider._api_get = AsyncMock(return_value=None) # type: ignore[method-assign]
55
56 with pytest.raises(MediaNotFoundError, match="No streaming ticket"):
57 await provider._get_stream_url("5034644330297-01_01")
58
59
60# ---------------------------------------------------------------------------
61# _get_radio_track
62# ---------------------------------------------------------------------------
63
64
65class TestGetRadioTrack:
66 """Tests for _get_radio_track."""
67
68 @pytest.mark.asyncio
69 async def test_returns_first_streamable_track(self, provider: MusicMeProvider) -> None:
70 """Test that the first streamable track barcode is returned."""
71 provider._api_get = AsyncMock( # type: ignore[method-assign]
72 return_value={
73 "results": {
74 "tracks": [
75 {"barcode": "AAA-01_01", "streamable": 1},
76 {"barcode": "BBB-01_01", "streamable": 2},
77 {"barcode": "CCC-01_01", "streamable": 2},
78 ]
79 }
80 }
81 )
82
83 result = await provider._get_radio_track("rd-123")
84 assert result == "BBB-01_01"
85
86 @pytest.mark.asyncio
87 async def test_raises_on_no_data(self, provider: MusicMeProvider) -> None:
88 """Test that None response raises MediaNotFoundError."""
89 provider._api_get = AsyncMock(return_value=None) # type: ignore[method-assign]
90
91 with pytest.raises(MediaNotFoundError, match="returned no data"):
92 await provider._get_radio_track("rd-123")
93
94 @pytest.mark.asyncio
95 async def test_raises_on_no_streamable_tracks(self, provider: MusicMeProvider) -> None:
96 """Test that no streamable tracks raises MediaNotFoundError."""
97 provider._api_get = AsyncMock( # type: ignore[method-assign]
98 return_value={
99 "results": {
100 "tracks": [
101 {"barcode": "AAA-01_01", "streamable": 1},
102 {"barcode": "BBB-01_01", "streamable": 0},
103 ]
104 }
105 }
106 )
107
108 with pytest.raises(MediaNotFoundError, match="No streamable tracks"):
109 await provider._get_radio_track("rd-123")
110
111
112# ---------------------------------------------------------------------------
113# get_stream_details
114# ---------------------------------------------------------------------------
115
116
117class TestGetStreamDetails:
118 """Tests for get_stream_details."""
119
120 @pytest.mark.asyncio
121 async def test_track_stream_details(self, provider: MusicMeProvider) -> None:
122 """Test stream details for a regular track."""
123 fake_url = f"{STREAM_BASE}/{PARTNER_ID}/TOKEN123.mp4"
124 provider._get_stream_url = AsyncMock(return_value=fake_url) # type: ignore[method-assign]
125
126 details = await provider.get_stream_details("5034644330297-01_01", MediaType.TRACK)
127
128 assert details.item_id == "5034644330297-01_01"
129 assert details.stream_type == StreamType.HTTP
130 assert details.audio_format.content_type == ContentType.MP4
131 assert details.audio_format.sample_rate == 44100
132 assert details.audio_format.channels == 2
133 assert details.path == fake_url
134 assert details.can_seek is True
135 provider._get_stream_url.assert_called_once_with("5034644330297-01_01")
136
137 @pytest.mark.asyncio
138 async def test_radio_resolves_track_first(self, provider: MusicMeProvider) -> None:
139 """Test that radio media type resolves to a track barcode before streaming."""
140 fake_url = f"{STREAM_BASE}/{PARTNER_ID}/TOKEN456.mp4"
141 provider._get_radio_track = AsyncMock(return_value="RESOLVED-01_01") # type: ignore[method-assign]
142 provider._get_stream_url = AsyncMock(return_value=fake_url) # type: ignore[method-assign]
143
144 details = await provider.get_stream_details("rd-1163935", MediaType.RADIO)
145
146 provider._get_radio_track.assert_called_once_with("rd-1163935")
147 provider._get_stream_url.assert_called_once_with("RESOLVED-01_01")
148 assert details.item_id == "rd-1163935"
149 assert details.path == fake_url
150