179 lines
7.2 KiB
Python
179 lines
7.2 KiB
Python
import math
|
|
import threading
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from soco import config as soco_config
|
|
from soco.data_structures import DidlAudioBroadcast, DidlMusicTrack, DidlPlaylistContainer, DidlResource
|
|
|
|
from app.sonos import SonosService
|
|
from app.web import create_app
|
|
|
|
|
|
class FakeZone:
|
|
def __init__(self, uid="RINCON_1", name="Wohnzimmer"):
|
|
self.uid, self.player_name, self.ip_address = uid, name, "192.168.1.20"
|
|
self._volume = 23
|
|
self.group = SimpleNamespace(coordinator=self, members=[self])
|
|
self.music_library = MagicMock()
|
|
for method in ("play", "pause", "stop", "join", "unjoin", "play_uri", "play_from_queue"):
|
|
setattr(self, method, MagicMock())
|
|
self.add_to_queue = MagicMock(return_value=4)
|
|
self.get_current_transport_info = MagicMock(return_value={"current_transport_state": "PLAYING"})
|
|
self.get_current_track_info = MagicMock(return_value={"title": "Titel", "artist": "Künstler"})
|
|
self.get_queue = MagicMock(return_value=[])
|
|
|
|
@property
|
|
def volume(self):
|
|
return self._volume
|
|
|
|
@volume.setter
|
|
def volume(self, value):
|
|
self._volume = value
|
|
|
|
|
|
@pytest.fixture
|
|
def zone():
|
|
return FakeZone()
|
|
|
|
|
|
@pytest.fixture
|
|
def service(monkeypatch, zone):
|
|
monkeypatch.setattr("app.sonos.soco.discover", MagicMock(return_value={zone}))
|
|
instance = SonosService(operation_timeout=1)
|
|
yield instance
|
|
instance.close()
|
|
|
|
|
|
def test_discovery_and_status(service):
|
|
result = service.zones()
|
|
assert result[0]["name"] == "Wohnzimmer"
|
|
assert result[0]["state"] == "PLAYING"
|
|
assert result[0]["track"]["title"] == "Titel"
|
|
|
|
|
|
def test_transport_and_volume(service, zone):
|
|
service.transport(zone.uid, "pause")
|
|
service.set_volume(zone.uid, 61)
|
|
zone.pause.assert_called_once_with()
|
|
assert zone.volume == 61
|
|
with pytest.raises(ValueError):
|
|
service.set_volume(zone.uid, 101)
|
|
|
|
|
|
def test_group_and_ungroup(monkeypatch):
|
|
coordinator, member = FakeZone("A", "Küche"), FakeZone("B", "Bad")
|
|
monkeypatch.setattr("app.sonos.soco.discover", MagicMock(return_value={coordinator, member}))
|
|
service = SonosService(operation_timeout=1)
|
|
service.group("A", ["B"])
|
|
service.ungroup("B")
|
|
member.join.assert_called_once_with(coordinator)
|
|
member.unjoin.assert_called_once_with()
|
|
|
|
|
|
def test_radio_favorite_uses_reference_and_direct_uri(service, zone):
|
|
resource = DidlResource("x-sonosapi-stream:test", "x-rincon-mp3radio:*:audio/mpeg:*")
|
|
reference = DidlAudioBroadcast("Radio", "FV:2", "item", resources=[resource])
|
|
zone.music_library.get_sonos_favorites.return_value = [SimpleNamespace(title="Radio", reference=reference)]
|
|
service.play_favorite(zone.uid, 0)
|
|
assert zone.play_uri.call_args.args == (resource.uri,)
|
|
assert "audioBroadcast" in zone.play_uri.call_args.kwargs["meta"]
|
|
zone.add_to_queue.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("item_type", [DidlMusicTrack, DidlPlaylistContainer])
|
|
def test_queueable_favorites_use_reference_and_queue_position(service, zone, item_type):
|
|
resource = DidlResource("x-rincon-cpcontainer:test", "x-rincon-playlist:*:*:*")
|
|
reference = item_type("Favorit", "FV:2", "item", resources=[resource])
|
|
zone.music_library.get_sonos_favorites.return_value = [SimpleNamespace(reference=reference)]
|
|
service.play_favorite(zone.uid, 0)
|
|
zone.add_to_queue.assert_called_once_with(reference)
|
|
zone.play_from_queue.assert_called_once_with(3)
|
|
|
|
|
|
def test_queue_and_stream(service, zone):
|
|
zone.get_queue.return_value = [SimpleNamespace(title="Lied", creator="Band")]
|
|
assert service.queue(zone.uid)[0]["creator"] == "Band"
|
|
service.play_queue(zone.uid, 0)
|
|
zone.play_from_queue.assert_called_once_with(0)
|
|
service.play_stream(zone.uid, "https://radio.example/stream.mp3")
|
|
zone.play_uri.assert_called_with("https://radio.example/stream.mp3", title="Web-Stream")
|
|
with pytest.raises(ValueError):
|
|
service.play_stream(zone.uid, "file:///etc/passwd")
|
|
|
|
|
|
def test_status_failure_is_isolated_and_queries_are_concurrent(monkeypatch):
|
|
slow, healthy = FakeZone("A", "Langsam"), FakeZone("B", "Schnell")
|
|
entered, release = threading.Event(), threading.Event()
|
|
|
|
def slow_info():
|
|
entered.set()
|
|
release.wait(0.5)
|
|
raise TimeoutError("timed out")
|
|
|
|
slow.get_current_transport_info.side_effect = slow_info
|
|
healthy.get_current_transport_info.side_effect = lambda: (release.set() or {"current_transport_state": "PLAYING"})
|
|
monkeypatch.setattr("app.sonos.soco.discover", MagicMock(return_value={slow, healthy}))
|
|
result = {item["uid"]: item for item in SonosService(operation_timeout=0.2).zones()}
|
|
assert entered.is_set()
|
|
assert result["A"]["state"] == "UNAVAILABLE"
|
|
assert result["B"]["state"] == "PLAYING"
|
|
|
|
|
|
def test_soco_timeout_no_persistent_executor_and_no_cache_lock_during_io(monkeypatch, zone):
|
|
monkeypatch.setattr("app.sonos.soco.discover", MagicMock(return_value={zone}))
|
|
service = SonosService(operation_timeout=0.25)
|
|
acquired = []
|
|
|
|
def pause():
|
|
acquired.append(service._cache_lock.acquire(blocking=False))
|
|
if acquired[-1]:
|
|
service._cache_lock.release()
|
|
|
|
zone.pause.side_effect = pause
|
|
service.transport(zone.uid, "pause")
|
|
assert soco_config.REQUEST_TIMEOUT == 0.25
|
|
assert not hasattr(service, "_executor")
|
|
assert acquired == [True]
|
|
|
|
|
|
def test_api_validation_and_error_mapping(service, zone):
|
|
client = create_app({"TESTING": True}, service).test_client()
|
|
assert client.get("/health").json == {"status": "ok"}
|
|
assert client.put(f"/api/zones/{zone.uid}/volume", json={"volume": 31}).status_code == 200
|
|
response = client.put(f"/api/zones/{zone.uid}/volume", json={"volume": "laut"})
|
|
assert response.status_code == 400 and response.json["error"] == "validation_error"
|
|
assert client.post(f"/api/zones/{zone.uid}/transport", data="{}", content_type="text/plain").status_code == 400
|
|
|
|
|
|
def test_api_never_discovers_real_devices(monkeypatch, zone):
|
|
discover = MagicMock(return_value={zone})
|
|
monkeypatch.setattr("app.sonos.soco.discover", discover)
|
|
client = create_app({"TESTING": True}, SonosService(operation_timeout=1)).test_client()
|
|
assert client.get("/api/zones").status_code == 200
|
|
discover.assert_called_once()
|
|
|
|
|
|
@pytest.mark.parametrize("name,value", [
|
|
("SONOS_OPERATION_TIMEOUT", 0), ("SONOS_OPERATION_TIMEOUT", math.inf),
|
|
("SONOS_DISCOVERY_TIMEOUT", -1), ("SONOS_DISCOVERY_TIMEOUT", math.nan),
|
|
])
|
|
def test_timeout_configuration_must_be_positive_and_finite(service, name, value):
|
|
with pytest.raises(ValueError):
|
|
create_app({"TESTING": True, name: value}, service)
|
|
|
|
|
|
def test_json_request_size_is_limited(service, zone):
|
|
client = create_app({"TESTING": True, "MAX_CONTENT_LENGTH": 64}, service).test_client()
|
|
response = client.post(f"/api/zones/{zone.uid}/transport", json={"action":"play", "padding":"x" * 100})
|
|
assert response.status_code == 413
|
|
|
|
|
|
def test_frontend_guards_stale_refreshes_and_preserves_members():
|
|
source = Path("app/static/app.js").read_text(encoding="utf-8")
|
|
assert "zoneController.abort()" in source
|
|
assert "request !== zoneRequest" in source
|
|
assert "selectedMembers.has(zone.uid)" in source
|