Marc Billow
b921bdbb28
Also drops appliance-specific wording from the new user-facing strings. The setup step said "your washer" and told people to "turn the dial", which is wrong for the DW5000C dishwasher that advertises the same tokens. The two that could have caused a wrong wash cycle: - The Download-course candidate was counted on every poll that saw a loaded one-time payload, not on the polls where one was actually loaded. Since a stale token is never evicted, it keeps being reported through however long the appliance then sits on some ordinary course -- so "most frequent" ranked by dwell time. Reproduced: one poll on Course_87 then 200 on Course_1B suggests 1B, and accepting the suggestion makes picking a download program start a Cotton wash. Now only a change of the payload counts, which is the moment the device is known to accept a program. - The Download-course dropdown had custom_value=True, contradicting its own comment, so a typed-in code went into the Course_ token of a real write unchecked. Off now, plus a server-side check against the device's own course list where the value is stored. Two that quietly broke things beyond this feature: - cycle_select now always supplies a display_fn (to label cloud programs), which defeated select._display's "no state table and no fallback -> return raw" exit. Every dryer, dishwasher and air dresser on an unrecognized course table would have had its options and state reshaped from '0E' to '0 E', breaking automations and recorder history. The exit now keys off whether anything actually named the value, not whether a fallback existed. - The synthetic cloud field reached diagnostics, which reads canonical_resources -- publishing user-typed program names in a dump people paste into issues, directly against the comment claiming it never could. Dropped at the redaction boundary, with a matching strip for the debug read service, which wants device state unredacted but shouldn't present our bookkeeping as something the appliance said. And two smaller ones: - The repair fired on any device advertising slots, so the DW5000C -- four advertised, none ever loaded -- got a permanent warning nothing the owner did in Home Assistant could clear. It now waits until a payload has been seen, which is the only evidence that household uses downloaded programs. - The name-collision check read only the translation catalog, missing the device's own personal-course labels, which the select renders identically.
387 lines
17 KiB
Python
387 lines
17 KiB
Python
"""Cloud "Download" programs on laundry devices (issue #342).
|
|
|
|
The store and blob parsing live in cloudcourse.py; how they reach the cycle
|
|
select lives in registry/capabilities/laundry.py. Both are covered here,
|
|
against the two real dumps in the corpus that carry these tokens:
|
|
|
|
washer_ww5000c_cloud the issue #342 reporter's WW5000C, sitting on
|
|
Download with a one-time Jeans override loaded over
|
|
a saved Sports default. Advertises nine programs.
|
|
washer_wa55a7700av a WA55A7700AV sitting on an ordinary local course,
|
|
with a saved program and the FFFF "none" sentinel in
|
|
the one-time slot. Advertises two programs.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from custom_components.localthings import cloudcourse
|
|
from custom_components.localthings.registry.capabilities import laundry, washer
|
|
from custom_components.localthings.registry.entities import SelectDesc
|
|
from tests.conftest import _load_device
|
|
|
|
# The reporter's nine captured programs, keyed by the slot byte the device
|
|
# itself advertises. Only used to drive the tests -- nothing in the shipped
|
|
# code carries a table of these (see cloudcourse.py's module docstring).
|
|
SPORTS = "0021550449284D134AA04C0035F004F005F0AC00"
|
|
JEANS = "001F6B0449284D134AA04C0035F004F005F0AC00"
|
|
TOWELS = "00040A0449404D134AB84C0035F004F005F0AC00"
|
|
|
|
|
|
def _cycle_desc():
|
|
return next(
|
|
e for e in washer.WASHER_COURSE.entities if e.key == "cycle" and isinstance(e, SelectDesc)
|
|
)
|
|
|
|
|
|
def _rep(options, cloud=None):
|
|
rep = {"x.com.samsung.da.options": list(options)}
|
|
if cloud is not None:
|
|
rep[cloudcourse.FIELD] = cloud
|
|
return rep
|
|
|
|
|
|
class TestBlobParsing:
|
|
def test_slot_is_byte_two(self):
|
|
"""Confirmed against every program in both dumps: byte 2 of a blob
|
|
is the slot id its own CloudExtraCourse_ token advertises."""
|
|
assert cloudcourse.slot_of(SPORTS) == "55"
|
|
assert cloudcourse.slot_of(JEANS) == "6B"
|
|
assert cloudcourse.slot_of(TOWELS) == "0A"
|
|
|
|
def test_ffff_prefix_is_not_a_program(self):
|
|
"""WA55A7700AV reports this while sitting on a local course -- it
|
|
means 'no one-time override', not a program, and its byte 2 is not
|
|
one of the slots that device advertises."""
|
|
sentinel = "FFFF010049004D004A804C0037F0AC00"
|
|
assert cloudcourse.is_loaded(sentinel) is False
|
|
assert cloudcourse.slot_of(sentinel) is None
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad",
|
|
[None, "", "zz", "0021", "0021550449284D134AA04C0035F004F005F0AC0"],
|
|
)
|
|
def test_malformed_blobs_yield_no_slot(self, bad):
|
|
assert cloudcourse.slot_of(bad) is None
|
|
|
|
def test_advertised_slots_preserve_device_order(self):
|
|
rep = _rep(["CloudExtraCourse_0A5C286B2D0C55301A"])
|
|
assert cloudcourse.advertised_slots(rep) == [
|
|
"0A",
|
|
"5C",
|
|
"28",
|
|
"6B",
|
|
"2D",
|
|
"0C",
|
|
"55",
|
|
"30",
|
|
"1A",
|
|
]
|
|
|
|
def test_no_cloud_tokens_means_unsupported(self):
|
|
assert cloudcourse.supports_cloud_courses(_rep(["Course_1C"])) is False
|
|
|
|
|
|
class TestRealDumps:
|
|
def test_reporter_dump_advertises_nine_and_learns_both_loaded_blobs(self):
|
|
"""One poll teaches two programs: the saved default and the loaded
|
|
one-time override are different programs on this dump."""
|
|
rep = _load_device("washer_ww5000c_cloud")["/course/vs/0"]
|
|
assert len(cloudcourse.advertised_slots(rep)) == 9
|
|
|
|
store = cloudcourse.CloudCourses()
|
|
assert store.observe(rep) is True
|
|
assert store.blob("55") == SPORTS
|
|
assert store.blob("6B") == JEANS
|
|
# Learned but unnamed -- nothing is offerable yet.
|
|
assert store.named() == {}
|
|
assert store.view() == {}
|
|
|
|
def test_reporter_dump_proposes_its_download_course(self):
|
|
rep = _load_device("washer_ww5000c_cloud")["/course/vs/0"]
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(rep)
|
|
assert store.download_candidates() == ["87"]
|
|
# A candidate is never used until confirmed.
|
|
assert store.download_course() is None
|
|
|
|
def test_wa55_learns_its_saved_program_but_not_the_sentinel(self):
|
|
rep = _load_device("washer_wa55a7700av")["/course/vs/0"]
|
|
assert cloudcourse.advertised_slots(rep) == ["59", "58"]
|
|
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(rep)
|
|
assert store.blob("59") == "001C590549164D114A224C2037F0AC22"
|
|
assert store.blob("01") is None # the FFFF sentinel's byte 2
|
|
|
|
def test_wa55_proposes_no_download_course(self):
|
|
"""It is sitting on an ordinary local course with no override
|
|
loaded, so there is nothing to infer -- exactly the case where
|
|
guessing a code would start the wrong cycle."""
|
|
rep = _load_device("washer_wa55a7700av")["/course/vs/0"]
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(rep)
|
|
assert store.download_candidates() == []
|
|
|
|
def test_a_dishwasher_advertises_programs_it_has_never_loaded(self):
|
|
"""DW5000C (issues #113/#123): CloudExtraCourse_ names four slots
|
|
with no CloudCourse_/OneTimeCloudCourse_ token anywhere in the array.
|
|
|
|
Two things at once -- the feature is not washer-only (DA_DW, not
|
|
DA_WM), and a device can advertise programs whose payloads have never
|
|
been observed. Nothing is learnable here, so nothing is offerable,
|
|
but the gap is still countable and still worth telling the user
|
|
about."""
|
|
rep = _load_device("dishwasher_dw5000c_cloud")["/course/vs/0"]
|
|
assert cloudcourse.advertised_slots(rep) == ["8E", "8D", "8F", "02"]
|
|
assert cloudcourse.supports_cloud_courses(rep) is True
|
|
|
|
store = cloudcourse.CloudCourses()
|
|
assert store.observe(rep) is False
|
|
assert store.view() == {}
|
|
assert cloudcourse.undiscovered(rep, store.snapshot()) == ["8E", "8D", "8F", "02"]
|
|
|
|
def test_byte_three_is_not_part_of_a_programs_identity(self):
|
|
"""Two WW5000C units on different firmware (issue #342's _B06C and
|
|
issues #259/#343's _B048) report the same program -- same id, same
|
|
slot, same field values, same tail -- differing only at byte 3.
|
|
|
|
This is the evidence against a shipped catalog of payloads: keyed on
|
|
program id, one unit's capture would have carried the other unit's
|
|
byte 3. Both are recorded and replayed per device instead, so
|
|
whatever byte 3 tracks never has to be understood."""
|
|
b06c = "00040A0449404D134AB84C0035F004F005F0AC00"
|
|
b048 = "00040A0649404D134AB84C0035F004F005F0AC00"
|
|
assert b06c != b048
|
|
# Same program by every identifier the device gives us.
|
|
assert cloudcourse.slot_of(b06c) == cloudcourse.slot_of(b048) == "0A"
|
|
assert b06c[:6] == b048[:6]
|
|
assert b06c[8:] == b048[8:]
|
|
# Learned separately per device, and each replays what it saw.
|
|
for blob in (b06c, b048):
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(_rep(["CloudExtraCourse_0A", f"CloudCourse_{blob}"]))
|
|
assert store.blob("0A") == blob
|
|
|
|
def test_a_sentinel_slot_byte_is_not_the_current_course(self):
|
|
"""The two sentinels in the corpus disagree about this -- WA55's byte
|
|
2 happens to equal its selected course, the WW5000C _B048's does not
|
|
(1B against Course_1C). Nothing keys off it: a sentinel is rejected
|
|
on its FFFF prefix, and its byte 2 is not an advertised slot either
|
|
way."""
|
|
wa55 = "FFFF010049004D004A804C0037F0AC00"
|
|
b048 = "FFFF1B0049004D004A804C0035F004F005F0AC00"
|
|
assert cloudcourse.slot_of(wa55) is None
|
|
assert cloudcourse.slot_of(b048) is None
|
|
# Different board widths, same sentinel rule.
|
|
assert len(wa55) != len(b048)
|
|
|
|
def test_undiscovered_counts_against_what_the_device_advertises(self):
|
|
rep = _load_device("washer_ww5000c_cloud")["/course/vs/0"]
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(rep)
|
|
# Both learned slots are still unnamed, so all nine are outstanding.
|
|
assert len(cloudcourse.undiscovered(rep, store.snapshot())) == 9
|
|
store.set_name("55", "Sports")
|
|
assert "55" not in cloudcourse.undiscovered(rep, store.snapshot())
|
|
assert len(cloudcourse.undiscovered(rep, store.snapshot())) == 8
|
|
|
|
|
|
class TestStoreRules:
|
|
def test_only_advertised_slots_are_recorded(self):
|
|
"""A blob for a slot this appliance doesn't list isn't a program it
|
|
offers."""
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(_rep(["CloudExtraCourse_55", f"OneTimeCloudCourse_{JEANS}"]))
|
|
assert store.blob("6B") is None
|
|
|
|
def test_a_relearned_blob_replaces_the_old_payload(self):
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(_rep(["CloudExtraCourse_55", f"CloudCourse_{SPORTS}"]))
|
|
rewritten = SPORTS.replace("F005F0AC00", "F005F0AC11")
|
|
assert store.observe(_rep(["CloudExtraCourse_55", f"CloudCourse_{rewritten}"])) is True
|
|
assert store.blob("55") == rewritten
|
|
|
|
def test_observing_the_same_rep_twice_changes_nothing(self):
|
|
store = cloudcourse.CloudCourses()
|
|
rep = _rep(["CloudExtraCourse_55", f"CloudCourse_{SPORTS}"])
|
|
assert store.observe(rep) is True
|
|
assert store.observe(rep) is False
|
|
|
|
def test_candidates_rank_by_program_loads_not_by_dwell_time(self):
|
|
"""A stale OneTimeCloudCourse_ token outlives the run it belonged to,
|
|
so it is still reported through every poll the appliance spends on
|
|
whatever ordinary course comes next. Counting polls would rank that
|
|
course first and suggest it as the Download course -- and accepting
|
|
the suggestion would start a real wash cycle. Only the poll where the
|
|
payload actually changed is evidence."""
|
|
store = cloudcourse.CloudCourses()
|
|
loaded = _rep(["CloudExtraCourse_55", "Course_87", f"OneTimeCloudCourse_{SPORTS}"])
|
|
stale = _rep(["CloudExtraCourse_55", "Course_1B", f"OneTimeCloudCourse_{SPORTS}"])
|
|
store.observe(loaded)
|
|
for _ in range(200): # ~100 minutes sitting on a cotton cycle
|
|
store.observe(stale)
|
|
assert store.download_candidates() == ["87"]
|
|
|
|
def test_a_second_distinct_load_is_counted(self):
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(_rep(["CloudExtraCourse_556B", "Course_87", f"OneTimeCloudCourse_{SPORTS}"]))
|
|
store.observe(_rep(["CloudExtraCourse_556B", "Course_87", f"OneTimeCloudCourse_{JEANS}"]))
|
|
assert store.download_candidates() == ["87"]
|
|
|
|
def test_view_is_empty_until_a_download_course_is_confirmed(self):
|
|
"""Both halves are required: without the course code there is no
|
|
write to build, so nothing should reach the select."""
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(_rep(["CloudExtraCourse_55", f"CloudCourse_{SPORTS}"]))
|
|
store.set_name("55", "Sports")
|
|
assert store.view() == {}
|
|
store.set_download_course("87")
|
|
assert store.view()["programs"] == {"55": {"blob": SPORTS, "name": "Sports"}}
|
|
|
|
def test_round_trips_through_the_entry(self):
|
|
store = cloudcourse.CloudCourses()
|
|
store.observe(_rep(["CloudExtraCourse_55", f"CloudCourse_{SPORTS}"]))
|
|
store.set_name("55", "Sports")
|
|
store.set_download_course("87")
|
|
restored = cloudcourse.CloudCourses(store.snapshot())
|
|
assert restored.snapshot() == store.snapshot()
|
|
assert restored.view() == store.view()
|
|
|
|
@pytest.mark.parametrize(
|
|
"junk",
|
|
[
|
|
"not a dict",
|
|
{"slots": "nope"},
|
|
{"slots": {"55": {"blob": "garbage", "name": "Sports"}}},
|
|
# blob's own byte 2 disagrees with the key it is filed under
|
|
{"slots": {"99": {"blob": SPORTS, "name": "Sports"}}},
|
|
],
|
|
)
|
|
def test_a_hand_edited_entry_cannot_crash_setup(self, junk):
|
|
assert cloudcourse.CloudCourses(junk).snapshot()["slots"] == {}
|
|
|
|
|
|
class TestCycleSelectIntegration:
|
|
"""The cloud programs ride in the ordinary cycle select -- on the
|
|
appliance's dial, Download is one course among the rest."""
|
|
|
|
def _cloud(self):
|
|
return {
|
|
"download_course": "87",
|
|
"programs": {"55": {"blob": SPORTS, "name": "Sports"}},
|
|
}
|
|
|
|
def test_local_courses_are_unchanged_without_cloud_data(self):
|
|
desc = _cycle_desc()
|
|
live = {"/wm/editcourse/vs/0": {"x.com.samsung.da.editCourseList": "EditCourseList_1C1D"}}
|
|
assert desc.options(live) == ["1C", "1D"]
|
|
|
|
def test_named_programs_join_the_option_list_after_local_courses(self):
|
|
desc = _cycle_desc()
|
|
live = {
|
|
"/wm/editcourse/vs/0": {"x.com.samsung.da.editCourseList": "EditCourseList_1C87"},
|
|
"/course/vs/0": _rep(["Course_87"], self._cloud()),
|
|
}
|
|
assert desc.options(live) == ["1C", "87", "cloud:55"]
|
|
|
|
def test_an_unnamed_program_is_never_offered(self):
|
|
desc = _cycle_desc()
|
|
live = {
|
|
"/wm/editcourse/vs/0": {"x.com.samsung.da.editCourseList": "EditCourseList_1C"},
|
|
"/course/vs/0": _rep(
|
|
["Course_87"],
|
|
{"download_course": "87", "programs": {}},
|
|
),
|
|
}
|
|
assert desc.options(live) == ["1C"]
|
|
|
|
def test_label_is_the_users_own_name(self):
|
|
desc = _cycle_desc()
|
|
resources = {"/course/vs/0": _rep(["Course_87"], self._cloud())}
|
|
assert desc.display_fn("cloud:55", resources) == "Sports"
|
|
|
|
def test_unknown_cloud_value_gets_no_invented_label(self):
|
|
desc = _cycle_desc()
|
|
resources = {"/course/vs/0": _rep(["Course_87"], self._cloud())}
|
|
assert desc.display_fn("cloud:99", resources) is None
|
|
|
|
def test_state_reports_the_loaded_program_while_on_download(self):
|
|
desc = _cycle_desc()
|
|
rep = _rep(["Course_87", f"OneTimeCloudCourse_{SPORTS}"], self._cloud())
|
|
assert desc.rep_fn(rep) == "cloud:55"
|
|
|
|
def test_state_falls_back_to_the_saved_program(self):
|
|
"""No one-time override loaded: the appliance runs the saved default,
|
|
which the reporter confirmed by leaving Download and returning."""
|
|
desc = _cycle_desc()
|
|
rep = _rep(
|
|
["Course_87", f"CloudCourse_{SPORTS}", "OneTimeCloudCourse_FFFF010049004D004A804C00"],
|
|
self._cloud(),
|
|
)
|
|
assert desc.rep_fn(rep) == "cloud:55"
|
|
|
|
def test_a_stale_program_token_is_not_reported_on_a_local_course(self):
|
|
"""Tokens are replaced by prefix and never evicted, so a one-time
|
|
program outlives its run. Reporting 'Sports' while a cotton cycle
|
|
runs would be a live lie about what the machine is doing."""
|
|
desc = _cycle_desc()
|
|
rep = _rep(["Course_1C", f"OneTimeCloudCourse_{SPORTS}"], self._cloud())
|
|
assert desc.rep_fn(rep) == "1C"
|
|
|
|
def test_selecting_a_program_switches_course_and_loads_it_in_one_write(self):
|
|
"""Confirmed on hardware: writing the program token alone while
|
|
another course is selected is silently ignored."""
|
|
desc = _cycle_desc()
|
|
rep = _rep(["Course_1C"], self._cloud())
|
|
path, body = desc.write_fn("cloud:55", rep)
|
|
assert path == ["course", "vs", "0"]
|
|
assert body == {"x.com.samsung.da.options": ["Course_87", f"OneTimeCloudCourse_{SPORTS}"]}
|
|
|
|
def test_selecting_a_local_course_still_writes_one_token(self):
|
|
desc = _cycle_desc()
|
|
rep = _rep(["Course_87"], self._cloud())
|
|
_, body = desc.write_fn("1C", rep)
|
|
assert body == {"x.com.samsung.da.options": ["Course_1C"]}
|
|
|
|
def test_no_write_without_a_confirmed_download_course(self):
|
|
desc = _cycle_desc()
|
|
rep = _rep(["Course_1C"], {"programs": {"55": {"blob": SPORTS, "name": "Sports"}}})
|
|
assert desc.write_fn("cloud:55", rep) is None
|
|
|
|
def test_no_write_for_an_unknown_program(self):
|
|
desc = _cycle_desc()
|
|
rep = _rep(["Course_1C"], self._cloud())
|
|
assert desc.write_fn("cloud:99", rep) is None
|
|
|
|
|
|
class TestOptionsMergePreservesSiblingTokens:
|
|
def test_two_token_write_merges_like_the_device_does(self):
|
|
"""The write carries only the changed tokens; merge_options_field is
|
|
what keeps the optimistic cache entry complete during the settle
|
|
window. Both tokens must land, and unrelated ones must survive."""
|
|
from custom_components.localthings.registry.capabilities.common import merge_options_field
|
|
|
|
cached = [
|
|
"DeviceType_0167",
|
|
"Course_1C",
|
|
f"CloudCourse_{SPORTS}",
|
|
"CloudExtraCourse_0A5C286B2D0C55301A",
|
|
f"OneTimeCloudCourse_{TOWELS}",
|
|
"GMT_04",
|
|
]
|
|
merged = merge_options_field(cached, ["Course_87", f"OneTimeCloudCourse_{SPORTS}"])
|
|
assert "Course_87" in merged
|
|
assert f"OneTimeCloudCourse_{SPORTS}" in merged
|
|
# The saved default and the device's own slot list are untouched.
|
|
assert f"CloudCourse_{SPORTS}" in merged
|
|
assert "CloudExtraCourse_0A5C286B2D0C55301A" in merged
|
|
assert "DeviceType_0167" in merged
|
|
|
|
def test_cloud_prefixes_do_not_poison_the_course_lookup(self):
|
|
"""'Course_' is a prefix of neither 'CloudCourse_' nor
|
|
'OneTimeCloudCourse_' only because option_value anchors at position
|
|
0 -- one character away from being wrong, so it is pinned."""
|
|
options = [f"CloudCourse_{SPORTS}", f"OneTimeCloudCourse_{JEANS}", "Course_1C"]
|
|
assert laundry.option_value(options, "Course") == "1C"
|
|
assert cloudcourse.option_value(options, "Course") == "1C"
|