/
/
/
1"""Tests for Sonos provider discovery."""
2
3import logging
4from unittest.mock import MagicMock
5
6import pytest
7from zeroconf import ServiceStateChange
8
9from music_assistant.providers.sonos.provider import SonosPlayerProvider
10
11
12def _make_provider(enabled: bool = False) -> tuple[SonosPlayerProvider, MagicMock]:
13 """Create a Sonos provider with mocked discovery dependencies."""
14 provider = SonosPlayerProvider.__new__(SonosPlayerProvider)
15 mass = MagicMock()
16 provider.mass = mass
17 provider.logger = logging.getLogger("test.sonos.discovery")
18 provider._ignored_disabled_players = set()
19 mass.config.get_raw_player_config_value.return_value = enabled
20 mass.players.get_player.return_value = None
21 return provider, mass
22
23
24def _make_discovery_info(player_id: str = "sonos_player") -> MagicMock:
25 """Create minimal Sonos mDNS discovery information."""
26 info = MagicMock()
27 info.decoded_properties = {"uuid": player_id}
28 return info
29
30
31@pytest.mark.asyncio
32async def test_disabled_discovery_is_not_scheduled_repeatedly(
33 caplog: pytest.LogCaptureFixture,
34) -> None:
35 """Test repeated announcements for a disabled player are ignored once."""
36 provider, mass = _make_provider()
37 info = _make_discovery_info()
38
39 with caplog.at_level(logging.DEBUG, logger=provider.logger.name):
40 await provider.on_mdns_service_state_change(
41 "Toilet._sonos._tcp.local.", ServiceStateChange.Added, info
42 )
43 await provider.on_mdns_service_state_change(
44 "Toilet._sonos._tcp.local.", ServiceStateChange.Updated, info
45 )
46
47 mass.call_later.assert_not_called()
48 ignored_records = [
49 record for record in caplog.records if "in discovery as it is disabled" in record.message
50 ]
51 assert len(ignored_records) == 1
52
53
54@pytest.mark.asyncio
55async def test_disabled_discovery_logs_again_after_reenable(
56 caplog: pytest.LogCaptureFixture,
57) -> None:
58 """Test a new disabled period produces a new discovery diagnostic."""
59 provider, mass = _make_provider()
60 info = _make_discovery_info()
61
62 with caplog.at_level(logging.DEBUG, logger=provider.logger.name):
63 await provider.on_mdns_service_state_change(
64 "Toilet._sonos._tcp.local.", ServiceStateChange.Added, info
65 )
66 mass.config.get_raw_player_config_value.return_value = True
67 await provider.on_mdns_service_state_change(
68 "Toilet._sonos._tcp.local.", ServiceStateChange.Updated, info
69 )
70 mass.config.get_raw_player_config_value.return_value = False
71 await provider.on_mdns_service_state_change(
72 "Toilet._sonos._tcp.local.", ServiceStateChange.Updated, info
73 )
74
75 mass.call_later.assert_called_once()
76 ignored_records = [
77 record for record in caplog.records if "in discovery as it is disabled" in record.message
78 ]
79 assert len(ignored_records) == 2
80