/
/
/
1"""Shared fixtures and stubs for Yandex Music provider tests."""
2
3from __future__ import annotations
4
5import asyncio
6import importlib
7import importlib.util
8import inspect
9import logging
10import sys
11from pathlib import Path
12from types import MethodType
13from typing import Any
14from unittest.mock import MagicMock
15
16import pytest
17from music_assistant_models.enums import MediaType
18from music_assistant_models.media_items import ItemMapping
19
20from music_assistant.mass import MusicAssistant
21
22_PROVIDER_PKG = "music_assistant.providers.yandex_music"
23
24
25def _alias_working_tree_provider(provider_dir: Path) -> None:
26 """
27 Alias the ``provider`` working-tree package onto the upstream import path.
28
29 In the provider-repo layout, tests must exercise the working tree, not the
30 provider snapshot baked into the venv's music_assistant install. In the
31 upstream (inlined) layout there is no sibling ``provider/`` directory and
32 the package under test IS the checkout itself, so the aliasing must no-op
33 instead of failing collection.
34
35 :param provider_dir: Path to the ``provider`` working-tree directory.
36 """
37 provider_dir = provider_dir.resolve()
38 if not provider_dir.is_dir():
39 return
40 existing = sys.modules.get(_PROVIDER_PKG)
41 if existing is not None:
42 # Something imported the provider before this conftest ran â silently
43 # testing the venv snapshot instead of the working tree must be fatal.
44 loaded_from = Path(getattr(existing, "__file__", "") or "").resolve().parent
45 if loaded_from != provider_dir:
46 raise RuntimeError(
47 f"{_PROVIDER_PKG} was already imported from {loaded_from}; "
48 f"tests must run against {provider_dir}"
49 )
50 return
51 spec = importlib.util.spec_from_file_location(
52 _PROVIDER_PKG,
53 provider_dir / "__init__.py",
54 submodule_search_locations=[str(provider_dir)],
55 )
56 if spec is None or spec.loader is None:
57 raise ImportError(f"cannot load provider package from {provider_dir}")
58 module = importlib.util.module_from_spec(spec)
59 sys.modules[_PROVIDER_PKG] = module
60 try:
61 spec.loader.exec_module(module)
62 except BaseException:
63 # Mirror the import machinery: a failed exec must not leave a
64 # half-initialized module registered under the package name.
65 del sys.modules[_PROVIDER_PKG]
66 raise
67 # Regular imports also bind the submodule as an attribute of its parent
68 # package; monkeypatch and friends resolve dotted paths via getattr.
69 parent = importlib.import_module("music_assistant.providers")
70 setattr(parent, "yandex_music", module) # noqa: B010
71
72
73# The assignment + is_dir() shape (not a bare call argument) keeps this
74# dereference visible to the rewrite-safe Rule C gate in CI.
75_PROVIDER_DIR = Path(__file__).resolve().parent.parent / "provider"
76if _PROVIDER_DIR.is_dir():
77 _alias_working_tree_provider(_PROVIDER_DIR)
78
79
80def provider_dir() -> Path:
81 """Directory of the provider package under test, in either layout."""
82 pkg = importlib.import_module(_PROVIDER_PKG)
83 pkg_file = pkg.__file__
84 assert pkg_file is not None # a real package always has a file
85 return Path(pkg_file).resolve().parent
86
87
88def use_real_create_task(mass: MagicMock | MusicAssistant) -> None:
89 """
90 Give a mocked Music Assistant the real task-creation implementation.
91
92 :param mass: The mock standing in for the Music Assistant instance.
93 """
94 mass._tracked_tasks = {}
95 mass.verify_event_loop_thread = MagicMock() # type: ignore[method-assign]
96 real_create_task = MethodType(MusicAssistant.create_task, mass)
97
98 def _create_task(target: Any, *args: Any, **kwargs: Any) -> Any:
99 if not (inspect.iscoroutine(target) or inspect.iscoroutinefunction(target)):
100 return MagicMock()
101 mass.loop = asyncio.get_running_loop()
102 return real_create_task(target, *args, **kwargs)
103
104 mass.create_task = MagicMock(side_effect=_create_task) # type: ignore[method-assign]
105
106
107class ProviderStub:
108 """
109 Minimal provider-like object for parser tests (no Mock).
110
111 Provides the minimal interface needed by parse_* functions.
112 """
113
114 domain = "yandex_music"
115 instance_id = "yandex_music_instance"
116
117 def __init__(self) -> None:
118 """Initialize stub with minimal client."""
119 self.client = type("ClientStub", (), {"user_id": 12345})()
120
121 def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping:
122 """Return ItemMapping for the given media type, key and name."""
123 return ItemMapping(
124 media_type=MediaType(media_type) if isinstance(media_type, str) else media_type,
125 item_id=key,
126 provider=self.instance_id,
127 name=name,
128 )
129
130
131class ConfigStub:
132 """Minimal config stub for provider tests."""
133
134 def __init__(self, values: dict[str, object] | None = None) -> None:
135 """Initialize with optional config values."""
136 self._values = values or {}
137
138 def get_value(self, key: str, default: object = None) -> object:
139 """Return config value or default."""
140 return self._values.get(key, default)
141
142
143class StreamingProviderStub:
144 """
145 Minimal provider stub for streaming tests (no Mock).
146
147 Provides the minimal interface needed by YandexMusicStreamingManager.
148 """
149
150 domain = "yandex_music"
151 instance_id = "yandex_music_instance"
152 logger = logging.getLogger("yandex_music_test_streaming")
153
154 def __init__(self) -> None:
155 """Initialize stub with minimal client."""
156 self.client = type("ClientStub", (), {"user_id": 12345})()
157 self.mass = type("MassStub", (), {})()
158 self.config = ConfigStub()
159 self._warning_count = 0
160
161 def _count_warning(self, *args: object, **kwargs: object) -> None:
162 """Track warning calls for test assertions."""
163 self._warning_count += 1
164
165
166class TrackingLogger:
167 """Logger that tracks calls for test assertions without using Mock."""
168
169 def __init__(self) -> None:
170 """Initialize with empty call counters."""
171 self._debug_count = 0
172 self._info_count = 0
173 self._warning_count = 0
174 self._error_count = 0
175
176 def debug(self, *args: object, **kwargs: object) -> None:
177 """Track debug calls."""
178 self._debug_count += 1
179
180 def info(self, *args: object, **kwargs: object) -> None:
181 """Track info calls."""
182 self._info_count += 1
183
184 def warning(self, *args: object, **kwargs: object) -> None:
185 """Track warning calls."""
186 self._warning_count += 1
187
188 def error(self, *args: object, **kwargs: object) -> None:
189 """Track error calls."""
190 self._error_count += 1
191
192
193class StreamingProviderStubWithTracking:
194 """
195 Provider stub with tracking logger for assertions.
196
197 Use this when you need to verify logging behavior.
198 """
199
200 domain = "yandex_music"
201 instance_id = "yandex_music_instance"
202
203 def __init__(self) -> None:
204 """Initialize stub with tracking logger."""
205 self.client = type("ClientStub", (), {"user_id": 12345})()
206 self.mass = type("MassStub", (), {})()
207 self.config = ConfigStub()
208 self.logger = TrackingLogger()
209
210
211# Minimal client-like object for yandex_music de_json (library requires client, not None)
212DE_JSON_CLIENT = type("ClientStub", (), {"report_unknown_fields": False})()
213
214
215@pytest.fixture
216def provider_stub() -> ProviderStub:
217 """Return a real provider stub (no Mock)."""
218 return ProviderStub()
219
220
221@pytest.fixture
222def streaming_provider_stub() -> StreamingProviderStub:
223 """Return a streaming provider stub (no Mock)."""
224 return StreamingProviderStub()
225
226
227@pytest.fixture
228def streaming_provider_stub_with_tracking() -> StreamingProviderStubWithTracking:
229 """Return a streaming provider stub with tracking logger."""
230 return StreamingProviderStubWithTracking()
231