Marc Billow
c731ecefe5
Power users can now pick a resource href from a live dropdown, view its current value, and POST a minimal patch straight to the device -- to pin down device-specific write behavior without waiting on a new release. Bypasses the remote-control block and all write_fn/validate_fn logic by design; the existing remote-control settings toggle moves behind the same options-flow menu.
138 lines
5.1 KiB
Python
138 lines
5.1 KiB
Python
"""Tests for LocalThingsCoordinator.async_raw_write -- the debug-only
|
|
arbitrary-href write primitive backing the options-flow debug panel
|
|
(issue #54). It deliberately bypasses the remote-control block and every
|
|
write_fn/validate_fn, so this only exercises the primitive itself: it
|
|
POSTs exactly the caller's body, reads the href back, and validates its
|
|
inputs.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock
|
|
|
|
import cbor2
|
|
import pytest
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from custom_components.localthings.const import (
|
|
CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
|
|
)
|
|
from custom_components.localthings.coordinator import LocalThingsCoordinator
|
|
|
|
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-----',
|
|
}
|
|
|
|
|
|
class _FakeRawWriteSession:
|
|
"""Stand-in for DtlsCoapSession: records every POST verbatim and
|
|
answers the follow-up GET with a canned representation -- no real
|
|
DTLS/network involved."""
|
|
|
|
def __init__(self, post_code: int = 0x44, get_rep: dict | None = None):
|
|
self.post_calls: list[tuple[list[str], bytes]] = []
|
|
self.get_calls: list[list[str]] = []
|
|
self._post_code = post_code
|
|
self._get_rep = {} if get_rep is None else get_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))
|
|
return 0x45, cbor2.dumps(self._get_rep)
|
|
|
|
def pace(self):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN, data=ENTRY_DATA, unique_id='localthings_RAWWRITE-TEST',
|
|
)
|
|
entry.add_to_hass(hass)
|
|
coord = LocalThingsCoordinator(hass, entry)
|
|
# async_raw_write kicks a full 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 the reference test setup does.
|
|
coord.async_request_refresh = AsyncMock()
|
|
return coord
|
|
|
|
|
|
async def test_raw_write_splits_href_and_posts_exact_body(coordinator) -> None:
|
|
fake = _FakeRawWriteSession(post_code=0x44, get_rep={'x.field': 'after'})
|
|
coordinator._session = fake
|
|
|
|
body = {'x.com.samsung.da.field': 'value', 'n': 1}
|
|
code, new_rep = await coordinator.async_raw_write('/course/vs/0', body)
|
|
|
|
assert len(fake.post_calls) == 1
|
|
posted_path, posted_bytes = fake.post_calls[0]
|
|
assert posted_path == ['course', 'vs', '0']
|
|
assert cbor2.loads(posted_bytes) == body
|
|
|
|
assert code == 0x44
|
|
assert new_rep == {'x.field': 'after'}
|
|
coordinator.async_request_refresh.assert_awaited_once()
|
|
|
|
|
|
async def test_raw_write_reads_href_back_for_ground_truth(coordinator) -> None:
|
|
fake = _FakeRawWriteSession(post_code=0x45, get_rep={'value': 'confirmed'})
|
|
coordinator._session = fake
|
|
|
|
code, new_rep = await coordinator.async_raw_write('/washer/vs/0', {'a': 1})
|
|
|
|
assert fake.get_calls == [['washer', 'vs', '0']]
|
|
assert code == 0x45
|
|
assert new_rep == {'value': 'confirmed'}
|
|
|
|
|
|
async def test_raw_write_returns_coap_code_from_post_not_get(coordinator) -> None:
|
|
"""The returned code reflects the write's own response, even though a
|
|
follow-up GET (which could carry a different code in principle) runs
|
|
right after it."""
|
|
fake = _FakeRawWriteSession(post_code=0x80, get_rep={'value': 1})
|
|
coordinator._session = fake
|
|
|
|
code, _ = await coordinator.async_raw_write('/test/vs/0', {'a': 1})
|
|
|
|
assert code == 0x80
|
|
|
|
|
|
async def test_raw_write_rejects_empty_dict(coordinator) -> None:
|
|
with pytest.raises(ServiceValidationError):
|
|
await coordinator.async_raw_write('/washer/vs/0', {})
|
|
|
|
|
|
async def test_raw_write_rejects_non_dict_payload(coordinator) -> None:
|
|
with pytest.raises(ServiceValidationError):
|
|
await coordinator.async_raw_write('/washer/vs/0', ['not', 'a', 'dict']) # type: ignore[arg-type]
|
|
|
|
|
|
async def test_raw_write_rejects_empty_href(coordinator) -> None:
|
|
with pytest.raises(ServiceValidationError):
|
|
await coordinator.async_raw_write('', {'a': 1})
|
|
|
|
|
|
async def test_raw_write_rejects_root_href(coordinator) -> None:
|
|
with pytest.raises(ServiceValidationError):
|
|
await coordinator.async_raw_write('/', {'a': 1})
|
|
|
|
|
|
async def test_raw_write_validation_errors_do_not_touch_the_session(coordinator) -> None:
|
|
"""A rejected call must fail before ever reaching the network -- no
|
|
session needs to be connected at all for validation to run."""
|
|
assert coordinator._session is None
|
|
|
|
with pytest.raises(ServiceValidationError):
|
|
await coordinator.async_raw_write('/washer/vs/0', {})
|
|
|
|
assert coordinator._session is None
|
|
coordinator.async_request_refresh.assert_not_awaited()
|