martin/musicmouse
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add "Mein Zimmer" room-control page (Home Assistant, proxied through the backend)

Browse Source 
Implements the room-control page from the design mockup: a scenes row above cards
for shutters, color lamps, and brightness-only lamps, all driven by a new
`general.ha` config section (server URL, token, ordered device/scene lists).

The backend proxies every Home Assistant call server-side (GET/POST /api/ha/...)
rather than the browser calling Home Assistant directly, so the long-lived token
never leaves the LAN device and Home Assistant's own CORS settings don't need to
know about musicmouse at all. Card kind (shutter/color/brightness-only) is
inferred at runtime from what Home Assistant reports about each entity, not
configured explicitly.

Also stops tracking python-backend/config.yml, which had drifted into the repo
despite its own header saying it shouldn't be - it now carries real credentials
locally and needs to stay untracked.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Martin Bauer
2026-08-27 23:45:29 +02:00
parent edb6e5e027
commit a8ed350aec
27 changed files with 1437 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
python-backend/tests/test_config.py
View File

@@ -52,6 +52,75 @@ def test_web_section_is_optional(config_dir: Path) -> None:
assert config.general.web.static_dir == (config_dir / "dist").resolve()
def test_ha_section_is_optional(config_dir: Path) -> None:
assert load_config(write_config(config_dir, VALID_CONFIG)).general.ha is None
def test_ha_device_and_scene_name_is_optional(config_dir: Path) -> None:
data = _config(
ha={
"url": "http://homeassistant.local:8123",
"token": "abc123",
"devices": [{"entity_id": "light.a"}],
"scenes": [{"entity_id": "scene.a"}],
}
)
config = load_config(write_config(config_dir, data))
assert config.general.ha is not None
assert config.general.ha.devices[0].name is None
assert config.general.ha.scenes[0].name is None
def test_ha_devices_and_scenes_preserve_config_order(config_dir: Path) -> None:
data = _config(
ha={
"url": "http://homeassistant.local:8123",
"token": "abc123",
"devices": [
{"entity_id": "cover.rollo"},
{"entity_id": "light.deckenlampe"},
{"entity_id": "light.beyond_links"},
],
"scenes": [
{"entity_id": "scene.aufwachen"},
{"entity_id": "scene.lesen"},
],
}
)
config = load_config(write_config(config_dir, data))
ha = config.general.ha
assert ha is not None
assert [d.entity_id for d in ha.devices] == [
"cover.rollo",
"light.deckenlampe",
"light.beyond_links",
]
assert [s.entity_id for s in ha.scenes] == ["scene.aufwachen", "scene.lesen"]
def test_ha_requires_devices_or_scenes(config_dir: Path) -> None:
message = _error(
config_dir,
_config(ha={"url": "http://ha", "token": "abc", "devices": [], "scenes": []}),
)
assert "configure at least one device or scene" in message
def test_ha_rejects_unknown_key_on_a_device(config_dir: Path) -> None:
message = _error(
config_dir,
_config(
ha={
"url": "http://ha",
"token": "abc",
"devices": [{"entity_id": "light.a", "kind": "beyond"}],
"scenes": [],
}
),
)
assert "unknown option" in message
# --------------------------------------------------------------------- error paths

179
python-backend/tests/test_web.py
View File

@@ -8,7 +8,8 @@ from __future__ import annotations
import asyncio
import contextlib
from collections.abc import AsyncIterator, Iterator
import json
from collections.abc import AsyncIterator
from pathlib import Path
import httpx2
@@ -19,7 +20,7 @@ from musicmouse.config import WebConfig, load_config
from musicmouse.services.web.service import build_app
from musicmouse.simulator.harness import Simulation, build_simulation
from tests.conftest import VALID_CONFIG, write_config
from tests.websocket_harness import websocket_connect
from tests.websocket_harness import WebSocketSession, websocket_connect
TRACK_SECONDS = 10.0
@@ -38,7 +39,7 @@ async def sim(config_dir: Path) -> AsyncIterator[Simulation]:
@pytest.fixture
def api(sim: Simulation, config_dir: Path) -> Iterator[FastAPI]:
async def api(sim: Simulation, config_dir: Path) -> AsyncIterator[FastAPI]:
"""The real ASGI app, on the test's own event loop."""
application, hub = build_app(sim.app, WebConfig(), config_dir / "config.yml")
hub.start()
@@ -46,6 +47,7 @@ def api(sim: Simulation, config_dir: Path) -> Iterator[FastAPI]:
yield application
finally:
hub.stop()
await application.state.ha_client.aclose()
@pytest.fixture
@@ -71,7 +73,8 @@ async def test_library_lists_every_album_with_its_tracks(client: Client) -> None
"Eule",
"Kinderparty Lieder",
"Conni in den Bergen",
"Wissen macht Ah",
"Alt",
"Neu",
}
album = await album_by_title(client, "Conni in den Bergen")
@@ -248,6 +251,23 @@ async def test_volume_needs_one_of_the_two_fields(client: Client) -> None:
# ---------------------------------------------------------------------- websocket
async def last_state(socket: WebSocketSession) -> dict:
"""Drain every currently queued frame and return the most recent state payload.
A single backend action can trigger more than one broadcast (e.g. a reaction to
``PlaybackChanged`` announcing its own state change); callers only care about
where things ended up, not how many frames it took to get there. Call only after
the bus has been drained, so every frame the action produced is already queued.
"""
state = None
while not socket.from_app.empty():
message = await socket.next_json()
if message["type"] == "state":
state = message["state"]
assert state is not None
return state
async def test_a_new_client_is_sent_a_snapshot_before_any_deltas(api: FastAPI) -> None:
"""State events only fire on change, so a tab that connects mid-track needs this."""
async with websocket_connect(api, "/api/ws") as socket:
@@ -272,6 +292,26 @@ async def test_a_state_change_reaches_every_client(api: FastAPI, sim: Simulation
assert message["state"]["active_figure"] == "fuchs"
async def test_switching_albums_while_playing_reaches_every_client(
api: FastAPI, client: Client, sim: Simulation
) -> None:
"""Regression: playing album B from track 0 while album A was already playing used
to never broadcast, leaving PlayView stuck on album A until a reload."""
first = await album_by_title(client, "Fuchs")
second = await album_by_title(client, "Eule")
async with websocket_connect(api, "/api/ws") as socket:
await socket.next_json()
await client.post("/api/play", json={"album_id": first["id"]})
await sim.bus.drain()
assert (await last_state(socket))["album_title"] == "Fuchs"
await client.post("/api/play", json={"album_id": second["id"]})
await sim.bus.drain()
assert (await last_state(socket))["album_title"] == "Eule"
async def test_a_refresh_tells_the_clients_to_reload_the_library(
api: FastAPI, client: Client, sim: Simulation
) -> None:
@@ -301,6 +341,137 @@ async def test_the_web_ui_follows_a_figure_placed_on_the_reader(
assert state["album_title"] == "Fuchs"
# ------------------------------------------------------------------------------- ha
HA_URL = "http://homeassistant.local:8123"
HA_TOKEN = "abc123"
VALID_CONFIG_WITH_HA: dict = {
**VALID_CONFIG,
"general": {
**VALID_CONFIG["general"],
"ha": {
"url": HA_URL,
"token": HA_TOKEN,
"devices": [
{"entity_id": "cover.rollo", "name": "Rollo"},
{"entity_id": "light.deckenlampe"},
],
"scenes": [{"entity_id": "scene.lesen", "name": "Lesen"}],
},
},
}
class FakeHomeAssistant:
"""Records every request the proxy makes and answers a couple of fixed routes."""
def __init__(self) -> None:
self.requests: list[httpx2.Request] = []
self.unreachable = False
def handler(self, request: httpx2.Request) -> httpx2.Response:
self.requests.append(request)
if self.unreachable:
raise httpx2.ConnectError("mock: connection refused")
if request.url.path == "/api/states/light.deckenlampe":
return httpx2.Response(200, json={"entity_id": "light.deckenlampe", "state": "on"})
if request.url.path == "/api/states/light.missing":
return httpx2.Response(404, json={"message": "Entity not found"})
if request.url.path == "/api/services/light/turn_on":
return httpx2.Response(200, json=[{"entity_id": "light.deckenlampe", "state": "on"}])
return httpx2.Response(404, json={"message": "unhandled in test"})
@pytest.fixture
async def fake_ha() -> FakeHomeAssistant:
return FakeHomeAssistant()
@pytest.fixture
async def client_with_ha(
config_dir: Path, fake_ha: FakeHomeAssistant
) -> AsyncIterator[httpx2.AsyncClient]:
"""A second client, built from a config with an ``ha:`` section and an ``ha_client``
pointed at ``fake_ha`` instead of the network, without touching the shared
``sim``/``api``/``client`` fixtures every other test in this file uses."""
config = load_config(write_config(config_dir, VALID_CONFIG_WITH_HA))
simulation = await build_simulation(config, track_duration=TRACK_SECONDS)
try:
ha_client = httpx2.AsyncClient(transport=httpx2.MockTransport(fake_ha.handler))
application, hub = build_app(
simulation.app, WebConfig(), config_dir / "config.yml", ha_client=ha_client
)
hub.start()
try:
transport = httpx2.ASGITransport(app=application)
async with httpx2.AsyncClient(transport=transport, base_url="http://mouse") as http:
yield http
finally:
hub.stop()
await ha_client.aclose()
finally:
await simulation.aclose()
async def test_ha_is_a_404_when_not_configured(client: Client) -> None:
assert (await client.get("/api/ha")).status_code == 404
assert (await client.get("/api/ha/states/light.x")).status_code == 404
assert (await client.post("/api/ha/services/light/turn_on", json={})).status_code == 404
async def test_ha_config_returns_ordered_lists_without_the_token(client_with_ha: Client) -> None:
body = (await client_with_ha.get("/api/ha")).json()
assert "url" not in body
assert "token" not in body
assert body["devices"] == [
{"entity_id": "cover.rollo", "name": "Rollo"},
{"entity_id": "light.deckenlampe", "name": None},
]
assert body["scenes"] == [{"entity_id": "scene.lesen", "name": "Lesen"}]
async def test_ha_states_proxies_to_home_assistant_with_the_token_attached(
client_with_ha: Client, fake_ha: FakeHomeAssistant
) -> None:
response = await client_with_ha.get("/api/ha/states/light.deckenlampe")
assert response.status_code == 200
assert response.json() == {"entity_id": "light.deckenlampe", "state": "on"}
assert len(fake_ha.requests) == 1
upstream = fake_ha.requests[0]
assert str(upstream.url) == f"{HA_URL}/api/states/light.deckenlampe"
assert upstream.headers["authorization"] == f"Bearer {HA_TOKEN}"
async def test_ha_states_passes_through_a_404_from_home_assistant(client_with_ha: Client) -> None:
response = await client_with_ha.get("/api/ha/states/light.missing")
assert response.status_code == 404
async def test_ha_services_proxies_the_body_with_the_token_attached(
client_with_ha: Client, fake_ha: FakeHomeAssistant
) -> None:
response = await client_with_ha.post(
"/api/ha/services/light/turn_on",
json={"entity_id": "light.deckenlampe", "brightness_pct": 80},
)
assert response.status_code == 200
upstream = fake_ha.requests[0]
assert str(upstream.url) == f"{HA_URL}/api/services/light/turn_on"
assert upstream.headers["authorization"] == f"Bearer {HA_TOKEN}"
assert json.loads(upstream.content) == {"entity_id": "light.deckenlampe", "brightness_pct": 80}
async def test_ha_proxy_reports_an_unreachable_home_assistant_as_a_502(
client_with_ha: Client, fake_ha: FakeHomeAssistant
) -> None:
fake_ha.unreachable = True
response = await client_with_ha.get("/api/ha/states/light.deckenlampe")
assert response.status_code == 502
# ------------------------------------------------- through a real uvicorn, not just ASGI