/
/
1"""Fixtures for Tidal provider tests."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator, Generator
6from typing import TYPE_CHECKING, Any
7from unittest.mock import AsyncMock, MagicMock, Mock, patch
8
9import pytest
10from music_assistant_models.media_items import ItemMapping
11
12from music_assistant.providers.tidal.media import TidalMediaManager
13
14if TYPE_CHECKING:
15 from music_assistant_models.enums import MediaType
16
17
18@pytest.fixture
19def provider_mock() -> Mock:
20 """Return a mock Tidal provider with an authenticated session and wired collaborators."""
21 provider = Mock()
22 provider.domain = "tidal"
23 provider.instance_id = "tidal_instance"
24
25 provider.auth.user_id = "12345"
26 provider.auth.country_code = "US"
27 provider.auth.access_token = "token"
28 provider.auth.session_id = "session"
29 provider.auth.user.profile_name = "Test User"
30 provider.auth.user.user_name = "Test User"
31 provider.auth.ensure_valid_token = AsyncMock(return_value=True)
32 provider.auth.refresh_token = AsyncMock()
33
34 provider.api = AsyncMock()
35 provider.api.get.return_value = {}
36 # paginate is an async generator, so it cannot be an AsyncMock child: it yields the
37 # items a test assigns to paginate.return_value
38 provider.api.paginate = MagicMock()
39
40 async def paginate(*_args: Any, **_kwargs: Any) -> AsyncGenerator[Any]:
41 for item in provider.api.paginate.return_value:
42 yield item
43
44 provider.api.paginate.side_effect = paginate
45 provider.api.paginate.return_value = []
46
47 provider.get_track = AsyncMock()
48
49 def get_item_mapping(media_type: MediaType, key: str, name: str) -> ItemMapping:
50 return ItemMapping(
51 media_type=media_type,
52 item_id=key,
53 provider=provider.instance_id,
54 name=name,
55 )
56
57 provider.get_item_mapping.side_effect = get_item_mapping
58
59 provider.mass.http_session = AsyncMock()
60 provider.mass.metadata.locale = "en_US"
61 provider.mass.config.get_provider_configs = AsyncMock(return_value=[])
62 provider.mass.cache.get = AsyncMock(return_value=None)
63 provider.mass.cache.set = AsyncMock()
64 provider.mass.cache.delete = AsyncMock()
65 provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None)
66
67 return provider
68
69
70@pytest.fixture
71def media_manager(provider_mock: Mock) -> TidalMediaManager:
72 """Return a TidalMediaManager instance."""
73 return TidalMediaManager(provider_mock)
74
75
76@pytest.fixture(autouse=True)
77def no_throttling() -> Generator[None]:
78 """
79 Disable rate limiting and retry backoff during tests.
80
81 The API client's throttler is class-level shared state: its real-time
82 rate window would otherwise carry over between tests and make every
83 test wait it out.
84
85 Note: the sleep patch targets the attribute on the shared asyncio
86 module, so asyncio.sleep is mocked process-wide while each test in
87 this directory runs. Keep that in mind for timing-dependent tests.
88 """
89 with (
90 patch(
91 "music_assistant.helpers.throttle_retry.Throttler.acquire",
92 new=AsyncMock(return_value=0.0),
93 ),
94 patch(
95 "music_assistant.helpers.throttle_retry.asyncio.sleep",
96 new_callable=AsyncMock,
97 ),
98 ):
99 yield
100