/
/
/
1"""Tests for aborting per-item stream responses of superseded sessions."""
2
3from __future__ import annotations
4
5from unittest.mock import MagicMock
6
7from music_assistant.controllers.streams.controller import StreamsController
8
9
10def _make_controller() -> StreamsController:
11 """Build a bare controller carrying only the open-streams registry."""
12 controller = StreamsController.__new__(StreamsController)
13 controller._open_item_streams = {}
14 return controller
15
16
17def test_only_responses_of_stale_sessions_are_aborted() -> None:
18 """The session that owns playback keeps its responses; every other one dies."""
19 controller = _make_controller()
20 stale, current = MagicMock(), MagicMock()
21 controller._open_item_streams["queue-1"] = [("old", stale), ("new", current)]
22
23 controller.close_superseded_item_streams("queue-1", "new")
24
25 stale.transport.abort.assert_called_once()
26 current.transport.abort.assert_not_called()
27
28
29def test_a_cleared_session_aborts_every_open_response() -> None:
30 """A stopped queue owns no session, so nothing may keep streaming."""
31 controller = _make_controller()
32 first, second = MagicMock(), MagicMock()
33 controller._open_item_streams["queue-1"] = [("a", first), ("b", second)]
34
35 controller.close_superseded_item_streams("queue-1", None)
36
37 first.transport.abort.assert_called_once()
38 second.transport.abort.assert_called_once()
39
40
41def test_an_already_gone_transport_is_left_alone() -> None:
42 """A response whose connection already closed has nothing left to abort."""
43 controller = _make_controller()
44 gone = MagicMock()
45 gone.transport = None
46 controller._open_item_streams["queue-1"] = [("old", gone)]
47
48 controller.close_superseded_item_streams("queue-1", "new")
49
50
51def test_a_queue_without_open_responses_is_a_noop() -> None:
52 """Rotating a session on an idle queue must not error."""
53 _make_controller().close_superseded_item_streams("queue-x", "s")
54