bhetherman/localthings
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5dbe990c1d06c3aaea3c54ebecdfedeaceee1db3
localthings/tests/test_services.py
T
Marc Billow 5dbe990c1d Add write_resource/read_resource services for probing write contracts (issue #300) 
The options-flow "Debug write" panel could only ever do one write to one
href per pass -- not enough for the issue #300 wall oven, whose board
discards settings writes while idle and only keeps them once a cycle is
already running. Finding what starts a cycle needs an ordered sequence of
writes across resources, with real settle delays between them, and a way
to check afterward whether anything actually held.

- coordinator.py: async_raw_write_sequence owns a whole ordered sequence
  under one _session_lock hold (so a poll can't interleave mid-sequence),
  with per-step settle and an optional delayed verify_after re-read done
  outside the lock. async_raw_write is now a one-item wrapper over it, so
  tests/test_coordinator_raw_write.py keeps passing unmodified. Also adds
  async_raw_read, a live GET bypassing the cache -- staleness is exactly
  what makes revert-testing unreliable.
- services.py (new): the two HA services. Device-target resolution scans
  loaded coordinators' MAIN/subdevice identifiers and requires exactly one
  match, so an area/label target can't silently fan a raw write out across
  several appliances. Canonical->actual href translation happens here, not
  in the coordinator, which stays subdevice-agnostic.
- services.yaml (new): selectors/descriptions for both services, inline
  per HA's custom-integration support -- keeps translations/en.json's
  mirror test (test_translations.py) green without touching all 6
  languages for a services block. New exception keys (write caps, device
  target resolution) still went into translations/*.json's existing
  exceptions section, mirrored across all 6 languages.
- __init__.py: adds async_setup to register the services once, process-wide.
- config_flow.py: the debug panel's async_step_debug_edit now calls
  write_resource instead of coord.async_raw_write directly, so there is
  exactly one code path that performs a raw write.
- README.md: new Part 5 documenting both services, with a worked
  write_resource example; points the capability-gap section at them.

tests/test_services.py (new): sequencing/ordering, settle timing, changed
vs. held (the reverted case is issue #300's own symptom), exactly-one-
device resolution, subdevice href translation, validation caps, and the
options-flow panel end to end through the service.
2026-08-07 22:08:28 +00:00

459 lines
17 KiB
Python
Raw Blame History

"""Tests for the write_resource/read_resource services (issue #300) and the
options-flow debug panel now that it goes through write_resource instead of
calling coordinator.async_raw_write directly (config_flow.py). See
tests/test_coordinator_raw_write.py for the underlying single-write
primitive's own session/validation tests -- these focus on the service
layer: device target resolution, subdevice href translation, ordered
multi-write sequencing, settle/verify_after, and the options-flow rewiring.
"""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
import cbor2
import pytest
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import device_registry as dr
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.localthings.config_flow import LocalThingsOptionsFlow
from custom_components.localthings.const import (
CONF_HOST,
CONF_LEAF_CERT_PEM,
CONF_LEAF_KEY_PEM,
CONF_PORT,
DOMAIN,
SERVICE_READ_RESOURCE,
SERVICE_WRITE_RESOURCE,
)
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.subdevices import Subdevice
from custom_components.localthings.services import async_setup_services
ENTRY_DATA = {
CONF_HOST: "10.0.0.199",
CONF_PORT: 49154,
CONF_LEAF_CERT_PEM: "-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----",
CONF_LEAF_KEY_PEM: "-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----",
}
_SLEEP_TARGET = "custom_components.localthings.coordinator.asyncio.sleep"
class _FakeSession:
"""Stand-in for DtlsCoapSession: records every POST verbatim and answers
GET from a per-href queue of canned representations, so a test can model
a value that changes across successive reads of the same href (the
write's own follow-up GET vs. a later verify_after re-read) -- no real
DTLS/network involved. Modeled on test_coordinator_raw_write.py's
_FakeRawWriteSession, extended for multi-step sequences."""
def __init__(self, post_code: int = 0x44):
self.post_calls: list[tuple[list[str], bytes]] = []
self.get_calls: list[list[str]] = []
self._post_code = post_code
self._get_reps: dict[str, list[dict]] = {}
def queue_get(self, href: str, rep: dict) -> None:
"""Queue one more canned rep for `href`'s next GET. Once an href's
queue is down to one entry, that entry keeps answering every
further GET -- a test only needs to queue the values that
actually change across calls."""
self._get_reps.setdefault(href.strip("/"), []).append(rep)
def post(self, path_segs, payload, timeout=None):
self.post_calls.append((list(path_segs), payload))
return self._post_code, b""
def get(self, path_segs, timeout=None):
self.get_calls.append(list(path_segs))
key = "/".join(path_segs)
queue = self._get_reps.get(key)
if not queue:
return 0x45, cbor2.dumps({})
rep = queue.pop(0) if len(queue) > 1 else queue[0]
return 0x45, cbor2.dumps(rep)
def pace(self):
pass
@pytest.fixture
def coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
entry = MockConfigEntry(
domain=DOMAIN,
data=ENTRY_DATA,
unique_id="localthings_SERVICES-TEST",
)
entry.add_to_hass(hass)
coord = LocalThingsCoordinator(hass, entry)
# async_raw_write_sequence kicks a refresh after the write; a real
# refresh would try to poll a session that doesn't exist for this unit
# test, so replace it with a no-op the same way
# test_coordinator_raw_write.py does.
coord.async_request_refresh = AsyncMock()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coord
async_setup_services(hass)
return coord
@pytest.fixture
def device_id(hass: HomeAssistant, coordinator: LocalThingsCoordinator) -> str:
"""Register MAIN's HA device under the same identifiers a real config
entry setup would have used (coordinator.device_info)."""
dev_reg = dr.async_get(hass)
device = dev_reg.async_get_or_create(
config_entry_id=coordinator._entry.entry_id,
identifiers=coordinator.device_info["identifiers"],
)
return device.id
async def _call_write(hass: HomeAssistant, device: str, **data):
return await hass.services.async_call(
DOMAIN,
SERVICE_WRITE_RESOURCE,
data,
target={"device_id": device},
blocking=True,
return_response=True,
)
async def _call_read(hass: HomeAssistant, device: str, **data):
return await hass.services.async_call(
DOMAIN,
SERVICE_READ_RESOURCE,
data,
target={"device_id": device},
blocking=True,
return_response=True,
)
# ----------------------------------------------------------------------
# write_resource: ordered multi-write sequencing
# ----------------------------------------------------------------------
async def test_write_resource_posts_in_order_with_exact_bodies(hass, coordinator, device_id):
fake = _FakeSession(post_code=0x44)
coordinator._session = fake
writes = [
{"href": "/mode/vs/0", "payload": {"a": 1}},
{"href": "/washer/vs/0", "payload": {"b": 2}},
{"href": "/mode/vs/0", "payload": {"a": 3}},
]
response = await _call_write(hass, device_id, writes=writes)
assert [path for path, _ in fake.post_calls] == [
["mode", "vs", "0"],
["washer", "vs", "0"],
["mode", "vs", "0"],
]
assert [cbor2.loads(body) for _, body in fake.post_calls] == [
{"a": 1},
{"b": 2},
{"a": 3},
]
assert response is not None
assert response["device_id"] == device_id
assert len(response["results"]) == 3
for result, write in zip(response["results"], writes, strict=True):
assert result["href"] == write["href"]
assert result["actual_href"] == write["href"] # MAIN -> identity transform
assert result["raw_code"] == 0x44
assert result["code"] == "2.04"
assert result["accepted"] is True
coordinator.async_request_refresh.assert_awaited_once()
async def test_write_resource_settle_honored_between_writes(hass, coordinator, device_id):
"""settle waits *before the next write*, not after the last one --
asserted on the recorded durations rather than a real sleep."""
fake = _FakeSession()
coordinator._session = fake
writes = [
{"href": "/a/vs/0", "payload": {"x": 1}, "settle": 2},
{"href": "/b/vs/0", "payload": {"x": 2}, "settle": 5},
{"href": "/c/vs/0", "payload": {"x": 3}, "settle": 9},
]
with patch(_SLEEP_TARGET, new_callable=AsyncMock) as mock_sleep:
await _call_write(hass, device_id, writes=writes)
assert [call.args[0] for call in mock_sleep.call_args_list] == [2.0, 5.0]
async def test_write_resource_changed_true_when_readback_matches_payload(
hass, coordinator, device_id
):
fake = _FakeSession()
fake.queue_get("mode/vs/0", {"x.field": "target"})
coordinator._session = fake
response = await _call_write(
hass, device_id, writes=[{"href": "/mode/vs/0", "payload": {"x.field": "target"}}]
)
assert response["results"][0]["changed"] is True
async def test_write_resource_changed_false_when_readback_differs(hass, coordinator, device_id):
fake = _FakeSession()
fake.queue_get("mode/vs/0", {"x.field": "unchanged"})
coordinator._session = fake
response = await _call_write(
hass, device_id, writes=[{"href": "/mode/vs/0", "payload": {"x.field": "target"}}]
)
assert response["results"][0]["changed"] is False
# ----------------------------------------------------------------------
# verify_after: the delayed re-read this feature exists for (issue #300)
# ----------------------------------------------------------------------
async def test_write_resource_verify_after_reports_held(hass, coordinator, device_id):
fake = _FakeSession()
fake.queue_get("mode/vs/0", {"x.field": "target"}) # write's own follow-up read
fake.queue_get("mode/vs/0", {"x.field": "target"}) # verify_after re-read: held
coordinator._session = fake
with patch(_SLEEP_TARGET, new_callable=AsyncMock) as mock_sleep:
response = await _call_write(
hass,
device_id,
writes=[{"href": "/mode/vs/0", "payload": {"x.field": "target"}}],
verify_after=30,
)
assert 30.0 in [call.args[0] for call in mock_sleep.call_args_list]
verified = response["verified"]["/mode/vs/0"]
assert verified["held"] is True
assert verified["rep"] == {"x.field": "target"}
async def test_write_resource_verify_after_reports_reverted(hass, coordinator, device_id):
"""Issue #300's own symptom: a Samsung wall oven board answers 2.04
Changed to a settings write while idle and then silently reverts it
once the follow-up read has already come back clean. verify_after's
whole purpose is catching exactly this -- `changed` (immediate) and
`held` (delayed) must be able to disagree."""
fake = _FakeSession()
fake.queue_get("mode/vs/0", {"x.field": "target"}) # looks accepted right after the write
fake.queue_get("mode/vs/0", {"x.field": "original"}) # reverted by the time verify_after fires
coordinator._session = fake
with patch(_SLEEP_TARGET, new_callable=AsyncMock):
response = await _call_write(
hass,
device_id,
writes=[{"href": "/mode/vs/0", "payload": {"x.field": "target"}}],
verify_after=30,
)
assert response["results"][0]["changed"] is True
verified = response["verified"]["/mode/vs/0"]
assert verified["held"] is False
assert verified["rep"] == {"x.field": "original"}
async def test_write_resource_no_verified_key_when_verify_after_is_zero(
hass, coordinator, device_id
):
fake = _FakeSession()
coordinator._session = fake
response = await _call_write(
hass, device_id, writes=[{"href": "/mode/vs/0", "payload": {"x": 1}}]
)
assert "verified" not in response
# ----------------------------------------------------------------------
# Device target resolution
# ----------------------------------------------------------------------
async def test_write_resource_zero_devices_rejected(hass, coordinator):
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_WRITE_RESOURCE,
{"writes": [{"href": "/a/vs/0", "payload": {"x": 1}}]},
blocking=True,
return_response=True,
)
async def test_write_resource_two_devices_rejected(hass, coordinator, device_id):
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_WRITE_RESOURCE,
{"writes": [{"href": "/a/vs/0", "payload": {"x": 1}}]},
target={"device_id": [device_id, "some-other-unrelated-device-id"]},
blocking=True,
return_response=True,
)
async def test_write_resource_unresolvable_device_rejected(hass, coordinator):
"""The device exists in the registry but no loaded coordinator claims
it -- e.g. a device belonging to a different integration entirely."""
dev_reg = dr.async_get(hass)
other_entry = MockConfigEntry(domain="other_domain")
other_entry.add_to_hass(hass)
device = dev_reg.async_get_or_create(
config_entry_id=other_entry.entry_id,
identifiers={("other_domain", "unrelated")},
)
with pytest.raises(ServiceValidationError):
await _call_write(hass, device.id, writes=[{"href": "/a/vs/0", "payload": {"x": 1}}])
# ----------------------------------------------------------------------
# Subdevice href translation (issue #177)
# ----------------------------------------------------------------------
async def test_write_resource_translates_href_for_indexed_subdevice(hass, coordinator):
sub = Subdevice(kind="indexed", key="1", seed_path=("device", "1"))
coordinator.subdevices = [sub]
dev_reg = dr.async_get(hass)
device = dev_reg.async_get_or_create(
config_entry_id=coordinator._entry.entry_id,
identifiers=coordinator.device_info_for(sub)["identifiers"],
)
fake = _FakeSession()
coordinator._session = fake
response = await _call_write(
hass, device.id, writes=[{"href": "/mode/vs/0", "payload": {"x": 1}}]
)
assert fake.post_calls[0][0] == ["mode", "vs", "1"]
result = response["results"][0]
assert result["href"] == "/mode/vs/0"
assert result["actual_href"] == "/mode/vs/1"
# ----------------------------------------------------------------------
# Validation caps
# ----------------------------------------------------------------------
async def test_write_resource_rejects_more_than_ten_writes(hass, coordinator, device_id):
writes = [{"href": f"/x/vs/{i}", "payload": {"a": i}} for i in range(11)]
with pytest.raises(ServiceValidationError):
await _call_write(hass, device_id, writes=writes)
async def test_write_resource_rejects_settle_above_thirty_seconds(hass, coordinator, device_id):
with pytest.raises(ServiceValidationError):
await _call_write(
hass,
device_id,
writes=[{"href": "/a/vs/0", "payload": {"x": 1}, "settle": 31}],
)
async def test_write_resource_rejects_empty_payload(hass, coordinator, device_id):
with pytest.raises(ServiceValidationError):
await _call_write(hass, device_id, writes=[{"href": "/a/vs/0", "payload": {}}])
async def test_write_resource_rejects_non_dict_payload(hass, coordinator, device_id):
with pytest.raises(ServiceValidationError):
await _call_write(
hass, device_id, writes=[{"href": "/a/vs/0", "payload": ["not", "a", "dict"]}]
)
async def test_write_resource_rejects_empty_href(hass, coordinator, device_id):
with pytest.raises(ServiceValidationError):
await _call_write(hass, device_id, writes=[{"href": "", "payload": {"x": 1}}])
# ----------------------------------------------------------------------
# read_resource
# ----------------------------------------------------------------------
async def test_read_resource_with_href_does_live_get(hass, coordinator, device_id):
fake = _FakeSession()
fake.queue_get("mode/vs/0", {"x.field": "live"})
coordinator._session = fake
response = await _call_read(hass, device_id, href="/mode/vs/0")
assert fake.get_calls == [["mode", "vs", "0"]]
assert response["href"] == "/mode/vs/0"
assert response["actual_href"] == "/mode/vs/0"
assert response["rep"] == {"x.field": "live"}
async def test_read_resource_without_href_returns_cached_snapshot_and_does_not_get(
hass, coordinator, device_id
):
fake = _FakeSession()
coordinator._session = fake
# Seed the cache the way a poll would -- the fake session's `get` is
# never touched by this test.
coordinator._observe.apply("/mode/vs/0", {"x.field": "cached"}, source="poll")
response = await _call_read(hass, device_id)
assert fake.get_calls == []
assert response["resources"]["/mode/vs/0"] == {"x.field": "cached"}
# ----------------------------------------------------------------------
# Options-flow debug panel, rewired onto write_resource (issue #300)
# ----------------------------------------------------------------------
async def test_options_flow_debug_write_goes_through_write_resource_service(
hass, coordinator, device_id
):
"""Drives LocalThingsOptionsFlow's real step methods directly (rather
than through hass.config_entries.options.async_init/async_configure,
which needs the integration resolvable through HA's loader -- not
exercised by any test in this repo and orthogonal to what this test is
actually checking): the panel's async_step_debug_edit must call the
write_resource service and get back a result the rest of the flow can
render, not coord.async_raw_write directly."""
fake = _FakeSession(post_code=0x44)
fake.queue_get("course/vs/0", {"x.field": "after"})
coordinator._session = fake
flow = LocalThingsOptionsFlow()
flow.hass = hass
flow.handler = coordinator._entry.entry_id
write_result = await flow.async_step_debug_write({"href": "/course/vs/0"})
assert write_result["step_id"] == "debug_edit"
edit_result = await flow.async_step_debug_edit({"payload": {"x.com.samsung.da.field": "value"}})
assert edit_result["step_id"] == "debug_result"
placeholders = edit_result["description_placeholders"]
assert placeholders is not None
assert placeholders["code"] == "2.04 (0x44)"
assert len(fake.post_calls) == 1
posted_path, posted_body = fake.post_calls[0]
assert posted_path == ["course", "vs", "0"]
assert cbor2.loads(posted_body) == {"x.com.samsung.da.field": "value"}
assert len(fake.post_calls) == 1
posted_path, posted_body = fake.post_calls[0]
assert posted_path == ["course", "vs", "0"]
assert cbor2.loads(posted_body) == {"x.com.samsung.da.field": "value"}
Reference in New Issue View Git Blame Copy Permalink