music-assistant-server

1"""Tests for the mammamiradio music provider."""
2
3from __future__ import annotations
4
5import json
6import os
7from pathlib import Path
8from typing import Any
9from unittest.mock import AsyncMock, MagicMock, patch
10
11import aiohttp
12import pytest
13from multidict import CIMultiDict
14from music_assistant_models.enums import ContentType, MediaType, ProviderFeature
15from music_assistant_models.errors import (
16    MediaNotFoundError,
17    ProviderUnavailableError,
18    SetupFailedError,
19)
20from music_assistant_models.media_items import Radio, SearchResults
21
22from music_assistant.models.music_provider import MusicProvider
23from music_assistant.models.setup_flow import SetupFlowError
24from music_assistant.providers.mammamiradio import (
25    CONF_MAMMAMIRADIO_URL,
26    DEFAULT_URL,
27    RADIO_ITEM_ID,
28    RADIO_NAME,
29    STREAM_METADATA_UPDATE_INTERVAL,
30    SUPPORTED_FEATURES,
31    MammamiradioProvider,
32    _audio_format_from_contract,
33    _host_display_names,
34    _normalize_base_url,
35    _stream_path_from_contract,
36    _supports_v1_schema,
37    _v1_to_stream_metadata,
38    setup,
39)
40from music_assistant.providers.mammamiradio import setup_flow as mammamiradio_setup_flow
41
42
43def _make_response_ctx(status: int = 200) -> MagicMock:
44    """Build an async-context-manager mock that yields a response with `status`."""
45    response = MagicMock()
46    response.status = status
47    ctx = MagicMock()
48    ctx.__aenter__ = AsyncMock(return_value=response)
49    ctx.__aexit__ = AsyncMock(return_value=False)
50    return ctx
51
52
53def _make_failing_ctx(exc: Exception) -> MagicMock:
54    """Build an async-context-manager mock whose __aenter__ raises ``exc``."""
55    ctx = MagicMock()
56    ctx.__aenter__ = AsyncMock(side_effect=exc)
57    ctx.__aexit__ = AsyncMock(return_value=False)
58    return ctx
59
60
61def _make_v1_ctx(payload: Any, status: int = 200, etag: str | None = None) -> MagicMock:
62    """Async-context-manager mock for the v1 endpoint, with case-insensitive headers (ETag)."""
63    response = MagicMock()
64    response.status = status
65    response.json = AsyncMock(return_value=payload)
66    response.headers = CIMultiDict({"ETag": etag}) if etag else CIMultiDict()
67    ctx = MagicMock()
68    ctx.__aenter__ = AsyncMock(return_value=response)
69    ctx.__aexit__ = AsyncMock(return_value=False)
70    return ctx
71
72
73def _make_bad_json_ctx(status: int = 200, exc: Exception | None = None) -> MagicMock:
74    """Async-context-manager mock whose body is not JSON (``.json()`` raises)."""
75    response = MagicMock()
76    response.status = status
77    response.json = AsyncMock(side_effect=exc or ValueError("not json"))
78    response.headers = CIMultiDict()
79    ctx = MagicMock()
80    ctx.__aenter__ = AsyncMock(return_value=response)
81    ctx.__aexit__ = AsyncMock(return_value=False)
82    return ctx
83
84
85# A representative v1 now-playing response (music segment) reused across tests.
86_V1_AUDIO_FORMAT: dict[str, Any] = {
87    "codec": "mp3",
88    "mime_type": "audio/mpeg",
89    "bitrate_kbps": 192,
90    "sample_rate_hz": 48000,
91    "channels": 2,
92}
93_V1_MUSIC: dict[str, Any] = {
94    "schema_version": "1",
95    "station": {
96        "name": "mammamiradio",
97        "hosts": [
98            {"engine_host": "gianni", "display_name": "Gianni"},
99            {"engine_host": "lucia", "display_name": "Lucia"},
100        ],
101    },
102    "stream": {"relative_url": "/stream", "audio_format": _V1_AUDIO_FORMAT},
103    "now_playing": {
104        "segment_class": "music",
105        "segment_type": "music",
106        "title": "Volare",
107        "artist": "Modugno",
108        "artwork": "http://art/volare.jpg",
109        "album": "Best Of",
110    },
111    "up_next": [
112        {
113            "segment_class": "voice",
114            "segment_type": "banter",
115            "title": "Chiacchiere",
116            "predicted": False,
117        }
118    ],
119    "session_state": "live",
120    "changed_at": 100.0,
121}
122
123
124@pytest.fixture
125def mass_mock() -> MagicMock:
126    """Return a mock MusicAssistant instance with an http_session."""
127    mass = MagicMock()
128    mass.http_session = MagicMock()
129    mass.config = MagicMock()
130    mass.config.get.return_value = {CONF_MAMMAMIRADIO_URL: "http://localhost:8000"}
131    mass.config.decrypt_string.side_effect = lambda value: value
132    # default: every request answers with a healthy v1 now-playing payload
133    mass.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC))
134    return mass
135
136
137@pytest.fixture
138def provider(mass_mock: MagicMock) -> MammamiradioProvider:
139    """Return a configured MammamiradioProvider for unit testing."""
140    manifest = MagicMock()
141    manifest.domain = "mammamiradio"
142    manifest.name = "mammamiradio"
143
144    config = MagicMock()
145    config.instance_id = "mammamiradio_test"
146    config.values = {}
147
148    def _get_value(key: str, default: Any = None) -> Any:
149        if key == CONF_MAMMAMIRADIO_URL:
150            return "legacy-config-must-not-be-used"
151        if key == "log_level":
152            return "GLOBAL"
153        return default
154
155    config.get_value.side_effect = _get_value
156
157    return MammamiradioProvider(mass_mock, manifest, config, SUPPORTED_FEATURES)
158
159
160@pytest.fixture
161async def initialized_provider(
162    provider: MammamiradioProvider, mass_mock: MagicMock
163) -> MammamiradioProvider:
164    """Return a provider that completed handle_async_init against a healthy v1 mock."""
165    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC))
166    await provider.handle_async_init()
167    # Install a fresh mock so tests using side_effect sequences don't have their
168    # first response consumed by the init probe.
169    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC))
170    return provider
171
172
173def _build_provider_with_url(
174    mass_mock: MagicMock, configured_url: str | None
175) -> MammamiradioProvider:
176    """Build a fresh provider instance configured with ``configured_url``."""
177    manifest = MagicMock()
178    manifest.domain = "mammamiradio"
179    manifest.name = "mammamiradio"
180    config = MagicMock()
181    config.instance_id = "mammamiradio_test"
182    config.values = {}
183
184    mass_mock.config.get.return_value = (
185        {CONF_MAMMAMIRADIO_URL: configured_url} if configured_url is not None else {}
186    )
187
188    def _get_value(key: str, default: Any = None) -> Any:
189        if key == CONF_MAMMAMIRADIO_URL:
190            return None if configured_url is None else "legacy-config-must-not-be-used"
191        if key == "log_level":
192            return "GLOBAL"
193        return default
194
195    config.get_value.side_effect = _get_value
196    return MammamiradioProvider(mass_mock, manifest, config, SUPPORTED_FEATURES)
197
198
199# ---------------------------------------------------------------------------
200# Setup flow
201# ---------------------------------------------------------------------------
202
203
204async def test_setup_flow_collects_single_url_field() -> None:
205    """The setup flow prefills and persists its one required URL field."""
206    session = MagicMock()
207    session.context.setup_data = {CONF_MAMMAMIRADIO_URL: "http://previous:8000"}
208    session.form = AsyncMock(return_value={CONF_MAMMAMIRADIO_URL: "http://mammamiradio.local:8000"})
209    session.finish = AsyncMock()
210
211    await mammamiradio_setup_flow.run_setup(session)
212
213    entries = session.form.await_args.args[0]
214    assert len(entries) == 1
215    entry = entries[0]
216    assert entry.key == CONF_MAMMAMIRADIO_URL
217    assert entry.required is True
218    assert entry.default_value == DEFAULT_URL
219    assert entry.value == "http://previous:8000"
220    assert entry.type.value == "string"
221    session.finish.assert_awaited_once_with(
222        {CONF_MAMMAMIRADIO_URL: "http://mammamiradio.local:8000"}
223    )
224
225
226async def test_setup_flow_retries_with_submitted_url_and_error() -> None:
227    """A setup failure redisplays the submitted URL and translated base error."""
228    session = MagicMock()
229    session.context.setup_data = {}
230    session.form = AsyncMock(
231        side_effect=[
232            {CONF_MAMMAMIRADIO_URL: "http://unreachable:8000"},
233            {CONF_MAMMAMIRADIO_URL: "http://mammamiradio.local:8000"},
234        ]
235    )
236    session.finish = AsyncMock(
237        side_effect=[
238            SetupFlowError("Unable to connect", translation_key="cannot_connect"),
239            None,
240        ]
241    )
242
243    await mammamiradio_setup_flow.run_setup(session)
244
245    retry_call = session.form.await_args_list[1]
246    retry_entry = retry_call.args[0][0]
247    assert retry_entry.value == "http://unreachable:8000"
248    assert retry_call.kwargs["errors"] == {"base": "cannot_connect"}
249    assert session.finish.await_count == 2
250
251
252async def test_setup_returns_provider_instance(mass_mock: MagicMock) -> None:
253    """The module-level setup() entrypoint constructs a MammamiradioProvider."""
254    manifest = MagicMock()
255    manifest.domain = "mammamiradio"
256    manifest.name = "mammamiradio"
257    config = MagicMock()
258    config.instance_id = "mammamiradio_test"
259    config.values = {}
260
261    def _get_value(key: str, default: Any = None) -> Any:
262        if key == CONF_MAMMAMIRADIO_URL:
263            return "http://localhost:8000"
264        if key == "log_level":
265            return "GLOBAL"
266        return default
267
268    config.get_value.side_effect = _get_value
269    prov = await setup(mass_mock, manifest, config)
270    assert isinstance(prov, MammamiradioProvider)
271
272
273# ---------------------------------------------------------------------------
274# Provider lifecycle
275# ---------------------------------------------------------------------------
276
277
278async def test_loaded_in_mass_adds_radio_after_base_hook(
279    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
280) -> None:
281    """Post-load setup calls the base hook before adding the station through core."""
282    events: list[str] = []
283
284    async def base_loaded(_provider: MusicProvider) -> None:
285        events.append("base")
286
287    async def add_item(_item: Radio) -> None:
288        events.append("add")
289
290    mass_mock.music.add_item_to_library = AsyncMock(side_effect=add_item)
291    with patch.object(MusicProvider, "loaded_in_mass", base_loaded):
292        await initialized_provider.loaded_in_mass()
293
294    assert events == ["base", "add"]
295    mass_mock.music.add_item_to_library.assert_awaited_once()
296    added = mass_mock.music.add_item_to_library.await_args.args[0]
297    assert isinstance(added, Radio)
298    assert added.item_id == RADIO_ITEM_ID
299    assert added.provider == initialized_provider.instance_id
300
301
302# ---------------------------------------------------------------------------
303# handle_async_init — the v1 now-playing contract probe
304# ---------------------------------------------------------------------------
305
306
307async def test_handle_async_init_v1_contract(
308    provider: MammamiradioProvider, mass_mock: MagicMock
309) -> None:
310    """A reachable v1 now-playing endpoint caches the base URL, audio format and stream path."""
311    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"a"'))
312    await provider.handle_async_init()
313    assert provider._base_url == "http://localhost:8000"
314    assert provider._audio_format_dict == _V1_AUDIO_FORMAT
315    assert provider._stream_path == "/stream"
316    called_url = mass_mock.http_session.get.call_args.args[0]
317    assert called_url == "http://localhost:8000/api/integrations/v1/now-playing"
318
319
320async def test_handle_async_init_none_config_uses_default_url(mass_mock: MagicMock) -> None:
321    """An absent URL config value falls back to DEFAULT_URL for the probe."""
322    prov = _build_provider_with_url(mass_mock, None)
323    await prov.handle_async_init()
324    called_url = mass_mock.http_session.get.call_args.args[0]
325    assert called_url == f"{DEFAULT_URL}/api/integrations/v1/now-playing"
326
327
328@pytest.mark.parametrize("status", [404, 405, 501])
329async def test_handle_async_init_missing_endpoint_raises_addon_too_old(
330    provider: MammamiradioProvider, mass_mock: MagicMock, status: int
331) -> None:
332    """A missing v1 endpoint (pre-2.13 addon) fails init with an actionable message."""
333    mass_mock.http_session.get = MagicMock(return_value=_make_response_ctx(status))
334    with pytest.raises(ProviderUnavailableError, match=r"2\.13"):
335        await provider.handle_async_init()
336    # One probe only — there is no legacy fallback endpoint to try.
337    assert mass_mock.http_session.get.call_count == 1
338
339
340async def test_handle_async_init_5xx_raises(
341    provider: MammamiradioProvider, mass_mock: MagicMock
342) -> None:
343    """A 5xx on the v1 probe fails init naming the HTTP status, not the 2.13 message."""
344    mass_mock.http_session.get = MagicMock(return_value=_make_response_ctx(503))
345    with pytest.raises(ProviderUnavailableError, match="HTTP 503") as excinfo:
346        await provider.handle_async_init()
347    assert "2.13" not in str(excinfo.value)
348    assert mass_mock.http_session.get.call_count == 1
349
350
351async def test_handle_async_init_non_json_body_raises_addon_too_old(
352    provider: MammamiradioProvider, mass_mock: MagicMock
353) -> None:
354    """A 200 whose body is not valid JSON fails init with the 2.13 message."""
355    mass_mock.http_session.get = MagicMock(return_value=_make_bad_json_ctx(200))
356    with pytest.raises(ProviderUnavailableError, match=r"2\.13"):
357        await provider.handle_async_init()
358    assert mass_mock.http_session.get.call_count == 1
359
360
361async def test_handle_async_init_array_payload_raises_addon_too_old(
362    provider: MammamiradioProvider, mass_mock: MagicMock
363) -> None:
364    """A 200 whose JSON body is an array (not an object) fails init with the 2.13 message."""
365    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(["not", "an", "object"]))
366    with pytest.raises(ProviderUnavailableError, match=r"2\.13"):
367        await provider.handle_async_init()
368    assert mass_mock.http_session.get.call_count == 1
369
370
371async def test_handle_async_init_content_type_error_raises_addon_too_old(
372    provider: MammamiradioProvider, mass_mock: MagicMock
373) -> None:
374    """
375    A non-JSON content-type on the v1 probe fails init with the 2.13 message.
376
377    ``aiohttp.ContentTypeError`` subclasses ``ClientError``, so a 200 HTML page
378    (e.g. from an ingress splash) must report "requires addon 2.13+" instead of
379    the generic "unreachable" error.
380    """
381    content_type_error = aiohttp.ContentTypeError(request_info=MagicMock(), history=())
382    mass_mock.http_session.get = MagicMock(return_value=_make_bad_json_ctx(exc=content_type_error))
383    with pytest.raises(ProviderUnavailableError, match=r"2\.13"):
384        await provider.handle_async_init()
385    assert mass_mock.http_session.get.call_count == 1
386
387
388@pytest.mark.parametrize(
389    "payload",
390    [
391        pytest.param({k: v for k, v in _V1_MUSIC.items() if k != "schema_version"}, id="absent"),
392        pytest.param({**_V1_MUSIC, "schema_version": None}, id="none"),
393    ],
394)
395async def test_handle_async_init_missing_schema_version_raises_addon_too_old(
396    provider: MammamiradioProvider, mass_mock: MagicMock, payload: dict[str, Any]
397) -> None:
398    """A payload without a usable schema_version fails init with the 2.13 message, not 'None'."""
399    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(payload))
400    with pytest.raises(ProviderUnavailableError, match=r"2\.13") as excinfo:
401        await provider.handle_async_init()
402    assert "None" not in str(excinfo.value)
403
404
405async def test_handle_async_init_unsupported_schema_names_version(
406    provider: MammamiradioProvider, mass_mock: MagicMock
407) -> None:
408    """A reachable but incompatible now-playing schema fails init naming the version."""
409    unsupported = {**_V1_MUSIC, "schema_version": "2"}
410    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(unsupported))
411    with pytest.raises(ProviderUnavailableError) as excinfo:
412        await provider.handle_async_init()
413    assert "schema_version '2'" in str(excinfo.value)
414
415
416async def test_handle_async_init_numeric_schema_version_raises_addon_too_old(
417    provider: MammamiradioProvider, mass_mock: MagicMock
418) -> None:
419    """A numeric schema_version is not a usable version and receives the 2.13 error."""
420    payload = {**_V1_MUSIC, "schema_version": 1}
421    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(payload))
422    with pytest.raises(ProviderUnavailableError, match=r"2\.13") as excinfo:
423        await provider.handle_async_init()
424    assert "schema_version" not in str(excinfo.value)
425
426
427async def test_handle_async_init_bool_schema_version_raises_addon_too_old(
428    provider: MammamiradioProvider, mass_mock: MagicMock
429) -> None:
430    """A bool schema_version is junk, not a version: init fails with the 2.13 message."""
431    payload = {**_V1_MUSIC, "schema_version": True}
432    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(payload))
433    with pytest.raises(ProviderUnavailableError, match=r"2\.13") as excinfo:
434        await provider.handle_async_init()
435    assert "True" not in str(excinfo.value)
436
437
438async def test_handle_async_init_without_stream_block_uses_defaults(
439    provider: MammamiradioProvider, mass_mock: MagicMock
440) -> None:
441    """A contract without a stream block still initializes with the published defaults."""
442    contract = {k: v for k, v in _V1_MUSIC.items() if k != "stream"}
443    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(contract))
444    await provider.handle_async_init()
445    details = await provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
446    assert details.path == "http://localhost:8000/stream"
447    assert details.audio_format.content_type == ContentType.MP3
448    assert details.audio_format.bit_rate == 192
449    assert details.audio_format.sample_rate == 48000
450    assert details.audio_format.channels == 2
451
452
453async def test_handle_async_init_raises_when_unreachable(
454    provider: MammamiradioProvider, mass_mock: MagicMock
455) -> None:
456    """
457    An unreachable addon fails init with a clean "unreachable" error.
458
459    The provider is the canonical place for liveness detection (matches
460    RadioBrowser's pattern). Raising here prevents MA from loading a
461    non-functional provider and surfaces a clean unavailable error to the
462    user instead of letting the stream URL fail silently inside ffmpeg.
463    """
464    mass_mock.http_session.get = MagicMock(
465        return_value=_make_failing_ctx(aiohttp.ClientConnectionError("nope"))
466    )
467    with pytest.raises(ProviderUnavailableError, match="unreachable"):
468        await provider.handle_async_init()
469
470
471async def test_handle_async_init_generic_client_error_still_raises(
472    provider: MammamiradioProvider, mass_mock: MagicMock
473) -> None:
474    """A plain aiohttp.ClientError (true connection failure) must still fail init."""
475    mass_mock.http_session.get = MagicMock(
476        return_value=_make_failing_ctx(aiohttp.ClientError("boom"))
477    )
478    with pytest.raises(ProviderUnavailableError, match="unreachable"):
479        await provider.handle_async_init()
480
481
482async def test_handle_async_init_raises_on_timeout(
483    provider: MammamiradioProvider, mass_mock: MagicMock
484) -> None:
485    """A timeout on the init probe surfaces as an "unreachable" ProviderUnavailableError."""
486    mass_mock.http_session.get = MagicMock(return_value=_make_failing_ctx(TimeoutError("slow")))
487    with pytest.raises(ProviderUnavailableError, match="unreachable"):
488        await provider.handle_async_init()
489
490
491# ---------------------------------------------------------------------------
492# browse() returns a single Radio entry
493# ---------------------------------------------------------------------------
494
495
496async def test_browse_returns_single_radio_entry(
497    initialized_provider: MammamiradioProvider,
498) -> None:
499    """browse() returns the one mammamiradio Radio object."""
500    items = await initialized_provider.browse("mammamiradio://")
501    assert len(items) == 1
502    radio = items[0]
503    assert isinstance(radio, Radio)
504    assert radio.item_id == RADIO_ITEM_ID
505    assert radio.name == RADIO_NAME
506    # Single ProviderMapping wired to this provider instance.
507    mappings = list(radio.provider_mappings)
508    assert len(mappings) == 1
509    assert mappings[0].provider_domain == "mammamiradio"
510    assert mappings[0].available is True
511
512
513# ---------------------------------------------------------------------------
514# search behaviour
515# ---------------------------------------------------------------------------
516
517
518async def test_search_exact_name_returns_entry(
519    initialized_provider: MammamiradioProvider,
520) -> None:
521    """search('mammamiradio') returns the Radio entry."""
522    results = await initialized_provider.search("mammamiradio", [MediaType.RADIO])
523    assert isinstance(results, SearchResults)
524    assert len(results.radio) == 1
525    assert results.radio[0].item_id == RADIO_ITEM_ID
526
527
528async def test_search_substring_returns_entry(
529    initialized_provider: MammamiradioProvider,
530) -> None:
531    """search('mamma') matches the entry by substring (case-insensitive)."""
532    results = await initialized_provider.search("MAMMA", [MediaType.RADIO])
533    assert len(results.radio) == 1
534    assert results.radio[0].item_id == RADIO_ITEM_ID
535
536
537async def test_search_display_name_returns_entry(
538    initialized_provider: MammamiradioProvider,
539) -> None:
540    """The full display name 'Mamma Mi Radio' (with spaces) matches the entry."""
541    results = await initialized_provider.search("Mamma Mi Radio", [MediaType.RADIO])
542    assert len(results.radio) == 1
543    assert results.radio[0].item_id == RADIO_ITEM_ID
544
545
546async def test_search_no_match_returns_empty(
547    initialized_provider: MammamiradioProvider,
548) -> None:
549    """search('zzz') returns no results."""
550    results = await initialized_provider.search("zzz", [MediaType.RADIO])
551    assert results.radio == []
552
553
554async def test_search_without_radio_media_type_returns_empty(
555    initialized_provider: MammamiradioProvider,
556) -> None:
557    """search() respects the media_types filter — no Radio in filter, no results."""
558    results = await initialized_provider.search("mammamiradio", [MediaType.TRACK])
559    assert results.radio == []
560
561
562async def test_search_empty_query_returns_empty(
563    initialized_provider: MammamiradioProvider,
564) -> None:
565    """Empty search string must return no results (not match-all)."""
566    results = await initialized_provider.search("", [MediaType.RADIO])
567    assert results.radio == []
568
569
570# ---------------------------------------------------------------------------
571# get_radio
572# ---------------------------------------------------------------------------
573
574
575async def test_get_radio_with_valid_id_returns_radio(
576    initialized_provider: MammamiradioProvider,
577) -> None:
578    """get_radio(valid_id) returns a fully-populated Radio."""
579    radio = await initialized_provider.get_radio(RADIO_ITEM_ID)
580    assert isinstance(radio, Radio)
581    assert radio.item_id == RADIO_ITEM_ID
582    assert radio.name == RADIO_NAME
583    assert radio.metadata.description
584    assert radio.metadata.genres
585    assert radio.metadata.languages == ["it"]
586
587
588async def test_get_radio_with_invalid_id_raises_media_not_found(
589    initialized_provider: MammamiradioProvider,
590) -> None:
591    """get_radio('bogus') raises MediaNotFoundError."""
592    with pytest.raises(MediaNotFoundError):
593        await initialized_provider.get_radio("does-not-exist")
594
595
596# ---------------------------------------------------------------------------
597# get_stream_details
598# ---------------------------------------------------------------------------
599
600
601async def test_get_stream_details_returns_mp3_format(
602    provider: MammamiradioProvider, mass_mock: MagicMock
603) -> None:
604    """
605    StreamDetails declares ContentType.MP3 at the addon's published bitrate.
606
607    When the contract omits ``stream.audio_format`` the published defaults
608    apply: MP3 @ 192 kbps / 48 kHz / stereo (matching the addon's AudioConfig
609    default, not the old hard-coded 128).
610    """
611    contract = {**_V1_MUSIC, "stream": {"relative_url": "/stream"}}
612    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(contract))
613    await provider.handle_async_init()
614    details = await provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
615    assert details.audio_format.content_type == ContentType.MP3
616    assert details.audio_format.bit_rate == 192
617    assert details.audio_format.sample_rate == 48000
618    assert details.audio_format.channels == 2
619    assert details.media_type == MediaType.RADIO
620
621
622async def test_get_stream_details_uses_configured_url_with_stream_suffix(
623    initialized_provider: MammamiradioProvider,
624) -> None:
625    """The stream path defaults to ``${url}/stream``."""
626    details = await initialized_provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
627    assert details.path == "http://localhost:8000/stream"
628    assert details.allow_seek is False
629    assert details.can_seek is False
630
631
632async def test_get_stream_details_uses_contract_relative_stream_path(
633    provider: MammamiradioProvider, mass_mock: MagicMock
634) -> None:
635    """The v1 consumer contract may publish the relative stream URL to expose."""
636    contract = {
637        **_V1_MUSIC,
638        "stream": {"relative_url": "/radio/live.mp3", "audio_format": _V1_AUDIO_FORMAT},
639    }
640    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(contract))
641    await provider.handle_async_init()
642    details = await provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
643    assert details.path == "http://localhost:8000/radio/live.mp3"
644
645
646async def test_get_stream_details_defaults_path_when_relative_url_absent(
647    provider: MammamiradioProvider, mass_mock: MagicMock
648) -> None:
649    """A contract stream block without relative_url falls back to the default /stream path."""
650    contract = {**_V1_MUSIC, "stream": {"audio_format": _V1_AUDIO_FORMAT}}
651    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(contract))
652    await provider.handle_async_init()
653    details = await provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
654    assert details.path == "http://localhost:8000/stream"
655
656
657async def test_get_stream_details_ignores_unsafe_contract_stream_path(
658    provider: MammamiradioProvider, mass_mock: MagicMock
659) -> None:
660    """An absolute contract URL must not turn the provider into a redirector."""
661    contract = {
662        **_V1_MUSIC,
663        "stream": {"relative_url": "https://example.invalid/stream", "audio_format": {}},
664    }
665    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(contract))
666    await provider.handle_async_init()
667    details = await provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
668    assert details.path == "http://localhost:8000/stream"
669
670
671async def test_get_stream_details_does_not_probe_at_stream_time(
672    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
673) -> None:
674    """
675    Stream-time has no HTTP probe; liveness is checked at init only.
676
677    This is intentional per MA convention — NTS/RadioBrowser/ORF Radiothek
678    all do the same. ``get_stream_details`` returns a passthrough
679    ``StreamDetails``; failures at the actual stream URL surface naturally
680    via MA's ffmpeg pipeline. Locks the contract that no http_session calls
681    happen during stream-details resolution.
682
683    Live metadata does not break this contract: ``get_stream_details`` only
684    *wires* the ``stream_metadata_update_callback`` + interval; the HTTP poll
685    happens later, inside the callback, never at stream-details time.
686    """
687    mass_mock.http_session.get = MagicMock()
688    mass_mock.http_session.head = MagicMock()
689    details = await initialized_provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
690    assert details.path == "http://localhost:8000/stream"
691    mass_mock.http_session.get.assert_not_called()
692    mass_mock.http_session.head.assert_not_called()
693    # The live-metadata callback is wired, but not invoked here.
694    assert details.stream_metadata_update_callback == initialized_provider._update_stream_metadata
695    assert details.stream_metadata_update_interval == STREAM_METADATA_UPDATE_INTERVAL
696
697
698async def test_get_stream_details_with_invalid_id_raises_media_not_found(
699    initialized_provider: MammamiradioProvider,
700) -> None:
701    """Unknown item id at stream time raises MediaNotFoundError, not unavailable."""
702    with pytest.raises(MediaNotFoundError):
703        await initialized_provider.get_stream_details("does-not-exist", MediaType.RADIO)
704
705
706async def test_get_stream_details_uses_contract_audio_format(
707    provider: MammamiradioProvider, mass_mock: MagicMock
708) -> None:
709    """
710    After init against a v1 contract, get_stream_details reflects the contract's format.
711
712    Locks the end-to-end plumbing with a NON-default format (the default test
713    payload is bit-identical to the fallback defaults, which would hide a
714    regression where _audio_format() ignored the cached contract).
715    """
716    contract = {
717        **_V1_MUSIC,
718        "stream": {
719            "relative_url": "/stream",
720            "audio_format": {
721                "codec": "aac",
722                "bitrate_kbps": 256,
723                "sample_rate_hz": 44100,
724                "channels": 1,
725            },
726        },
727    }
728    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(contract, etag='W/"a"'))
729    await provider.handle_async_init()
730    details = await provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
731    assert details.audio_format.content_type == ContentType.AAC
732    assert details.audio_format.bit_rate == 256
733    assert details.audio_format.sample_rate == 44100
734    assert details.audio_format.channels == 1
735
736
737# ---------------------------------------------------------------------------
738# URL hygiene + supported features (locks the contract for the discovery flow)
739# ---------------------------------------------------------------------------
740
741
742async def test_supported_features_are_browse_and_search_only() -> None:
743    """SUPPORTED_FEATURES is exactly {BROWSE, SEARCH} (matches SomaFM)."""
744    assert {ProviderFeature.BROWSE, ProviderFeature.SEARCH} == SUPPORTED_FEATURES
745
746
747async def test_trailing_slash_in_configured_url_does_not_double_up(
748    mass_mock: MagicMock,
749) -> None:
750    """A configured URL with a trailing slash should still produce ``${url}/stream``."""
751    prov = _build_provider_with_url(mass_mock, "http://localhost:8000/")
752    await prov.handle_async_init()
753    probe_url = mass_mock.http_session.get.call_args.args[0]
754    assert probe_url == "http://localhost:8000/api/integrations/v1/now-playing"
755    details = await prov.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
756    assert details.path == "http://localhost:8000/stream"
757
758
759async def test_query_string_in_configured_url_is_stripped(
760    mass_mock: MagicMock,
761) -> None:
762    """A configured URL with a query string must not corrupt the probe/stream URLs."""
763    prov = _build_provider_with_url(mass_mock, "http://localhost:8000?foo=bar")
764    await prov.handle_async_init()
765    probe_url = mass_mock.http_session.get.call_args.args[0]
766    assert probe_url == "http://localhost:8000/api/integrations/v1/now-playing"
767    details = await prov.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
768    assert details.path == "http://localhost:8000/stream"
769
770
771async def test_credentials_in_configured_url_are_stripped(
772    mass_mock: MagicMock,
773) -> None:
774    """A pasted token in URL userinfo must not reach the probe or stream URLs."""
775    prov = _build_provider_with_url(
776        mass_mock, "http://admin:secret-token@localhost:8000?admin_token=also-secret"
777    )
778    await prov.handle_async_init()
779    probe_url = mass_mock.http_session.get.call_args.args[0]
780    assert probe_url == "http://localhost:8000/api/integrations/v1/now-playing"
781    assert "secret" not in probe_url
782    assert "@" not in probe_url
783    details = await prov.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
784    assert details.path == "http://localhost:8000/stream"
785
786
787async def test_fragment_in_configured_url_is_stripped(
788    mass_mock: MagicMock,
789) -> None:
790    """A configured URL with a fragment must not corrupt the probe/stream URLs."""
791    prov = _build_provider_with_url(mass_mock, "http://localhost:8000#frag")
792    await prov.handle_async_init()
793    probe_url = mass_mock.http_session.get.call_args.args[0]
794    assert probe_url == "http://localhost:8000/api/integrations/v1/now-playing"
795    details = await prov.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
796    assert details.path == "http://localhost:8000/stream"
797
798
799@pytest.mark.parametrize(
800    ("raw", "expected"),
801    [
802        ("http://localhost:8000", "http://localhost:8000"),
803        ("https://radio.example.test", "https://radio.example.test"),
804        ("https://radio.example.test/mamma/", "https://radio.example.test/mamma"),
805        ("http://[::1]:8000/", "http://[::1]:8000"),
806        ("http://user:secret@host:8000?token=x#frag", "http://host:8000"),
807        ("  http://localhost:8000  ", "http://localhost:8000"),
808        ("HTTP://localhost:8000", "http://localhost:8000"),
809    ],
810)
811def test_normalize_base_url_accepts(raw: str, expected: str) -> None:
812    """Valid http(s) base URLs normalize to a sanitized scheme://host[:port][/path]."""
813    assert _normalize_base_url(raw) == expected
814
815
816@pytest.mark.parametrize(
817    "raw",
818    [
819        "localhost:8000",
820        "//localhost:8000",
821        "ftp://localhost:8000",
822        "http:///stream",
823        "",
824        "   ",
825        "http://[::1:8000",
826        "http://localhost:99999",
827        "http://localhost:notaport",
828        "http://local host:8000",
829        "http://local\thost:8000",
830        "http://host\\evil:8000",
831    ],
832)
833def test_normalize_base_url_rejects_bad_urls(raw: str) -> None:
834    """Any string that is not a full http(s) URL with a hostname raises ValueError."""
835    with pytest.raises(ValueError, match="base URL"):
836        _normalize_base_url(raw)
837
838
839@pytest.mark.parametrize("raw", [123, True, None, ["http://localhost:8000"]])
840def test_normalize_base_url_rejects_non_strings(raw: Any) -> None:
841    """Provider-visible non-string values raise TypeError instead of being coerced."""
842    with pytest.raises(TypeError, match="base URL"):
843        _normalize_base_url(raw)
844
845
846@pytest.mark.parametrize(
847    ("value", "expected"),
848    [
849        ("1", True),
850        (1, False),
851        (True, False),
852        (1.0, False),
853        ("2", False),
854        (None, False),
855    ],
856)
857def test_supports_v1_schema(value: Any, expected: bool) -> None:
858    """Only a supported version string counts; other value types never do."""
859    assert _supports_v1_schema(value) is expected
860
861
862@pytest.mark.parametrize(
863    "value",
864    ["//evil/stream", "radio/live.mp3", "/", None, "", "//["],
865)
866def test_stream_path_from_contract_rejects_unsafe_values(value: Any) -> None:
867    """Protocol-relative, schemeless-relative, or unparsable values fall back to /stream."""
868    assert _stream_path_from_contract(value) == "/stream"
869
870
871@pytest.mark.parametrize(
872    ("hosts", "expected"),
873    [
874        ("Gianni", None),
875        ({"display_name": "Gianni"}, None),
876        ([{"engine_host": "gianni"}], "gianni"),
877    ],
878)
879def test_host_display_names_variants(hosts: Any, expected: str | None) -> None:
880    """Non-list hosts yield no byline; a dict host without display_name uses engine_host."""
881    assert _host_display_names(hosts) == expected
882
883
884async def test_invalid_base_url_fails_setup_before_http(mass_mock: MagicMock) -> None:
885    """A schemeless URL raises a provider-localized SetupFailedError before any request."""
886    prov = _build_provider_with_url(mass_mock, "localhost:8000")
887    with pytest.raises(SetupFailedError) as excinfo:
888        await prov.handle_async_init()
889    assert excinfo.value.translation_key == "invalid_base_url"
890    assert excinfo.value.translation_owner == "provider.mammamiradio"
891    mass_mock.http_session.get.assert_not_called()
892
893
894async def test_cached_base_url_survives_config_replacement(mass_mock: MagicMock) -> None:
895    """
896    The base URL is bound at init; a later config swap does not affect polling.
897
898    Base ``Provider.update_config`` replaces ``self.config`` immediately and
899    schedules the reload later; an already-resolved stream must keep polling the
900    URL it was initialized with instead of raising through the callback.
901    """
902    prov = _build_provider_with_url(mass_mock, "http://radio.example.test:8000")
903    await prov.handle_async_init()
904    details = await prov.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
905    assert details.path == "http://radio.example.test:8000/stream"
906
907    bad_config = MagicMock()
908    bad_config.get_value.return_value = "not a url"
909    prov.config = bad_config
910    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC))
911    await prov._update_stream_metadata(details, 0)  # must not raise
912    called_url = mass_mock.http_session.get.call_args.args[0]
913    assert called_url == "http://radio.example.test:8000/api/integrations/v1/now-playing"
914
915
916# ---------------------------------------------------------------------------
917# v1 now-playing contract — audio format
918# ---------------------------------------------------------------------------
919
920
921def test_audio_format_from_contract_reads_published_format() -> None:
922    """The audio format comes from the v1 contract: MP3 / 192 kbps / 48 kHz / stereo."""
923    fmt = _audio_format_from_contract(_V1_AUDIO_FORMAT)
924    assert fmt.content_type == ContentType.MP3
925    assert fmt.bit_rate == 192
926    assert fmt.sample_rate == 48000
927    assert fmt.channels == 2
928
929
930def test_audio_format_from_contract_defaults_when_absent() -> None:
931    """Missing/None contract falls back to the addon's published defaults."""
932    fmt = _audio_format_from_contract(None)
933    assert fmt.content_type == ContentType.MP3
934    assert fmt.bit_rate == 192
935    assert fmt.sample_rate == 48000
936    assert fmt.channels == 2
937
938
939def test_audio_format_from_contract_unknown_codec_defaults_to_mp3() -> None:
940    """An unrecognized codec degrades to MP3 rather than ContentType.UNKNOWN."""
941    fmt = _audio_format_from_contract({"codec": "weird", "bitrate_kbps": 96})
942    assert fmt.content_type == ContentType.MP3
943    assert fmt.bit_rate == 96
944
945
946def test_audio_format_from_contract_honors_alternate_codec() -> None:
947    """A real alternate codec is parsed (forward-compat for a future addon encoder)."""
948    fmt = _audio_format_from_contract(
949        {"codec": "aac", "bitrate_kbps": 256, "sample_rate_hz": 44100, "channels": 1}
950    )
951    assert fmt.content_type == ContentType.AAC
952    assert fmt.bit_rate == 256
953    assert fmt.sample_rate == 44100
954    assert fmt.channels == 1
955
956
957@pytest.mark.parametrize("bitrate", [True, 0, -5])
958def test_audio_format_from_contract_non_positive_bitrate_defaults(bitrate: Any) -> None:
959    """A bool / zero / negative bitrate_kbps falls back to the published 192 default."""
960    fmt = _audio_format_from_contract({"codec": "mp3", "bitrate_kbps": bitrate})
961    assert fmt.bit_rate == 192
962
963
964# ---------------------------------------------------------------------------
965# v1 now-playing contract — `_v1_to_stream_metadata` (pure mapping)
966# ---------------------------------------------------------------------------
967
968
969def _v1_payload(now_playing: Any, *, up_next: Any = None, station: Any = None) -> dict[str, Any]:
970    """Build a minimal v1 response around a ``now_playing`` block."""
971    return {
972        "schema_version": "1",
973        "station": station if station is not None else {"name": "mammamiradio", "hosts": []},
974        "stream": {"relative_url": "/stream", "audio_format": _V1_AUDIO_FORMAT},
975        "now_playing": now_playing,
976        "up_next": up_next if up_next is not None else [],
977        "session_state": "live" if now_playing is not None else "empty_queue",
978        "changed_at": 1.0,
979    }
980
981
982def test_v1_music_maps_title_artist_artwork_album() -> None:
983    """A music segment surfaces title / artist / artwork / album from the contract."""
984    sm = _v1_to_stream_metadata(_V1_MUSIC, show_upcoming=False)
985    assert sm.title == "Volare"
986    assert sm.artist == "Modugno"
987    assert sm.image_url == "http://art/volare.jpg"
988    assert sm.album == "Best Of"
989
990
991def test_v1_music_without_album_has_no_album() -> None:
992    """A music segment without an album stays album-less (no station-name fallback)."""
993    now = {k: v for k, v in _V1_MUSIC["now_playing"].items() if k != "album"}
994    sm = _v1_to_stream_metadata(_v1_payload(now), show_upcoming=False)
995    assert sm.title == "Volare"
996    assert sm.album is None
997
998
999def test_v1_music_non_string_fields_coerced() -> None:
1000    """Non-string artist / artwork / album from untrusted JSON coerce to None, not garbage."""
1001    now = {
1002        "segment_class": "music",
1003        "segment_type": "music",
1004        "title": "X",
1005        "artist": 42,
1006        "artwork": ["not", "a", "url"],
1007        "album": {"not": "a string"},
1008    }
1009    sm = _v1_to_stream_metadata(_v1_payload(now), show_upcoming=False)
1010    assert sm.title == "X"
1011    assert sm.artist is None
1012    assert sm.image_url is None
1013    assert sm.album is None
1014
1015
1016@pytest.mark.parametrize(
1017    ("artwork", "expected"),
1018    [
1019        ("javascript:alert(1)", None),
1020        ("file:///etc/passwd", None),
1021        ("relative/art.jpg", None),
1022        ("http://", None),
1023        ("http:///pathonly.jpg", None),
1024        ("http://[bad/art.jpg", None),
1025        ("http://art/ok.jpg", "http://art/ok.jpg"),
1026        ("https://art/ok.jpg", "https://art/ok.jpg"),
1027        ("HTTPS://art/ok.jpg", "HTTPS://art/ok.jpg"),
1028    ],
1029)
1030def test_v1_artwork_non_http_scheme_dropped(artwork: str, expected: str | None) -> None:
1031    """Only http(s) artwork URLs may reach MA media surfaces."""
1032    now = {**_V1_MUSIC["now_playing"], "artwork": artwork}
1033    sm = _v1_to_stream_metadata(_v1_payload(now), show_upcoming=False)
1034    assert sm.image_url == expected
1035
1036
1037def test_v1_voice_uses_now_playing_host() -> None:
1038    """A voice segment uses the contract's top-level ``host`` byline as the artist."""
1039    now = {
1040        "segment_class": "voice",
1041        "segment_type": "banter",
1042        "title": "Host banter",
1043        "host": "Gianni",
1044    }
1045    sm = _v1_to_stream_metadata(_v1_payload(now), show_upcoming=False)
1046    assert sm.title == "Host banter"
1047    assert sm.artist == "Gianni"
1048
1049
1050def test_v1_voice_host_string_wins_over_station_hosts() -> None:
1051    """
1052    A populated now_playing.host takes precedence over station.hosts.
1053
1054    Mirrors the live news_flash case (host="Giulia") where the byline must be the
1055    single reading host, not the full station roster.
1056    """
1057    now = {
1058        "segment_class": "voice",
1059        "segment_type": "news_flash",
1060        "title": "Notizie",
1061        "host": "Giulia",
1062    }
1063    station = {
1064        "name": "mammamiradio",
1065        "hosts": [
1066            {"engine_host": "m", "display_name": "Marco"},
1067            {"engine_host": "l", "display_name": "Lucia"},
1068        ],
1069    }
1070    sm = _v1_to_stream_metadata(_v1_payload(now, station=station), show_upcoming=False)
1071    assert sm.artist == "Giulia"
1072
1073
1074def test_v1_voice_without_host_falls_back_to_station_display_names() -> None:
1075    """The real banter fix: station hosts are display_name dicts, not strings."""
1076    now = {"segment_class": "voice", "segment_type": "banter", "title": None, "host": None}
1077    station = {
1078        "name": "mammamiradio",
1079        "hosts": [
1080            {"engine_host": "g", "display_name": "Gianni"},
1081            {"engine_host": "l", "display_name": "Lucia"},
1082        ],
1083    }
1084    sm = _v1_to_stream_metadata(_v1_payload(now, station=station), show_upcoming=False)
1085    assert sm.title == "Host banter"
1086    assert sm.artist == "Gianni, Lucia"
1087
1088
1089def test_v1_banter_string_hosts_join() -> None:
1090    """station.hosts given as plain strings still joins the names (hardening)."""
1091    now = {"segment_class": "voice", "segment_type": "banter", "title": None, "host": None}
1092    station = {"name": "mammamiradio", "hosts": ["Gianni", "Lucia"]}
1093    sm = _v1_to_stream_metadata(_v1_payload(now, station=station), show_upcoming=False)
1094    assert sm.artist == "Gianni, Lucia"
1095
1096
1097def test_v1_voice_empty_hosts_falls_back_to_station_name() -> None:
1098    """A voice segment with an empty hosts roster and no host byline uses the station name."""
1099    now = {"segment_class": "voice", "segment_type": "banter", "title": None}
1100    sm = _v1_to_stream_metadata(_v1_payload(now), show_upcoming=False)
1101    assert sm.title == "Host banter"
1102    assert sm.artist == "mammamiradio"
1103
1104
1105def test_v1_interstitial_titles_with_station_artist() -> None:
1106    """An interstitial (ad / station id) carries its label and the station as artist."""
1107    now = {"segment_class": "interstitial", "segment_type": "ad", "title": "Ad break"}
1108    sm = _v1_to_stream_metadata(_v1_payload(now), show_upcoming=False)
1109    assert sm.title == "Ad break"
1110    assert sm.artist == "mammamiradio"
1111
1112
1113def test_v1_unavailable_renders_idle_station_frame() -> None:
1114    """An 'unavailable' segment renders the station name and suppresses the description."""
1115    now = {"segment_class": "unavailable", "segment_type": "skipping", "title": None}
1116    sm = _v1_to_stream_metadata(_v1_payload(now, up_next=[{"title": "Next"}]), show_upcoming=True)
1117    assert sm.title == "mammamiradio"
1118    assert sm.description is None
1119
1120
1121def test_v1_no_now_playing_is_idle() -> None:
1122    """session_state stopped/empty_queue (now_playing null) renders the station name."""
1123    sm = _v1_to_stream_metadata(_v1_payload(None), show_upcoming=True)
1124    assert sm.title == "mammamiradio"
1125    assert sm.description is None
1126
1127
1128def test_v1_unknown_segment_class_renders_idle_not_leak() -> None:
1129    """A future additive segment_class degrades to the idle station frame, not a leak."""
1130    now = {"segment_class": "future_thing", "segment_type": "promo", "title": "Promo X"}
1131    sm = _v1_to_stream_metadata(_v1_payload(now), show_upcoming=False)
1132    assert sm.title == "mammamiradio"
1133
1134
1135def test_v1_up_next_description_when_show_upcoming() -> None:
1136    """The 'Up next' frame surfaces the next item's title."""
1137    sm = _v1_to_stream_metadata(_V1_MUSIC, show_upcoming=True)
1138    assert sm.description == "Up next: Chiacchiere"
1139
1140
1141def test_v1_no_up_next_description_on_now_frame() -> None:
1142    """The 'Now' frame carries no description (no 'A casa' in the v1 contract)."""
1143    sm = _v1_to_stream_metadata(_V1_MUSIC, show_upcoming=False)
1144    assert sm.description is None
1145
1146
1147def test_v1_up_next_non_dict_entry_ignored() -> None:
1148    """A non-dict first up_next entry is skipped without raising; no 'Up next' line."""
1149    sm = _v1_to_stream_metadata(
1150        _v1_payload(_V1_MUSIC["now_playing"], up_next=["just-a-string"]), show_upcoming=True
1151    )
1152    assert sm.title == "Volare"
1153    assert sm.description is None
1154
1155
1156def test_v1_up_next_unavailable_entry_suppressed() -> None:
1157    """An idle 'unavailable' up-next entry never renders as an 'Up next' line."""
1158    up_next = [{"segment_class": "unavailable", "segment_type": "skipping", "title": "Nothing"}]
1159    sm = _v1_to_stream_metadata(
1160        _v1_payload(_V1_MUSIC["now_playing"], up_next=up_next), show_upcoming=True
1161    )
1162    assert sm.title == "Volare"
1163    assert sm.description is None
1164
1165
1166@pytest.mark.parametrize("up_title", [None, ""])
1167def test_v1_up_next_missing_title_suppresses_description(up_title: str | None) -> None:
1168    """An up-next entry without a usable title renders no 'Up next' line."""
1169    up_next = [{"segment_class": "music", "segment_type": "music", "title": up_title}]
1170    sm = _v1_to_stream_metadata(
1171        _v1_payload(_V1_MUSIC["now_playing"], up_next=up_next), show_upcoming=True
1172    )
1173    assert sm.description is None
1174
1175
1176def test_v1_mapper_is_total_against_malformed_payload() -> None:
1177    """Non-dict now_playing / non-list up_next / non-dict station never raise."""
1178    payload = {"station": ["x"], "now_playing": ["not", "a", "dict"], "up_next": {"bad": 1}}
1179    sm = _v1_to_stream_metadata(payload, show_upcoming=True)
1180    assert sm.title == "Mamma Mi Radio"
1181    assert sm.description is None
1182
1183
1184@pytest.mark.parametrize("seg_class", [[], {}, 7, None, True])
1185def test_v1_mapper_non_string_segment_class_never_raises(seg_class: Any) -> None:
1186    """A non-str segment_class renders the station frame (unhashable values must not raise)."""
1187    payload = {
1188        "station": {"name": "Mamma Mi Radio"},
1189        "now_playing": {"segment_class": seg_class, "title": "x"},
1190        "up_next": [{"segment_class": "music", "title": "Next"}],
1191    }
1192    sm = _v1_to_stream_metadata(payload, show_upcoming=True)
1193    assert sm.title == "Mamma Mi Radio"
1194    assert sm.description is None
1195
1196
1197# ---------------------------------------------------------------------------
1198# v1 now-playing contract — `_update_stream_metadata` callback (stateful)
1199# ---------------------------------------------------------------------------
1200
1201
1202async def _details_for(provider: MammamiradioProvider) -> Any:
1203    """Resolve a StreamDetails object to drive the metadata callback against."""
1204    return await provider.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
1205
1206
1207async def test_v1_callback_populates_from_contract(
1208    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1209) -> None:
1210    """The v1 callback polls the contract endpoint and sets stream_metadata."""
1211    details = await _details_for(initialized_provider)
1212    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"v1"'))
1213    await initialized_provider._update_stream_metadata(details, 0)
1214    assert details.stream_metadata.title == "Volare"
1215    assert details.stream_metadata.artist == "Modugno"
1216    assert mass_mock.http_session.get.call_args.args[0].endswith("/api/integrations/v1/now-playing")
1217
1218
1219async def test_v1_callback_alternates_now_then_upnext(
1220    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1221) -> None:
1222    """Call 1 renders the Now frame; call 2 (same segment) flips to the Up-next frame."""
1223    details = await _details_for(initialized_provider)
1224    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"v1"'))
1225    await initialized_provider._update_stream_metadata(details, 0)
1226    desc_now = details.stream_metadata.description
1227    assert desc_now is None
1228    await initialized_provider._update_stream_metadata(details, 0)
1229    desc_next = details.stream_metadata.description
1230    assert desc_next == "Up next: Chiacchiere"
1231    # Per-stream state lives under its own namespace in StreamDetails.data so it
1232    # can never collide with keys MA core stashes there (e.g. HLS bookkeeping).
1233    state = details.data["mammamiradio"]
1234    assert {"v1_segment", "show_upcoming", "v1_etag", "v1_last"} <= set(state)
1235
1236
1237async def test_v1_callback_304_reuses_cache_and_keeps_alternating(
1238    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1239) -> None:
1240    """A 304 reuses the cached payload (conditional poll) and still flips the view."""
1241    details = await _details_for(initialized_provider)
1242    mass_mock.http_session.get = MagicMock(
1243        side_effect=[
1244            _make_v1_ctx(_V1_MUSIC, status=200, etag='W/"v1"'),
1245            _make_v1_ctx(None, status=304),
1246        ]
1247    )
1248    await initialized_provider._update_stream_metadata(details, 0)  # 200 -> Now frame
1249    desc_now = details.stream_metadata.description
1250    assert desc_now is None
1251    await initialized_provider._update_stream_metadata(details, 0)  # 304 -> Up-next from cache
1252    desc_next = details.stream_metadata.description
1253    assert desc_next == "Up next: Chiacchiere"
1254    # The second request was conditional on the stored ETag.
1255    second_call = mass_mock.http_session.get.call_args_list[1]
1256    assert second_call.kwargs["headers"]["If-None-Match"] == 'W/"v1"'
1257
1258
1259async def test_v1_callback_swallows_unreachable_keeps_prior(
1260    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1261) -> None:
1262    """A mid-stream connection failure must not raise, keeps the prior frame, drops the ETag."""
1263    details = await _details_for(initialized_provider)
1264    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"v1"'))
1265    await initialized_provider._update_stream_metadata(details, 0)
1266    prior = details.stream_metadata
1267    assert details.data["mammamiradio"]["v1_etag"] == 'W/"v1"'
1268    mass_mock.http_session.get = MagicMock(
1269        return_value=_make_failing_ctx(aiohttp.ClientConnectionError("nope"))
1270    )
1271    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1272    assert details.stream_metadata is prior
1273    # The stored validator is dropped on the ClientError leg too (mirrors the
1274    # poisoned-ETag recovery on the ValueError leg).
1275    assert "v1_etag" not in details.data["mammamiradio"]
1276
1277
1278async def test_v1_callback_swallows_timeout_keeps_prior(
1279    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1280) -> None:
1281    """A mid-stream poll timeout must not raise and keeps the prior frame."""
1282    details = await _details_for(initialized_provider)
1283    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"v1"'))
1284    await initialized_provider._update_stream_metadata(details, 0)
1285    prior = details.stream_metadata
1286    mass_mock.http_session.get = MagicMock(return_value=_make_failing_ctx(TimeoutError("slow")))
1287    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1288    assert details.stream_metadata is prior
1289
1290
1291async def test_v1_callback_http_error_keeps_prior(
1292    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1293) -> None:
1294    """A mid-stream 5xx from the v1 endpoint must not raise and keeps the prior frame."""
1295    details = await _details_for(initialized_provider)
1296    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"v1"'))
1297    await initialized_provider._update_stream_metadata(details, 0)
1298    prior = details.stream_metadata
1299    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(None, status=503))
1300    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1301    assert details.stream_metadata is prior
1302
1303
1304async def test_v1_callback_non_dict_payload_keeps_prior(
1305    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1306) -> None:
1307    """A JSON array from the v1 endpoint mid-stream is ignored, prior frame kept."""
1308    details = await _details_for(initialized_provider)
1309    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"v1"'))
1310    await initialized_provider._update_stream_metadata(details, 0)
1311    prior = details.stream_metadata
1312    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(["unexpected", "array"]))
1313    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1314    assert details.stream_metadata is prior
1315
1316
1317async def test_v1_callback_bad_json_keeps_prior(
1318    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1319) -> None:
1320    """A non-JSON body from the v1 endpoint mid-stream is ignored, prior frame kept."""
1321    details = await _details_for(initialized_provider)
1322    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"v1"'))
1323    await initialized_provider._update_stream_metadata(details, 0)
1324    prior = details.stream_metadata
1325    mass_mock.http_session.get = MagicMock(return_value=_make_bad_json_ctx())
1326    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1327    assert details.stream_metadata is prior
1328
1329
1330async def test_v1_callback_poisoned_etag_cleared_and_recovers(
1331    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1332) -> None:
1333    """
1334    A stored ETag that fails at request time is dropped so the next tick recovers.
1335
1336    A poisoned validator (e.g. control characters from a broken proxy) raises on
1337    every conditional request; the provider must pop it instead of freezing the
1338    metadata forever.
1339    """
1340    details = await _details_for(initialized_provider)
1341    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag="bad\r\netag"))
1342    await initialized_provider._update_stream_metadata(details, 0)
1343    prior = details.stream_metadata
1344    assert details.data["mammamiradio"]["v1_etag"] == "bad\r\netag"
1345
1346    mass_mock.http_session.get = MagicMock(side_effect=ValueError("invalid header value"))
1347    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1348    assert details.stream_metadata is prior
1349    assert "v1_etag" not in details.data["mammamiradio"]
1350
1351    recovered = {**_V1_MUSIC, "now_playing": {**_V1_MUSIC["now_playing"], "title": "Recovered"}}
1352    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(recovered))
1353    await initialized_provider._update_stream_metadata(details, 0)
1354    assert details.stream_metadata.title == "Recovered"
1355    # The recovery request went out unconditionally (no stored validator left).
1356    assert "If-None-Match" not in mass_mock.http_session.get.call_args.kwargs["headers"]
1357
1358
1359async def test_v1_callback_mapper_exception_propagates(
1360    initialized_provider: MammamiradioProvider,
1361    mass_mock: MagicMock,
1362    monkeypatch: pytest.MonkeyPatch,
1363) -> None:
1364    """
1365    A mapper raise escapes the callback instead of being swallowed.
1366
1367    The mapper is total by construction, so any raise is a genuine bug. The
1368    raise escapes into MA's fire-and-forget callback task, where asyncio
1369    reports an unretrieved task exception at the default log level and nothing
1370    crashes. The show_upcoming flip is skipped so the next successful tick
1371    renders the frame the failed tick would have.
1372    """
1373    details = await _details_for(initialized_provider)
1374    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC, etag='W/"a"'))
1375    await initialized_provider._update_stream_metadata(details, 0)
1376    prior = details.stream_metadata
1377    flip_before = details.data["mammamiradio"]["show_upcoming"]
1378
1379    def _boom(*_args: Any, **_kwargs: Any) -> Any:
1380        raise RuntimeError("v1 mapper blew up")
1381
1382    monkeypatch.setattr("music_assistant.providers.mammamiradio._v1_to_stream_metadata", _boom)
1383    with pytest.raises(RuntimeError, match="v1 mapper blew up"):
1384        await initialized_provider._update_stream_metadata(details, 0)
1385    assert details.stream_metadata is prior
1386    assert details.data["mammamiradio"]["show_upcoming"] == flip_before
1387
1388
1389async def test_v1_callback_resets_alternation_on_segment_change(
1390    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1391) -> None:
1392    """
1393    A new segment resets the alternation back to the Now frame.
1394
1395    Keyed on segment identity (segment_type/title/started_at), not the addon's
1396    ``changed_at`` clock, so a mid-segment queue-append does not snap to Now.
1397    """
1398    details = await _details_for(initialized_provider)
1399    other = {**_V1_MUSIC, "now_playing": {**_V1_MUSIC["now_playing"], "title": "OtherSong"}}
1400    mass_mock.http_session.get = MagicMock(
1401        side_effect=[
1402            _make_v1_ctx(_V1_MUSIC, etag='W/"a"'),
1403            _make_v1_ctx(_V1_MUSIC, etag='W/"a"'),
1404            _make_v1_ctx(_V1_MUSIC, etag='W/"a"'),
1405            _make_v1_ctx(other, etag='W/"b"'),
1406        ]
1407    )
1408    await initialized_provider._update_stream_metadata(details, 0)  # Now
1409    d1 = details.stream_metadata.description
1410    await initialized_provider._update_stream_metadata(details, 0)  # Up-next
1411    d2 = details.stream_metadata.description
1412    await initialized_provider._update_stream_metadata(details, 0)  # Now again
1413    d3 = details.stream_metadata.description
1414    # Segment change while show_upcoming is True — only the reset logic can
1415    # produce a "Now" frame here; without it this call would render "Up next".
1416    await initialized_provider._update_stream_metadata(details, 0)
1417    d4 = details.stream_metadata.description
1418    assert d1 is None
1419    assert d2 == "Up next: Chiacchiere"
1420    assert d3 is None
1421    assert d4 is None
1422    assert details.stream_metadata.title == "OtherSong"
1423
1424
1425async def test_v1_callback_cold_cache_304_keeps_prior(
1426    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1427) -> None:
1428    """A 304 with no cached payload (cold cache) is a no-op, not a crash."""
1429    details = await _details_for(initialized_provider)
1430    assert details.stream_metadata is None
1431    mass_mock.http_session.get = MagicMock(
1432        return_value=_make_v1_ctx(None, status=304, etag='W/"x"')
1433    )
1434    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1435    assert details.stream_metadata is None
1436
1437
1438async def test_v1_callback_idle_no_now_playing(
1439    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1440) -> None:
1441    """A live response with session_state empty_queue / now_playing=null renders idle, no raise."""
1442    details = await _details_for(initialized_provider)
1443    idle: dict[str, Any] = {
1444        "schema_version": "1",
1445        "station": {"name": "mammamiradio", "hosts": []},
1446        "stream": {"relative_url": "/stream", "audio_format": _V1_AUDIO_FORMAT},
1447        "now_playing": None,
1448        "up_next": [],
1449        "session_state": "empty_queue",
1450        "changed_at": 0.0,
1451    }
1452    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(idle, etag='W/"i"'))
1453    await initialized_provider._update_stream_metadata(details, 0)  # must not raise
1454    assert details.stream_metadata.title == "mammamiradio"
1455    assert details.stream_metadata.description is None
1456
1457
1458async def test_v1_callback_without_etag_polls_unconditionally(
1459    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1460) -> None:
1461    """
1462    If the addon omits the ETag header, the provider degrades gracefully.
1463
1464    Each tick is a fresh 200 with no If-None-Match sent; polling keeps working
1465    without the 304 optimization.
1466    """
1467    details = await _details_for(initialized_provider)
1468    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC))  # no etag
1469    await initialized_provider._update_stream_metadata(details, 0)
1470    assert details.stream_metadata.title == "Volare"
1471    await initialized_provider._update_stream_metadata(details, 0)
1472    second = mass_mock.http_session.get.call_args_list[1]
1473    assert "If-None-Match" not in second.kwargs.get("headers", {})
1474
1475
1476async def test_v1_callback_resets_on_artist_change_same_title(
1477    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1478) -> None:
1479    """
1480    Segment identity includes artist/host, not just type+title.
1481
1482    ``started_at`` is None when the addon does not know the segment start, so
1483    two consecutive segments sharing type and title (e.g. a same-title cover)
1484    must still be told apart by the other contract fields.
1485    """
1486    details = await _details_for(initialized_provider)
1487    cover = {**_V1_MUSIC, "now_playing": {**_V1_MUSIC["now_playing"], "artist": "Cover Band"}}
1488    mass_mock.http_session.get = MagicMock(
1489        side_effect=[
1490            _make_v1_ctx(_V1_MUSIC, etag='W/"a"'),
1491            _make_v1_ctx(cover, etag='W/"b"'),
1492        ]
1493    )
1494    await initialized_provider._update_stream_metadata(details, 0)  # Now (show_upcoming -> True)
1495    await initialized_provider._update_stream_metadata(details, 0)  # new artist -> reset
1496    desc = details.stream_metadata.description
1497    assert desc is None
1498    assert details.stream_metadata.artist == "Cover Band"
1499
1500
1501async def test_v1_callback_drops_stale_etag_when_header_disappears(
1502    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1503) -> None:
1504    """A 200 without an ETag clears the stored validator; polling becomes unconditional."""
1505    details = await _details_for(initialized_provider)
1506    mass_mock.http_session.get = MagicMock(
1507        side_effect=[
1508            _make_v1_ctx(_V1_MUSIC, etag='W/"a"'),
1509            _make_v1_ctx(_V1_MUSIC),  # ETag header disappears
1510            _make_v1_ctx(_V1_MUSIC),
1511        ]
1512    )
1513    await initialized_provider._update_stream_metadata(details, 0)
1514    await initialized_provider._update_stream_metadata(details, 0)
1515    second = mass_mock.http_session.get.call_args_list[1]
1516    assert second.kwargs["headers"]["If-None-Match"] == 'W/"a"'
1517    await initialized_provider._update_stream_metadata(details, 0)
1518    third = mass_mock.http_session.get.call_args_list[2]
1519    assert "If-None-Match" not in third.kwargs.get("headers", {})
1520
1521
1522async def test_v1_callback_unsupported_schema_keeps_prior(
1523    initialized_provider: MammamiradioProvider, mass_mock: MagicMock
1524) -> None:
1525    """
1526    A drifted now-playing schema is ignored mid-stream instead of mapped loosely.
1527
1528    Init raises on an unsupported schema; the mid-stream callback instead
1529    ignores the response and keeps the prior frame.
1530    """
1531    details = await _details_for(initialized_provider)
1532    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(_V1_MUSIC))
1533    await initialized_provider._update_stream_metadata(details, 0)
1534    prior = details.stream_metadata
1535    unsupported = {**_V1_MUSIC, "schema_version": "2"}
1536    mass_mock.http_session.get = MagicMock(return_value=_make_v1_ctx(unsupported))
1537    await initialized_provider._update_stream_metadata(details, 0)
1538    assert details.stream_metadata is prior
1539
1540
1541# ---------------------------------------------------------------------------
1542# Shared contract fixture (cross-repo golden copy)
1543# ---------------------------------------------------------------------------
1544
1545
1546def test_golden_fixture_maps_cleanly() -> None:
1547    """
1548    The cross-repo golden fixture renders through the v1 mapper.
1549
1550    The same bytes live in the addon repo (tests/integrations/golden/); the
1551    addon's contract-drift CI checksum-compares the two copies, so this test is
1552    the provider-side half of the shared contract fixture.
1553    """
1554    fixture = Path(__file__).parent / "fixtures" / "v1_now_playing.json"
1555    payload = json.loads(fixture.read_text(encoding="utf-8"))
1556    assert _supports_v1_schema(payload["schema_version"])
1557
1558    now_frame = _v1_to_stream_metadata(payload, show_upcoming=False)
1559    assert now_frame.title == "Volare"
1560    assert now_frame.artist == "Domenico Modugno"
1561    assert now_frame.album == "Mr Volare"
1562    assert now_frame.image_url == "https://example.test/art.jpg"
1563    assert now_frame.description is None
1564
1565    upcoming_frame = _v1_to_stream_metadata(payload, show_upcoming=True)
1566    assert upcoming_frame.description == "Up next: Sapore di Sale — Gino Paoli"
1567
1568
1569# ---------------------------------------------------------------------------
1570# Live integration smoke (opt-in via MAMMAMIRADIO_LIVE_URL)
1571# ---------------------------------------------------------------------------
1572
1573
1574async def test_live_stream_smoke() -> None:
1575    """
1576    Live smoke test against a running mammamiradio addon. Skipped by default.
1577
1578    Opt in by setting ``MAMMAMIRADIO_LIVE_URL`` (e.g. ``http://localhost:8000``).
1579    Verifies the init probe, browse, stream-details resolution, and the
1580    metadata fetch/mapping against a live addon; the audio path itself is not
1581    exercised.
1582    """
1583    live_url = os.environ.get("MAMMAMIRADIO_LIVE_URL")
1584    if not live_url:
1585        pytest.skip("MAMMAMIRADIO_LIVE_URL not set")
1586
1587    async with aiohttp.ClientSession() as session:
1588        mass = MagicMock()
1589        mass.http_session = session
1590        prov = _build_provider_with_url(mass, live_url)
1591
1592        # Init validates the v1 now-playing contract; it raises when the addon
1593        # is unreachable or older than 2.13 (no legacy fallback).
1594        await prov.handle_async_init()
1595        # Browse returns exactly one Radio entry.
1596        items = await prov.browse("mammamiradio://")
1597        assert len(items) == 1
1598        assert isinstance(items[0], Radio)
1599        # Stream details succeed against the live addon.
1600        details = await prov.get_stream_details(RADIO_ITEM_ID, MediaType.RADIO)
1601        assert isinstance(details.path, str)
1602        assert details.path.endswith("/stream")
1603        assert details.audio_format.content_type == ContentType.MP3
1604        # The live-metadata callback always polls the v1 now-playing endpoint.
1605        await prov._update_stream_metadata(details, 0)
1606        assert details.stream_metadata is not None
1607        assert details.stream_metadata.title
1608