/
/
/
1"""Tests for Podcast Index library operations."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import cast
7from unittest.mock import MagicMock, patch
8
9from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, ProviderConfig
10from music_assistant_models.enums import ConfigEntryType, MediaType, ProviderType
11from music_assistant_models.media_items import Podcast
12
13from music_assistant.providers.podcast_index.constants import CONF_STORED_PODCASTS
14from music_assistant.providers.podcast_index.provider import PodcastIndexProvider
15
16
17async def test_concurrent_add_and_remove_preserve_both_updates() -> None:
18 """Concurrent library changes retain every intended stored-podcast update."""
19 removed_feed = "https://example.com/removed.xml"
20 added_feed = "https://example.com/added.xml"
21 provider, config_store = _provider([removed_feed])
22 lookup_barrier = _LookupBarrier(
23 {"remove": removed_feed, "add": added_feed},
24 operation_count=2,
25 )
26 with patch.object(
27 PodcastIndexProvider,
28 "_get_feed_url_for_podcast",
29 side_effect=lookup_barrier.lookup,
30 ):
31 remove_task = asyncio.create_task(provider.library_remove("remove", MediaType.PODCAST))
32 add_task = asyncio.create_task(provider.library_add(_podcast("add")))
33
34 await lookup_barrier.wait_until_blocked()
35 lookup_barrier.release()
36
37 remove_result, add_result = await asyncio.gather(remove_task, add_task)
38 assert remove_result is True
39 assert add_result is True
40 assert config_store.stored_podcasts == [added_feed]
41
42
43class _ConfigStore:
44 """Store Podcast Index configuration values for provider tests."""
45
46 def __init__(self, stored_podcasts: list[str]) -> None:
47 self.stored_podcasts = list(stored_podcasts)
48
49 def get_raw_provider_config_value(
50 self, provider_instance: str, key: str, default: ConfigValueType = None
51 ) -> ConfigValueType:
52 """Return the current stored configuration value."""
53 assert provider_instance == "podcast_index--test"
54 assert key == CONF_STORED_PODCASTS
55 return list(self.stored_podcasts)
56
57 def set_raw_provider_config_value(
58 self,
59 provider_instance: str,
60 key: str,
61 value: ConfigValueType,
62 encrypted: bool = False,
63 immediate: bool = False,
64 ) -> None:
65 """Persist a provider configuration value."""
66 assert provider_instance == "podcast_index--test"
67 assert key == CONF_STORED_PODCASTS
68 assert isinstance(value, list)
69 assert all(isinstance(item, str) for item in value)
70 self.stored_podcasts = cast("list[str]", value)
71
72
73class _LookupBarrier:
74 """Block feed lookups until every concurrent operation has reached the lookup."""
75
76 def __init__(self, feed_urls: dict[str, str], operation_count: int) -> None:
77 self._feed_urls = feed_urls
78 self._operation_count = operation_count
79 self._blocked_count = 0
80 self._all_blocked = asyncio.Event()
81 self._release = asyncio.Event()
82
83 async def lookup(self, podcast_id: str) -> str:
84 """Return a feed URL after all expected lookups are blocked."""
85 self._blocked_count += 1
86 if self._blocked_count == self._operation_count:
87 self._all_blocked.set()
88 await self._release.wait()
89 return self._feed_urls[podcast_id]
90
91 async def wait_until_blocked(self) -> None:
92 """Wait until all expected operations are blocked in their feed lookup."""
93 await asyncio.wait_for(self._all_blocked.wait(), timeout=1)
94
95 def release(self) -> None:
96 """Release every blocked feed lookup."""
97 self._release.set()
98
99
100def _provider(stored_podcasts: list[str]) -> tuple[PodcastIndexProvider, _ConfigStore]:
101 """Create a Podcast Index provider backed by an in-memory configuration store."""
102 config_store = _ConfigStore(stored_podcasts)
103 config = ProviderConfig(
104 values={
105 CONF_STORED_PODCASTS: ConfigEntry(
106 key=CONF_STORED_PODCASTS,
107 type=ConfigEntryType.STRING,
108 multi_value=True,
109 value=list(stored_podcasts),
110 )
111 },
112 type=ProviderType.MUSIC,
113 domain="podcast_index",
114 instance_id="podcast_index--test",
115 )
116 provider = object.__new__(PodcastIndexProvider)
117 provider.mass = MagicMock()
118 provider.mass.config = config_store
119 provider.config = config
120 provider.logger = MagicMock()
121 return provider, config_store
122
123
124def _podcast(item_id: str) -> Podcast:
125 """Create a Podcast Index podcast for library tests."""
126 return Podcast(
127 item_id=item_id,
128 provider="podcast_index--test",
129 name=f"Podcast {item_id}",
130 provider_mappings=set(),
131 )
132