diff --git a/custom_components/localthings/config_flow.py b/custom_components/localthings/config_flow.py index fe72327..48bdb87 100644 --- a/custom_components/localthings/config_flow.py +++ b/custom_components/localthings/config_flow.py @@ -2,6 +2,7 @@ from __future__ import annotations import datetime +import json import logging import re import selectors @@ -15,6 +16,10 @@ from homeassistant import config_entries from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult from homeassistant.helpers.selector import ( + ObjectSelector, + SelectSelector, + SelectSelectorConfig, + SelectSelectorMode, TextSelector, TextSelectorConfig, TextSelectorType, @@ -396,22 +401,44 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): class LocalThingsOptionsFlow(config_entries.OptionsFlow): - """Per-device override of the remote-control-off write block (issue - #54). The block exists because most devices reject writes outright - while remote control is off and a clear error beats a silent + """Per-device options: the remote-control-off write-block override + (issue #54) plus a debug panel for writing an arbitrary body to an + arbitrary resource href, so a user can pin down device-specific write + behavior without waiting on a new release. + + The remote-control override exists because most devices reject writes + outright while remote control is off and a clear error beats a silent device-side rejection -- but not every model actually enforces that, so this lets a user who's confirmed their device accepts writes anyway turn the block off for just that device rather than it being - hardcoded on for everyone.""" + hardcoded on for everyone. The debug panel goes further: it bypasses + that block (and every write_fn/validate_fn) entirely, sending exactly + the body the user types to whatever href they pick. + """ + + def __init__(self) -> None: + self._debug_href: str = "" + self._debug_result: tuple[int, dict] | None = None + + def _coordinator(self): + return self.hass.data.get(DOMAIN, {}).get(self.config_entry.entry_id) async def async_step_init( self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + return self.async_show_menu( + step_id="init", + menu_options=["settings", "debug_write"], + ) + + async def async_step_settings( + self, user_input: dict[str, Any] | None = None ) -> FlowResult: if user_input is not None: return self.async_create_entry(data=user_input) return self.async_show_form( - step_id="init", + step_id="settings", data_schema=vol.Schema({ vol.Required( CONF_BYPASS_REMOTE_CONTROL, @@ -421,3 +448,96 @@ class LocalThingsOptionsFlow(config_entries.OptionsFlow): ): bool, }), ) + + async def async_step_debug_write( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + coord = self._coordinator() + if coord is None: + return self.async_abort(reason="not_loaded") + + if user_input is not None: + self._debug_href = user_input["href"] + return await self.async_step_debug_edit() + + hrefs = sorted(coord.last_resources.keys()) + return self.async_show_form( + step_id="debug_write", + data_schema=vol.Schema({ + vol.Required("href"): SelectSelector(SelectSelectorConfig( + options=hrefs, + custom_value=True, + mode=SelectSelectorMode.DROPDOWN, + )), + }), + ) + + def _show_debug_edit_form( + self, href: str, current: dict, errors: dict[str, str], payload, + ) -> FlowResult: + return self.async_show_form( + step_id="debug_edit", + data_schema=vol.Schema({ + vol.Required( + "payload", default=(payload if payload is not None else {}), + ): ObjectSelector(), + }), + errors=errors, + description_placeholders={ + "href": href, + "current_value": ( + json.dumps(current, indent=2, ensure_ascii=False) + if current else "(no cached value)" + ), + }, + ) + + async def async_step_debug_edit( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + coord = self._coordinator() + if coord is None: + return self.async_abort(reason="not_loaded") + + href = self._debug_href + current = coord.resource(href) + + if user_input is not None: + payload = user_input.get("payload") or {} + if not isinstance(payload, dict) or not payload: + return self._show_debug_edit_form( + href, current, {"payload": "empty_payload"}, payload + ) + try: + code, new_rep = await coord.async_raw_write(href, payload) + except Exception: # noqa: BLE001 - surfaced to the user below + _LOGGER.exception("debug raw write failed for %s", href) + return self._show_debug_edit_form( + href, current, {"base": "write_failed"}, payload + ) + self._debug_result = (code, new_rep) + return await self.async_step_debug_result() + + return self._show_debug_edit_form(href, current, {}, None) + + async def async_step_debug_result( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + code, new_rep = self._debug_result or (0, {}) + return self.async_show_menu( + step_id="debug_result", + menu_options=["debug_write", "finish"], + description_placeholders={ + "code": f"{code >> 5}.{code & 0x1f:02d} ({code:#04x})", + "new_value": ( + json.dumps(new_rep, indent=2, ensure_ascii=False) + if new_rep else "(no value returned)" + ), + }, + ) + + async def async_step_finish( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + # Close the flow without altering saved options. + return self.async_create_entry(data=dict(self.config_entry.options)) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index 94e8fc8..0e3d44a 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -630,3 +630,58 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._log.error("command failed for %s: %s", write_href, e) else: await self.async_request_refresh() + + # ------------------------------------------------------------------ + # Debug raw write (issue #54): a power-user escape hatch for the + # options-flow debug panel, letting a user POST an arbitrary partial + # body to an arbitrary href to pin down device-specific write behavior + # without waiting on a new release. Deliberately bypasses the + # remote-control block and every write_fn/validate_fn above -- that's + # the whole point, so use with care. + # ------------------------------------------------------------------ + + def _raw_write_blocking(self, path_segs: list[str], body: dict, href: str) -> tuple[int, dict]: + """Debug primitive: POST an arbitrary patch, then read the href + back for ground truth. Blocking -- runs in executor.""" + if self._session is None: + self._connect_session() + sess = self._session + if sess is None: + raise RuntimeError("no session") + code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=self._POST_TIMEOUT_S) + self._log.warning("DEBUG raw write POST %s %r → code %#04x", href, body, code) + new_rep: dict = {} + try: + sess.pace() + rcode, payload = sess.get(path_segs, timeout=10.0) + if rcode == 0x45 and payload: + rep = cbor2.loads(payload) + if isinstance(rep, dict): + self._observe.apply(href, rep, source='poll') + new_rep = rep + except Exception as e: + self._log.debug("raw write follow-up read failed: %s", e) + return code, new_rep + + async def async_raw_write(self, href: str, body: dict) -> tuple[int, dict]: + """Debug-only arbitrary write (issue #54). Bypasses the + remote-control block and all write_fn/validate_fn logic; sends + `body` verbatim as a partial-rep PATCH to `href`. Returns + (coap_code, new_rep) where new_rep is the href's value read back + right after the write. Used by the options-flow debug panel to + help users pin down device-specific write behavior without a new + release.""" + if not isinstance(body, dict) or not body: + raise ServiceValidationError("Debug write payload must be a non-empty object.") + path_segs = [s for s in href.strip('/').split('/') if s] + if not path_segs: + raise ServiceValidationError("A resource href is required.") + norm_href = '/' + '/'.join(path_segs) + async with self._session_lock: + code, new_rep = await self.hass.async_add_executor_job( + self._raw_write_blocking, path_segs, body, norm_href + ) + # Hasten a full summary poll so entities on other resources catch + # up too -- a debug write can affect siblings, not just its href. + await self.async_request_refresh() + return code, new_rep diff --git a/custom_components/localthings/strings.json b/custom_components/localthings/strings.json index d375db1..de1ac17 100644 --- a/custom_components/localthings/strings.json +++ b/custom_components/localthings/strings.json @@ -250,11 +250,47 @@ "step": { "init": { "title": "Local Things Options", + "menu_options": { + "settings": "Remote control write settings", + "debug_write": "Debug: write to a resource" + } + }, + "settings": { + "title": "Remote control write settings", "description": "Some devices accept certain writes (e.g. default detergent/softener dosing on a washer) even while reporting remote control off. By default, LocalThings blocks every write with a clear error whenever a device reports remote control off, rather than letting the device silently reject it. Only enable this if you've confirmed writes actually work on this device with remote control off -- otherwise you'll trade that clear error for a silent failure.", "data": { "bypass_remote_control_lock": "Allow writes even when remote control is reported off" } + }, + "debug_write": { + "title": "Debug: write to a resource", + "description": "Power-user tool for pinning down device-specific write behavior. Pick the resource (href) you want to write to, or type a custom one that isn't listed. This bypasses the remote-control-off block and sends exactly the fields you provide -- it can misconfigure your appliance, so use it deliberately.", + "data": { + "href": "Resource href" + } + }, + "debug_edit": { + "title": "Debug write: {href}", + "description": "Current value of `{href}`:\n```\n{current_value}\n```\nEnter ONLY the field(s) you want to change below. What you enter is sent to the device as-is (a partial update); it is NOT merged with the fields shown above, so don't paste the whole value back. Mind the types: a numeric string like \"1\" must stay quoted -- a bare 1 is sent as an integer, which some devices reject.", + "data": { + "payload": "Payload to write" + } + }, + "debug_result": { + "title": "Debug write result", + "description": "The device returned CoAP code {code}. A 2.xx class means the write was accepted; 4.xx/5.xx means it was rejected. The resource now reads:\n```\n{new_value}\n```\nIf the value is unchanged, the device likely dropped the write. Pick what to do next:", + "menu_options": { + "debug_write": "Write another resource", + "finish": "Finish" + } } + }, + "error": { + "empty_payload": "Enter at least one field to write.", + "write_failed": "The write failed. Check the Home Assistant logs for details." + }, + "abort": { + "not_loaded": "This device isn't connected yet. Try again once it has loaded." } }, "issues": { diff --git a/custom_components/localthings/translations/en.json b/custom_components/localthings/translations/en.json index d375db1..de1ac17 100644 --- a/custom_components/localthings/translations/en.json +++ b/custom_components/localthings/translations/en.json @@ -250,11 +250,47 @@ "step": { "init": { "title": "Local Things Options", + "menu_options": { + "settings": "Remote control write settings", + "debug_write": "Debug: write to a resource" + } + }, + "settings": { + "title": "Remote control write settings", "description": "Some devices accept certain writes (e.g. default detergent/softener dosing on a washer) even while reporting remote control off. By default, LocalThings blocks every write with a clear error whenever a device reports remote control off, rather than letting the device silently reject it. Only enable this if you've confirmed writes actually work on this device with remote control off -- otherwise you'll trade that clear error for a silent failure.", "data": { "bypass_remote_control_lock": "Allow writes even when remote control is reported off" } + }, + "debug_write": { + "title": "Debug: write to a resource", + "description": "Power-user tool for pinning down device-specific write behavior. Pick the resource (href) you want to write to, or type a custom one that isn't listed. This bypasses the remote-control-off block and sends exactly the fields you provide -- it can misconfigure your appliance, so use it deliberately.", + "data": { + "href": "Resource href" + } + }, + "debug_edit": { + "title": "Debug write: {href}", + "description": "Current value of `{href}`:\n```\n{current_value}\n```\nEnter ONLY the field(s) you want to change below. What you enter is sent to the device as-is (a partial update); it is NOT merged with the fields shown above, so don't paste the whole value back. Mind the types: a numeric string like \"1\" must stay quoted -- a bare 1 is sent as an integer, which some devices reject.", + "data": { + "payload": "Payload to write" + } + }, + "debug_result": { + "title": "Debug write result", + "description": "The device returned CoAP code {code}. A 2.xx class means the write was accepted; 4.xx/5.xx means it was rejected. The resource now reads:\n```\n{new_value}\n```\nIf the value is unchanged, the device likely dropped the write. Pick what to do next:", + "menu_options": { + "debug_write": "Write another resource", + "finish": "Finish" + } } + }, + "error": { + "empty_payload": "Enter at least one field to write.", + "write_failed": "The write failed. Check the Home Assistant logs for details." + }, + "abort": { + "not_loaded": "This device isn't connected yet. Try again once it has loaded." } }, "issues": { diff --git a/tests/localthings/test_config_flow.py b/tests/localthings/test_config_flow.py index f21ee54..bc7de8b 100644 --- a/tests/localthings/test_config_flow.py +++ b/tests/localthings/test_config_flow.py @@ -279,6 +279,20 @@ def test_probe_marks_washer_as_recognized(monkeypatch): assert recognized is True +async def test_options_flow_init_shows_menu(hass: HomeAssistant) -> None: + """The options flow's entry point is now a menu (issue #54's debug + panel lives alongside the remote-control settings), not the settings + form directly.""" + entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}') + entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(entry.entry_id) + + assert result['type'] == FlowResultType.MENU + assert result['step_id'] == 'init' + assert set(result['menu_options']) == {'settings', 'debug_write'} + + async def test_options_flow_default_is_off(hass: HomeAssistant) -> None: """The bypass defaults to False, so devices that never touch this option see no change in the remote-control write block.""" @@ -286,9 +300,12 @@ async def test_options_flow_default_is_off(hass: HomeAssistant) -> None: entry.add_to_hass(hass) result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'settings'} + ) assert result['type'] == FlowResultType.FORM - assert result['step_id'] == 'init' + assert result['step_id'] == 'settings' assert result['data_schema']({})[CONF_BYPASS_REMOTE_CONTROL] is False @@ -299,6 +316,9 @@ async def test_options_flow_can_enable_bypass(hass: HomeAssistant) -> None: entry.add_to_hass(hass) result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'settings'} + ) result = await hass.config_entries.options.async_configure( result['flow_id'], user_input={CONF_BYPASS_REMOTE_CONTROL: True} ) @@ -317,5 +337,140 @@ async def test_options_flow_reflects_previously_saved_value(hass: HomeAssistant) entry.add_to_hass(hass) result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'settings'} + ) assert result['data_schema']({})[CONF_BYPASS_REMOTE_CONTROL] is True + + +async def test_options_flow_debug_write_shows_hrefs_from_coordinator( + hass: HomeAssistant, mock_coordinator_session, +) -> None: + """The debug panel's href dropdown is populated from the live + coordinator's cached resources, not a static list.""" + entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}') + entry.add_to_hass(hass) + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'debug_write'} + ) + + assert result['type'] == FlowResultType.FORM + assert result['step_id'] == 'debug_write' + + +async def test_options_flow_debug_write_aborts_when_device_not_loaded( + hass: HomeAssistant, +) -> None: + """No coordinator in hass.data (device never finished loading) means + the debug panel has nothing to write to or read from.""" + entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}') + entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'debug_write'} + ) + + assert result['type'] == FlowResultType.ABORT + assert result['reason'] == 'not_loaded' + + +async def test_options_flow_debug_edit_writes_and_shows_result( + hass: HomeAssistant, mock_coordinator_session, +) -> None: + """Picking an href, then submitting a payload, drives + coordinator.async_raw_write and lands on the result menu with the + device's response.""" + entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}') + entry.add_to_hass(hass) + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'debug_write'} + ) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'href': '/washer/vs/0'}, + ) + assert result['type'] == FlowResultType.FORM + assert result['step_id'] == 'debug_edit' + + with patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator.async_raw_write', + return_value=(0x44, {'a': 1}), + ): + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'payload': {'a': 1}}, + ) + + assert result['type'] == FlowResultType.MENU + assert result['step_id'] == 'debug_result' + assert result['description_placeholders']['code'] == '2.04 (0x44)' + + +async def test_options_flow_debug_edit_rejects_empty_payload( + hass: HomeAssistant, mock_coordinator_session, +) -> None: + entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}') + entry.add_to_hass(hass) + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'debug_write'} + ) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'href': '/washer/vs/0'}, + ) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'payload': {}}, + ) + + assert result['type'] == FlowResultType.FORM + assert result['step_id'] == 'debug_edit' + assert result['errors'] == {'payload': 'empty_payload'} + + +async def test_options_flow_finish_preserves_existing_options( + hass: HomeAssistant, mock_coordinator_session, +) -> None: + """Finishing from the debug-result menu must not clobber a previously + saved remote-control bypass setting.""" + entry = MockConfigEntry( + domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}', + options={CONF_BYPASS_REMOTE_CONTROL: True}, + ) + entry.add_to_hass(hass) + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + result = await hass.config_entries.options.async_init(entry.entry_id) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'debug_write'} + ) + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'href': '/washer/vs/0'}, + ) + with patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator.async_raw_write', + return_value=(0x44, {'a': 1}), + ): + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'payload': {'a': 1}}, + ) + assert result['type'] == FlowResultType.MENU + assert result['step_id'] == 'debug_result' + + result = await hass.config_entries.options.async_configure( + result['flow_id'], user_input={'next_step_id': 'finish'} + ) + + assert result['type'] == FlowResultType.CREATE_ENTRY + assert entry.options[CONF_BYPASS_REMOTE_CONTROL] is True diff --git a/tests/test_coordinator_raw_write.py b/tests/test_coordinator_raw_write.py new file mode 100644 index 0000000..516e05e --- /dev/null +++ b/tests/test_coordinator_raw_write.py @@ -0,0 +1,137 @@ +"""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()