/
/
/
1"""Tests for the audio overlay mixing in StreamsAudio."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from typing import TYPE_CHECKING, Any
7from unittest.mock import AsyncMock, MagicMock
8
9from music_assistant_models.enums import MediaType, StreamType
10from music_assistant_models.media_items import AudioFormat, ItemMapping
11
12from music_assistant.controllers.streams.audio import StreamsAudio, overlay_active
13
14if TYPE_CHECKING:
15 import pytest
16
17_PCM_FORMAT = AudioFormat(sample_rate=44100, bit_depth=16, channels=2)
18_MUSIC_CHUNKS = [b"chunk1", b"chunk2"]
19_OVERLAY_URL = "http://ambient.example/rain.mp3"
20
21
22def _make_streams_audio(provider: MagicMock | None = None) -> StreamsAudio:
23 """Build a StreamsAudio whose mass resolves the overlay provider to the given mock."""
24 mass = MagicMock()
25 mass.get_provider = MagicMock(return_value=provider)
26 return StreamsAudio(mass)
27
28
29def _make_queue(
30 *, enabled: bool = True, source: ItemMapping | None = None, volume: int = 100
31) -> MagicMock:
32 """Build a PlayerQueue double carrying just the overlay fields."""
33 queue = MagicMock()
34 queue.overlay_enabled = enabled
35 queue.overlay_source = source
36 queue.overlay_volume = volume
37 return queue
38
39
40def _make_source_mapping() -> ItemMapping:
41 return ItemMapping(
42 media_type=MediaType.SOUND_EFFECT,
43 item_id="rain",
44 provider="ambient",
45 name="Rain",
46 )
47
48
49async def _music_stream() -> AsyncGenerator[bytes]:
50 for chunk in _MUSIC_CHUNKS:
51 yield chunk
52
53
54# --- overlay_active ---
55
56
57def test_overlay_active() -> None:
58 """The overlay is only active when enabled AND a source is selected."""
59 assert overlay_active(_make_queue(enabled=True, source=_make_source_mapping()))
60 assert not overlay_active(_make_queue(enabled=True, source=None))
61 assert not overlay_active(_make_queue(enabled=False, source=_make_source_mapping()))
62
63
64# --- get_overlay_mixed_stream ---
65
66
67async def test_passthrough_when_provider_unavailable() -> None:
68 """An unavailable overlay provider degrades to unmodified music playback."""
69 audio = _make_streams_audio(provider=None)
70 queue = _make_queue(source=_make_source_mapping())
71 result = [
72 chunk async for chunk in audio.get_overlay_mixed_stream(queue, _music_stream(), _PCM_FORMAT)
73 ]
74 assert result == _MUSIC_CHUNKS
75
76
77async def test_passthrough_when_streamdetails_fail() -> None:
78 """An error resolving the overlay source degrades to unmodified music playback."""
79 provider = MagicMock()
80 provider.get_stream_details = AsyncMock(side_effect=RuntimeError("boom"))
81 audio = _make_streams_audio(provider=provider)
82 queue = _make_queue(source=_make_source_mapping())
83 result = [
84 chunk async for chunk in audio.get_overlay_mixed_stream(queue, _music_stream(), _PCM_FORMAT)
85 ]
86 assert result == _MUSIC_CHUNKS
87
88
89async def test_passthrough_when_stream_type_unsupported() -> None:
90 """An overlay source with a non file/url stream type degrades to unmodified playback."""
91 provider = MagicMock()
92 provider.get_stream_details = AsyncMock(
93 return_value=MagicMock(stream_type=StreamType.CUSTOM, path=None)
94 )
95 audio = _make_streams_audio(provider=provider)
96 queue = _make_queue(source=_make_source_mapping())
97 result = [
98 chunk async for chunk in audio.get_overlay_mixed_stream(queue, _music_stream(), _PCM_FORMAT)
99 ]
100 assert result == _MUSIC_CHUNKS
101
102
103async def test_passthrough_when_local_file_missing() -> None:
104 """An overlay source pointing at a non-existing file degrades to unmodified playback."""
105 provider = MagicMock()
106 provider.get_stream_details = AsyncMock(
107 return_value=MagicMock(stream_type=StreamType.LOCAL_FILE, path="/does/not/exist.wav")
108 )
109 audio = _make_streams_audio(provider=provider)
110 queue = _make_queue(source=_make_source_mapping())
111 result = [
112 chunk async for chunk in audio.get_overlay_mixed_stream(queue, _music_stream(), _PCM_FORMAT)
113 ]
114 assert result == _MUSIC_CHUNKS
115
116
117async def test_mixes_when_source_resolves(monkeypatch: pytest.MonkeyPatch) -> None:
118 """A resolvable overlay source is handed to the ffmpeg mixer with the queue's volume."""
119 provider = MagicMock()
120 provider.get_stream_details = AsyncMock(
121 return_value=MagicMock(stream_type=StreamType.HTTP, path=_OVERLAY_URL)
122 )
123 audio = _make_streams_audio(provider=provider)
124 queue = _make_queue(source=_make_source_mapping(), volume=55)
125
126 mixer_kwargs: dict[str, Any] = {}
127
128 async def _fake_mixer(**kwargs: Any) -> AsyncGenerator[bytes]:
129 mixer_kwargs.update(kwargs)
130 async for chunk in kwargs["audio_input"]:
131 yield b"mixed:" + chunk
132
133 monkeypatch.setattr(
134 "music_assistant.controllers.streams.audio.get_ffmpeg_overlay_stream", _fake_mixer
135 )
136 result = [
137 chunk async for chunk in audio.get_overlay_mixed_stream(queue, _music_stream(), _PCM_FORMAT)
138 ]
139 assert result == [b"mixed:" + chunk for chunk in _MUSIC_CHUNKS]
140 assert mixer_kwargs["overlay_input"] == _OVERLAY_URL
141 assert mixer_kwargs["overlay_volume"] == 55
142 assert mixer_kwargs["pcm_format"] is _PCM_FORMAT
143