mirror of
https://github.com/Matysh/houseplan-card
synced 2026-07-31 08:28:31 +00:00
Matysh
c1e3cdb768
- upload/plan_set tests asserted the old public /houseplan_files/... paths and only run in CI, so the B1 change surfaced there - +test for the fail-closed admin check (audit B2/T4: the authorization boundary had zero coverage)
128 lines
4.7 KiB
Python
128 lines
4.7 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()
|
|
assert resp["success"] and resp["result"]["url"].startswith("/api/houseplan/content/plans/_/s1.png?v=")
|
|
|
|
|
|
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
|