282 lines
11 KiB
Python
282 lines
11 KiB
Python
import math
|
|
import struct
|
|
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._bass = 1
|
|
self._treble = -2
|
|
self._loudness = True
|
|
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
|
|
|
|
@property
|
|
def bass(self):
|
|
return self._bass
|
|
|
|
@bass.setter
|
|
def bass(self, value):
|
|
self._bass = value
|
|
|
|
@property
|
|
def treble(self):
|
|
return self._treble
|
|
|
|
@treble.setter
|
|
def treble(self, value):
|
|
self._treble = value
|
|
|
|
@property
|
|
def loudness(self):
|
|
return self._loudness
|
|
|
|
@loudness.setter
|
|
def loudness(self, value):
|
|
self._loudness = 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"
|
|
assert result[0]["bass"] == 1
|
|
assert result[0]["treble"] == -2
|
|
assert result[0]["loudness"] is True
|
|
|
|
|
|
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_equalizer_controls(service, zone):
|
|
service.set_eq(zone.uid, "bass", -6)
|
|
service.set_eq(zone.uid, "treble", 7)
|
|
service.set_eq(zone.uid, "loudness", False)
|
|
assert (zone.bass, zone.treble, zone.loudness) == (-6, 7, False)
|
|
with pytest.raises(ValueError):
|
|
service.set_eq(zone.uid, "bass", 11)
|
|
with pytest.raises(ValueError):
|
|
service.set_eq(zone.uid, "loudness", 1)
|
|
with pytest.raises(ValueError):
|
|
service.set_eq(zone.uid, "balance", 0)
|
|
|
|
|
|
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
|
|
assert client.put(f"/api/zones/{zone.uid}/eq", json={"control":"bass", "value":4}).status_code == 200
|
|
assert zone.bass == 4
|
|
assert client.put(f"/api/zones/{zone.uid}/eq", json={"control":"treble", "value":99}).status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize(("payload", "status"), [
|
|
({"control": "bass"}, 400),
|
|
({"control": "bass", "value": None}, 400),
|
|
({"control": "bass", "value": True}, 400),
|
|
({"control": "bass", "value": 1.5}, 400),
|
|
({"control": "bass", "value": -10}, 200),
|
|
({"control": "treble", "value": 10}, 200),
|
|
({"control": "loudness"}, 400),
|
|
({"control": "loudness", "value": None}, 400),
|
|
])
|
|
def test_equalizer_api_value_validation(service, zone, payload, status):
|
|
client = create_app({"TESTING": True}, service).test_client()
|
|
assert client.put(f"/api/zones/{zone.uid}/eq", json=payload).status_code == status
|
|
|
|
|
|
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()
|
|
|
|
|
|
def test_favicon_is_linked_and_served(service):
|
|
client = create_app({"TESTING": True}, service).test_client()
|
|
page = client.get("/")
|
|
assert b'rel="icon" type="image/x-icon" href="/static/favicon.ico"' in page.data
|
|
assert b'sizes="16x16 32x32 48x48 64x64"' in page.data
|
|
favicon = client.get("/static/favicon.ico")
|
|
assert favicon.status_code == 200
|
|
assert favicon.mimetype in {"image/vnd.microsoft.icon", "image/x-icon"}
|
|
reserved, image_type, image_count = struct.unpack_from("<HHH", favicon.data)
|
|
assert (reserved, image_type, image_count) == (0, 1, 4)
|
|
entries = [struct.unpack_from("<BBBBHHII", favicon.data, 6 + index * 16) for index in range(image_count)]
|
|
dimensions = {(width or 256, height or 256) for width, height, *_ in entries}
|
|
assert dimensions == {(16, 16), (32, 32), (48, 48), (64, 64)}
|
|
for *_, byte_count, image_offset in entries:
|
|
assert byte_count > 0
|
|
assert image_offset >= 6 + image_count * 16
|
|
assert image_offset + byte_count <= len(favicon.data)
|
|
|
|
|
|
@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
|
|
|
|
|
|
def test_frontend_serializes_eq_writes_and_reloads_confirmed_values():
|
|
source = Path("app/static/app.js").read_text(encoding="utf-8")
|
|
assert "if (pendingEq.has(key)) return" in source
|
|
assert "pendingEq.add(key); input.disabled = true" in source
|
|
assert source.count("await load(false)") >= 3
|
|
assert "loudness.dataset.value !== 'true'" in source
|
|
assert "value:!zone.loudness" not in source
|
|
|
|
|
|
def test_zone_cards_render_name_directly_without_transport_status():
|
|
source = Path("app/static/app.js").read_text(encoding="utf-8")
|
|
styles = Path("app/static/style.css").read_text(encoding="utf-8")
|
|
assert "stateName" not in source
|
|
assert "el('span', 'state'" not in source
|
|
assert "WIEDERGABE" not in source
|
|
assert "zone-head" not in source
|
|
assert ".zone-head" not in styles
|
|
assert "card.append(name, track, controls, volume, eq, zoneActions)" in source
|
|
assert "Kein aktueller Titel" not in source
|
|
assert "zone.state === 'PLAYING' && zone.track.title" in source
|