/
/
/
1"""Shared fixtures for the streams controller tests."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import AsyncMock, patch
7
8import pytest
9
10from music_assistant.controllers.streams.controller import StreamsController
11from music_assistant.controllers.tasks import TasksController
12
13if TYPE_CHECKING:
14 from collections.abc import AsyncGenerator, Iterator
15
16 from music_assistant.mass import MusicAssistant
17
18
19@pytest.fixture
20async def streams_controller(mass_minimal: MusicAssistant) -> AsyncGenerator[StreamsController]:
21 """
22 Yield a StreamsController attached to a minimal server, closed afterwards.
23
24 :param mass_minimal: Minimal MusicAssistant instance.
25 """
26 mass_minimal.tasks = TasksController(mass_minimal)
27 await mass_minimal.tasks.setup(await mass_minimal.config.get_core_config("tasks"))
28 streams = StreamsController(mass_minimal)
29 mass_minimal.streams = streams
30 try:
31 yield streams
32 finally:
33 # close unconditionally: a failed assertion must not leave the socket bound
34 await streams.close()
35 await mass_minimal.tasks.close()
36
37
38@pytest.fixture
39def streamserver_fallback(
40 streams_controller: StreamsController,
41) -> Iterator[AsyncMock]:
42 """
43 Make the streamserver report a successful fallback to all interfaces.
44
45 :param streams_controller: StreamsController whose server should report the fallback.
46 """
47 server = streams_controller._server
48
49 async def setup(*, bind_port: int, **_kwargs: object) -> None:
50 server._bind_ip = None
51 server._bind_port = bind_port
52
53 with patch.object(server, "setup", AsyncMock(side_effect=setup)) as setup_mock:
54 yield setup_mock
55