/
/
/
1"""Test Nugs.net stream url building for regular, promo/trial and inactive subscriptions."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.errors import AudioError
10
11from music_assistant.providers.nugs import NugsProvider
12
13USER_DATA = {"userId": "user-1"}
14SUBSCRIPTION_BASE = {
15 "startedAt": "08/01/2026 00:00:00",
16 "endsAt": "09/01/2026 00:00:00",
17 "legacySubscriptionId": "legacy-1",
18}
19
20
21def _stub_get_data(provider: NugsProvider, subscription: dict[str, Any]) -> None:
22 """Attach a _get_data stub returning the given subscription payload."""
23
24 async def _fake(nugs_api: str, _endpoint: str, **_kwargs: Any) -> Any:
25 if nugs_api == "subscription":
26 return subscription
27 return USER_DATA
28
29 provider._get_data = AsyncMock(side_effect=_fake) # type: ignore[method-assign]
30
31
32def _stub_http_session(provider: NugsProvider) -> MagicMock:
33 """Stub the http session so the subplayer call returns a fixed stream link."""
34 response = AsyncMock()
35 response.raise_for_status = lambda: None
36 response.text = AsyncMock(return_value='{"streamLink": "https://stream.test/track.m3u8"}')
37 get_ctx = AsyncMock()
38 get_ctx.__aenter__ = AsyncMock(return_value=response)
39 get_ctx.__aexit__ = AsyncMock(return_value=False)
40 session_get = MagicMock(return_value=get_ctx)
41 mass: Any = provider.mass
42 mass.http_session.get = session_get
43 return session_get
44
45
46@pytest.mark.asyncio
47async def test_stream_url_uses_regular_plan(provider: NugsProvider) -> None:
48 """A regular subscription passes its own plan id as the cost plan access list."""
49 _stub_get_data(provider, {**SUBSCRIPTION_BASE, "plan": {"id": "plan-42"}})
50 session_get = _stub_http_session(provider)
51
52 assert await provider._get_stream_url("123") == "https://stream.test/track.m3u8"
53 assert session_get.call_args.kwargs["params"]["subCostplanIDAccessList"] == "plan-42"
54
55
56@pytest.mark.asyncio
57async def test_stream_url_falls_back_to_promo_plan(provider: NugsProvider) -> None:
58 """A trial/promo subscription has plan set to None and its plan id under promo."""
59 _stub_get_data(
60 provider,
61 {**SUBSCRIPTION_BASE, "plan": None, "promo": {"plan": {"id": "promo-7"}}},
62 )
63 session_get = _stub_http_session(provider)
64
65 assert await provider._get_stream_url("123") == "https://stream.test/track.m3u8"
66 assert session_get.call_args.kwargs["params"]["subCostplanIDAccessList"] == "promo-7"
67
68
69@pytest.mark.asyncio
70async def test_stream_url_without_any_plan_raises(provider: NugsProvider) -> None:
71 """Without a regular or promo plan a clear audio error is raised."""
72 _stub_get_data(provider, {**SUBSCRIPTION_BASE, "plan": None, "promo": None})
73
74 with pytest.raises(AudioError):
75 await provider._get_stream_url("123")
76