/
/
/
1"""Fixtures for Tidal provider tests."""
2
3from __future__ import annotations
4
5from collections.abc import Generator
6from typing import TYPE_CHECKING
7from unittest.mock import AsyncMock, 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
37 provider.get_track = AsyncMock()
38 # Churn-healing collaborators: the cache-only redirect defaults to identity
39 # (id is not known-stale) and the reactive resolver to unresolvable.
40 provider.redirect_cached_id = AsyncMock(side_effect=lambda item_id: item_id)
41 provider.resolve_live_track_id = AsyncMock(return_value=None)
42
43 def get_item_mapping(media_type: MediaType, key: str, name: str) -> ItemMapping:
44 return ItemMapping(
45 media_type=media_type,
46 item_id=key,
47 provider=provider.instance_id,
48 name=name,
49 )
50
51 provider.get_item_mapping.side_effect = get_item_mapping
52
53 provider.mass.http_session = AsyncMock()
54 provider.mass.metadata.locale = "en_US"
55 provider.mass.config.get_provider_configs = AsyncMock(return_value=[])
56 provider.mass.cache.get = AsyncMock(return_value=None)
57 provider.mass.cache.set = AsyncMock()
58 provider.mass.cache.delete = AsyncMock()
59 provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None)
60
61 return provider
62
63
64@pytest.fixture
65def media_manager(provider_mock: Mock) -> TidalMediaManager:
66 """Return a TidalMediaManager instance."""
67 return TidalMediaManager(provider_mock)
68
69
70@pytest.fixture(autouse=True)
71def no_throttling() -> Generator[None]:
72 """
73 Disable rate limiting and retry backoff during tests.
74
75 The API client's throttler is class-level shared state: its real-time
76 rate window would otherwise carry over between tests and make every
77 test wait it out.
78
79 Note: the sleep patch targets the attribute on the shared asyncio
80 module, so asyncio.sleep is mocked process-wide while each test in
81 this directory runs. Keep that in mind for timing-dependent tests.
82 """
83 with (
84 patch(
85 "music_assistant.helpers.throttle_retry.Throttler.acquire",
86 new=AsyncMock(return_value=0.0),
87 ),
88 patch(
89 "music_assistant.helpers.throttle_retry.asyncio.sleep",
90 new_callable=AsyncMock,
91 ),
92 ):
93 yield
94