/
/
1"""Tests for the tracks controller explicit filter."""
2
3from collections.abc import AsyncGenerator
4from typing import Any
5from unittest.mock import patch
6
7import pytest
8
9from music_assistant.controllers.music import MusicController
10from music_assistant.mass import MusicAssistant
11
12
13@pytest.fixture
14async def music(mass_minimal: MusicAssistant) -> AsyncGenerator[MusicController]:
15 """Return a music controller attached to the minimal mass instance."""
16 controller = MusicController(mass_minimal)
17 mass_minimal.music = controller
18 yield controller
19 if controller._database:
20 await controller._database.close()
21
22
23@pytest.mark.asyncio
24async def test_explicit_filter_true_generates_sql(music: MusicController) -> None:
25 """Test that explicit=True generates correct SQL filter."""
26 captured_parts: list[str] = []
27
28 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
29 captured_parts.extend(kwargs.get("extra_query_parts", []))
30 return []
31
32 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
33 await music.tracks.library_items(explicit=True, limit=10)
34 assert any(
35 "json_extract(tracks.metadata, '$.explicit') = 1" in part for part in captured_parts
36 )
37
38
39@pytest.mark.asyncio
40async def test_explicit_filter_false_generates_sql(music: MusicController) -> None:
41 """Test that explicit=False generates correct SQL filter."""
42 captured_parts: list[str] = []
43
44 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
45 captured_parts.extend(kwargs.get("extra_query_parts", []))
46 return []
47
48 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
49 await music.tracks.library_items(explicit=False, limit=10)
50 assert any("IS NULL" in part and "= 0" in part for part in captured_parts)
51
52
53@pytest.mark.asyncio
54async def test_explicit_filter_none_generates_no_sql(music: MusicController) -> None:
55 """Test that explicit=None generates no explicit filter."""
56 captured_parts: list[str] = []
57
58 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
59 captured_parts.extend(kwargs.get("extra_query_parts", []))
60 return []
61
62 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
63 await music.tracks.library_items(explicit=None, limit=10)
64 assert not any("explicit" in part.lower() for part in captured_parts)
65
66
67@pytest.mark.asyncio
68async def test_explicit_filter_default_is_none(music: MusicController) -> None:
69 """Test that omitting explicit parameter behaves like explicit=None."""
70 captured_parts: list[str] = []
71
72 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
73 captured_parts.extend(kwargs.get("extra_query_parts", []))
74 return []
75
76 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
77 await music.tracks.library_items(limit=10)
78 assert not any("explicit" in part.lower() for part in captured_parts)
79
80
81@pytest.mark.asyncio
82async def test_by_prov_id_batches_item_ids_into_in_clause(music: MusicController) -> None:
83 """provider_item_ids builds a single parameterized IN (...) subquery."""
84 captured_parts: list[str] = []
85 captured_params: dict[str, Any] = {}
86
87 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
88 captured_parts.extend(kwargs.get("extra_query_parts", []))
89 captured_params.update(kwargs.get("extra_query_params", {}))
90 return []
91
92 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
93 await music.tracks.get_library_items_by_prov_id(
94 provider_instance_id_or_domain="spotify",
95 provider_item_ids=["x", "y", "z"],
96 )
97
98 subquery = " ".join(captured_parts)
99 assert "provider_mappings.provider_item_id IN (:item_id_0, :item_id_1, :item_id_2)" in subquery
100 assert captured_params["prov_id"] == "spotify"
101 assert [captured_params[f"item_id_{i}"] for i in range(3)] == ["x", "y", "z"]
102
103
104@pytest.mark.asyncio
105async def test_by_prov_id_empty_item_ids_matches_nothing(music: MusicController) -> None:
106 """An explicit empty provider_item_ids returns [] (not the whole provider library)."""
107 ran = False
108
109 async def mock_query(*_args: Any, **_kwargs: Any) -> list[Any]:
110 nonlocal ran
111 ran = True
112 return []
113
114 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
115 result = await music.tracks.get_library_items_by_prov_id(
116 provider_instance_id_or_domain="spotify", provider_item_ids=[]
117 )
118
119 assert result == []
120 assert ran is False # short-circuits before it can build an unconstrained query
121