/
/
/
1"""Tests for the Snapcast provider."""
2
3from __future__ import annotations
4
5import logging
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import pytest
9
10from music_assistant.constants import CONF_LOG_LEVEL, VERBOSE_LOG_LEVEL
11from music_assistant.models.player_provider import PlayerProvider
12from music_assistant.providers.snapcast.provider import SnapCastProvider
13
14
15def _provider_with_log_level(level: int) -> SnapCastProvider:
16 provider = SnapCastProvider.__new__(SnapCastProvider)
17 provider.logger = logging.getLogger("snapcast-provider-test")
18 provider.logger.setLevel(level)
19 return provider
20
21
22@pytest.mark.parametrize(
23 ("provider_level", "snapcast_level"),
24 [
25 (logging.INFO, logging.WARNING),
26 (logging.DEBUG, logging.INFO),
27 (VERBOSE_LOG_LEVEL, logging.DEBUG),
28 ],
29)
30def test_snapcast_library_log_level(provider_level: int, snapcast_level: int) -> None:
31 """Snapcast logging stays quieter than the provider unless verbose logging is enabled."""
32 provider = _provider_with_log_level(provider_level)
33
34 with patch("music_assistant.providers.snapcast.provider.logging.getLogger") as get_logger:
35 provider._set_snapcast_log_level()
36
37 get_logger.assert_called_once_with("snapcast")
38 get_logger.return_value.setLevel.assert_called_once_with(snapcast_level)
39
40
41async def test_log_level_config_update_realigns_snapcast_logger() -> None:
42 """A log-level-only config update immediately realigns Snapcast logging."""
43 provider = _provider_with_log_level(logging.INFO)
44 config = MagicMock()
45 changed_keys = {f"values/{CONF_LOG_LEVEL}"}
46
47 with (
48 patch.object(PlayerProvider, "update_config", new_callable=AsyncMock) as update_config,
49 patch.object(SnapCastProvider, "_set_snapcast_log_level") as set_log_level,
50 ):
51 await provider.update_config(config, changed_keys)
52
53 update_config.assert_awaited_once_with(config, changed_keys)
54 set_log_level.assert_called_once_with()
55
56
57async def test_unload_unregisters_mdns_services() -> None:
58 """Unloading the provider unregisters the mdns records of the built-in Snapserver."""
59 provider = _provider_with_log_level(logging.INFO)
60 provider.mass = MagicMock()
61 unregister = AsyncMock()
62 provider.mass.discovery.aiozc.async_unregister_service = unregister
63 provider._snapserver = MagicMock(clients=[])
64 provider._snapcast_ma_streams = {}
65 provider._snapserver_runner = None
66 infos = {"-http": MagicMock(), "": MagicMock()}
67 provider._zc_services = dict(infos)
68
69 await provider.unload()
70
71 assert [call.args[0] for call in unregister.await_args_list] == list(infos.values())
72 assert provider._zc_services == {}
73
74
75def test_lost_connection_arms_the_reload_under_the_load_task_id() -> None:
76 """A lost SnapServer connection arms the reload so a (re)load starting first cancels it."""
77 provider = _provider_with_log_level(logging.INFO)
78 provider.config = MagicMock(instance_id="snapcast--test")
79 provider.mass = MagicMock(closing=False)
80 provider._stop_called = False
81
82 provider._handle_disconnect(ConnectionError("connection lost"))
83
84 retry = provider.mass.call_later.call_args
85 assert retry.args == (5, provider.mass.load_provider, "snapcast--test")
86 assert retry.kwargs == {"allow_retry": True, "task_id": "load_provider_snapcast--test"}
87