/
/
/
1"""Tests for what a failing Podcast Index call reports back."""
2
3from __future__ import annotations
4
5import logging
6from typing import Any
7from unittest.mock import MagicMock, patch
8
9import aiohttp
10import pytest
11from music_assistant_models.errors import (
12 InvalidDataError,
13 LoginFailed,
14 ProviderUnavailableError,
15)
16
17from music_assistant.providers.podcast_index.constants import MAX_ERROR_DETAIL_LENGTH
18from music_assistant.providers.podcast_index.helpers import make_api_request
19from music_assistant.providers.podcast_index.provider import PodcastIndexProvider
20
21API_KEY = "key"
22API_SECRET = "secret"
23
24
25class _FakeResponse:
26 def __init__(self, status: int, body: str = "", payload: Any = None) -> None:
27 self.status = status
28 self._body = body
29 self._payload = payload
30
31 async def text(self) -> str:
32 return self._body
33
34 async def json(self) -> Any:
35 if self._payload is None:
36 raise aiohttp.ContentTypeError(MagicMock(), ())
37 return self._payload
38
39
40class _FakeRequestContext:
41 def __init__(self, response: _FakeResponse) -> None:
42 self._response = response
43
44 async def __aenter__(self) -> _FakeResponse:
45 return self._response
46
47 async def __aexit__(self, *exc_info: object) -> bool:
48 return False
49
50
51def _mass(response: _FakeResponse) -> MagicMock:
52 """Return a Music Assistant stub whose http session answers with the given response."""
53 mass = MagicMock()
54 mass.http_session.get = MagicMock(return_value=_FakeRequestContext(response))
55 return mass
56
57
58async def _request(response: _FakeResponse, logger: logging.Logger | None = None) -> Any:
59 return await make_api_request(
60 _mass(response), API_KEY, API_SECRET, "stats/current", logger=logger
61 )
62
63
64async def test_rejected_credentials_quote_the_api() -> None:
65 """A refused key reports what Podcast Index said about it, not just the status."""
66 response = _FakeResponse(401, body="Invalid authorization header")
67
68 with pytest.raises(LoginFailed, match="Invalid authorization header") as err:
69 await _request(response)
70
71 assert "401" in str(err.value)
72
73
74async def test_a_failure_without_a_body_still_reports_the_status() -> None:
75 """An error that says nothing must not leave a dangling separator behind."""
76 response = _FakeResponse(500, body=" ")
77
78 with pytest.raises(ProviderUnavailableError) as err:
79 await _request(response)
80
81 assert str(err.value) == "API request failed (HTTP 500)"
82
83
84async def test_a_long_error_is_cut_to_a_readable_length() -> None:
85 """A page of markup is quoted only as far as it stays readable."""
86 response = _FakeResponse(403, body="x" * (MAX_ERROR_DETAIL_LENGTH * 2))
87
88 with pytest.raises(ProviderUnavailableError) as err:
89 await _request(response)
90
91 assert str(err.value).endswith("...")
92 assert len(str(err.value)) < MAX_ERROR_DETAIL_LENGTH * 2
93
94
95async def test_a_refusal_carrying_a_reason_reports_it() -> None:
96 """An answer that reports failure in its payload surfaces that description."""
97 response = _FakeResponse(200, payload={"status": "false", "description": "no such feed"})
98
99 with pytest.raises(InvalidDataError, match="no such feed"):
100 await _request(response)
101
102
103async def test_a_failure_is_logged_for_support(caplog: pytest.LogCaptureFixture) -> None:
104 """A failing call records the endpoint, the status and the reason at debug level."""
105 logger = logging.getLogger("test.podcast_index")
106 response = _FakeResponse(401, body="Invalid authorization header")
107
108 with caplog.at_level(logging.DEBUG, logger=logger.name), pytest.raises(LoginFailed):
109 await _request(response, logger=logger)
110
111 assert "stats/current" in caplog.text
112 assert "401" in caplog.text
113 assert "Invalid authorization header" in caplog.text
114
115
116async def test_credentials_are_never_logged(caplog: pytest.LogCaptureFixture) -> None:
117 """The key and secret must stay out of anything a user is asked to share."""
118 logger = logging.getLogger("test.podcast_index")
119 # the live API names the header rather than the value, but the body is not ours to trust
120 response = _FakeResponse(401, body=f"key {API_KEY} with secret {API_SECRET} was refused")
121
122 with caplog.at_level(logging.DEBUG, logger=logger.name), pytest.raises(LoginFailed) as err:
123 await _request(response, logger=logger)
124
125 assert API_KEY not in caplog.text
126 assert API_SECRET not in caplog.text
127 assert API_KEY not in str(err.value)
128 assert API_SECRET not in str(err.value)
129
130
131async def test_a_successful_call_is_logged(caplog: pytest.LogCaptureFixture) -> None:
132 """A call that worked records that it did, so a working setup is recognisable."""
133 logger = logging.getLogger("test.podcast_index")
134 response = _FakeResponse(200, payload={"status": "true", "count": 3})
135
136 with caplog.at_level(logging.DEBUG, logger=logger.name):
137 data = await _request(response, logger=logger)
138
139 assert data["count"] == 3
140 assert "stats/current" in caplog.text
141
142
143async def test_a_single_item_call_is_not_reported_as_empty(
144 caplog: pytest.LogCaptureFixture,
145) -> None:
146 """Endpoints returning one item carry no count, which is not the same as returning none."""
147 logger = logging.getLogger("test.podcast_index")
148 response = _FakeResponse(200, payload={"status": "true", "episode": {"id": 1}})
149
150 with caplog.at_level(logging.DEBUG, logger=logger.name):
151 await _request(response, logger=logger)
152
153 assert "no items" not in caplog.text
154 assert "stats/current succeeded" in caplog.text
155
156
157def _browse_provider() -> PodcastIndexProvider:
158 """Create a provider whose API calls can be stubbed out."""
159 provider = object.__new__(PodcastIndexProvider)
160 provider.mass = MagicMock()
161 provider.logger = MagicMock()
162 return provider
163
164
165@pytest.mark.parametrize(
166 ("browse", "args"),
167 [
168 (PodcastIndexProvider._browse_trending, ()),
169 (PodcastIndexProvider._browse_category_podcasts, ("comedy",)),
170 ],
171)
172async def test_browsing_reports_rejected_credentials(browse: Any, args: tuple[Any, ...]) -> None:
173 """A rejected key must surface, not leave the shelf looking empty."""
174 provider = _browse_provider()
175 with (
176 patch.object(PodcastIndexProvider, "_api_request", side_effect=LoginFailed("key refused")),
177 patch.object(
178 PodcastIndexProvider, "_fetch_podcasts", side_effect=LoginFailed("key refused")
179 ),
180 pytest.raises(LoginFailed),
181 ):
182 await browse.__wrapped__(provider, *args)
183