/
/
/
1"""Tests for the MPD player setup flow."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator, Callable
6from typing import cast
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from music_assistant_models.config_entries import PlayerConfig
11from music_assistant_models.enums import FlowStepType
12
13from music_assistant.constants import CONF_PASSWORD, CONF_PLAYERS
14from music_assistant.mass import MusicAssistant
15from music_assistant.providers.mpd.player import MPDPlayer
16
17
18@pytest.fixture
19async def flow_mass(mass_minimal: MusicAssistant) -> AsyncGenerator[MusicAssistant]:
20 """
21 Provide a minimal server suitable for driving player setup flows through the engine.
22
23 Builds on mass_minimal (config controller only) and stubs the players controller
24 surface the flow engine touches, mirroring the flow_mass fixture in
25 tests/controllers/config/test_setup_flows.py.
26 """
27 mass_minimal.players = MagicMock()
28 mass_minimal.players.on_player_config_change = AsyncMock()
29 try:
30 yield mass_minimal
31 finally:
32 for flow in list(mass_minimal.config._setup_flows.values()):
33 await mass_minimal.config._abort_flow(flow, reason="aborted")
34 if (sweep_handle := mass_minimal.config._flow_sweep_handle) is not None:
35 sweep_handle.cancel()
36
37
38async def test_get_config_entries_excludes_password(
39 make_mpd_player: Callable[..., MPDPlayer],
40) -> None:
41 """The password is flow-managed now and must not appear in the options config entries."""
42 player = make_mpd_player()
43 entries = await player.get_config_entries()
44 assert CONF_PASSWORD not in {entry.key for entry in entries}
45
46
47async def test_on_config_updated_reads_password_via_setup_value(
48 make_mpd_player: Callable[..., MPDPlayer],
49) -> None:
50 """on_config_updated resolves the password from setup_data, not the options config."""
51 player = make_mpd_player()
52 cast("MagicMock", player.mass.config.get).return_value = {CONF_PASSWORD: "encrypted-blob"}
53 cast("MagicMock", player.mass.config.decrypt_string).return_value = "secret"
54 with patch.object(player, "_connect", AsyncMock()):
55 await player.on_config_updated()
56 assert player.password == "secret"
57 assert player.needs_setup is False
58 assert player.setup_reason is None
59
60
61async def test_setup_flow_collects_and_persists_password(
62 flow_mass: MusicAssistant, make_mpd_player: Callable[..., MPDPlayer]
63) -> None:
64 """The setup flow shows a single password field and persists it encrypted."""
65 player_id = "mpd_test"
66 player = make_mpd_player(player_id)
67 flow_mass.config.set(
68 f"{CONF_PLAYERS}/{player_id}",
69 {"player_id": player_id, "provider": player.provider_id, "enabled": True},
70 )
71 player_config = PlayerConfig(values={}, provider=player.provider_id, player_id=player_id)
72 with (
73 patch.object(flow_mass.players, "get_player", return_value=player),
74 patch.object(flow_mass.config, "get_player_config", AsyncMock(return_value=player_config)),
75 ):
76 step = await flow_mass.config.setup_player(player_id)
77 assert step.type == FlowStepType.FORM
78 assert {entry.key for entry in step.entries} == {CONF_PASSWORD}
79 finish_step = await flow_mass.config.submit_setup_flow(
80 step.flow_id, {CONF_PASSWORD: "secret"}
81 )
82 assert finish_step.type == FlowStepType.FINISH
83 assert finish_step.result == {"player_id": player_id}
84 raw_conf = flow_mass.config.get(f"{CONF_PLAYERS}/{player_id}")
85 assert flow_mass.config.decrypt_string(raw_conf["setup_data"][CONF_PASSWORD]) == "secret"
86