bhetherman/localthings
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into claude/home-assistant-i18n-iwwyvr

Browse Source
This commit is contained in:
Marc Billow
2026-07-24 15:23:10 -05:00
committed by GitHub
parent 10c850f2b9 ee77e3bda8
commit 30d38d855c
14 changed files with 288 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/localthings/coordinator.py
+19 -2
View File
@@ -22,7 +22,7 @@ from smartthings_local.ocf.state_cache import StateCache
from .registry.batch import parse_device0_batch
from .registry.by_type import for_device, for_device_by_model, for_device_by_resources
from .registry.capabilities.common import remote_control_enabled
from .registry.capabilities.common import merge_options_field, remote_control_enabled
from .registry.discovery import discover, BoundEntity
from .registry import CAPABILITIES
from .registry.adapter import flatten
@@ -611,7 +611,24 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# introduced its own races around overlapping writes to the same
# href. Simpler and safer to just hold the guard for the full,
# generously-sized window and let it expire on its own.
self._observe.apply(write_href, body, source='optimistic')
# write_fn bodies that touch x.com.samsung.da.options carry only the
# changed token(s) now (issue #54: confirmed sufficient on the wire --
# the device merges by prefix itself), not the whole packed array.
# observe.apply()'s field-level {**cached, **rep} merge doesn't know
# that -- handed the bare token list, it would replace the cached
# field outright and wipe every sibling option for the rest of the
# settle window. Pre-merge it here the same way the device does, so
# the optimistic cache entry stays complete; the minimal `body` below
# is still exactly what goes out over the wire.
optimistic_body = body
new_options = body.get('x.com.samsung.da.options')
if isinstance(new_options, list):
cached_options = (self._cache.get(write_href) or {}).get('x.com.samsung.da.options')
optimistic_body = {
**body,
'x.com.samsung.da.options': merge_options_field(cached_options, new_options),
}
self._observe.apply(write_href, optimistic_body, source='optimistic')
self._observe.mark_write_pending(
write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S
)
custom_components/localthings/registry/capabilities/air_purifier.py
+13 -8
View File
@@ -8,11 +8,11 @@ dishwasher.DIAGNOSIS -- identical field/write contract
(x.com.samsung.da.diagnosisStart, 'Ready' on both dumps).
/mode/vs/0's x.com.samsung.da.options array packs multiple independent
'<Prefix>_<value>' flags into one list -- the same packed-list/RMW contract
laundry.py's option_value/replace_in_options already model for
/course/vs/0's options[] (reused directly below, just against this family's
own href). Per issue #56's follow-up (five diagnostics dumps captured with
the physical unit set to Auto/Sleep/Low/Medium/High):
'<Prefix>_<value>' flags into one list -- the same packed-list contract
laundry.py's option_value/option_write already model for /course/vs/0's
options[] (reused directly below, just against this family's own href). Per
issue #56's follow-up (five diagnostics dumps captured with the physical unit
set to Auto/Sleep/Low/Medium/High):
Light_On / Light_Off -- a plain on/off flag; MODE below models it as a
real switch, RMW-replacing just that one entry.
Comode_Off -- read 'Off' on *every* one of the five dumps,
@@ -44,7 +44,7 @@ stable capture -- see the issue #56 discussion for what's needed.
from ..capability import Capability
from ..entities import BinarySensorDesc, SensorDesc, SwitchDesc
from .common import int_or_none, sensor_item_value
from .laundry import bool_option_exists, bool_option_value, option_value, replace_in_options
from .laundry import bool_option_exists, bool_option_value, option_value, option_write
_AIR_QUALITY_SENSORS = (
('dust', 'mdi:blur', 'Dust'),
@@ -135,9 +135,14 @@ AIRFLOW_VS_FALLBACK = Capability(
def _light_write(payload, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
# option_write's single-token write is confirmed on a washer's
# /course/vs/0 (issue #54), NOT independently on this family's
# /mode/vs/0 -- extrapolated on the assumption the same vendor field
# merges the same way everywhere. If some unit replaces the field
# outright instead, this would drop Comode/OptionCode alongside it on
# the next light toggle; revisit if a real device report surfaces that.
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, 'Light', payload),
'x.com.samsung.da.options': option_write('Light', payload),
}
custom_components/localthings/registry/capabilities/common.py
+27
View File
@@ -65,6 +65,33 @@ def _active_alarm_codes(items):
return ', '.join(codes) if codes else 'none'
def merge_options_field(cached, new_tokens):
"""Merge freshly-written `<Prefix>_<Value>` tokens into a cached
x.com.samsung.da.options[]-style array the same way the device itself
merges them: match by prefix, replace if present, append if not.
Confirmed on real hardware (issue #54) that a write only needs to carry
the changed token(s), not the whole array -- see laundry.option_write /
oven._option_write for the write side. This is the read side of that
same fact: coordinator.async_send_command uses it to keep the
optimistic cache entry for the written href complete (every sibling
option still present) during the write-settle window, since the wire
body it applies straight to the cache no longer carries them."""
merged = list(cached or [])
for token in new_tokens or ():
if not isinstance(token, str) or '_' not in token:
continue
prefix = token.split('_', 1)[0]
replaced = False
for i, o in enumerate(merged):
if isinstance(o, str) and o.startswith(prefix + '_'):
merged[i] = token
replaced = True
if not replaced:
merged.append(token)
return merged
def sensor_item_value(items, sensor_type, index=0):
"""Pull one reading out of a `/sensors/vs/0`-style items[] list -- each
item is `{type, value: [...]}`; `index` picks which slot of a possibly
custom_components/localthings/registry/capabilities/laundry.py
+15 -12
View File
@@ -147,9 +147,13 @@ BUZZER_SOUND = Capability(
# Cycle selection over /course/vs/0.
#
# The selected course and every other user-tunable option ride in the
# x.com.samsung.da.options array on /course/vs/0 as `<Prefix>_<value>` tokens;
# a write is a read-modify-write of that whole array (cycle_write). The set of
# *selectable* courses is not hardcoded -- it's read live from
# x.com.samsung.da.options array on /course/vs/0 as `<Prefix>_<value>` tokens.
# Confirmed on real hardware (issue #54): a write only needs to carry the one
# changed token -- `{'x.com.samsung.da.options': ['SoftenerLevelCtrl_2']}` --
# the device matches by prefix, evicts the stale token, and merges the result
# into the array itself. No read-modify-write of the whole array needed (see
# option_write). The set of *selectable* courses is not hardcoded -- it's read
# live from
# x.com.samsung.da.editCourseList on /wm/editcourse/vs/0 (cycle_options), so we
# never show a course a given model doesn't have or hide one it does. Course
# codes are uppercase hex; display names live in translations under
@@ -258,17 +262,17 @@ def _course_codes_from_supported_options(course_rep):
return []
def replace_in_options(options, prefix, new_value):
return [f"{prefix}_{new_value}" if isinstance(o, str) and o.startswith(prefix + '_') else o
for o in options]
def option_write(prefix, new_value):
"""A one-token x.com.samsung.da.options write -- see the module comment
above cycle_options for why this doesn't read/rewrite the whole array."""
return [f'{prefix}_{new_value}']
def cycle_write(p, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, 'Course', p),
'x.com.samsung.da.options': option_write('Course', p),
}
@@ -342,11 +346,10 @@ def bool_option_write(prefix):
def write(p, rep, href=None):
if p not in ('On', 'Off'):
return None
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, prefix, p),
'x.com.samsung.da.options': option_write(prefix, p),
}
return write
custom_components/localthings/registry/capabilities/oven.py
+21 -20
View File
@@ -122,10 +122,17 @@ def _option_value(options, prefix):
return None
def _replace_in_options(options, prefix, new_value):
"""Return a new options list with the `<prefix>_*` slot replaced."""
return [f"{prefix}_{new_value}" if o.startswith(prefix + '_') else o
for o in options]
def _option_write(prefix, new_value):
"""A one-token x.com.samsung.da.options write, mirroring
laundry.option_write. NOT independently confirmed on an oven -- issue
#54 only confirmed prefix-merge-on-write for a washer's /course/vs/0.
This extrapolates that same vendor field/contract to the oven's
/mode/vs/0, on the assumption the firmware handles the array the same
way there. If that assumption is wrong for some oven, a device that
replaces the field outright instead of merging would drop every other
option in it (Sound/fastpreheat/etc.) on the next write -- revisit if a
real device report surfaces that."""
return [f'{prefix}_{new_value}']
# ---------------------------------------------------------------------------
@@ -175,47 +182,41 @@ def _oven_mode_write(p, rep, href=None):
def _lamp_write(p, rep, href=None):
if p not in ('On', 'Off'):
return None
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'UpperLamp', p),
'x.com.samsung.da.options': _option_write('UpperLamp', p),
}
def _sound_write(p, rep, href=None):
if p not in ('On', 'Off'):
return None
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'Sound', p),
'x.com.samsung.da.options': _option_write('Sound', p),
}
def _fastpreheat_write(p, rep, href=None):
if p not in ('On', 'Off'):
return None
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'fastpreheat', p),
'x.com.samsung.da.options': _option_write('fastpreheat', p),
}
def _naturalsteam_write(p, rep, href=None):
if p not in ('On', 'Off'):
return None
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
if not any(o.startswith('NaturalSteam_') for o in opts):
opts = opts + [f'NaturalSteam_{p}']
else:
opts = _replace_in_options(opts, 'NaturalSteam', p)
return ['mode', 'vs', '0'], {'x.com.samsung.da.options': opts}
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _option_write('NaturalSteam', p),
}
# ---------------------------------------------------------------------------
custom_components/localthings/registry/capabilities/washer.py
+3 -4
View File
@@ -19,7 +19,7 @@ from ..capability import Capability
from ..entities import BinarySensorDesc, SelectDesc, SensorDesc
from .laundry import (
bool_option_exists, bool_option_switch, cycle_options, cycle_select, hex_pairs, option_value,
replace_in_options,
option_write,
)
# ---------------------------------------------------------------------------
@@ -215,8 +215,7 @@ def _dosing_level(prefix):
def _level_write(prefix):
def write(p, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
# `p` is the zero-padded supported code the UI selected (e.g. '03');
# the device stores the level un-padded (e.g. '3'), matching how it
@@ -226,7 +225,7 @@ def _level_write(prefix):
except (TypeError, ValueError):
native = p
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, prefix, native),
'x.com.samsung.da.options': option_write(prefix, native),
}
return write
custom_components/localthings/translations/en.json
+22 -10
View File
@@ -361,8 +361,16 @@
"washer_cycle_table_02": {
"name": "Cycle",
"state": {
"01": "Normal",
"04": "Quick Wash",
"17": "Downloaded",
"1b": "Cotton",
"1c": "Eco 40-60",
"1d": "Super Speed",
"1e": "15' Quick Wash",
"1f": "Intense Cold",
"20": "Hygiene Steam",
"21": "Colours",
"21": "Colors",
"22": "Wool",
"23": "Outdoor",
"24": "Towels",
@@ -371,6 +379,9 @@
"27": "Rinse+Spin",
"28": "Drain/Spin",
"29": "Drum Clean+",
"2d": "Silent Wash",
"2e": "Baby Care",
"2f": "Activewear",
"30": "Cloudy Day",
"32": "Shirts",
"33": "Bedding",
@@ -378,17 +389,18 @@
"37": "Air Wash",
"38": "Cotton Dry",
"39": "Synthetics Dry",
"53": "Heavy Duty",
"55": "Activewear",
"57": "Delicate",
"5e": "Rinse+Spin",
"65": "Colors",
"66": "Denim",
"96": "Less Microfiber",
"1c": "Eco 40-60",
"1d": "Super Speed",
"1b": "Cotton",
"1e": "15' Quick Wash",
"2f": "Activewear",
"2e": "Baby Care",
"2d": "Silent Wash",
"7c": "Whites",
"7d": "Bedding/Waterproof",
"7e": "Self-Clean",
"86": "Deep Wash",
"8f": "Intense Cold",
"1f": "Intense Cold"
"96": "Less Microfiber"
}
},
"washer_dry_level": {
tests/test_air_purifier_capabilities.py
+5 -6
View File
@@ -73,9 +73,10 @@ def test_diagnosis_reuses_dishwasher_capability():
def test_light_switch_write_contract():
"""The display-light switch RMW-replaces only the 'Light_*' entry in the
packed /mode/vs/0 options list (via laundry.replace_in_options), leaving
the other flags and the list order untouched."""
"""The display-light switch writes only the changed 'Light_*' token
(via laundry.option_write) -- confirmed on real hardware (issue #54)
that the device merges by prefix itself, so no read-modify-write of the
whole packed /mode/vs/0 options list is needed."""
desc = next(e for e in air_purifier.MODE.entities if e.key == 'display_light')
rep = {'x.com.samsung.da.options': [
'Comode_Off', 'Blooming_0', 'Light_On', 'OptionCode_60282',
@@ -83,9 +84,7 @@ def test_light_switch_write_contract():
assert desc.rep_fn(rep) is True
assert desc.write_fn('Off', rep) == (
['mode', 'vs', '0'],
{'x.com.samsung.da.options': [
'Comode_Off', 'Blooming_0', 'Light_Off', 'OptionCode_60282',
]},
{'x.com.samsung.da.options': ['Light_Off']},
)
tests/test_common_capabilities.py
+31
View File
@@ -26,6 +26,37 @@ def test_common_caps_discover_on_dishwasher(dishwasher_resources):
assert 'power_switch' in keys
class TestMergeOptionsField:
"""merge_options_field() is the read side of issue #54's finding: a
write only needs to carry the changed token, so the coordinator uses
this to keep its optimistic cache entry complete (every sibling option
still present) without waiting on a real poll."""
def test_replaces_matching_prefix(self):
cached = ['DeviceType_0167', 'Course_16', 'GMT_04']
assert common.merge_options_field(cached, ['Course_1D']) == [
'DeviceType_0167', 'Course_1D', 'GMT_04',
]
def test_appends_when_prefix_absent(self):
cached = ['DeviceType_0167']
assert common.merge_options_field(cached, ['NaturalSteam_On']) == [
'DeviceType_0167', 'NaturalSteam_On',
]
def test_merges_multiple_tokens_independently(self):
cached = ['DetergentLevelCtrl_1', 'SoftenerLevelCtrl_0', 'GMT_04']
merged = common.merge_options_field(cached, ['SoftenerLevelCtrl_2'])
assert merged == ['DetergentLevelCtrl_1', 'SoftenerLevelCtrl_2', 'GMT_04']
def test_handles_missing_cache(self):
assert common.merge_options_field(None, ['Course_1D']) == ['Course_1D']
def test_ignores_malformed_tokens(self):
cached = ['Course_16']
assert common.merge_options_field(cached, ['nounderscore']) == ['Course_16']
# ---------------------------------------------------------------------------
# OCF-native / vendor '-vs' fallback pairs (power, kids-lock, remote control).
# ---------------------------------------------------------------------------
tests/test_coordinator_send_command.py
+94
View File
@@ -0,0 +1,94 @@
"""Tests for LocalThingsCoordinator.async_send_command's optimistic-apply
step, specifically for x.com.samsung.da.options[] writes (issue #54).
The write itself only needs to carry the single changed token -- confirmed
on real hardware, the device merges by prefix and evicts the stale token
itself. But observe.ObserveManager.apply() does a shallow {**cached, **rep}
field merge, so handing it that same minimal single-token body would
overwrite the *whole* cached options[] field, wiping every sibling option
(other courses/levels/toggles packed into the same array) until the next
real poll lands. async_send_command must pre-merge the token into the
cached array itself before applying it optimistically, while still POSTing
only the minimal body over the wire.
"""
from __future__ import annotations
from unittest.mock import AsyncMock
import cbor2
import pytest
from homeassistant.core import HomeAssistant
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
from custom_components.localthings.registry.capabilities import laundry
from custom_components.localthings.registry.discovery import BoundEntity
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 _FakeSendSession:
def __init__(self):
self.post_calls: list[tuple[list[str], bytes]] = []
def post(self, path_segs, payload, timeout=None):
self.post_calls.append((list(path_segs), payload))
return 0x44, b''
def pace(self):
pass
@pytest.fixture
def coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
entry = MockConfigEntry(
domain=DOMAIN, data=ENTRY_DATA, unique_id='localthings_SENDCMD-TEST',
)
entry.add_to_hass(hass)
coord = LocalThingsCoordinator(hass, entry)
coord.async_request_refresh = AsyncMock()
coord._session = _FakeSendSession()
return coord
async def test_options_write_posts_only_the_changed_token(coordinator) -> None:
href = '/course/vs/0'
coordinator._observe.apply(href, {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04'],
}, source='poll')
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
bound = BoundEntity(href=href, capability=None, desc=desc)
await coordinator.async_send_command(bound, '1D')
posted_path, posted_bytes = coordinator._session.post_calls[0]
assert posted_path == ['course', 'vs', '0']
assert cbor2.loads(posted_bytes) == {'x.com.samsung.da.options': ['Course_1D']}
async def test_options_write_optimistic_cache_keeps_sibling_tokens(coordinator) -> None:
"""The regression this guards: applying the minimal wire body straight
to the cache would wipe DeviceType_0167/GMT_04 until the next poll."""
href = '/course/vs/0'
coordinator._observe.apply(href, {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04'],
}, source='poll')
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
bound = BoundEntity(href=href, capability=None, desc=desc)
await coordinator.async_send_command(bound, '1D')
cached = coordinator._cache.get(href)
assert cached['x.com.samsung.da.options'] == [
'DeviceType_0167', 'Course_1D', 'GMT_04',
]
tests/test_dishwasher_capabilities.py
+1 -3
View File
@@ -28,9 +28,7 @@ class TestCycleOptions:
rep = {'x.com.samsung.da.options': ['DeviceType_0001', 'Course_0E', 'GMT_04']}
path, body = desc.write_fn('90', rep)
assert path == ['course', 'vs', '0']
assert body == {
'x.com.samsung.da.options': ['DeviceType_0001', 'Course_90', 'GMT_04'],
}
assert body == {'x.com.samsung.da.options': ['Course_90']}
class TestDishwasherOptions:
tests/test_laundry_capabilities.py
+5 -4
View File
@@ -153,14 +153,15 @@ class TestCycleSelect:
live = {'/wm/editcourse/vs/0': {'x.com.samsung.da.editCourseList': 'EditCourseList_16'}}
assert desc.exists_fn({}, live) is True
def test_cycle_write_rmw_on_options(self):
def test_cycle_write_is_single_token(self):
"""Confirmed on real hardware (issue #54): the device merges a
single-token options[] write by prefix itself, so the write only
needs to carry the changed token, not the whole rewritten array."""
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
rep = {'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04']}
path, body = desc.write_fn('1D', rep)
assert path == ['course', 'vs', '0']
assert body == {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1D', 'GMT_04'],
}
assert body == {'x.com.samsung.da.options': ['Course_1D']}
def test_cycle_write_noop_without_options(self):
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
tests/test_oven_capabilities.py
+14 -13
View File
@@ -137,7 +137,10 @@ def test_oven_mode_rejects_unknown():
# ---------------------------------------------------------------------------
# OVEN_MODE options-array RMW (lamp, sound, fast_preheat, natural_steam)
# OVEN_MODE options-array writes (lamp, sound, fast_preheat, natural_steam).
# Confirmed on real hardware (issue #54): a write only needs to carry the
# changed token -- the device matches by prefix, evicts the stale token, and
# merges the result into the array itself. No read-modify-write needed.
# ---------------------------------------------------------------------------
def _mode_rep(*extra_opts):
@@ -146,12 +149,11 @@ def _mode_rep(*extra_opts):
]}
def test_lamp_write_replaces_slot():
def test_lamp_write_is_single_token():
desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'lamp')
path, body = desc.write_fn('On', _mode_rep())
opts = body['x.com.samsung.da.options']
assert 'UpperLamp_On' in opts
assert 'UpperLamp_Off' not in opts
assert path == ['mode', 'vs', '0']
assert body == {'x.com.samsung.da.options': ['UpperLamp_On']}
def test_lamp_write_requires_existing_options():
@@ -159,20 +161,19 @@ def test_lamp_write_requires_existing_options():
assert desc.write_fn('On', {}) is None
def test_sound_write_preserves_other_options():
def test_sound_write_is_single_token():
desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'sound')
path, body = desc.write_fn('Off', _mode_rep())
opts = body['x.com.samsung.da.options']
assert 'Sound_Off' in opts
assert 'UpperLamp_Off' in opts # other slot unchanged
assert body == {'x.com.samsung.da.options': ['Sound_Off']}
def test_natural_steam_appended_if_absent():
"""NaturalSteam slot is absent until first write — write_fn must append it."""
def test_natural_steam_write_is_single_token():
"""NaturalSteam's slot may be absent from the live rep until first
write -- the single-token write covers both the insert and replace case
identically, since the device merges by prefix either way."""
desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'natural_steam')
path, body = desc.write_fn('On', _mode_rep()) # no NaturalSteam_* in rep
opts = body['x.com.samsung.da.options']
assert any(o.startswith('NaturalSteam_') for o in opts)
assert body == {'x.com.samsung.da.options': ['NaturalSteam_On']}
# ---------------------------------------------------------------------------
tests/test_washer_capabilities.py
+18 -13
View File
@@ -101,13 +101,14 @@ class TestWasherCourse:
assert desc.exists_fn({}, live) is True
def test_cycle_write(self):
"""Confirmed on real hardware (issue #54): the write only needs to
carry the changed token -- the device matches by prefix, evicts the
stale token, and merges the result into the array itself."""
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
rep = {'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1C', 'GMT_04']}
path, body = desc.write_fn('1D', rep)
assert path == ['course', 'vs', '0']
assert body == {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1D', 'GMT_04'],
}
assert body == {'x.com.samsung.da.options': ['Course_1D']}
class TestDrumClean:
@@ -226,20 +227,22 @@ class TestDetergentSoftenerDosing:
def test_quantity_write(self):
"""The UI selects a padded supported code ('01'); the write posts the
un-padded device code ('1'), mirroring how the device reports it."""
un-padded device code ('1'), mirroring how the device reports it.
Confirmed on real hardware (issue #54): the write only needs to carry
the changed token -- the device matches by prefix, evicts the stale
token, and merges the result into the array itself. No need to read
the current array back and rewrite it whole."""
rep = {'x.com.samsung.da.options': list(_DOSING_OPTIONS)}
path, body = self._desc('detergent_quantity').write_fn('01', rep)
assert path == ['course', 'vs', '0']
assert 'DetergentLevelCtrl_1' in body['x.com.samsung.da.options']
assert 'DetergentLevelCtrl_3' not in body['x.com.samsung.da.options']
# untouched siblings survive the read-modify-write
assert 'SoftenerLevelCtrl_3' in body['x.com.samsung.da.options']
assert body == {'x.com.samsung.da.options': ['DetergentLevelCtrl_1']}
def test_hardness_write(self):
rep = {'x.com.samsung.da.options': list(_DOSING_OPTIONS)}
path, body = self._desc('softener_concentration').write_fn('03', rep)
assert path == ['course', 'vs', '0']
assert 'SoftenerLevel2Ctrl_3' in body['x.com.samsung.da.options']
assert body == {'x.com.samsung.da.options': ['SoftenerLevel2Ctrl_3']}
def test_low_reservoir_off_when_alarm_off(self):
rep = {'x.com.samsung.da.options': _DOSING_OPTIONS}
@@ -301,18 +304,20 @@ class TestWashOptionToggles:
"""write_fn receives the same 'On'/'Off' string switch.py sends
(not a bool) -- covers a bug where an earlier `'On' if p else 'Off'`
implementation always wrote 'On', since any non-empty string
(including 'Off') is truthy."""
(including 'Off') is truthy.
The write body carries only the changed token (issue #54: confirmed
the device merges by prefix itself), not the whole options array."""
for key in self._keys():
prefix = self._prefix(key)
rep = {'x.com.samsung.da.options': [f'{prefix}_Off', 'GMT_02']}
path, body = self._desc(key).write_fn('On', rep)
assert path == ['course', 'vs', '0']
assert f'{prefix}_On' in body['x.com.samsung.da.options']
assert 'GMT_02' in body['x.com.samsung.da.options']
assert body == {'x.com.samsung.da.options': [f'{prefix}_On']}
rep = {'x.com.samsung.da.options': [f'{prefix}_On']}
path, body = self._desc(key).write_fn('Off', rep)
assert f'{prefix}_Off' in body['x.com.samsung.da.options']
assert body == {'x.com.samsung.da.options': [f'{prefix}_Off']}
assert f'{prefix}_On' not in body['x.com.samsung.da.options']
def test_write_rejects_non_on_off_payload(self):