/
/
/
1"""Tests for Emby library listing validation."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator, Callable
6from typing import Any
7from unittest.mock import AsyncMock, Mock
8
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import InvalidDataError
12
13from music_assistant.providers.emby import SUPPORTED_FEATURES, EmbyProvider
14from music_assistant.providers.emby.const import ITEM_KEY_ID, ITEM_KEY_MEDIA_STREAMS, ITEMS
15
16
17@pytest.fixture
18def provider() -> EmbyProvider:
19 """Create an Emby provider with mocked dependencies."""
20 mass = Mock()
21 manifest = Mock()
22 manifest.domain = "emby"
23 config = Mock()
24 config.instance_id = "emby--test123"
25 config.name = "Emby Test"
26 config.enabled = True
27 config.get_value.return_value = "GLOBAL"
28 result = EmbyProvider(mass, manifest, config, SUPPORTED_FEATURES)
29 result._user_id = "user"
30 return result
31
32
33@pytest.mark.parametrize(
34 "method_name",
35 [
36 "get_library_artists",
37 "get_library_albums",
38 "get_library_tracks",
39 "get_library_playlists",
40 ],
41)
42@pytest.mark.parametrize("response", [{}, {ITEMS: None}, {ITEMS: {}}])
43async def test_malformed_item_list_raises(
44 provider: EmbyProvider,
45 method_name: str,
46 response: dict[str, Any],
47) -> None:
48 """A missing or wrong-typed Items field must abort the library sync."""
49 provider._get_music_libraries = AsyncMock( # type: ignore[method-assign]
50 return_value=[{ITEM_KEY_ID: "library"}]
51 )
52 provider._get = AsyncMock(return_value=response) # type: ignore[method-assign]
53
54 listing: Callable[[], AsyncGenerator[Any]] = getattr(provider, method_name)
55 with pytest.raises(InvalidDataError):
56 _ = [item async for item in listing()]
57
58
59async def test_empty_item_list_is_a_valid_empty_library(provider: EmbyProvider) -> None:
60 """An explicit empty Items list remains a valid empty library."""
61 provider._get_music_libraries = AsyncMock( # type: ignore[method-assign]
62 return_value=[{ITEM_KEY_ID: "library"}]
63 )
64 provider._get = AsyncMock(return_value={ITEMS: []}) # type: ignore[method-assign]
65
66 assert [track async for track in provider.get_library_tracks()] == []
67
68
69@pytest.mark.parametrize(
70 ("method_name", "media_type"),
71 [
72 ("get_library_artists", MediaType.ARTIST),
73 ("get_library_albums", MediaType.ALBUM),
74 ("get_library_tracks", MediaType.TRACK),
75 ("get_library_playlists", MediaType.PLAYLIST),
76 ],
77)
78@pytest.mark.parametrize("entry", [None, {}, {ITEM_KEY_ID: {}}])
79async def test_non_object_entry_is_reported(
80 provider: EmbyProvider,
81 method_name: str,
82 media_type: MediaType,
83 entry: Any,
84) -> None:
85 """An unidentifiable malformed entry must hold back deletions for its listing."""
86 provider._get_music_libraries = AsyncMock( # type: ignore[method-assign]
87 return_value=[{ITEM_KEY_ID: "library"}]
88 )
89 provider._get = AsyncMock( # type: ignore[method-assign]
90 side_effect=[{ITEMS: [entry]}, {ITEMS: []}]
91 )
92 provider.report_skipped_sync_item = Mock() # type: ignore[method-assign]
93
94 listing: Callable[[], AsyncGenerator[Any]] = getattr(provider, method_name)
95 assert [item async for item in listing()] == []
96
97 provider.report_skipped_sync_item.assert_called_once()
98 args = provider.report_skipped_sync_item.call_args.args
99 assert args[:2] == (media_type, None)
100 assert isinstance(args[2], InvalidDataError)
101
102
103@pytest.mark.parametrize("media_streams", [None, {}, []])
104async def test_streamless_track_is_reported(provider: EmbyProvider, media_streams: Any) -> None:
105 """A track without media streams must be reported instead of silently disappearing."""
106 provider._get_music_libraries = AsyncMock( # type: ignore[method-assign]
107 return_value=[{ITEM_KEY_ID: "library"}]
108 )
109 provider._get = AsyncMock( # type: ignore[method-assign]
110 side_effect=[
111 {
112 ITEMS: [
113 {
114 ITEM_KEY_ID: "track-1",
115 ITEM_KEY_MEDIA_STREAMS: media_streams,
116 }
117 ]
118 },
119 {ITEMS: []},
120 ]
121 )
122 provider.report_skipped_sync_item = Mock() # type: ignore[method-assign]
123
124 assert [track async for track in provider.get_library_tracks()] == []
125
126 args = provider.report_skipped_sync_item.call_args.args
127 assert args[:2] == (MediaType.TRACK, "track-1")
128 assert isinstance(args[2], InvalidDataError)
129
130
131async def test_malformed_music_library_list_raises(provider: EmbyProvider) -> None:
132 """A malformed media-folder response must not make every listing look empty."""
133 provider._get = AsyncMock(return_value={}) # type: ignore[method-assign]
134
135 with pytest.raises(InvalidDataError):
136 await provider._get_music_libraries()
137
138
139async def test_malformed_music_library_entry_raises(provider: EmbyProvider) -> None:
140 """A non-object media folder must abort every listing that depends on it."""
141 provider._get = AsyncMock(return_value={ITEMS: [None]}) # type: ignore[method-assign]
142
143 with pytest.raises(InvalidDataError):
144 await provider._get_music_libraries()
145