music-assistant-server

1"""Shared fixtures for the streams controller tests."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7import pytest
8
9from music_assistant.controllers.streams.controller import StreamsController
10from music_assistant.controllers.tasks import TasksController
11
12if TYPE_CHECKING:
13    from collections.abc import AsyncGenerator
14
15    from music_assistant.mass import MusicAssistant
16
17
18@pytest.fixture
19async def streams_controller(mass_minimal: MusicAssistant) -> AsyncGenerator[StreamsController]:
20    """
21    Yield a StreamsController attached to a minimal server, closed afterwards.
22
23    :param mass_minimal: Minimal MusicAssistant instance.
24    """
25    mass_minimal.tasks = TasksController(mass_minimal)
26    await mass_minimal.tasks.setup(await mass_minimal.config.get_core_config("tasks"))
27    streams = StreamsController(mass_minimal)
28    mass_minimal.streams = streams
29    try:
30        yield streams
31    finally:
32        # close unconditionally: a failed assertion must not leave the socket bound
33        await streams.close()
34        await mass_minimal.tasks.close()
35