/
/
/
1"""End-to-end integration test for the streaming background scan."""
2
3from __future__ import annotations
4
5import asyncio
6from pathlib import Path
7from typing import TYPE_CHECKING
8from unittest.mock import AsyncMock, MagicMock
9
10import pytest
11from music_assistant_models.enums import ContentType, MediaType
12from music_assistant_models.media_items import AudioFormat
13
14from music_assistant.constants import CONF_LOG_LEVEL
15from music_assistant.controllers.streams.audio_analysis import AudioAnalysisController
16from music_assistant.helpers.ffmpeg import FFMpeg
17from music_assistant.models.audio_analysis import AudioAnalysisData
18from music_assistant.providers.loudness_analysis.provider import (
19 CONF_WRITE_REPLAYGAIN_TAGS,
20 LoudnessAnalysisProvider,
21)
22
23if TYPE_CHECKING:
24 from collections.abc import AsyncGenerator
25
26 from music_assistant_models.streamdetails import StreamDetails
27
28
29FIXTURE_AUDIO = Path(__file__).parent.parent / "fixtures" / "audio" / "short_test.flac"
30
31
32async def _real_get_media_stream(
33 sd: StreamDetails, pcm_format: AudioFormat, **_kwargs: object
34) -> AsyncGenerator[bytes]:
35 """
36 Real-ffmpeg stand-in for mass.streams.audio.get_media_stream.
37
38 Mirrors the wait-then-close pattern in audio.py:466-528 so close() doesn't
39 hit the SIGINT path on Windows when the process is still running.
40 """
41 assert isinstance(sd.path, str)
42 proc = FFMpeg(
43 audio_input=sd.path,
44 input_format=sd.audio_format,
45 output_format=pcm_format,
46 collect_log_history=True,
47 )
48 try:
49 await proc.start()
50 async for chunk in proc.iter_chunked(pcm_format.pcm_sample_size):
51 yield chunk
52 await proc.wait_with_timeout(5)
53 finally:
54 await proc.close()
55
56
57@pytest.mark.skipif(not FIXTURE_AUDIO.exists(), reason="fixture FLAC missing")
58async def test_streaming_background_scan_loudness_end_to_end() -> None:
59 """
60 Drive a real LoudnessAnalysisProvider through _run_background_streaming_for_track.
61
62 Verifies:
63 - An audio_analysis row is written (captured via mocked set_audio_analysis)
64 - The analysis contains a plausible loudness value
65 - The session is removed from _active_sessions
66 """
67 captured_rows: list[tuple[str, str, AudioAnalysisData]] = []
68
69 async def _capture_set(**kwargs: object) -> None:
70 captured_rows.append(
71 (
72 str(kwargs["item_id"]),
73 str(kwargs["aa_provider_domain"]),
74 kwargs["analysis"], # type: ignore[arg-type]
75 )
76 )
77
78 mass = MagicMock()
79 mass.streams.audio_analysis.set_audio_analysis = AsyncMock(side_effect=_capture_set)
80 mass.streams.audio_analysis.get_audio_analysis_version = AsyncMock(return_value=None)
81
82 manifest = MagicMock()
83 manifest.domain = "loudness_analysis"
84 config = MagicMock()
85 config.instance_id = "loudness_analysis_test"
86 config.values = {}
87 # write_replaygain_tags=False so post_analysis is a no-op; log_level must be a valid string
88 config.get_value = MagicMock(
89 side_effect=lambda key: {
90 CONF_LOG_LEVEL: "GLOBAL",
91 CONF_WRITE_REPLAYGAIN_TAGS: False,
92 }.get(key, "GLOBAL")
93 )
94
95 provider = LoudnessAnalysisProvider(mass, manifest, config, supported_features=set())
96 # domain comes from manifest.domain, instance_id from config.instance_id (both already set)
97 provider.available = True
98
99 # Wire get_provider so the controller can look up the provider by instance_id
100 mass.get_provider = MagicMock(return_value=provider)
101
102 # create_task schedules real asyncio tasks so finalize can run
103 created_tasks: list[asyncio.Task[None]] = []
104
105 def _create_task(coro: object) -> asyncio.Task[None]:
106 task: asyncio.Task[None] = asyncio.create_task(coro) # type: ignore[arg-type]
107 created_tasks.append(task)
108 return task
109
110 mass.create_task = MagicMock(side_effect=_create_task)
111 mass.logger.getChild = MagicMock(return_value=MagicMock())
112
113 mass.streams.audio.get_media_stream = _real_get_media_stream
114
115 streams = MagicMock()
116 streams.mass = mass
117 controller = AudioAnalysisController(streams)
118
119 streamdetails = MagicMock()
120 streamdetails.path = str(FIXTURE_AUDIO)
121 streamdetails.uri = f"track://test/{FIXTURE_AUDIO.name}"
122 streamdetails.audio_format = AudioFormat(
123 content_type=ContentType.FLAC,
124 sample_rate=44100,
125 bit_depth=16,
126 channels=1,
127 )
128 streamdetails.item_id = "fixture-track"
129 streamdetails.provider = "filesystem_local"
130 streamdetails.media_type = MediaType.TRACK
131 # None is not VolumeNormalizationMode.DISABLED, so loudness analysis proceeds
132 streamdetails.volume_normalization_mode = None
133
134 try:
135 await controller._run_background_streaming_for_track(streamdetails, [provider])
136 finalize_tasks = [
137 task for task in created_tasks if task is not controller._idle_unload_task
138 ]
139 await asyncio.wait_for(asyncio.gather(*finalize_tasks), timeout=5)
140 finally:
141 await controller.close()
142
143 # Assertions
144 assert len(captured_rows) == 1, f"expected exactly one analysis row; got {len(captured_rows)}"
145 item_id, aa_domain, analysis = captured_rows[0]
146 assert item_id == "fixture-track"
147 assert aa_domain == "loudness_analysis"
148 assert analysis.loudness_integrated is not None, "loudness_integrated must be populated"
149 measured = analysis.loudness_integrated
150 assert -70.0 < measured < 0.0, (
151 f"loudness {measured} LUFS is outside the plausible range (-70, 0)"
152 )
153
154 assert streamdetails.uri not in controller._active_sessions, (
155 "session must be cleaned up from _active_sessions"
156 )
157