/
/
/
1"""Tests for malformed NetEase library listing responses."""
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.neteasecloudmusic import NeteaseCloudMusicProvider
14
15
16@pytest.mark.parametrize(
17 ("method_name", "response"),
18 [
19 ("get_library_artists", {"data": {"unexpected": []}}),
20 ("get_library_albums", {"data": {"unexpected": []}}),
21 ("get_library_tracks", {"unexpected": []}),
22 ("get_library_playlists", {"unexpected": []}),
23 ],
24)
25async def test_malformed_listing_field_raises(
26 provider: NeteaseCloudMusicProvider,
27 method_name: str,
28 response: dict[str, Any],
29) -> None:
30 """A missing item list must abort the sync instead of looking empty."""
31 provider._client = AsyncMock()
32 provider._client.get.return_value = response
33
34 listing: Callable[[], AsyncGenerator[Any]] = getattr(provider, method_name)
35 with pytest.raises(InvalidDataError):
36 _ = [item async for item in listing()]
37
38
39@pytest.mark.parametrize(
40 ("method_name", "response", "media_type"),
41 [
42 ("get_library_artists", {"artists": [None]}, MediaType.ARTIST),
43 ("get_library_albums", {"albums": [None]}, MediaType.ALBUM),
44 ("get_library_playlists", {"playlist": [None]}, MediaType.PLAYLIST),
45 ("get_library_artists", {"artists": [{"id": {}}]}, MediaType.ARTIST),
46 ("get_library_albums", {"albums": [{"id": {}}]}, MediaType.ALBUM),
47 ("get_library_playlists", {"playlist": [{"id": {}}]}, MediaType.PLAYLIST),
48 ],
49)
50async def test_malformed_listing_entry_is_reported(
51 provider: NeteaseCloudMusicProvider,
52 method_name: str,
53 response: dict[str, Any],
54 media_type: MediaType,
55) -> None:
56 """An unidentifiable malformed entry must hold back deletions for the listing."""
57 provider._client = AsyncMock()
58 provider._client.get.return_value = response
59 provider.report_skipped_sync_item = Mock() # type: ignore[method-assign]
60
61 listing: Callable[[], AsyncGenerator[Any]] = getattr(provider, method_name)
62 assert [item async for item in listing()] == []
63
64 provider.report_skipped_sync_item.assert_called_once()
65 args = provider.report_skipped_sync_item.call_args.args
66 assert args[:2] == (media_type, None)
67 assert isinstance(args[2], InvalidDataError)
68
69
70async def test_malformed_liked_track_id_is_reported(
71 provider: NeteaseCloudMusicProvider,
72) -> None:
73 """An invalid liked-track id must not silently disappear from the authoritative list."""
74 provider._client = AsyncMock()
75 provider._client.get.return_value = {"ids": [None]}
76 provider.report_skipped_sync_item = Mock() # type: ignore[method-assign]
77
78 assert [track async for track in provider.get_library_tracks()] == []
79
80 provider.report_skipped_sync_item.assert_called_once()
81 args = provider.report_skipped_sync_item.call_args.args
82 assert args[:2] == (MediaType.TRACK, None)
83 assert isinstance(args[2], InvalidDataError)
84