music-assistant-server

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