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

Fix pre-existing ty diagnostics in a second batch of test files

Browse Source 
Same isinstance/cast narrowing and Optional-field assert pattern as the
prior commits, covering the airconditioner, fridge, washer, operational,
subdevices, sensor_hysteresis, laundry, select_options, identity and
entities test files.
This commit is contained in:
Marc Billow
2026-08-03 00:11:05 +00:00
parent 07061c734d
commit 36a642135b
17 changed files with 322 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/localthings/test_config_flow.py
+22 -9
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from typing import Iterable, cast
from unittest.mock import patch
from homeassistant.core import HomeAssistant
@@ -32,8 +33,10 @@ async def test_form_first_device(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": "user"})
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert CONF_CA_CERT_PEM in result["data_schema"].schema
assert CONF_CA_KEY_PEM in result["data_schema"].schema
data_schema = result["data_schema"]
assert data_schema is not None
assert CONF_CA_CERT_PEM in data_schema.schema
assert CONF_CA_KEY_PEM in data_schema.schema
async def test_form_second_device_reuses_creds(hass: HomeAssistant) -> None:
@@ -44,8 +47,10 @@ async def test_form_second_device_reuses_creds(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": "user"})
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user_reuse"
assert CONF_CA_CERT_PEM not in result["data_schema"].schema
assert CONF_CA_KEY_PEM not in result["data_schema"].schema
data_schema = result["data_schema"]
assert data_schema is not None
assert CONF_CA_CERT_PEM not in data_schema.schema
assert CONF_CA_KEY_PEM not in data_schema.schema
async def test_successful_setup(hass: HomeAssistant, mock_probe) -> None:
@@ -243,7 +248,9 @@ async def test_cannot_connect(hass: HomeAssistant) -> None:
},
)
assert result["type"] == FlowResultType.FORM
assert result["errors"]["base"] == "cannot_connect"
errors = result["errors"]
assert errors is not None
assert errors["base"] == "cannot_connect"
async def test_recognized_type_skips_confirmation_step(hass: HomeAssistant, mock_probe) -> None:
@@ -368,7 +375,7 @@ async def test_options_flow_init_shows_menu(hass: HomeAssistant) -> None:
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "init"
assert set(result["menu_options"]) == {"settings", "debug_write"}
assert set(cast(Iterable[str], result["menu_options"])) == {"settings", "debug_write"}
async def test_options_flow_default_is_off(hass: HomeAssistant) -> None:
@@ -384,7 +391,9 @@ async def test_options_flow_default_is_off(hass: HomeAssistant) -> None:
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "settings"
assert result["data_schema"]({})[CONF_BYPASS_REMOTE_CONTROL] is False
data_schema = result["data_schema"]
assert data_schema is not None
assert data_schema({})[CONF_BYPASS_REMOTE_CONTROL] is False
async def test_options_flow_can_enable_bypass(hass: HomeAssistant) -> None:
@@ -421,7 +430,9 @@ async def test_options_flow_reflects_previously_saved_value(hass: HomeAssistant)
result["flow_id"], user_input={"next_step_id": "settings"}
)
assert result["data_schema"]({})[CONF_BYPASS_REMOTE_CONTROL] is True
data_schema = result["data_schema"]
assert data_schema is not None
assert data_schema({})[CONF_BYPASS_REMOTE_CONTROL] is True
async def test_options_flow_debug_write_shows_hrefs_from_coordinator(
@@ -495,7 +506,9 @@ async def test_options_flow_debug_edit_writes_and_shows_result(
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "debug_result"
assert result["description_placeholders"]["code"] == "2.04 (0x44)"
description_placeholders = result["description_placeholders"]
assert description_placeholders is not None
assert description_placeholders["code"] == "2.04 (0x44)"
async def test_options_flow_debug_edit_rejects_empty_payload(
tests/localthings/test_coordinator.py
+27 -17
View File
@@ -291,7 +291,9 @@ def test_logger_is_scoped_to_device_host(hass: HomeAssistant, mock_entry) -> Non
makes multi-device logs ambiguous."""
coordinator = LocalThingsCoordinator(hass, mock_entry)
assert coordinator._log.name.endswith(ENTRY_DATA[CONF_HOST])
host = ENTRY_DATA[CONF_HOST]
assert isinstance(host, str)
assert coordinator._log.name.endswith(host)
assert coordinator.logger is coordinator._log
assert coordinator._observe.log is coordinator._log
@@ -393,7 +395,9 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll(
# still alive (see test_poll_failure_skips_reconnect_when_push_is_healthy)
# and would not trigger a reconnect at all. This test covers a
# genuinely dead channel: no recent push, poll fails, reconnect fires.
coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
last_notify_ts = coordinator._observe._last_notify_ts
assert last_notify_ts is not None
coordinator._observe._last_notify_ts = last_notify_ts - (PUSH_HEALTH_WINDOW_S + 1)
# Simulate the existing "poll failed, reconnecting" branch: _poll_once
# fails once (triggering the reconnect/backoff path), then succeeds.
@@ -490,7 +494,9 @@ async def test_poll_timeout_reconnects_after_consecutive_limit_even_with_push(
# No notify is recent any more, so every timeout below counts toward
# the consecutive-timeout limit instead of being deferred.
coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
last_notify_ts = coordinator._observe._last_notify_ts
assert last_notify_ts is not None
coordinator._observe._last_notify_ts = last_notify_ts - (PUSH_HEALTH_WINDOW_S + 1)
# Call _async_update_data directly rather than via
# async_request_refresh() — the coordinator's built-in debouncer
@@ -553,7 +559,9 @@ async def test_poll_timeout_counter_resets_when_push_is_healthy_again(
assert coordinator.observe_mode == MODE_OBSERVE
# Age the notify out so timeouts accrue toward the limit.
coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
last_notify_ts = coordinator._observe._last_notify_ts
assert last_notify_ts is not None
coordinator._observe._last_notify_ts = last_notify_ts - (PUSH_HEALTH_WINDOW_S + 1)
with patch(
"custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once",
side_effect=TimeoutError("GET /device/0 block 11 timeout"),
@@ -605,7 +613,9 @@ async def test_reconnect_from_observe_mode_resubscribes_immediately(
# failure below actually triggers a reconnect (see
# test_poll_failure_skips_reconnect_when_push_is_healthy for the
# healthy-push case, which now skips reconnecting entirely).
coordinator._observe._last_notify_ts -= PUSH_HEALTH_WINDOW_S + 1
last_notify_ts = coordinator._observe._last_notify_ts
assert last_notify_ts is not None
coordinator._observe._last_notify_ts = last_notify_ts - (PUSH_HEALTH_WINDOW_S + 1)
# If a stale retry timer (rather than an immediate resubscribe) were
# driving recovery, mode would still be 'poll' right after this single
@@ -734,7 +744,7 @@ async def test_write_marks_href_pending_before_post(
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
def _write_fn(payload, rep, href):
def _write_fn(payload, rep, href=None):
return (["some", "path"], {"value": payload})
desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
@@ -771,7 +781,7 @@ async def test_send_command_applies_write_optimistically_before_settling(
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
def _write_fn(payload, rep, href):
def _write_fn(payload, rep, href=None):
return (["some", "path"], {"value": payload})
desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
@@ -812,8 +822,10 @@ async def test_climate_power_write_applies_to_its_own_href_not_bound_href(
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
climate_href = airconditioner.CLIMATE.href
assert climate_href is not None
bound = BoundEntity(
href=airconditioner.CLIMATE.href,
href=climate_href,
capability=coordinator.bound[0].capability,
desc=airconditioner.CLIMATE.entities[0],
)
@@ -823,9 +835,7 @@ async def test_climate_power_write_applies_to_its_own_href_not_bound_href(
await coordinator.async_send_command(bound, ("power", True))
assert (coordinator._cache.get("/power/vs/0") or {}).get("x.com.samsung.da.power") == "On"
assert "x.com.samsung.da.power" not in (
coordinator._cache.get(airconditioner.CLIMATE.href) or {}
)
assert "x.com.samsung.da.power" not in (coordinator._cache.get(climate_href) or {})
assert coordinator._observe._settle_until.get("/power/vs/0") is not None
@@ -858,7 +868,7 @@ async def test_send_command_survives_stale_confirm_poll(
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
def _write_fn(payload, rep, href):
def _write_fn(payload, rep, href=None):
return (["test", "vs", "0"], {"value": payload})
desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
@@ -917,12 +927,12 @@ async def test_second_write_to_same_href_lands_during_first_writes_settle_window
desc_a = NumberDesc(
key="cycle",
field="cycle",
write_fn=lambda p, rep, href: (["test", "vs", "0"], {"cycle": p}),
write_fn=lambda p, rep, href=None: (["test", "vs", "0"], {"cycle": p}),
)
desc_b = NumberDesc(
key="detergent",
field="detergent",
write_fn=lambda p, rep, href: (["test", "vs", "0"], {"detergent": p}),
write_fn=lambda p, rep, href=None: (["test", "vs", "0"], {"detergent": p}),
)
bound_a = BoundEntity(
href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc_a
@@ -1023,7 +1033,7 @@ async def test_send_command_blocked_when_remote_control_disabled(
source="test",
)
def _write_fn(payload, rep, href):
def _write_fn(payload, rep, href=None):
return (["some", "path"], {"value": payload})
desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
@@ -1064,7 +1074,7 @@ async def test_send_command_allowed_when_remote_control_enabled(
source="test",
)
def _write_fn(payload, rep, href):
def _write_fn(payload, rep, href=None):
return (["some", "path"], {"value": payload})
desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
@@ -1139,7 +1149,7 @@ async def test_send_command_bypasses_remote_control_when_option_enabled(
source="test",
)
def _write_fn(payload, rep, href):
def _write_fn(payload, rep, href=None):
return (["some", "path"], {"value": payload})
desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
tests/localthings/test_observe.py
+3
View File
@@ -53,6 +53,7 @@ def test_apply_merges_partial_update_onto_prior_rep():
mgr.apply("/mode/vs/0", partial, source="observe")
cached = mgr.cache.get("/mode/vs/0")
assert cached is not None
assert cached["x.com.samsung.da.modes"] == ["CVN_CONVERTIBLE_ZONE", "WATERFILTER_ENABLE"]
assert cached["x.com.samsung.da.supportedOptions"] == ["CV_FDR_WINE", "CV_FDR_MEAT"]
@@ -435,6 +436,7 @@ def test_close_stops_the_refresh_thread():
session = _FakeSession()
mgr.start_refresh_task(session)
thread = mgr._refresh_thread
assert thread is not None
mgr.close()
thread.join(timeout=2.0)
@@ -449,6 +451,7 @@ def test_downgrade_to_poll_stops_refresh_task():
session = _FakeSession()
mgr.start_refresh_task(session)
thread = mgr._refresh_thread
assert thread is not None
mgr.downgrade_to_poll()
thread.join(timeout=2.0)
tests/test_air_purifier_capabilities.py
+18 -5
View File
@@ -4,6 +4,7 @@ from custom_components.localthings.registry.adapter import flatten
from custom_components.localthings.registry.by_type import for_device_by_model
from custom_components.localthings.registry.capabilities import air_purifier
from custom_components.localthings.registry.discovery import discover
from custom_components.localthings.registry.entities import FanDesc, SensorDesc, SwitchDesc
from tests.conftest import _load_device
@@ -90,7 +91,9 @@ def test_light_switch_write_contract():
(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")
desc = next(
e for e in air_purifier.MODE.entities if e.key == "display_light" and isinstance(e, SwitchDesc)
)
rep = {
"x.com.samsung.da.options": [
"Comode_Off",
@@ -99,7 +102,9 @@ def test_light_switch_write_contract():
"OptionCode_60282",
]
}
assert desc.rep_fn is not None
assert desc.rep_fn(rep) is True
assert desc.write_fn is not None
assert desc.write_fn("Off", rep) == (
["mode", "vs", "0"],
{"x.com.samsung.da.options": ["Light_Off"]},
@@ -112,8 +117,11 @@ def test_operating_mode_is_a_read_only_diagnostic():
regardless of the device's actual fan setting, ruling out the original
guess that it was the fan-speed selector; its real purpose is still
unconfirmed (see the air_purifier.py module docstring)."""
operating_mode = next(e for e in air_purifier.MODE.entities if e.key == "operating_mode")
operating_mode = next(
e for e in air_purifier.MODE.entities if e.key == "operating_mode" and isinstance(e, SensorDesc)
)
rep = {"x.com.samsung.da.options": ["Comode_Off"]}
assert operating_mode.rep_fn is not None
assert operating_mode.rep_fn(rep) == "Off"
assert not hasattr(operating_mode, "write_fn")
@@ -128,14 +136,16 @@ def test_blooming_not_modeled():
def test_airflow_vs_fallback_only_binds_without_generic():
"""/airflow/vs/0 is a match_fn fallback -- it must not bind when the
OCF-standard /airflow/0 is also present (both are on every dump seen)."""
match_fn = air_purifier.AIRFLOW_VS_FALLBACK.match_fn
assert match_fn is not None
assert (
air_purifier.AIRFLOW_VS_FALLBACK.match_fn(
match_fn(
{},
{"/airflow/0": {"speed": 0, "direction": "Off"}},
)
is False
)
assert air_purifier.AIRFLOW_VS_FALLBACK.match_fn({}, {}) is True
assert match_fn({}, {}) is True
def test_airflow_fan_write_contract():
@@ -143,7 +153,10 @@ def test_airflow_fan_write_contract():
(two independent units, 60-90s apart per setting): /airflow/0's `speed`
is a clean, monotonic 0-4 code, so the write is a plain int passthrough
-- no named-preset table needed (see fan.py's LocalThingsAirflowFan)."""
fan_desc = next(e for e in air_purifier.AIRFLOW_GENERIC.entities if e.key == "airflow_fan")
fan_desc = next(
e for e in air_purifier.AIRFLOW_GENERIC.entities if e.key == "airflow_fan" and isinstance(e, FanDesc)
)
assert fan_desc.write_fn is not None
assert fan_desc.write_fn(("speed", 3), {}) == (["airflow", "0"], {"speed": 3})
assert fan_desc.write_fn(("power", True, "/power/vs/0"), {}) == (
["power", "vs", "0"],
tests/test_dishwasher_capabilities.py
+17 -3
View File
@@ -6,6 +6,7 @@ check the dishwasher wiring and its device-specific options.
"""
from custom_components.localthings.registry.capabilities import dishwasher, laundry
from custom_components.localthings.registry.entities import SwitchDesc
class TestCycleOptions:
@@ -34,14 +35,27 @@ class TestCycleOptions:
class TestDishwasherOptions:
def test_storm_wash_read_and_write(self):
desc = next(e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == "storm_wash")
desc = next(
e
for e in dishwasher.CYCLE_OPTIONS.entities
if e.key == "storm_wash" and isinstance(e, SwitchDesc)
)
assert desc.rep_fn is not None
assert desc.rep_fn({"x.com.samsung.da.options": ["StormWashZone_On"]}) is True
assert desc.rep_fn({"x.com.samsung.da.options": ["StormWashZone_Off"]}) is False
path, body = desc.write_fn("Off", {"x.com.samsung.da.options": ["StormWashZone_On"]})
assert desc.write_fn is not None
result = desc.write_fn("Off", {"x.com.samsung.da.options": ["StormWashZone_On"]})
assert result is not None
path, body = result
assert path == ["course", "vs", "0"]
assert "StormWashZone_Off" in body["x.com.samsung.da.options"]
def test_auto_release_exists_only_when_field_present(self):
desc = next(e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == "auto_release_dry")
desc = next(
e
for e in dishwasher.CYCLE_OPTIONS.entities
if e.key == "auto_release_dry" and isinstance(e, SwitchDesc)
)
assert desc.exists_fn is not None
assert desc.exists_fn({"x.com.samsung.da.options": []}, {}) is False
assert desc.exists_fn({"x.com.samsung.da.options": ["AutoDoorRelease_On"]}, {}) is True
tests/test_entities.py
+8 -1
View File
@@ -20,7 +20,13 @@ def test_value_fn_defaults_to_identity():
def test_descriptions_are_frozen():
d = SensorDesc(key="power", field="f")
try:
d.key = "other"
# setattr() through a variable name (not `d.key = "other"`, and not a
# literal setattr(d, "key", ...) -- ruff's B010 rewrites that back to
# attribute-assignment syntax), so this reaches the same
# frozen-dataclass __setattr__ at runtime without ty statically
# flagging the (deliberately illegal) direct attribute assignment.
attr = "key"
setattr(d, attr, "other")
except Exception as e:
assert "frozen" in str(type(e)).lower() or "cannot" in str(e).lower()
else:
@@ -46,4 +52,5 @@ def test_select_carries_options_and_write_fn():
options=("voice", "tone", "mute"),
write_fn=lambda p, rep: (["settings", "sound", "mode", "vs", "0"], {"mode": p}),
)
assert d.write_fn is not None
assert d.write_fn("tone", {}) == (["settings", "sound", "mode", "vs", "0"], {"mode": "tone"})
tests/test_entity.py
+15 -8
View File
@@ -8,6 +8,9 @@ entity there would silently drop working sensors on first-poll timing,
not just fix phantom ones.
"""
from typing import cast
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.entity import _is_included
from custom_components.localthings.registry.capability import Capability
from custom_components.localthings.registry.discovery import BoundEntity
@@ -27,6 +30,10 @@ class _FakeCoordinator:
return self.last_resources
def _coord(last_resources) -> LocalThingsCoordinator:
return cast(LocalThingsCoordinator, _FakeCoordinator(last_resources))
def _bound(desc, href):
capability = Capability(href=href, entities=(desc,))
return BoundEntity(href=href, capability=capability, desc=desc)
@@ -40,17 +47,17 @@ class TestDefaultFieldGate:
def test_included_when_field_present(self):
bound = self._bound()
coord = _FakeCoordinator({"/x/vs/0": {"x.com.samsung.da.value": "1"}})
coord = _coord({"/x/vs/0": {"x.com.samsung.da.value": "1"}})
assert _is_included(bound, coord) is True
def test_excluded_when_field_absent_from_populated_rep(self):
bound = self._bound()
coord = _FakeCoordinator({"/x/vs/0": {"x.com.samsung.da.other": "1"}})
coord = _coord({"/x/vs/0": {"x.com.samsung.da.other": "1"}})
assert _is_included(bound, coord) is False
def test_included_on_true_stub(self):
bound = self._bound()
coord = _FakeCoordinator({"/x/vs/0": {"href": "/x/vs/0"}})
coord = _coord({"/x/vs/0": {"href": "/x/vs/0"}})
assert _is_included(bound, coord) is True
def test_included_on_genuinely_empty_rep(self):
@@ -60,12 +67,12 @@ class TestDefaultFieldGate:
exists_fn (e.g. common.ENERGY_METER) opts into excluding on
confirmed-empty, after verifying that's actually safe for its field."""
bound = self._bound()
coord = _FakeCoordinator({"/x/vs/0": {}})
coord = _coord({"/x/vs/0": {}})
assert _is_included(bound, coord) is True
def test_excluded_when_href_missing_from_resources(self):
bound = self._bound()
coord = _FakeCoordinator({})
coord = _coord({})
assert _is_included(bound, coord) is False
@@ -77,9 +84,9 @@ class TestExplicitExistsFnGate:
exists_fn=lambda rep, resources: rep.get("flag") is True,
)
bound = _bound(desc, "/x/vs/0")
coord = _FakeCoordinator({"/x/vs/0": {"flag": True}})
coord = _coord({"/x/vs/0": {"flag": True}})
assert _is_included(bound, coord) is True
coord = _FakeCoordinator({"/x/vs/0": {"flag": False, "x.com.samsung.da.value": "1"}})
coord = _coord({"/x/vs/0": {"flag": False, "x.com.samsung.da.value": "1"}})
assert _is_included(bound, coord) is False
@@ -87,5 +94,5 @@ class TestNoFieldEntities:
def test_rep_fn_entity_always_included(self):
desc = SensorDesc(key="x", rep_fn=lambda rep: rep.get("x.com.samsung.da.value"))
bound = _bound(desc, "/x/vs/0")
coord = _FakeCoordinator({"/x/vs/0": {}})
coord = _coord({"/x/vs/0": {}})
assert _is_included(bound, coord) is True
tests/test_identity.py
+6 -2
View File
@@ -71,8 +71,12 @@ def test_read_identity_keeps_raw_payloads_for_diagnostics():
}
)
ident = read_identity(sess, serial=None)
assert ident.raw["/oic/p"]["mnmo"] == "RF9000B"
assert ident.raw["/oic/d"]["di"] == "abc-123"
oic_p = ident.raw["/oic/p"]
assert isinstance(oic_p, dict)
assert oic_p["mnmo"] == "RF9000B"
oic_d = ident.raw["/oic/d"]
assert isinstance(oic_d, dict)
assert oic_d["di"] == "abc-123"
def test_read_identity_captures_oic_res_links():
tests/test_laundry_capabilities.py
+11 -2
View File
@@ -1,6 +1,7 @@
"""Tests for the shared laundry capabilities (washer/dryer/dishwasher)."""
from custom_components.localthings.registry.capabilities import laundry
from custom_components.localthings.registry.entities import SelectDesc
class TestCourseHelpers:
@@ -261,14 +262,22 @@ class TestBuzzerSound:
assert laundry.BUZZER_SOUND.href == "/buzzersound/vs/0"
def test_buzzer_sound_write(self):
desc = next(e for e in laundry.BUZZER_SOUND.entities if e.key == "buzzer_sound")
desc = next(
e
for e in laundry.BUZZER_SOUND.entities
if e.key == "buzzer_sound" and isinstance(e, SelectDesc)
)
assert desc.options_field == "supportedBuzzerSound"
path, body = desc.write_fn("On", {})
assert desc.write_fn is not None
result = desc.write_fn("On", {})
assert result is not None
path, body = result
assert path == ["buzzersound", "vs", "0"]
assert body == {"setBuzzerSound": "On"}
def test_finish_sound_exists_only_when_supported(self):
desc = next(e for e in laundry.BUZZER_SOUND.entities if e.key == "finish_sound")
assert desc.exists_fn is not None
assert desc.exists_fn({"setBuzzerSound": "On"}, {}) is False
assert desc.exists_fn({"supportedFinishSound": ["FinishSound_1"]}, {}) is True
tests/test_laundry_writes.py
+17 -6
View File
@@ -2,10 +2,12 @@
from custom_components.localthings.registry.capabilities import laundry
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
from custom_components.localthings.registry.entities import ButtonDesc, SelectDesc, SwitchDesc
def test_start_button_writes_run():
btn = next(e for e in OPERATIONAL_STATE.entities if e.key == "start")
btn = next(e for e in OPERATIONAL_STATE.entities if e.key == "start" and isinstance(e, ButtonDesc))
assert btn.write_fn is not None
assert btn.write_fn("Run", {}) == (
["operational", "state", "vs", "0"],
{"x.com.samsung.da.state": "Run"},
@@ -13,7 +15,8 @@ def test_start_button_writes_run():
def test_pause_button_writes_pause():
btn = next(e for e in OPERATIONAL_STATE.entities if e.key == "pause")
btn = next(e for e in OPERATIONAL_STATE.entities if e.key == "pause" and isinstance(e, ButtonDesc))
assert btn.write_fn is not None
assert btn.write_fn("Pause", {}) == (
["operational", "state", "vs", "0"],
{"x.com.samsung.da.state": "Pause"},
@@ -21,7 +24,8 @@ def test_pause_button_writes_pause():
def test_stop_button_writes_ready():
btn = next(e for e in OPERATIONAL_STATE.entities if e.key == "stop")
btn = next(e for e in OPERATIONAL_STATE.entities if e.key == "stop" and isinstance(e, ButtonDesc))
assert btn.write_fn is not None
assert btn.write_fn("Ready", {}) == (
["operational", "state", "vs", "0"],
{"x.com.samsung.da.state": "Ready"},
@@ -29,7 +33,8 @@ def test_stop_button_writes_ready():
def test_sound_mode_write_valid():
desc = next(e for e in laundry.SOUND_MODE.entities if e.key == "sound_mode")
desc = next(e for e in laundry.SOUND_MODE.entities if e.key == "sound_mode" and isinstance(e, SelectDesc))
assert desc.write_fn is not None
assert desc.write_fn("mute", {}) == (
["settings", "sound", "mode", "vs", "0"],
{"mode": "mute"},
@@ -38,7 +43,10 @@ def test_sound_mode_write_valid():
def test_led_brightness_write_valid():
desc = next(e for e in laundry.DOOR_LED.entities if e.key == "led_brightness")
desc = next(
e for e in laundry.DOOR_LED.entities if e.key == "led_brightness" and isinstance(e, SelectDesc)
)
assert desc.write_fn is not None
assert desc.write_fn("Low", {}) == (
["doorled", "light", "vs", "0"],
{"setBrightness": "Low"},
@@ -47,7 +55,10 @@ def test_led_brightness_write_valid():
def test_led_night_light_write_valid():
desc = next(e for e in laundry.DOOR_LED.entities if e.key == "led_night_light")
desc = next(
e for e in laundry.DOOR_LED.entities if e.key == "led_night_light" and isinstance(e, SwitchDesc)
)
assert desc.write_fn is not None
assert desc.write_fn("On", {}) == (
["doorled", "light", "vs", "0"],
{"setNightLight": "On"},
tests/test_operational_capability.py
+29 -4
View File
@@ -1,6 +1,7 @@
"""Unit tests for operational state capabilities."""
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
from custom_components.localthings.registry.entities import NumberDesc
def test_machine_state_maps_samsung_to_ocf():
@@ -17,11 +18,13 @@ class TestProgressPercentage:
def test_zeroed_when_not_active(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "progress_percentage")
assert desc.rep_fn is not None
rep = {"x.com.samsung.da.state": "Ready", "x.com.samsung.da.progressPercentage": "1"}
assert desc.rep_fn(rep) == 0
def test_passes_through_when_active(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "progress_percentage")
assert desc.rep_fn is not None
rep = {"x.com.samsung.da.state": "Run", "x.com.samsung.da.progressPercentage": "42"}
assert desc.rep_fn(rep) == 42
@@ -31,6 +34,7 @@ class TestCompletionMinutes:
def test_completion_minutes_parsing(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "completion_minutes")
assert desc.rep_fn is not None
# 1 hour 25 mins 30 secs -> 85 mins + 1 sec ceiling = 86 mins
rep = {"x.com.samsung.da.remainingTime": "01:25:30"}
@@ -42,6 +46,7 @@ class TestCompletionMinutes:
def test_completion_minutes_fallback_key(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "completion_minutes")
assert desc.rep_fn is not None
rep = {"remainingTime": "00:45:00"}
assert desc.rep_fn(rep) == 45
@@ -49,6 +54,7 @@ class TestCompletionMinutes:
"""Firmware freezes remainingTime at '00:01:00' when progress reaches 'Finish'.
Should return 0 to prevent stuck values."""
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "completion_minutes")
assert desc.rep_fn is not None
rep = {
"x.com.samsung.da.progress": "Finish",
"x.com.samsung.da.remainingTime": "00:01:00",
@@ -57,6 +63,7 @@ class TestCompletionMinutes:
def test_completion_minutes_missing_or_invalid(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "completion_minutes")
assert desc.rep_fn is not None
rep = {}
assert desc.rep_fn(rep) is None
@@ -71,6 +78,7 @@ class TestFinishTime:
def test_seconds_and_microseconds_are_zeroed(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "finish_time")
assert desc.rep_fn is not None
rep = {
"x.com.samsung.da.state": "Run",
"x.com.samsung.da.remainingTime": "00:29:00",
@@ -81,6 +89,7 @@ class TestFinishTime:
def test_stable_across_polls_within_same_minute(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "finish_time")
assert desc.rep_fn is not None
rep = {
"x.com.samsung.da.state": "Run",
"x.com.samsung.da.remainingTime": "00:29:00",
@@ -97,6 +106,7 @@ class TestDelayFieldFallback:
)
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "delay_start_hours")
assert desc.rep_fn is not None
rep = {"x.com.samsung.da.delayEndTime": "02:30:00"}
assert desc.rep_fn(rep) == 2.5
@@ -106,6 +116,7 @@ class TestDelayFieldFallback:
)
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "delay_start_hours")
assert desc.rep_fn is not None
rep = {
"x.com.samsung.da.delayStartTime": "01:00:00",
"x.com.samsung.da.delayEndTime": "02:00:00",
@@ -117,9 +128,16 @@ class TestDelayFieldFallback:
OPERATIONAL_STATE,
)
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "delay_start_hours")
desc = next(
e
for e in OPERATIONAL_STATE.entities
if e.key == "delay_start_hours" and isinstance(e, NumberDesc)
)
assert desc.write_fn is not None
rep = {"x.com.samsung.da.delayEndTime": "00:00:00"}
path, body = desc.write_fn(1.5, rep)
result = desc.write_fn(1.5, rep)
assert result is not None
path, body = result
assert path == ["operational", "state", "vs", "0"]
assert body == {"x.com.samsung.da.delayEndTime": "1:30:00"}
@@ -128,7 +146,14 @@ class TestDelayFieldFallback:
OPERATIONAL_STATE,
)
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "delay_start_hours")
desc = next(
e
for e in OPERATIONAL_STATE.entities
if e.key == "delay_start_hours" and isinstance(e, NumberDesc)
)
assert desc.write_fn is not None
rep = {"x.com.samsung.da.delayStartTime": "00:00:00"}
_path, body = desc.write_fn(1.5, rep)
result = desc.write_fn(1.5, rep)
assert result is not None
_path, body = result
assert body == {"x.com.samsung.da.delayStartTime": "1:30:00"}
tests/test_range_capabilities.py
+31 -6
View File
@@ -4,6 +4,7 @@ from custom_components.localthings.registry.adapter import flatten
from custom_components.localthings.registry.by_type import for_device_by_model
from custom_components.localthings.registry.capabilities import range as range_caps
from custom_components.localthings.registry.discovery import discover
from custom_components.localthings.registry.entities import SelectDesc, SensorDesc, SwitchDesc
from tests.conftest import _load_device
@@ -68,14 +69,21 @@ def test_unreported_burners_gated_out():
def test_burner_power_level_write_is_read_modify_write():
desc = next(e for e in range_caps.COOKTOP_STATUS.entities if e.key == "burner_1_power_level")
desc = next(
e
for e in range_caps.COOKTOP_STATUS.entities
if e.key == "burner_1_power_level" and isinstance(e, SelectDesc)
)
rep = {
"burnerList": [
{"burnerNumber": 0, "powerLevel": "3"},
{"burnerNumber": 1, "powerLevel": "0"},
]
}
path, body = desc.write_fn("boost", rep)
assert desc.write_fn is not None
result = desc.write_fn("boost", rep)
assert result is not None
path, body = result
assert path == ["cooktop", "status", "vs", "0"]
burners = {b["burnerNumber"]: b["powerLevel"] for b in body["burnerList"]}
assert burners[1] == "boost"
@@ -83,8 +91,13 @@ def test_burner_power_level_write_is_read_modify_write():
def test_burner_power_level_write_rejects_missing_burner():
desc = next(e for e in range_caps.COOKTOP_STATUS.entities if e.key == "burner_2_power_level")
desc = next(
e
for e in range_caps.COOKTOP_STATUS.entities
if e.key == "burner_2_power_level" and isinstance(e, SelectDesc)
)
rep = {"burnerList": [{"burnerNumber": 0, "powerLevel": "0"}]}
assert desc.write_fn is not None
assert desc.write_fn("5", rep) is None
@@ -109,17 +122,29 @@ def test_cooktop_power_is_read_only(): # issue #86
def test_cooktop_child_lock_write(): # issue #86
desc = next(e for e in range_caps.COOKTOP_STATUS.entities if e.key == "cooktop_child_lock")
desc = next(
e
for e in range_caps.COOKTOP_STATUS.entities
if e.key == "cooktop_child_lock" and isinstance(e, SwitchDesc)
)
assert desc.value_fn("on") is True
assert desc.value_fn("off") is False
path, body = desc.write_fn("On", {})
assert desc.write_fn is not None
result = desc.write_fn("On", {})
assert result is not None
path, body = result
assert path == ["cooktop", "status", "vs", "0"]
assert body == {"childLock": "on"}
assert desc.write_fn("Bogus", {}) is None
def test_probe_status_temperature_unit_reads_live_field(): # issue #86
desc = next(e for e in range_caps.PROBE_STATUS.entities if e.key == "probe_temperature")
desc = next(
e
for e in range_caps.PROBE_STATUS.entities
if e.key == "probe_temperature" and isinstance(e, SensorDesc)
)
assert desc.unit_fn is not None
assert desc.unit_fn({"temperatureUnit": "C"}) == "°C"
assert desc.unit_fn({"temperatureUnit": "F"}) == "°F"
assert desc.unit_fn({}) == "°C"
tests/test_select_options.py
+4 -3
View File
@@ -3,8 +3,9 @@
and callable forms of SelectDesc.options.
"""
from typing import ClassVar
from typing import ClassVar, cast
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.capability import Capability
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import SelectDesc
@@ -27,7 +28,7 @@ class _FakeCoordinator:
def _make_select(desc, href, last_resources):
capability = Capability(href=href, entities=(desc,))
bound = BoundEntity(href=href, capability=capability, desc=desc)
return LocalThingsSelect(_FakeCoordinator(last_resources), bound)
return LocalThingsSelect(cast(LocalThingsCoordinator, _FakeCoordinator(last_resources)), bound)
def test_static_options_unaffected():
@@ -107,7 +108,7 @@ async def test_unknown_vendor_option_round_trips_to_exact_raw_value():
capability = Capability(href="/x/vs/0", entities=(desc,))
bound = BoundEntity(href="/x/vs/0", capability=capability, desc=desc)
coordinator = _WritableCoordinator({})
entity = LocalThingsSelect(coordinator, bound)
entity = LocalThingsSelect(cast(LocalThingsCoordinator, coordinator), bound)
assert entity.options[-1] == "Future Vendor Mode"
await entity.async_select_option("Future Vendor Mode")
tests/test_sensor_hysteresis.py
+10 -4
View File
@@ -3,8 +3,10 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from typing import cast
from custom_components.localthings.const import CONF_FINISH_TIME_HYSTERESIS_MINUTES
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.sensor import LocalThingsSensor
@@ -32,12 +34,14 @@ class _FakeCoordinator:
def _sensor(threshold_minutes=3):
coordinator = _FakeCoordinator(threshold_minutes)
href = OPERATIONAL_STATE.href
assert href is not None
bound = BoundEntity(
href=OPERATIONAL_STATE.href,
href=href,
capability=OPERATIONAL_STATE,
desc=_FINISH_TIME_DESC,
)
sensor = LocalThingsSensor(coordinator, bound)
sensor = LocalThingsSensor(cast(LocalThingsCoordinator, coordinator), bound)
return sensor, coordinator
@@ -92,12 +96,14 @@ def test_non_hysteresis_sensor_is_unaffected():
"""A SensorDesc without hysteresis=True reads straight through, unchanged."""
machine_state_desc = next(e for e in OPERATIONAL_STATE.entities if e.key == "machine_state")
coordinator = _FakeCoordinator(threshold_minutes=3)
href = OPERATIONAL_STATE.href
assert href is not None
bound = BoundEntity(
href=OPERATIONAL_STATE.href,
href=href,
capability=OPERATIONAL_STATE,
desc=machine_state_desc,
)
sensor = LocalThingsSensor(coordinator, bound)
sensor = LocalThingsSensor(cast(LocalThingsCoordinator, coordinator), bound)
coordinator.data = {"machine_state": "active"}
assert sensor.native_value == "active"
tests/test_subdevice_discovery.py
+27 -11
View File
@@ -8,8 +8,11 @@ together, including device_info_for/via_device and the "no phantom
from __future__ import annotations
from typing import Any, cast
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from smartthings_local.protocol.dtls_session import DtlsCoapSession
from custom_components.localthings.const import (
CONF_HOST,
@@ -52,7 +55,7 @@ async def _discover_with(
without the polling/reconnect machinery around it -- see coordinator.py's
_enumerate_subdevices_blocking/_run_discovery. `_discover` below is the
fixture-file-backed convenience wrapper most tests want."""
coordinator._session = FakeCoapSession(seeds)
coordinator._session = cast(DtlsCoapSession, FakeCoapSession(seeds))
# _connect_session (skipped here -- the session is pre-set) is what
# normally populates _identity via read_identity; set it directly with
# the fixture's real /oic/res so enumeration sees the same links a live
@@ -88,7 +91,7 @@ async def _discover(coordinator: LocalThingsCoordinator, name: str) -> None:
await _discover_with(coordinator, resources, oic_res, seeds)
def _climate_bound(coordinator, subdevice_key: str):
def _climate_bound(coordinator, subdevice_key: str | None):
from custom_components.localthings.registry.subdevices import MAIN
for b in coordinator.bound:
@@ -369,12 +372,16 @@ async def test_flat_subdevice_materializes_and_repolls_end_to_end(hass: HomeAssi
# Re-poll: a fresh reading under the prefix should reach
# canonical_resources through _poll_subdevice_seed's flat-mode branch,
# not just sit frozen at the one-time enumeration snapshot.
coordinator._session.seeds[f"/{_SUB_UUID}/temperature/current/0"] = {
# FakeCoapSession's `seeds` is typed `dict[str, list]` for the common
# batch-list shape, but (per its own docstring) also legitimately holds
# plain Property maps for probe-style hrefs like these two.
seeds_map = cast("dict[str, Any]", cast(FakeCoapSession, coordinator._session).seeds)
seeds_map[f"/{_SUB_UUID}/temperature/current/0"] = {
"range": [18.0, 30.0],
"units": "C",
"temperature": 27.5,
}
coordinator._session.seeds[f"/{_SUB_UUID}/option/autoclean/vs/0"] = {
seeds_map[f"/{_SUB_UUID}/option/autoclean/vs/0"] = {
"x.com.samsung.da.settingStatus": "On",
}
refreshed = coordinator._poll_subdevice_seed(subdevice)
@@ -437,7 +444,7 @@ def test_poll_subdevice_seed_collection_mode_unaffected_by_flat_fallback(
],
}
)
coordinator._session = sess
coordinator._session = cast(DtlsCoapSession, sess)
subdevice = Subdevice(kind="prefixed", key=_SUB_UUID, seed_path=(_SUB_UUID, "device", "0"))
result = coordinator._poll_subdevice_seed(subdevice)
@@ -463,7 +470,7 @@ def test_poll_subdevice_seed_flat_mode_polls_each_href_individually(
# (_SUB_UUID, 'power', 'vs', '0') deliberately absent -> drops out.
}
)
coordinator._session = sess
coordinator._session = cast(DtlsCoapSession, sess)
subdevice = Subdevice(
kind="prefixed",
key=_SUB_UUID,
@@ -496,7 +503,7 @@ def test_poll_subdevice_seed_flat_mode_skips_hrefs_covered_by_hot_warm_subpolls(
(_SUB_UUID, "power", "vs", "0"): {"power": "On"},
}
)
coordinator._session = sess
coordinator._session = cast(DtlsCoapSession, sess)
coordinator._warm_hrefs = [f"/{_SUB_UUID}/mode/vs/0"]
subdevice = Subdevice(
kind="prefixed",
@@ -528,10 +535,19 @@ async def test_multidevice_probe_never_reaches_discovery_or_the_cache(
"""
resources, _oic, _seeds = _load_device_full("washer_flexwash")
coordinator = _coordinator(hass)
coordinator._session = FakeCoapSession(
{
"/multidevice/vs/0": {"x.com.samsung.da.numofsubdevice": "2"},
}
# /multidevice/vs/0 here is a plain Property map, not a batch list -- see
# FakeCoapSession's own docstring on the two shapes its `seeds` values can
# take; its `seeds` param type only names the more common (list) shape.
coordinator._session = cast(
DtlsCoapSession,
FakeCoapSession(
cast(
"dict[str, list]",
{
"/multidevice/vs/0": {"x.com.samsung.da.numofsubdevice": "2"},
},
)
),
)
coordinator._identity = DeviceIdentity(
manufacturer="Samsung Electronics",
tests/test_subdevices.py
+10 -7
View File
@@ -6,8 +6,11 @@ UUID-prefixed tree) this module unifies.
from __future__ import annotations
from typing import cast
import cbor2
from custom_components.localthings.registry.by_type import DeviceRegistry
from custom_components.localthings.registry.capability import Capability
from custom_components.localthings.registry.entities import BinarySensorDesc, SensorDesc
from custom_components.localthings.registry.subdevices import (
@@ -621,7 +624,7 @@ def test_discover_partitioned_main_pass_excludes_subdevice_hrefs_from_unbound():
discover_partitioned(
resources,
[sub1],
lambda r, **_: reg,
lambda r, **_: cast(DeviceRegistry, reg),
fallback_capabilities={},
log=unbound.append,
)
@@ -752,7 +755,7 @@ def test_discover_partitioned_no_subdevices_matches_plain_discover():
bound_via_helper, _, materialized, skipped = discover_partitioned(
resources,
[],
lambda r, **_: reg,
lambda r, **_: cast(DeviceRegistry, reg),
fallback_capabilities={},
)
bound_direct = discover(resources, reg.capabilities, reg.pattern_capabilities)
@@ -789,7 +792,7 @@ def test_discover_partitioned_skips_candidate_with_no_live_primary_entity():
bound, _, materialized, skipped = discover_partitioned(
resources,
[unit2],
lambda r, **_: reg,
lambda r, **_: cast(DeviceRegistry, reg),
fallback_capabilities={},
)
assert materialized == []
@@ -828,7 +831,7 @@ def test_discover_partitioned_materializes_candidate_with_live_primary_entity():
bound, _, materialized, skipped = discover_partitioned(
resources,
[sub1],
lambda r, **_: reg,
lambda r, **_: cast(DeviceRegistry, reg),
fallback_capabilities={},
)
assert materialized == [sub1]
@@ -871,7 +874,7 @@ def test_discover_partitioned_skips_candidate_whose_only_live_primary_is_a_meter
bound, _, materialized, skipped = discover_partitioned(
resources,
[unit1],
lambda r, **_: reg,
lambda r, **_: cast(DeviceRegistry, reg),
fallback_capabilities={},
)
assert materialized == []
@@ -910,7 +913,7 @@ def test_discover_partitioned_meter_carve_out_does_not_gate_out_a_live_subdevice
bound, _, materialized, skipped = discover_partitioned(
resources,
[unit1],
lambda r, **_: reg,
lambda r, **_: cast(DeviceRegistry, reg),
fallback_capabilities={},
)
assert materialized == [unit1]
@@ -938,7 +941,7 @@ def test_discover_partitioned_skipped_candidate_contributes_no_hot_warm_hrefs():
discover_partitioned(
resources,
[unit2],
lambda r, **_: reg,
lambda r, **_: cast(DeviceRegistry, reg),
fallback_capabilities={},
tier_log=lambda href, tier: tiers.append(href),
)
tests/test_washer_capabilities.py
+67 -12
View File
@@ -8,6 +8,7 @@ fallback pairs and the energy meter in test_common_capabilities.py.
from datetime import UTC
from custom_components.localthings.registry.capabilities import laundry, washer
from custom_components.localthings.registry.entities import SelectDesc
def _rep_by_href(cap, href):
@@ -20,25 +21,50 @@ class TestWasherSettings:
assert washer.WASHER_SETTINGS.href == "/washer/vs/0"
def test_wash_temperature_read(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == "wash_temperature")
desc = next(
e
for e in washer.WASHER_SETTINGS.entities
if e.key == "wash_temperature" and isinstance(e, SelectDesc)
)
assert desc.field == "x.com.samsung.da.waterTemperature"
assert desc.options_field == "x.com.samsung.da.supportedWaterTemperature"
def test_wash_temperature_write(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == "wash_temperature")
path, body = desc.write_fn("60", {})
desc = next(
e
for e in washer.WASHER_SETTINGS.entities
if e.key == "wash_temperature" and isinstance(e, SelectDesc)
)
assert desc.write_fn is not None
result = desc.write_fn("60", {})
assert result is not None
path, body = result
assert path == ["washer", "vs", "0"]
assert body == {"x.com.samsung.da.waterTemperature": "60"}
def test_spin_speed_write(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == "spin_speed")
path, body = desc.write_fn("1400", {})
desc = next(
e
for e in washer.WASHER_SETTINGS.entities
if e.key == "spin_speed" and isinstance(e, SelectDesc)
)
assert desc.write_fn is not None
result = desc.write_fn("1400", {})
assert result is not None
path, body = result
assert path == ["washer", "vs", "0"]
assert body == {"x.com.samsung.da.spinLevel": "1400"}
def test_rinse_cycles_write(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == "rinse_cycles")
path, body = desc.write_fn("3", {})
desc = next(
e
for e in washer.WASHER_SETTINGS.entities
if e.key == "rinse_cycles" and isinstance(e, SelectDesc)
)
assert desc.write_fn is not None
result = desc.write_fn("3", {})
assert result is not None
path, body = result
assert path == ["washer", "vs", "0"]
assert body == {"x.com.samsung.da.rinseCycles": "3"}
@@ -49,12 +75,20 @@ class TestDryLevel:
def test_exists_only_when_supported_dry_level_present(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == "dry_level")
assert desc.exists_fn is not None
assert desc.exists_fn({"x.com.samsung.da.supportedDryLevel": ["None", "30"]}, {}) is True
assert desc.exists_fn({}, {}) is False
def test_write(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == "dry_level")
path, body = desc.write_fn("Cupboard", {})
desc = next(
e
for e in washer.WASHER_SETTINGS.entities
if e.key == "dry_level" and isinstance(e, SelectDesc)
)
assert desc.write_fn is not None
result = desc.write_fn("Cupboard", {})
assert result is not None
path, body = result
assert path == ["washer", "vs", "0"]
assert body == {"x.com.samsung.da.dryLevel": "Cupboard"}
@@ -82,15 +116,21 @@ class TestWasherCourse:
translations/en.json via translation_key, not Python (see select.py's
_display())."""
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == "cycle")
assert desc.rep_fn is not None
rep = {"x.com.samsung.da.options": ["DeviceType_0167", "Course_1C", "GMT_04"]}
assert desc.rep_fn(rep) == "1C"
def test_missing_course_option_returns_none(self):
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == "cycle")
assert desc.rep_fn is not None
assert desc.rep_fn({"x.com.samsung.da.options": ["GMT_04"]}) is None
def test_cycle_desc_uses_cycle_options_callable(self):
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == "cycle")
desc = next(
e
for e in washer.WASHER_COURSE.entities
if e.key == "cycle" and isinstance(e, SelectDesc)
)
assert desc.options is laundry.cycle_options
def test_exists_only_when_edit_course_list_is_live(self):
@@ -98,6 +138,7 @@ class TestWasherCourse:
when a device actually populates editCourseList (see
_cycle_options's docstring for why MostUsed_ isn't used either)."""
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == "cycle")
assert desc.exists_fn is not None
assert desc.exists_fn({}, {}) is False
assert desc.exists_fn({}, {"/wm/editcourse/vs/0": {}}) is False
live = {"/wm/editcourse/vs/0": {"x.com.samsung.da.editCourseList": "EditCourseList_1C"}}
@@ -107,9 +148,16 @@ class TestWasherCourse:
"""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")
desc = next(
e
for e in washer.WASHER_COURSE.entities
if e.key == "cycle" and isinstance(e, SelectDesc)
)
assert desc.write_fn is not None
rep = {"x.com.samsung.da.options": ["DeviceType_0167", "Course_1C", "GMT_04"]}
path, body = desc.write_fn("1D", rep)
result = desc.write_fn("1D", rep)
assert result is not None
path, body = result
assert path == ["course", "vs", "0"]
assert body == {"x.com.samsung.da.options": ["Course_1D"]}
@@ -121,6 +169,7 @@ class TestDrumClean:
desc = next(
e for e in washer.WASHER_COURSE.entities if e.key == "drum_clean_cycles_remaining"
)
assert desc.rep_fn is not None
rep = {"x.com.samsung.da.options": ["WashingTimes_3", "DrumCleanProposal_40"]}
assert desc.rep_fn(rep) == 37
@@ -128,6 +177,7 @@ class TestDrumClean:
desc = next(
e for e in washer.WASHER_COURSE.entities if e.key == "drum_clean_cycles_remaining"
)
assert desc.rep_fn is not None
rep = {"x.com.samsung.da.options": ["WashingTimes_50", "DrumCleanProposal_40"]}
assert desc.rep_fn(rep) == 0
@@ -135,12 +185,14 @@ class TestDrumClean:
desc = next(
e for e in washer.WASHER_COURSE.entities if e.key == "drum_clean_cycles_remaining"
)
assert desc.rep_fn is not None
assert desc.rep_fn({"x.com.samsung.da.options": []}) is None
def test_cycles_remaining_exists_only_when_computable(self):
desc = next(
e for e in washer.WASHER_COURSE.entities if e.key == "drum_clean_cycles_remaining"
)
assert desc.exists_fn is not None
assert desc.exists_fn({"x.com.samsung.da.options": []}, {}) is False
rep = {"x.com.samsung.da.options": ["WashingTimes_3", "DrumCleanProposal_40"]}
assert desc.exists_fn(rep, {}) is True
@@ -149,6 +201,7 @@ class TestDrumClean:
"""DrumCleanLog_2026-07-01T20:18:07 -> a UTC-aware datetime,
matching the same screenshot's '10 days ago' (as of 2026-07-11)."""
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == "drum_clean_last_cleaned")
assert desc.rep_fn is not None
rep = {"x.com.samsung.da.options": ["DrumCleanLog_2026-07-01T20:18:07"]}
from datetime import datetime
@@ -156,7 +209,9 @@ class TestDrumClean:
def test_last_cleaned_missing(self):
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == "drum_clean_last_cleaned")
assert desc.rep_fn is not None
assert desc.rep_fn({"x.com.samsung.da.options": []}) is None
assert desc.exists_fn is not None
assert desc.exists_fn({"x.com.samsung.da.options": []}, {}) is False