music-assistant-server

1"""Tests for the AudioAnalysisController."""
2
3from __future__ import annotations
4
5import asyncio
6import unittest.mock
7from unittest.mock import AsyncMock, MagicMock
8
9import pytest
10from music_assistant_models.enums import ContentType, MediaType
11from music_assistant_models.media_items import AudioFormat
12
13from music_assistant.controllers.streams.audio_analysis import (
14    REALTIME_ANALYSIS_MAX_SESSIONS,
15    AudioAnalysisController,
16)
17from music_assistant.controllers.streams.audio_buffer import AudioBuffer, AudioBufferDiscarded
18from music_assistant.models.audio_analysis_provider import (
19    AnalysisSessionData,
20    AudioAnalysisProvider,
21)
22
23TEST_PCM_FORMAT = AudioFormat(
24    content_type=ContentType.PCM_S16LE,
25    sample_rate=44100,
26    bit_depth=16,
27    channels=2,
28)
29ONE_SECOND_CHUNK = b"\x00" * TEST_PCM_FORMAT.pcm_sample_size
30
31
32def _create_mock_provider(
33    instance_id: str = "prov_1",
34    name: str = "TestProvider",
35    domain: str = "test_domain",
36) -> MagicMock:
37    prov = MagicMock(spec=AudioAnalysisProvider)
38    prov.instance_id = instance_id
39    prov.name = name
40    prov.domain = domain
41    prov.available = True
42    prov.start_analysis = AsyncMock(return_value=True)
43    prov.process_pcm_chunk = AsyncMock()
44    prov.finalize = AsyncMock()
45    prov.cancel = AsyncMock()
46    return prov
47
48
49async def _send_chunks(audio_buffer: AudioBuffer, num_chunks: int) -> None:
50    """Fill the buffer with PCM chunks then signal EOF; the analysis worker reads from it."""
51    for _ in range(num_chunks):
52        await audio_buffer._put(ONE_SECOND_CHUNK)
53    await audio_buffer._set_eof()
54
55
56@pytest.fixture
57def mock_provider() -> MagicMock:
58    """Return a single mock AudioAnalysisProvider."""
59    return _create_mock_provider()
60
61
62@pytest.fixture
63def mock_mass(mock_provider: MagicMock) -> MagicMock:
64    """Return a mock MusicAssistant instance wired to mock_provider."""
65    mass = MagicMock()
66    mass._created_tasks = []
67
68    def _track_create_task(coro):  # type: ignore[no-untyped-def]
69        task = asyncio.ensure_future(coro)
70        mass._created_tasks.append(task)
71        return task
72
73    mass.create_task = MagicMock(side_effect=_track_create_task)
74    mass.get_providers = MagicMock(return_value=[mock_provider])
75    mass.get_provider = MagicMock(return_value=mock_provider)
76    return mass
77
78
79async def _await_tasks(mock_mass: MagicMock) -> None:
80    """Await all tasks created via mock_mass.create_task, including nested ones."""
81    awaited = 0
82    while awaited < len(mock_mass._created_tasks):
83        pending = mock_mass._created_tasks[awaited:]
84        awaited = len(mock_mass._created_tasks)
85        for task in pending:
86            # The idle-model monitor is a long-lived loop, not a one-shot; cancel it rather
87            # than block here waiting for it to finish.
88            coro = task.get_coro()
89            if coro is not None and "_monitor_idle_models" in getattr(coro, "__qualname__", ""):
90                task.cancel()
91        await asyncio.gather(*pending, return_exceptions=True)
92
93
94@pytest.fixture
95def mock_streams(mock_mass: MagicMock) -> MagicMock:
96    """Return a mock StreamsController with mass attached."""
97    streams = MagicMock()
98    streams.mass = mock_mass
99    return streams
100
101
102@pytest.fixture
103def mock_stream_details() -> MagicMock:
104    """Return mock StreamDetails for a test track."""
105    sd = MagicMock()
106    sd.seek_position = 0
107    sd.provider = "test_prov"
108    sd.media_type = "track"
109    sd.item_id = "test_123"
110    sd.uri = "test_prov://track/test_123"
111    # unknown duration: the completeness guard only applies when a duration is known
112    sd.duration = None
113    return sd
114
115
116@pytest.fixture
117def controller(mock_streams: MagicMock) -> AudioAnalysisController:
118    """Return an AudioAnalysisController wired to mock_streams."""
119    return AudioAnalysisController(mock_streams)
120
121
122@pytest.fixture
123def audio_buffer() -> AudioBuffer:
124    """Return a fresh AudioBuffer with test PCM format."""
125    return AudioBuffer(TEST_PCM_FORMAT)
126
127
128# -- Early returns --
129
130
131@pytest.mark.asyncio
132async def test_start_analysis_no_providers(
133    controller: AudioAnalysisController,
134    audio_buffer: AudioBuffer,
135    mock_stream_details: MagicMock,
136    mock_mass: MagicMock,
137) -> None:
138    """No providers available means no callbacks registered and no sessions."""
139    mock_mass.get_providers.return_value = []
140    await controller.start_analysis(audio_buffer, mock_stream_details)
141    assert len(audio_buffer._cancel_callbacks) == 0
142    assert len(controller._active_sessions) == 0
143
144
145@pytest.mark.asyncio
146async def test_start_analysis_duplicate_session(
147    controller: AudioAnalysisController,
148    audio_buffer: AudioBuffer,
149    mock_stream_details: MagicMock,
150) -> None:
151    """Second call with the same stream details is a no-op."""
152    await controller.start_analysis(audio_buffer, mock_stream_details)
153    assert len(controller._active_sessions) == 1
154
155    buf2 = AudioBuffer(TEST_PCM_FORMAT)
156    await controller.start_analysis(buf2, mock_stream_details)
157    # No callbacks registered on the second buffer
158    assert len(buf2._cancel_callbacks) == 0
159    # Still only one session
160    assert len(controller._active_sessions) == 1
161
162
163@pytest.mark.asyncio
164async def test_start_analysis_all_providers_fail(
165    controller: AudioAnalysisController,
166    audio_buffer: AudioBuffer,
167    mock_stream_details: MagicMock,
168    mock_provider: MagicMock,
169) -> None:
170    """All providers raising on start_analysis means no session is created."""
171    mock_provider.start_analysis.side_effect = RuntimeError("init failed")
172    await controller.start_analysis(audio_buffer, mock_stream_details)
173    assert len(controller._active_sessions) == 0
174    assert len(audio_buffer._cancel_callbacks) == 0
175
176
177# -- Happy path --
178
179
180@pytest.mark.asyncio
181async def test_chunks_delivered_to_provider(
182    controller: AudioAnalysisController,
183    audio_buffer: AudioBuffer,
184    mock_stream_details: MagicMock,
185    mock_provider: MagicMock,
186    mock_mass: MagicMock,
187) -> None:
188    """Provider receives all PCM chunks via the worker."""
189    await controller.start_analysis(audio_buffer, mock_stream_details)
190    await _send_chunks(audio_buffer, 3)
191    await _await_tasks(mock_mass)
192    assert mock_provider.process_pcm_chunk.call_count == 3
193
194
195@pytest.mark.asyncio
196async def test_finalize_called_on_eof(
197    controller: AudioAnalysisController,
198    audio_buffer: AudioBuffer,
199    mock_stream_details: MagicMock,
200    mock_provider: MagicMock,
201    mock_mass: MagicMock,
202) -> None:
203    """After EOF, provider.finalize is called with the session key."""
204    await controller.start_analysis(audio_buffer, mock_stream_details)
205    await _send_chunks(audio_buffer, 2)
206    await _await_tasks(mock_mass)
207    session_key = "test_prov://track/test_123"
208    mock_provider.finalize.assert_called_once_with(session_key)
209
210
211@pytest.mark.asyncio
212async def test_multiple_providers_receive_chunks(
213    controller: AudioAnalysisController,
214    audio_buffer: AudioBuffer,
215    mock_stream_details: MagicMock,
216    mock_mass: MagicMock,
217) -> None:
218    """Two providers both get all chunks and finalize."""
219    prov_a = _create_mock_provider(instance_id="prov_a", name="ProvA")
220    prov_b = _create_mock_provider(instance_id="prov_b", name="ProvB")
221    mock_mass.get_providers.return_value = [prov_a, prov_b]
222
223    def _get_prov(pid: str) -> MagicMock:
224        return {"prov_a": prov_a, "prov_b": prov_b}.get(pid, prov_a)
225
226    mock_mass.get_provider = MagicMock(side_effect=_get_prov)
227
228    await controller.start_analysis(audio_buffer, mock_stream_details)
229    await _send_chunks(audio_buffer, 3)
230    await _await_tasks(mock_mass)
231
232    assert prov_a.process_pcm_chunk.call_count == 3
233    assert prov_b.process_pcm_chunk.call_count == 3
234    prov_a.finalize.assert_called_once()
235    prov_b.finalize.assert_called_once()
236
237
238@pytest.mark.asyncio
239async def test_session_cleaned_up_after_finalize(
240    controller: AudioAnalysisController,
241    audio_buffer: AudioBuffer,
242    mock_stream_details: MagicMock,
243    mock_mass: MagicMock,
244) -> None:
245    """Internal dicts are empty after finalize completes."""
246    await controller.start_analysis(audio_buffer, mock_stream_details)
247    await _send_chunks(audio_buffer, 1)
248    await _await_tasks(mock_mass)
249    assert len(controller._active_sessions) == 0
250    assert len(controller._workers) == 0
251
252
253# -- Cancel path --
254
255
256@pytest.mark.asyncio
257async def test_cancel_on_buffer_clear(
258    controller: AudioAnalysisController,
259    audio_buffer: AudioBuffer,
260    mock_stream_details: MagicMock,
261    mock_provider: MagicMock,
262    mock_mass: MagicMock,
263) -> None:
264    """Clearing the buffer triggers provider.cancel."""
265    await controller.start_analysis(audio_buffer, mock_stream_details)
266    await audio_buffer._put(ONE_SECOND_CHUNK)
267    await audio_buffer._put(ONE_SECOND_CHUNK)
268    await audio_buffer.clear()
269    await _await_tasks(mock_mass)
270    session_key = "test_prov://track/test_123"
271    mock_provider.cancel.assert_called_once_with(session_key)
272
273
274@pytest.mark.asyncio
275async def test_session_cleaned_up_after_cancel(
276    controller: AudioAnalysisController,
277    audio_buffer: AudioBuffer,
278    mock_stream_details: MagicMock,
279    mock_mass: MagicMock,
280) -> None:
281    """Internal dicts are empty after cancel."""
282    await controller.start_analysis(audio_buffer, mock_stream_details)
283    await audio_buffer._put(ONE_SECOND_CHUNK)
284    await audio_buffer.clear()
285    await _await_tasks(mock_mass)
286    assert len(controller._active_sessions) == 0
287    assert len(controller._workers) == 0
288
289
290@pytest.mark.asyncio
291async def test_worker_cancelled_on_buffer_clear(
292    controller: AudioAnalysisController,
293    audio_buffer: AudioBuffer,
294    mock_stream_details: MagicMock,
295    mock_mass: MagicMock,
296) -> None:
297    """Worker task is cancelled when buffer is cleared."""
298    await controller.start_analysis(audio_buffer, mock_stream_details)
299    session_key = "test_prov://track/test_123"
300    worker = controller._workers.get(session_key)
301    assert worker is not None
302    await audio_buffer.clear()
303    await _await_tasks(mock_mass)
304    assert worker.cancelled() or worker.done()
305
306
307@pytest.mark.asyncio
308async def test_realtime_sessions_capped_evicting_oldest(
309    controller: AudioAnalysisController,
310    mock_provider: MagicMock,
311    mock_mass: MagicMock,
312) -> None:
313    """Starting past the cap evicts the oldest realtime session and cancels its providers."""
314    keys: list[str] = []
315    for i in range(REALTIME_ANALYSIS_MAX_SESSIONS + 1):
316        sd = MagicMock()
317        sd.uri = f"test://track/{i}"
318        sd.provider = "test"
319        sd.item_id = f"t{i}"
320        sd.media_type = MediaType.TRACK
321        sd.queue_id = "queue-1"  # same queue, so the per-queue cap applies
322        keys.append(sd.uri)
323        await controller.start_analysis(AudioBuffer(TEST_PCM_FORMAT), sd)
324
325    # Only the cap's worth survive; the oldest was evicted, the newest kept.
326    assert len(controller._workers) == REALTIME_ANALYSIS_MAX_SESSIONS
327    assert keys[0] not in controller._workers
328    assert keys[0] not in controller._active_sessions
329    assert keys[-1] in controller._workers
330    mock_provider.cancel.assert_any_call(keys[0])
331
332    # Drain the pending reader/cancel tasks so nothing leaks past the test.
333    for task in mock_mass._created_tasks:
334        if not task.done():
335            task.cancel()
336    await asyncio.gather(*mock_mass._created_tasks, return_exceptions=True)
337
338
339@pytest.mark.asyncio
340async def test_realtime_session_cap_is_per_queue(
341    controller: AudioAnalysisController,
342    mock_provider: MagicMock,
343    mock_mass: MagicMock,
344) -> None:
345    """Concurrent queues are capped independently; one queue's burst can't evict another's."""
346
347    async def _start(uri: str, queue_id: str) -> None:
348        sd = MagicMock()
349        sd.uri = uri
350        sd.provider = "test"
351        sd.item_id = uri
352        sd.media_type = MediaType.TRACK
353        sd.queue_id = queue_id
354        await controller.start_analysis(AudioBuffer(TEST_PCM_FORMAT), sd)
355
356    # Fill two queues to the cap each (interleaved).
357    for i in range(REALTIME_ANALYSIS_MAX_SESSIONS):
358        await _start(f"a://{i}", "queueA")
359        await _start(f"b://{i}", "queueB")
360
361    # Both queues are fully populated and nothing was evicted across queues.
362    assert len(controller._workers) == 2 * REALTIME_ANALYSIS_MAX_SESSIONS
363    mock_provider.cancel.assert_not_called()
364
365    # One more in queueA evicts queueA's oldest only; queueB is untouched.
366    await _start("a://new", "queueA")
367    assert "a://0" not in controller._workers
368    assert "b://0" in controller._workers
369    mock_provider.cancel.assert_called_once_with("a://0")
370
371    for task in mock_mass._created_tasks:
372        if not task.done():
373            task.cancel()
374    await asyncio.gather(*mock_mass._created_tasks, return_exceptions=True)
375
376
377def _unloadable_provider() -> MagicMock:
378    prov = MagicMock(spec=AudioAnalysisProvider)
379    prov.has_unloadable_models = True
380    prov.available = True
381    prov.unload_idle_models = AsyncMock()
382    return prov
383
384
385@pytest.mark.asyncio
386async def test_idle_monitor_unloads_models_when_idle(
387    controller: AudioAnalysisController,
388    monkeypatch: pytest.MonkeyPatch,
389) -> None:
390    """With no active sessions past the idle timeout, unloadable providers are unloaded."""
391    prov = _unloadable_provider()
392    monkeypatch.setattr(controller.__class__, "providers", property(lambda _self: [prov]))
393    monkeypatch.setattr(
394        "music_assistant.controllers.streams.audio_analysis.MODEL_IDLE_CHECK_INTERVAL_SECONDS", 0.01
395    )
396    monkeypatch.setattr(
397        "music_assistant.controllers.streams.audio_analysis.MODEL_IDLE_UNLOAD_SECONDS", 0.0
398    )
399
400    controller._mark_analysis_activity()  # starts the monitor
401    assert controller._idle_unload_task is not None
402    await asyncio.wait_for(controller._idle_unload_task, timeout=2.0)
403
404    prov.unload_idle_models.assert_awaited_once()
405
406
407@pytest.mark.asyncio
408async def test_idle_monitor_keeps_models_while_sessions_active(
409    controller: AudioAnalysisController,
410    monkeypatch: pytest.MonkeyPatch,
411) -> None:
412    """Models are not unloaded while an analysis session is still active."""
413    prov = _unloadable_provider()
414    monkeypatch.setattr(controller.__class__, "providers", property(lambda _self: [prov]))
415    monkeypatch.setattr(
416        "music_assistant.controllers.streams.audio_analysis.MODEL_IDLE_CHECK_INTERVAL_SECONDS", 0.01
417    )
418    monkeypatch.setattr(
419        "music_assistant.controllers.streams.audio_analysis.MODEL_IDLE_UNLOAD_SECONDS", 0.0
420    )
421
422    controller._active_sessions["sess"] = {"prov"}
423    controller._mark_analysis_activity()
424    await asyncio.sleep(0.05)  # several monitor ticks
425
426    prov.unload_idle_models.assert_not_called()
427
428    assert controller._idle_unload_task is not None
429    controller._idle_unload_task.cancel()
430    await asyncio.gather(controller._idle_unload_task, return_exceptions=True)
431
432
433@pytest.mark.asyncio
434async def test_idle_monitor_keeps_models_while_finalize_in_flight(
435    controller: AudioAnalysisController,
436    mock_mass: MagicMock,
437    monkeypatch: pytest.MonkeyPatch,
438) -> None:
439    """An in-flight finalize keeps the models loaded, and releases them once it completes."""
440    prov = _unloadable_provider()
441    prov.instance_id = "prov_1"
442    release = asyncio.Event()
443
444    async def _blocking_finalize(_session_id: str) -> None:
445        await release.wait()
446
447    prov.finalize = AsyncMock(side_effect=_blocking_finalize)
448    mock_mass.get_provider = MagicMock(return_value=prov)
449    monkeypatch.setattr(controller.__class__, "providers", property(lambda _self: [prov]))
450    monkeypatch.setattr(
451        "music_assistant.controllers.streams.audio_analysis.MODEL_IDLE_CHECK_INTERVAL_SECONDS", 0.01
452    )
453    monkeypatch.setattr(
454        "music_assistant.controllers.streams.audio_analysis.MODEL_IDLE_UNLOAD_SECONDS", 0.0
455    )
456
457    controller._active_sessions["sess"] = {"prov_1"}
458    controller._mark_analysis_activity()
459    controller._finalize_providers("sess")
460    assert "sess" not in controller._active_sessions
461
462    await asyncio.sleep(0.05)  # several monitor ticks while the finalize is still running
463    prov.unload_idle_models.assert_not_called()
464
465    release.set()
466    assert controller._idle_unload_task is not None
467    await asyncio.wait_for(controller._idle_unload_task, timeout=2.0)
468    prov.unload_idle_models.assert_awaited_once()
469
470
471# -- Edge cases --
472
473
474@pytest.mark.asyncio
475async def test_worker_drops_session_when_falling_behind(
476    controller: AudioAnalysisController,
477    mock_stream_details: MagicMock,
478    mock_provider: MagicMock,
479    mock_mass: MagicMock,
480) -> None:
481    """If a chunk has been evicted before analysis reads it, the session is dropped, not finalized."""
482    # A buffer that yields one chunk, then reports the next as already evicted.
483    fake_buffer = MagicMock()
484    fake_buffer.pcm_format = TEST_PCM_FORMAT
485    fake_buffer.first_buffered_chunk = 0
486    fake_buffer.read_chunk_for_analysis = AsyncMock(
487        side_effect=[ONE_SECOND_CHUNK, AudioBufferDiscarded]
488    )
489
490    await controller.start_analysis(fake_buffer, mock_stream_details)
491    await _await_tasks(mock_mass)
492
493    session_key = "test_prov://track/test_123"
494    # The one available chunk was processed; falling behind drops the session (cancel, no finalize).
495    mock_provider.process_pcm_chunk.assert_called_once()
496    mock_provider.cancel.assert_called_once_with(session_key)
497    mock_provider.finalize.assert_not_called()
498    assert session_key not in controller._active_sessions
499    assert session_key not in controller._workers
500
501
502@pytest.mark.asyncio
503async def test_provider_error_during_chunk_processing_evicts_provider(
504    controller: AudioAnalysisController,
505    audio_buffer: AudioBuffer,
506    mock_stream_details: MagicMock,
507    mock_provider: MagicMock,
508    mock_mass: MagicMock,
509) -> None:
510    """
511    Provider that raises in process_pcm_chunk is evicted from the session.
512
513    The first chunk processes successfully. The second chunk's exception
514    triggers eviction. The third chunk is not delivered. The provider's
515    cancel hook is dispatched (replaces finalize for evicted providers).
516    """
517    call_count = 0
518
519    async def _flaky_process(_session_id: str, _chunk: bytes) -> None:
520        nonlocal call_count
521        call_count += 1
522        if call_count == 2:
523            raise RuntimeError("transient error")
524
525    mock_provider.process_pcm_chunk = AsyncMock(side_effect=_flaky_process)
526    await controller.start_analysis(audio_buffer, mock_stream_details)
527    await _send_chunks(audio_buffer, 3)
528    await _await_tasks(mock_mass)
529
530    # Provider was called twice: chunk 1 (success), chunk 2 (raised → evicted)
531    assert call_count == 2
532    # Evicted provider does NOT get finalize, but does get cancel
533    mock_provider.finalize.assert_not_called()
534    mock_provider.cancel.assert_called_once()
535
536
537@pytest.mark.asyncio
538async def test_provider_error_during_start(
539    controller: AudioAnalysisController,
540    audio_buffer: AudioBuffer,
541    mock_stream_details: MagicMock,
542    mock_mass: MagicMock,
543) -> None:
544    """One provider fails start_analysis, another succeeds. Successful one works."""
545    prov_fail = _create_mock_provider(instance_id="prov_fail", name="FailProv")
546    prov_fail.start_analysis.side_effect = RuntimeError("init failed")
547    prov_ok = _create_mock_provider(instance_id="prov_ok", name="OkProv")
548    mock_mass.get_providers.return_value = [prov_fail, prov_ok]
549    mock_mass.get_provider = MagicMock(return_value=prov_ok)
550
551    await controller.start_analysis(audio_buffer, mock_stream_details)
552    await _send_chunks(audio_buffer, 2)
553    await _await_tasks(mock_mass)
554
555    assert prov_ok.process_pcm_chunk.call_count == 2
556    prov_ok.finalize.assert_called_once()
557    prov_fail.process_pcm_chunk.assert_not_called()
558    prov_fail.finalize.assert_not_called()
559
560
561@pytest.mark.asyncio
562async def test_slow_provider_removed_after_timeout(
563    controller: AudioAnalysisController,
564    audio_buffer: AudioBuffer,
565    mock_stream_details: MagicMock,
566    mock_mass: MagicMock,
567) -> None:
568    """A provider that exceeds the chunk timeout is removed from the session."""
569    prov_slow = _create_mock_provider(instance_id="prov_slow", name="SlowProv")
570    never_done = asyncio.Event()
571
572    async def _hang(_session_id: str, _chunk: bytes) -> None:
573        await never_done.wait()
574
575    prov_slow.process_pcm_chunk = AsyncMock(side_effect=_hang)
576
577    prov_fast = _create_mock_provider(instance_id="prov_fast", name="FastProv")
578    mock_mass.get_providers.return_value = [prov_slow, prov_fast]
579
580    def _get_prov(pid: str) -> MagicMock:
581        return {"prov_slow": prov_slow, "prov_fast": prov_fast}[pid]
582
583    mock_mass.get_provider = MagicMock(side_effect=_get_prov)
584
585    with unittest.mock.patch(
586        "music_assistant.controllers.streams.audio_analysis.CHUNK_HANG_GUARD_SECONDS",
587        0.1,
588    ):
589        await controller.start_analysis(audio_buffer, mock_stream_details)
590        await _send_chunks(audio_buffer, 3)
591        await _await_tasks(mock_mass)
592
593    assert prov_fast.process_pcm_chunk.call_count == 3
594    prov_fast.finalize.assert_called_once()
595    assert prov_slow.process_pcm_chunk.call_count == 1
596
597
598@pytest.mark.asyncio
599async def test_provider_rejects_analysis(
600    controller: AudioAnalysisController,
601    mock_mass: MagicMock,
602    mock_stream_details: MagicMock,
603    mock_provider: MagicMock,
604) -> None:
605    """Controller skips provider when start_analysis returns False."""
606    mock_provider.start_analysis = AsyncMock(return_value=False)
607    mock_mass.get_providers.return_value = [mock_provider]
608
609    audio_buffer = AudioBuffer(TEST_PCM_FORMAT)
610    await controller.start_analysis(audio_buffer, mock_stream_details)
611
612    assert not controller._active_sessions
613
614
615@pytest.mark.asyncio
616async def test_finalize_cleans_up_provider_sessions() -> None:
617    """Verify provider._sessions is cleaned up after finalize."""
618    provider = MagicMock(spec=AudioAnalysisProvider)
619    provider.logger = MagicMock()
620    provider._sessions = {"test_session": MagicMock(spec=AnalysisSessionData)}
621    provider._finalize_tasks = set()
622    provider.unloading = False
623    provider._finalize = AsyncMock(return_value=None)
624
625    await AudioAnalysisProvider.finalize(provider, "test_session")
626
627    provider._finalize.assert_called_once_with("test_session")
628    assert "test_session" not in provider._sessions
629
630
631@pytest.mark.asyncio
632async def test_provider_start_analysis_uses_media_type_for_version_gating() -> None:
633    """Version gating must include media_type so analyses do not collide across item types."""
634    provider = MagicMock(spec=AudioAnalysisProvider)
635    provider.mass = MagicMock()
636    provider.mass.streams.audio_analysis.get_audio_analysis_version = AsyncMock(return_value=None)
637    provider._sessions = {}
638    provider._start_analysis = AsyncMock(return_value=True)
639    provider.domain = "test_domain"
640    provider.analysis_version = 1
641    provider.max_analysis_duration = None
642    provider.unloading = False
643
644    streamdetails = MagicMock()
645    streamdetails.item_id = "shared_id"
646    streamdetails.provider = "test_prov"
647    streamdetails.media_type = MediaType.RADIO
648
649    accepted = await AudioAnalysisProvider.start_analysis(
650        provider,
651        session_id="session_1",
652        streamdetails=streamdetails,
653        audio_format=TEST_PCM_FORMAT,
654    )
655
656    assert accepted is True
657    provider.mass.streams.audio_analysis.get_audio_analysis_version.assert_awaited_once_with(
658        "shared_id",
659        "test_prov",
660        "test_domain",
661        media_type=MediaType.RADIO,
662    )
663
664
665@pytest.mark.asyncio
666async def test_finalize_swallows_finalize_exception_and_cleans_up() -> None:
667    """
668    Verify provider._sessions is cleaned up even when _finalize raises.
669
670    The finalize wrapper catches _finalize exceptions and logs ERROR; it must
671    not propagate them to the controller, and must still pop the session.
672    """
673    provider = MagicMock(spec=AudioAnalysisProvider)
674    provider.logger = MagicMock()
675    provider._sessions = {"test_session": MagicMock(spec=AnalysisSessionData)}
676    provider._finalize_tasks = set()
677    provider.unloading = False
678    provider._finalize = AsyncMock(side_effect=RuntimeError("analysis failed"))
679
680    # MUST NOT raise — exception is swallowed and logged
681    await AudioAnalysisProvider.finalize(provider, "test_session")
682
683    assert "test_session" not in provider._sessions
684    provider.logger.error.assert_called_once()
685