mirror of
https://github.com/rstrouse/ESPSomfy-RTS.git
synced 2026-10-31 13:52:15 +01:00
77237111ea
The edit routes become ops 26-44 (addRoom ... groupSortOrder): ProtoEdits turns an op's payload into the route's request, full names and real booleans, and loop() applies it with applyModelEdit() as the route does. The reply goes to the client that sent it, with an add's new id, or the route's reason and a shade's validation code; the HTTP routes keep sending requestResult. getNextRoom, getNextShade, getNextGroup and groupOptions (45-48) are answered by loop() from the shade model, through a new LoopQueue item. Messages that arrive in pieces, a frame over one TCP segment or several frames, are put back together by ProtoAssembly: one at a time across clients, up to 4 KB, in a buffer held only while one is on its way. A message in pieces is now told from its first frame's opcode, which also stops a fragmented message's last frame drawing a second reply. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
115 lines
5 KiB
Python
115 lines
5 KiB
Python
"""The edit ops and the reads an edit form needs (docs/PROTOCOL.md). Without
|
|
--edit nothing changes: edits are refused, or set what is already there."""
|
|
import pytest
|
|
|
|
from test_commands import is_ok, op, pack, synced_model
|
|
from test_messages import error_name, reply_to
|
|
|
|
EDIT_OPS = ("addRoom", "addShade", "addGroup", "saveRoom", "saveShade", "saveGroup", "deleteRoom", "deleteShade",
|
|
"deleteGroup", "linkRemote", "unlinkRemote", "linkToGroup", "unlinkFromGroup", "linkRepeater",
|
|
"unlinkRepeater", "setPaired", "roomSortOrder", "shadeSortOrder", "groupSortOrder")
|
|
READ_OPS = ("getNextRoom", "getNextShade", "getNextGroup", "groupOptions")
|
|
|
|
|
|
def field(schema, reply, name):
|
|
return reply["d"].get(schema.keys[name], schema.default(name))
|
|
|
|
|
|
def unused(items, key):
|
|
return next(i for i in range(250, 0, -1) if i not in {x[key] for x in items})
|
|
|
|
|
|
def test_every_edit_and_read_op_is_in_the_schema(schema):
|
|
for name in EDIT_OPS + READ_OPS:
|
|
assert 16 <= schema.type_numbers[name] <= 127, name
|
|
|
|
|
|
def test_the_next_shade_gets_a_free_id_and_address(connect, schema):
|
|
c, shades, groups = synced_model(connect, schema)
|
|
reply = op(c, schema, "getNextShade", {})
|
|
assert is_ok(schema, reply)
|
|
assert field(schema, reply, "shadeId") not in {s["shadeId"] for s in shades}
|
|
address = field(schema, reply, "remoteAddress")
|
|
assert address and address not in {s.get("remoteAddress") for s in shades + groups}
|
|
|
|
|
|
def test_the_next_room_and_group(connect, schema):
|
|
c, _, groups = synced_model(connect, schema)
|
|
assert is_ok(schema, op(c, schema, "getNextRoom", {}))
|
|
reply = op(c, schema, "getNextGroup", {})
|
|
assert is_ok(schema, reply)
|
|
assert field(schema, reply, "groupId") not in {g["groupId"] for g in groups}
|
|
|
|
|
|
def test_group_options_are_the_shades_it_lacks(connect, schema):
|
|
c, shades, groups = synced_model(connect, schema)
|
|
if not groups:
|
|
pytest.skip("the device has no groups")
|
|
group = groups[0]
|
|
reply = op(c, schema, "groupOptions", {"groupId": group["groupId"]})
|
|
assert is_ok(schema, reply)
|
|
expected = {s["shadeId"] for s in shades} - set(group.get("linkedShades", []))
|
|
assert set(field(schema, reply, "ids")) == expected
|
|
|
|
|
|
def test_group_options_of_a_missing_group(connect, schema):
|
|
c, _, groups = synced_model(connect, schema)
|
|
reply = op(c, schema, "groupOptions", {"groupId": unused(groups, "groupId")})
|
|
assert error_name(schema, reply) == "notFound"
|
|
|
|
|
|
@pytest.mark.parametrize("name, payload, why", [
|
|
("saveRoom", {"zz": 1}, "Unknown field zz"),
|
|
("setPaired", {"s": 1, "i": 2}, "paired is not a boolean (1 or 0)"),
|
|
("shadeSortOrder", {}, "ids is missing"),
|
|
])
|
|
def test_a_bad_payload_is_refused_before_it_is_queued(connect, schema, name, payload, why):
|
|
c, _, _ = synced_model(connect, schema)
|
|
msg_id = c.send(schema.type_numbers[name], payload, with_id=True)
|
|
reply = reply_to(c, msg_id)
|
|
assert error_name(schema, reply) == "badRequest"
|
|
assert reply["d"][schema.keys["desc"]] == why
|
|
|
|
|
|
def test_an_edit_of_a_missing_shade_says_why(connect, schema):
|
|
"""Refused by the edit itself on loop(), as its route would be."""
|
|
c, shades, _ = synced_model(connect, schema)
|
|
reply = op(c, schema, "saveShade", {"shadeId": unused(shades, "shadeId"), "name": "x"})
|
|
assert error_name(schema, reply) == "badRequest"
|
|
assert reply["d"][schema.keys["desc"]] == "Shade Id not found."
|
|
|
|
|
|
def test_a_sort_order_set_to_what_it_is(connect, schema):
|
|
c, shades, _ = synced_model(connect, schema)
|
|
if not shades:
|
|
pytest.skip("the device has no shades")
|
|
ids = [s["shadeId"] for s in sorted(shades, key=lambda s: s.get("sortOrder", 0))]
|
|
reply = op(c, schema, "shadeSortOrder", {"ids": ids})
|
|
assert is_ok(schema, reply)
|
|
order = c.recv_until(lambda f: f["t"] == schema.type_numbers["shadeOrder"])[-1]
|
|
assert order["d"][schema.keys["ids"]] == ids
|
|
|
|
|
|
def test_a_long_edit_is_put_back_together(connect, schema):
|
|
"""3 KB arrives in pieces: the reply carries its id, so it was read whole."""
|
|
c, shades, _ = synced_model(connect, schema)
|
|
padding = " " * 3000
|
|
msg_id = 4242
|
|
c.send_text('{"v":2,"t":%d,"id":%d,"d":{"%s":%d%s}}' % (
|
|
schema.type_numbers["saveShade"], msg_id, schema.keys["shadeId"], unused(shades, "shadeId"), padding))
|
|
reply = reply_to(c, msg_id)
|
|
assert reply["d"][schema.keys["desc"]] == "Shade Id not found."
|
|
|
|
|
|
def test_adding_and_deleting_a_room(connect, schema, editing):
|
|
"""With --edit: a room is added, then deleted again."""
|
|
c, _, _ = synced_model(connect, schema)
|
|
reply = op(c, schema, "addRoom", {"name": "e2e room"})
|
|
assert is_ok(schema, reply), reply
|
|
room_id = field(schema, reply, "roomId")
|
|
assert room_id
|
|
added = c.recv_until(lambda f: f["t"] == schema.type_numbers["room"])[-1]
|
|
assert schema.expand(added["d"])["name"] == "e2e room"
|
|
assert is_ok(schema, op(c, schema, "deleteRoom", {"roomId": room_id}))
|
|
removed = c.recv_until(lambda f: f["t"] == schema.type_numbers["roomRemoved"])[-1]
|
|
assert removed["d"][schema.keys["roomId"]] == room_id
|