Files
houseplan-card/tests_backend/test_ha_websocket.py
T
Matysh 3d41fe16b8 v1.45.3: 'value instead of an icon' could never be saved (issue #3)
The device editor has offered display='value' since v1.26.0; MARKER_SCHEMA
accepted only badge/ripple/icon_ripple. Picking it produced

  not a valid value for dictionary value @ data['config']['markers'][n]['display']

and since one rejected marker fails the whole config write, the user could not
save the plan at all until the setting was undone. Reported by @RemyRoux with
the exact error text, 2026-07-27 — a year and a half after the feature shipped.

The schema now accepts it, and the class of bug is closed rather than the
instance: DISPLAY_MODES, TAP_ACTIONS, SPACE_FILL_MODES and ROOM_FILL_MODES are
exported from src/logic.ts, the editors render their options from them, and a
backend test parses those lists out of the TypeScript source and asserts the
schema accepts every one (and rejects a bogus value). Reverting the one-word
schema fix fails that test, which is the check that was missing.

Plus an HA-harness test saving a config that contains a value-display marker —
the exact call the user's card was making.
Docs: CHANGELOG.md + CHANGELOG.ru.md + TESTING.md + STATUS.md.
2026-07-28 00:23:37 +03:00

507 lines
19 KiB
Python

"""WebSocket API tests (CI): layout ops, config rev conflict, not_ready gate."""
import pytest
@pytest.fixture(autouse=True)
def _enable_custom_integrations(enable_custom_integrations):
"""Allow loading custom_components in the test hass."""
yield
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from pytest_homeassistant_custom_component.typing import WebSocketGenerator
from custom_components.houseplan.const import DOMAIN
async def _setup(hass: HomeAssistant) -> MockConfigEntry:
entry = MockConfigEntry(domain=DOMAIN, title="House Plan", data={}, options={})
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
async def test_layout_roundtrip(hass: HomeAssistant, hass_ws_client: WebSocketGenerator) -> None:
await _setup(hass)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "houseplan/layout/get"})
resp = await client.receive_json()
assert resp["success"] and resp["result"]["layout"] == {}
await client.send_json_auto_id(
{"type": "houseplan/layout/set", "layout": {"dev1": {"s": "f1", "x": 0.5, "y": 0.5}}}
)
assert (await client.receive_json())["success"]
await client.send_json_auto_id(
{"type": "houseplan/layout/update", "device_id": "dev2", "pos": {"s": "f1", "x": 0.1, "y": 0.2}}
)
assert (await client.receive_json())["success"]
await client.send_json_auto_id({"type": "houseplan/layout/delete", "device_id": "dev1"})
assert (await client.receive_json())["success"]
await client.send_json_auto_id({"type": "houseplan/layout/get"})
resp = await client.receive_json()
assert set(resp["result"]["layout"]) == {"dev2"}
async def test_config_rev_conflict(hass: HomeAssistant, hass_ws_client: WebSocketGenerator) -> None:
await _setup(hass)
client = await hass_ws_client(hass)
cfg = {"spaces": [], "markers": [], "settings": {}}
await client.send_json_auto_id({"type": "houseplan/config/set", "config": cfg, "expected_rev": 0})
resp = await client.receive_json()
assert resp["success"] and resp["result"]["rev"] == 1
# stale expected_rev must be rejected with `conflict`
await client.send_json_auto_id({"type": "houseplan/config/set", "config": cfg, "expected_rev": 0})
resp = await client.receive_json()
assert not resp["success"] and resp["error"]["code"] == "conflict"
await client.send_json_auto_id({"type": "houseplan/config/get"})
resp = await client.receive_json()
assert resp["result"]["rev"] == 1
async def test_not_ready_without_entry(hass: HomeAssistant, hass_ws_client: WebSocketGenerator) -> None:
"""WS commands answer not_ready when the integration has no loaded entry."""
# register only the WS commands, without an entry
from custom_components.houseplan import websocket_api as hp_ws
hp_ws.async_register(hass)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "houseplan/layout/get"})
resp = await client.receive_json()
assert not resp["success"] and resp["error"]["code"] == "not_ready"
async def test_plan_set_validates(hass: HomeAssistant, hass_ws_client: WebSocketGenerator) -> None:
await _setup(hass)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{"type": "houseplan/plan/set", "space_id": "../evil", "ext": "png", "data": "aGk="}
)
resp = await client.receive_json()
assert not resp["success"] and resp["error"]["code"] == "invalid_space_id"
await client.send_json_auto_id(
{"type": "houseplan/plan/set", "space_id": "s1", "ext": "png", "data": "%%%not-base64%%%"}
)
resp = await client.receive_json()
assert not resp["success"] and resp["error"]["code"] == "invalid_data"
await client.send_json_auto_id(
{"type": "houseplan/plan/set", "space_id": "s1", "ext": "png", "data": "aGVsbG8="}
)
resp = await client.receive_json()
url = resp["result"]["url"]
assert resp["success"]
# versioned name: "<space>.<token>.<ext>" (review R2-1)
name = url.rsplit("/", 1)[-1]
assert name.startswith("s1.") and name.endswith(".png") and len(name.split(".")) == 3
async def test_admin_check_fails_closed(hass, hass_ws_client):
"""audit B2/T4: with no config entry the policy is unknown — deny writes.
This used to allow them: plan uploads slipped through during a reload.
"""
from custom_components.houseplan import websocket_api as wsapi
class _User:
is_admin = False
class _Conn:
user = _User()
# no entry at all → non-admin must be refused
assert wsapi._check_write(hass, _Conn()) is False
class _Admin:
is_admin = True
class _AdminConn:
user = _Admin()
assert wsapi._check_write(hass, _AdminConn()) is True
async def test_files_migrate_copies_and_reports_mapping(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""review CR-2/CR-3: migrate COPIES, never overwrites, and reports the mapping."""
import os
from custom_components.houseplan.const import FILES_DIR
await _setup(hass)
client = await hass_ws_client(hass)
base = hass.config.path(FILES_DIR)
src = os.path.join(base, "old1")
dst = os.path.join(base, "new1")
def _prepare() -> None:
os.makedirs(src, exist_ok=True)
os.makedirs(dst, exist_ok=True)
with open(os.path.join(src, "m.pdf"), "wb") as fh:
fh.write(b"SOURCE")
# a DIFFERENT file already owns the name in the destination
with open(os.path.join(dst, "m.pdf"), "wb") as fh:
fh.write(b"OTHER")
await hass.async_add_executor_job(_prepare)
await client.send_json_auto_id(
{"type": "houseplan/files/migrate", "from_id": "old1", "to_id": "new1"}
)
resp = await client.receive_json()
assert resp["success"], resp
mapping = resp["result"]["mapping"]
assert mapping["m.pdf"] != "m.pdf" # renamed instead of overwriting
def _read_all() -> tuple[bool, bytes, bytes]:
with open(os.path.join(dst, "m.pdf"), "rb") as fh:
other = fh.read()
with open(os.path.join(dst, mapping["m.pdf"]), "rb") as fh:
copied = fh.read()
return os.path.isfile(os.path.join(src, "m.pdf")), other, copied
src_kept, other, copied = await hass.async_add_executor_job(_read_all)
assert src_kept, "migrate must COPY, not move (review CR-2)"
assert other == b"OTHER" and copied == b"SOURCE"
# cleanup runs only after the config is safely committed
await client.send_json_auto_id({"type": "houseplan/files/cleanup", "marker_id": "old1"})
resp2 = await client.receive_json()
assert resp2["success"] and resp2["result"]["removed"] is True
assert not await hass.async_add_executor_job(lambda: os.path.isdir(src))
async def _cfg(spaces: list[dict]) -> dict:
"""Minimal accepted configuration with the given spaces."""
return {
"spaces": [
{"id": sp["id"], "title": sp["id"], "plan_url": sp.get("plan_url"),
"aspect": 1.4, "view_box": [0, 0, 1, 1], "rooms": []}
for sp in spaces
],
"markers": [],
}
async def _save(client, config, expected_rev):
await client.send_json_auto_id(
{"type": "houseplan/config/set", "config": config, "expected_rev": expected_rev}
)
return await client.receive_json()
async def _upload(client, space_id, data=b"x", ext="png"):
import base64 as _b64
await client.send_json_auto_id({
"type": "houseplan/plan/set", "space_id": space_id, "ext": ext,
"data": _b64.b64encode(data).decode(),
})
resp = await client.receive_json()
return resp["result"]["url"], resp["result"]["url"].rsplit("/", 1)[-1]
async def test_plan_upload_does_not_touch_the_previous_file(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""review R2-1: a rejected config write must leave the stored plan intact.
The upload used to overwrite "<space>.<ext>" (and unlink the other
extension) BEFORE the revision-checked config write, so a conflict left the
live plan replaced — or, with a new extension, pointing at a deleted file.
"""
from pathlib import Path
from custom_components.houseplan.const import PLANS_DIR
await _setup(hass)
client = await hass_ws_client(hass)
plans = Path(hass.config.path(PLANS_DIR))
plans.mkdir(parents=True, exist_ok=True)
for stale in plans.glob("s9.*"):
stale.unlink()
legacy = plans / "s9.svg" # what an older version stored, still referenced
legacy.write_bytes(b"<svg>old</svg>")
url0 = "/api/houseplan/content/plans/_/s9.svg"
resp = await _save(client, await _cfg([{"id": "s9", "plan_url": url0}]), 0)
rev = resp["result"]["rev"]
assert legacy.is_file()
url1, first = await _upload(client, "s9", b"hello")
# config write rejected (stale revision): nothing on disk may change
bad = await _save(client, await _cfg([{"id": "s9", "plan_url": url1}]), rev - 1)
assert not bad["success"] and bad["error"]["code"] == "conflict"
assert legacy.read_bytes() == b"<svg>old</svg>"
assert (plans / first).read_bytes() == b"hello"
# a second attempt neither overwrites the first nor the legacy file
url2, second = await _upload(client, "s9", b"world")
assert second != first
assert (plans / first).read_bytes() == b"hello"
assert legacy.is_file()
# config accepted → the superseded file goes, the referenced one stays.
# `first` is a rejected upload: it is young, so it is kept for now.
ok = await _save(client, await _cfg([{"id": "s9", "plan_url": url2}]), rev)
assert ok["success"]
assert (plans / second).read_bytes() == b"world"
assert not legacy.exists(), "the superseded plan is collected"
assert (plans / first).is_file(), "a fresh unreferenced upload is NOT collected"
async def test_late_commit_of_one_client_never_deletes_another_client_s_plan(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""review R3-1: collection belongs to the commit, not to a client's request.
With the previous `plan/cleanup(keep=...)` command, this interleaving left
the accepted configuration pointing at a file that had just been deleted:
A: upload PA, config/set(PA) accepted
B: upload PB, config/set(PB) accepted
A: cleanup(keep=PA) -> removes PB
Collection now runs inside config/set under the write lock, so a late
client cannot express an opinion about a revision it never saw.
"""
from pathlib import Path
from custom_components.houseplan.const import PLANS_DIR
await _setup(hass)
a = await hass_ws_client(hass)
b = await hass_ws_client(hass)
plans = Path(hass.config.path(PLANS_DIR))
plans.mkdir(parents=True, exist_ok=True)
for stale in plans.glob("r1.*"):
stale.unlink()
url0, p0 = await _upload(a, "r1", b"zero")
rev = (await _save(a, await _cfg([{"id": "r1", "plan_url": url0}]), 0))["result"]["rev"]
url_a, pa = await _upload(a, "r1", b"aaa")
rev_a = (await _save(a, await _cfg([{"id": "r1", "plan_url": url_a}]), rev))["result"]["rev"]
assert not (plans / p0).exists(), "P0 was superseded by A"
url_b, pb = await _upload(b, "r1", b"bbb")
ok = await _save(b, await _cfg([{"id": "r1", "plan_url": url_b}]), rev_a)
assert ok["success"]
# the accepted configuration points at PB, and PB is on disk
assert (plans / pb).read_bytes() == b"bbb"
assert not (plans / pa).exists(), "PA was superseded by B's commit"
async def test_commit_does_not_collect_another_client_s_uncommitted_upload(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""review R3-1, second interleaving: B uploads, A commits, then B commits.
A's commit must not remove PB — B has not written its configuration yet, so
PB is unreferenced but belongs to a live transaction. Age is the guard.
"""
from pathlib import Path
from custom_components.houseplan.const import PLANS_DIR
await _setup(hass)
a = await hass_ws_client(hass)
b = await hass_ws_client(hass)
plans = Path(hass.config.path(PLANS_DIR))
plans.mkdir(parents=True, exist_ok=True)
for stale in plans.glob("r2.*"):
stale.unlink()
url0, _p0 = await _upload(a, "r2", b"zero")
rev = (await _save(a, await _cfg([{"id": "r2", "plan_url": url0}]), 0))["result"]["rev"]
_url_b, pb = await _upload(b, "r2", b"bbb") # B uploads, does not commit
url_a, pa = await _upload(a, "r2", b"aaa")
rev_a = (await _save(a, await _cfg([{"id": "r2", "plan_url": url_a}]), rev))["result"]["rev"]
assert (plans / pb).is_file(), "an uncommitted upload survives someone else's commit"
# B now commits on top of A's revision — its file is still there
ok = await _save(b, await _cfg([{"id": "r2", "plan_url": "/api/houseplan/content/plans/_/" + pb}]), rev_a)
assert ok["success"]
assert (plans / pb).read_bytes() == b"bbb"
assert not (plans / pa).exists()
async def test_abandoned_uploads_are_collected_once_old(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""A rejected upload must not accumulate forever — but only age may free it."""
import os
import time
from pathlib import Path
from custom_components.houseplan.const import PLANS_DIR, PLAN_ORPHAN_TTL_S
await _setup(hass)
client = await hass_ws_client(hass)
plans = Path(hass.config.path(PLANS_DIR))
plans.mkdir(parents=True, exist_ok=True)
for stale in plans.glob("r3.*"):
stale.unlink()
url0, p0 = await _upload(client, "r3", b"zero")
rev = (await _save(client, await _cfg([{"id": "r3", "plan_url": url0}]), 0))["result"]["rev"]
_url, orphan = await _upload(client, "r3", b"abandoned")
old = time.time() - PLAN_ORPHAN_TTL_S - 60
os.utime(plans / orphan, (old, old))
ok = await _save(client, await _cfg([{"id": "r3", "plan_url": url0}]), rev)
assert ok["success"]
assert not (plans / orphan).exists(), "an aged, unreferenced upload is collected"
assert (plans / p0).is_file(), "the referenced plan is never touched"
async def test_collection_ignores_files_that_are_not_plans(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""The plans directory may hold nothing else, but be sure we only take ours."""
import os
import time
from pathlib import Path
from custom_components.houseplan.const import PLANS_DIR, PLAN_ORPHAN_TTL_S
await _setup(hass)
client = await hass_ws_client(hass)
plans = Path(hass.config.path(PLANS_DIR))
plans.mkdir(parents=True, exist_ok=True)
old = time.time() - PLAN_ORPHAN_TTL_S - 60
for name in ("notes.txt", "deep.name.with.dots.png", "readme"):
(plans / name).write_bytes(b"x")
os.utime(plans / name, (old, old))
rev = (await _save(client, await _cfg([{"id": "r4", "plan_url": None}]), 0))["result"]["rev"]
assert rev
assert (plans / "notes.txt").is_file()
assert (plans / "deep.name.with.dots.png").is_file()
assert (plans / "readme").is_file()
async def test_a_marker_showing_its_value_can_be_saved(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""issue #3: display='value' was rejected, and one bad marker fails the lot.
A user could not save the configuration at all after setting any sensor to
"value instead of an icon" — the editor offered the option, the schema had
never heard of it.
"""
await _setup(hass)
client = await hass_ws_client(hass)
cfg = await _cfg([{"id": "f1", "plan_url": None}])
cfg["markers"] = [
{"id": "sensor.t", "binding": "entity:sensor.t", "display": "value"},
{"id": "sensor.h", "binding": "entity:sensor.h", "display": "badge"},
]
ok = await _save(client, cfg, 0)
assert ok["success"], ok.get("error")
await client.send_json_auto_id({"type": "houseplan/config/get"})
got = await client.receive_json()
assert [m["display"] for m in got["result"]["config"]["markers"]] == ["value", "badge"]
async def test_a_failing_collector_does_not_undo_an_accepted_save(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, monkeypatch
) -> None:
"""review R4-1: garbage collection runs behind an already durable write.
If it raised, the client got an error for a revision the store had already
accepted — and its retry then failed with `conflict`, because the server had
moved on. The commit stands and the event fires regardless.
"""
from custom_components.houseplan import websocket_api as wsapi
await _setup(hass)
client = await hass_ws_client(hass)
events = []
hass.bus.async_listen("houseplan_config_updated", lambda ev: events.append(ev.data))
def _boom(*_a, **_k):
raise OSError("the plans directory is on fire")
monkeypatch.setattr(wsapi, "collect_plans", _boom)
cfg = await _cfg([{"id": "r5", "plan_url": None}])
ok = await _save(client, cfg, 0)
assert ok["success"], "an accepted revision must be reported as accepted"
rev = ok["result"]["rev"]
await hass.async_block_till_done()
assert events and events[-1]["rev"] == rev, "the update event still fires"
# the store really holds the new revision, and the reported rev is usable
await client.send_json_auto_id({"type": "houseplan/config/get"})
got = await client.receive_json()
assert got["result"]["rev"] == rev
assert [sp["id"] for sp in got["result"]["config"]["spaces"]] == ["r5"]
monkeypatch.undo()
again = await _save(client, cfg, rev)
assert again["success"], "the next CAS on the reported revision goes through"
async def test_content_signed_path_opens_without_a_bearer_header(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, hass_client_no_auth
) -> None:
"""B1 follow-up: a browser <image>/<a> sends no Authorization header.
The unsigned url must be refused and the signed one must work — otherwise
plan backgrounds and PDF links 401 on a real dashboard (reproduced live,
2026-07-27).
"""
import os
from custom_components.houseplan.const import CONTENT_URL, PLANS_DIR
await _setup(hass)
plans = hass.config.path(PLANS_DIR)
def _write() -> None:
os.makedirs(plans, exist_ok=True)
with open(os.path.join(plans, "s1.png"), "wb") as fh:
fh.write(b"PNGDATA")
await hass.async_add_executor_job(_write)
path = f"{CONTENT_URL}/plans/_/s1.png"
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "houseplan/content/sign", "paths": [path]})
resp = await client.receive_json()
assert resp["success"], resp
signed = resp["result"]["urls"][path]
assert "authSig=" in signed
http = await hass_client_no_auth()
assert (await http.get(path)).status == 401 # unsigned: refused
ok = await http.get(signed)
assert ok.status == 200 and await ok.read() == b"PNGDATA"
# only our own endpoint may be signed
await client.send_json_auto_id(
{"type": "houseplan/content/sign", "paths": ["/api/other/secret"]}
)
resp2 = await client.receive_json()
assert resp2["success"] and resp2["result"]["urls"] == {}