diff --git a/.gitignore b/.gitignore index e05d7b9..d9e701a 100644 --- a/.gitignore +++ b/.gitignore @@ -35,3 +35,7 @@ SomfyController.ino.esp32s2.bin # Temporary and Backup Files *.orig *.bak + +# Python (the e2e suite and build scripts) +__pycache__/ +.pytest_cache/ diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index af07ab6..b9d4fa2 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -1,51 +1,89 @@ # Socket protocol 2 -**Status:** outline. The type numbers, field keys and payloads are filled in with the firmware protocol core. Why the protocol looks like this: [ADR 0005](adr/0005-one-websocket-protocol.md). Terms: [GLOSSARY.md](../GLOSSARY.md). +Why the protocol looks like this: [ADR 0005](adr/0005-one-websocket-protocol.md). Terms: [GLOSSARY.md](../GLOSSARY.md). + +**The table of types, fields, error codes and feeds is defined once, in [src/ProtoSchema.h](../src/ProtoSchema.h).** The device sends it to every client as the schema, and `test/fixtures/proto2/schema.json` is a copy recorded by the e2e suite (`pytest test/e2e --record`). This page describes how the protocol works, not every field. ## Connection -- `ws://host/ws` on port 80. Under Secured the upgrade needs `?apikey=` ([ADR 0006](adr/0006-two-security-modes-eleven-http-routes.md)). -- Optional `?schema=`: the client already has that schema and doesn't need it again. -- Text frames only. A binary encoding, if one comes, would be asked for at the upgrade (for example `?enc=bin`) and use binary frames. +- `ws://host/ws` on port 80. Under Secured the upgrade needs `?apikey=` from `/login`; without a valid one it is refused with 401 ([ADR 0006](adr/0006-two-security-modes-eleven-http-routes.md)). +- Optional `?schema=`: the client already has the schema with that hash and doesn't need it again. +- At most 5 clients. A sixth is closed as soon as it connects. +- Text frames only, and each message in one frame: a fragmented message is answered with a bad request. A binary encoding, if one comes, would be asked for at the upgrade (for example `?enc=bin`) and use binary frames. ## Envelope Every frame, in both directions, is one JSON object: ```json -{"v":2,"t":20,"id":17,"d":{}} +{"v":2,"t":129,"id":17,"d":{"s":3,"p":40}} ``` | Key | Meaning | |---|---| -| `v` | protocol version | -| `t` | message type, a number from the table below | -| `id` | only on an op and its reply: the client's number for the op | -| `d` | the payload, with short keys from the schema | +| `v` | protocol version, 2 | +| `t` | message type, a number from the schema | +| `id` | only on a client message that wants a reply, and on that reply | +| `d` | the payload, with keys from the schema's field dictionary | -Within `d`, booleans are `1`/`0` and fields at their default are left out. Keys are one or two characters from one dictionary shared by every type. +Within `d`: +- keys are one or two characters, from one dictionary shared by every type; +- booleans are `1` and `0`; +- a field at its default is left out: a client fills it in from the schema; +- lists and objects use the same dictionary for the keys inside them. + +The keys of `hello` (`fw`, `sh`) and of the schema itself (`T`, `F`, `E`, `Q`) never change, since a client needs them before it has a schema. ## Message types -| Range | Sender | Kinds | +| Range | Sender | Types | |---|---|---| -| 1–15 | either | control: `hello`, `reply`, `next`, `sync`, `syncEnd`, `subscribe`, `unsubscribe` | -| 16–127 | client | ops: the commands, reads, model edits and `getSettings`/`saveSettings` | -| 128–255 | device | events: `shadeState`, `shade`, `group`, `room`, `*Removed`, `settingsChanged`, status, feeds | +| 1–15 | control | `hello` 1, `reply` 2, `next` 3, `sync` 4, `syncEnd` 5, `subscribe` 6, `unsubscribe` 7 | +| 16–127 | client ops | none yet; commands, model edits and settings follow | +| 128–255 | device events | `shadeState`, `shade`, `shadeRemoved`, `group`, `groupState`, `groupRemoved`, `room`, `roomRemoved`, `roomOrder`, `shadeOrder`, `groupOrder`, `shadeCommand`, `memory`, `wifi`, `ethernet`, `frame`, `frequencyScan`, `repeaters` | A number is never reused. A retired type stays reserved. ## Connect sequence -1. The device sends `hello`: protocol version, firmware version, schema hash. -2. Unless the client passed the matching `?schema=`, the sync starts with the schema: types, fields (key, name, kind, default) and error codes. -3. Sync: header, rooms, shades, groups, repeaters, status. Each message is at most 4 KB, items are never split, and the client sends `next` after each one. -4. `syncEnd`. From then on, events keep the client current. Events also arrive during the sync, so the client upserts by id. +1. The device sends `hello`: `{"fw":"","sh":""}`. +2. The device sends the first `sync` frame straight away. Unless the client passed the matching `?schema=`, the sync starts with the schema: + - `T`: types, as `[number, name]` + - `F`: fields, as `[key, name, kind, default]`, kind being `b` bool, `i` integer, `n` number, `s` string, `l` list, `o` object + - `E`: error codes, as `[number, name]` + - `Q`: feeds, as `[number, name]` +3. Then the device: `header` (limits, starting address, what a new shade gets, server id, hostname, model, chip, firmware, connection type), `rooms`, `shades`, `groups`, `repeaters` and `status` (memory, and WiFi or Ethernet). A frame is at most 4 KB, an item is never split across frames, and several sections can share one. +4. The client sends `next` after each `sync` frame, and the device answers with the following one. With no `next` for 30 s the device closes the socket. +5. `syncEnd` follows the last `sync` frame. + +The client can send `sync` at any time to get the shade model and status again (without the schema). + +Events go out to every client from the moment it connects, so they can arrive during the sync, and even before the schema. A client keeps any frame it can't decode yet until it has the schema. Each sync frame is formatted from the live model, so an item in a later frame already holds every change before it. A client treats sync items and events alike: insert or update by id, and ignore removing an id it doesn't have. + +## Events + +| Event | When | Payload | +|---|---|---| +| `shadeState` | a shade moved or its state changed | id, position, target, direction, favorite, tilt, flags, rolling code | +| `shade` | a shade was added or a setting changed (edit, pairing, linked remote, group, its room deleted) | the whole shade, as in the sync | +| `group`, `groupState` | the same for a group: its settings or linked shades, or its flags and direction | | +| `room` | a room was added or saved | the whole room | +| `*Removed` | a shade, group or room was deleted | its id | +| `roomOrder`, `shadeOrder`, `groupOrder` | the sort order changed | `is`: the ids in order | +| `repeaters` | a repeater was linked or unlinked | the list | +| `shadeCommand` | a remote or the device commanded a shade | id, address, command, source, rolling code, source address | +| `memory`, `wifi`, `ethernet` | every 30 s, WiFi only when the signal moved | | ## Replies -Every op gets exactly one reply, to its sender only: `{"ok":1, ...}`, or `{"ok":0,"code":,"desc":"..."}`. The error codes are bad request, unknown op, unauthorized, not found, busy, radio unavailable and internal; their numbers are in the schema. +A client message with an `id` gets exactly one `reply` with that `id`, to that client only: `ok` 1 on success, or an error `code` and a `desc` for people. Error codes are in the schema: bad request, unknown op, unauthorized, not found, busy, radio unavailable, internal. + +A malformed message gets a bad request, with its `id` when it had one. A type a client may not send gets an unknown op. ## Feeds -`subscribe`/`unsubscribe` with a feed: `frames` (received remote frames) or `frequencyScan`. Feed events go only to subscribed clients. +`subscribe` and `unsubscribe` take a feed number in `fe`: `frames` (received remote frames, with their pulses) or `frequencyScan`. Their events go only to clients subscribed to them. + +## Inside the device + +async_tcp reads client frames and answers with fixed frames that hold nothing of the shade model: `hello`, errors and `ok`. Everything formatted from the shade model goes out from `loop()`, which is the only task that reads and writes it ([ADR 0002](adr/0002-loop-task-is-the-only-writer-of-the-shade-model.md)). Sync frames and events share the socket event buffer, so protocol 2 needs no more static memory. diff --git a/test/e2e/README.md b/test/e2e/README.md new file mode 100644 index 0000000..daa8602 --- /dev/null +++ b/test/e2e/README.md @@ -0,0 +1,17 @@ +# e2e suite + +Runs against a real device on your network, by hand, never in CI. It checks socket protocol 2 ([PROTOCOL.md](../../docs/PROTOCOL.md)) and that the device survives the loads that used to run its heap out. + +```sh +~/.platformio/penv/Scripts/uv run --no-project --with pytest -- pytest test/e2e --device 192.168.1.120 +``` + +| Option | Use | +|---|---| +| `--device` | the device's address (required) | +| `--password`, `--username` / `--pin` | the credentials of a Secured device | +| `--record` | save hello, the sync and the schema to `test/fixtures/proto2/`, for the web UI and Home Assistant tests | + +A plain run sends nothing on air, changes no settings and doesn't reboot the device. It does fill every socket slot for a few seconds, and replays a page load. + +The tests run in whatever security mode the device is in; the suite never changes it. To test both modes, run the suite once in each. diff --git a/test/e2e/conftest.py b/test/e2e/conftest.py new file mode 100644 index 0000000..fda0430 --- /dev/null +++ b/test/e2e/conftest.py @@ -0,0 +1,84 @@ +"""Options and fixtures for the e2e suite. See README.md.""" +import os +import sys +import time + +import pytest + +sys.path.insert(0, os.path.dirname(__file__)) +from proto_client import ProtoClient, Schema, http_json, run_sync # noqa: E402 + + +def pytest_addoption(parser): + parser.addoption("--device", required=True, help="the device's address, e.g. 192.168.1.120") + parser.addoption("--password", default=None, help="the device's password, when it is Secured with one") + parser.addoption("--username", default="", help="the username that goes with --password") + parser.addoption("--pin", default=None, help="the device's PIN, when it is Secured with one") + parser.addoption("--record", action="store_true", help="save what the device sends to test/fixtures/proto2") + + +@pytest.fixture(scope="session") +def host(request): + return request.config.getoption("--device") + + +@pytest.fixture(scope="session") +def login_context(host): + return http_json(host, "/loginContext") + + +@pytest.fixture(scope="session") +def secured(login_context): + return login_context.get("type", 0) != 0 + + +@pytest.fixture(scope="session") +def token(request, host, secured): + """The token for this PC, or None when the device isn't Secured.""" + if not secured: + return None + password = request.config.getoption("--password") + pin = request.config.getoption("--pin") + if password is None and pin is None: + pytest.skip("the device is Secured: pass --password or --pin") + body = {"username": request.config.getoption("--username"), "password": password or "", "pin": pin or ""} + return http_json(host, "/login", method="PUT", body=body)["apiKey"] + + +@pytest.fixture +def connect(host, token): + """Opens protocol 2 clients and closes them after the test.""" + clients = [] + + def open_client(**params): + if token and "apikey" not in params: + params["apikey"] = token + c = ProtoClient(host, params=params or None) + clients.append(c) + return c + + yield open_client + for c in clients: + c.close() + # The device frees a slot when it sees the disconnect; the next test may + # need every slot. + time.sleep(0.5) + + +@pytest.fixture(scope="session") +def schema(host, token): + """The device's schema, from one full sync.""" + s = Schema() + c = ProtoClient(host, params={"apikey": token} if token else None) + try: + assert c.status == 101, "the device has no /ws: is it running protocol 2 firmware?" + run_sync(c, s) + finally: + c.close() + assert s.complete() and s.hash + return s + + +@pytest.fixture(scope="session") +def recording(request): + return request.config.getoption("--record") diff --git a/test/e2e/proto_client.py b/test/e2e/proto_client.py new file mode 100644 index 0000000..8fa1783 --- /dev/null +++ b/test/e2e/proto_client.py @@ -0,0 +1,232 @@ +"""A socket protocol 2 client for the e2e suite (docs/PROTOCOL.md). + +Only the standard library, so the suite needs nothing but pytest. Frames are +read on a background thread into a queue; tests take them with recv(). +""" +import base64 +import json +import os +import queue +import socket +import struct +import threading +import time +import urllib.parse +import urllib.request + +PROTO_VERSION = 2 + + +class SocketClosed(Exception): + pass + + +class Schema: + """The device's schema: numbers and short keys to names, and defaults.""" + + def __init__(self): + self.hash = None + self.types = {} # number -> name + self.type_numbers = {} + self.fields = {} # key -> (name, kind, default) + self.keys = {} # name -> key + self.errors = {} # number -> name + self.error_numbers = {} + self.feeds = {} + self.feed_numbers = {} + + def add(self, d): + for number, name in d.get("T", []): + self.types[number] = name + self.type_numbers[name] = number + for key, name, kind, default in d.get("F", []): + self.fields[key] = (name, kind, default) + self.keys[name] = key + for number, name in d.get("E", []): + self.errors[number] = name + self.error_numbers[name] = number + for number, name in d.get("Q", []): + self.feeds[number] = name + self.feed_numbers[name] = number + + def complete(self): + return bool(self.types and self.fields and self.errors) + + def expand(self, value): + """Short keys to names, 1/0 to booleans; omitted fields stay omitted.""" + if isinstance(value, list): + return [self.expand(v) for v in value] + if not isinstance(value, dict): + return value + out = {} + for key, v in value.items(): + name, kind, _ = self.fields.get(key, (key, None, None)) + if kind == "b" and isinstance(v, int): + v = bool(v) + out[name] = self.expand(v) + return out + + def default(self, name): + key = self.keys[name] + _, kind, default = self.fields[key] + if kind == "b": + return bool(default) + if kind == "s": + return "" + if kind == "l": + return [] + return default + + +class ProtoClient: + def __init__(self, host, port=80, path="/ws", params=None, timeout=10): + self.host = host + query = ("?" + urllib.parse.urlencode(params)) if params else "" + self.sock = socket.create_connection((host, port), timeout=timeout) + key = base64.b64encode(os.urandom(16)).decode() + self.sock.sendall(( + f"GET {path}{query} HTTP/1.1\r\nHost: {host}:{port}\r\nUpgrade: websocket\r\n" + f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode()) + head = b"" + while b"\r\n\r\n" not in head: + chunk = self.sock.recv(1024) + if not chunk: + raise SocketClosed("closed during the handshake") + head += chunk + status_line, self._rest = head.split(b"\r\n\r\n", 1) + self.status = int(status_line.split(b" ")[1]) + self.frames = queue.Queue() + self.raw = [] # every text frame as received, for fixtures + self.closed = threading.Event() + if self.status == 101: + threading.Thread(target=self._reader, daemon=True).start() + else: + self.sock.close() + self.closed.set() + self._next_id = 1 + + def _recvn(self, n): + while len(self._rest) < n: + chunk = self.sock.recv(4096) + if not chunk: + raise SocketClosed() + self._rest += chunk + out, self._rest = self._rest[:n], self._rest[n:] + return out + + def _reader(self): + self.sock.settimeout(None) + try: + while True: + b1, b2 = self._recvn(2) + n = b2 & 0x7F + if n == 126: + n = struct.unpack(">H", self._recvn(2))[0] + elif n == 127: + n = struct.unpack(">Q", self._recvn(8))[0] + data = self._recvn(n) + opcode = b1 & 0x0F + if opcode == 1: + text = data.decode() + self.raw.append(text) + self.frames.put(json.loads(text)) + elif opcode == 8: + break + except (OSError, SocketClosed, ValueError): + pass + self.closed.set() + + def send_text(self, text): + """A client frame must be masked.""" + payload = text.encode() + mask = os.urandom(4) + header = bytes([0x81]) + n = len(payload) + if n < 126: + header += bytes([0x80 | n]) + elif n < 65536: + header += bytes([0x80 | 126]) + struct.pack(">H", n) + else: + header += bytes([0x80 | 127]) + struct.pack(">Q", n) + masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + self.sock.sendall(header + mask + masked) + + def send(self, type_number, d=None, with_id=False): + frame = {"v": PROTO_VERSION, "t": type_number} + msg_id = None + if with_id: + msg_id = self._next_id + self._next_id += 1 + frame["id"] = msg_id + if d is not None: + frame["d"] = d + self.send_text(json.dumps(frame, separators=(",", ":"))) + return msg_id + + def recv(self, timeout=10): + try: + return self.frames.get(timeout=timeout) + except queue.Empty: + raise TimeoutError("no frame in %ss" % timeout) + + def recv_until(self, predicate, timeout=10): + """Frames up to and including the first one predicate accepts.""" + end = time.time() + timeout + seen = [] + while True: + left = end - time.time() + if left <= 0: + raise TimeoutError("not seen in %ss; got types %s" % (timeout, [f.get("t") for f in seen])) + f = self.recv(left) + seen.append(f) + if predicate(f): + return seen + + def close(self): + try: + self.sock.sendall(bytes([0x88, 0x80]) + os.urandom(4)) + except OSError: + pass + self.sock.close() + + +def run_sync(client, schema=None, timeout=20, hello=True): + """Takes hello (unless hello=False, for a resync) and a whole sync, + sending next after each sync frame. + + Returns (hello, sync frames, events seen meanwhile). Fills schema, and + its hash, from the sync when it carries one. + """ + first = client.recv(timeout) if hello else None + if hello: + assert first["t"] == 1, first + if schema is not None: + schema.hash = first["d"]["sh"] + sync_frames, events = [], [] + end = time.time() + timeout + while True: + f = client.recv(max(0.1, end - time.time())) + if f["t"] == 4: + sync_frames.append(f) + if schema is not None: + schema.add(f["d"]) + client.send(3) + elif f["t"] == 5: + return first, sync_frames, events + else: + events.append(f) + + +def http_json(host, path, port=80, method="GET", body=None, headers=None, timeout=20): + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request(f"http://{host}:{port}{path}", data=data, method=method, headers=headers or {}) + if data is not None: + req.add_header("Content-Type", "application/json") + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.loads(r.read()) + + +def heap(host): + """free, min and uptime from /discovery on 8081 (protocol 1, until it goes).""" + m = http_json(host, "/discovery", port=8081)["memory"] + return {"free": m["free"], "min": m["min"], "max": m["max"], "uptime": m["uptime"]} diff --git a/test/e2e/test_auth.py b/test/e2e/test_auth.py new file mode 100644 index 0000000..41b8f00 --- /dev/null +++ b/test/e2e/test_auth.py @@ -0,0 +1,38 @@ +"""The socket under the device's security mode (docs/adr/0006-two-security-modes-eleven-http-routes.md). + +The suite never changes the device's security itself, so it tests the mode it +finds: pass --password or --pin for a Secured device. +""" +import pytest + +from proto_client import ProtoClient + + +def test_none_lets_anyone_connect(host, secured): + if secured: + pytest.skip("the device is Secured") + c = ProtoClient(host) + assert c.status == 101 + c.close() + + +def test_secured_refuses_a_socket_without_a_token(host, secured): + if not secured: + pytest.skip("the device's security is None") + c = ProtoClient(host) + assert c.status == 401 + + +def test_secured_refuses_a_wrong_token(host, secured): + if not secured: + pytest.skip("the device's security is None") + c = ProtoClient(host, params={"apikey": "0" * 64}) + assert c.status == 401 + + +def test_secured_accepts_the_token(host, secured, token): + if not secured: + pytest.skip("the device's security is None") + c = ProtoClient(host, params={"apikey": token}) + assert c.status == 101 + c.close() diff --git a/test/e2e/test_messages.py b/test/e2e/test_messages.py new file mode 100644 index 0000000..9179e39 --- /dev/null +++ b/test/e2e/test_messages.py @@ -0,0 +1,60 @@ +"""What the device answers to client frames: errors and feed subscriptions.""" +from proto_client import run_sync + + +def synced(connect, schema): + c = connect(schema=schema.hash) + run_sync(c) + return c + + +def reply_to(c, msg_id, timeout=5): + return c.recv_until(lambda f: f["t"] == 2 and f.get("id") == msg_id, timeout)[-1] + + +def error_name(schema, reply): + return schema.errors[reply["d"][schema.keys["code"]]] + + +def test_malformed_json_is_a_bad_request(connect, schema): + c = synced(connect, schema) + c.send_text("not json") + reply = c.recv_until(lambda f: f["t"] == 2)[-1] + assert error_name(schema, reply) == "badRequest" + + +def test_a_wrong_version_is_a_bad_request(connect, schema): + c = synced(connect, schema) + c.send_text('{"v":1,"t":3,"id":7}') + assert error_name(schema, reply_to(c, 7)) == "badRequest" + + +def test_an_event_type_from_a_client_is_unknown(connect, schema): + c = synced(connect, schema) + msg_id = c.send(schema.type_numbers["shadeState"], {}, with_id=True) + assert error_name(schema, reply_to(c, msg_id)) == "unknownOp" + + +def test_subscribing_to_a_feed_is_ok(connect, schema): + c = synced(connect, schema) + for feed in ("frames", "frequencyScan"): + msg_id = c.send(schema.type_numbers["subscribe"], {schema.keys["feed"]: schema.feed_numbers[feed]}, with_id=True) + reply = reply_to(c, msg_id) + assert reply["d"].get(schema.keys["ok"]) == 1 + msg_id = c.send(schema.type_numbers["unsubscribe"], {schema.keys["feed"]: schema.feed_numbers[feed]}, with_id=True) + assert reply_to(c, msg_id)["d"].get(schema.keys["ok"]) == 1 + + +def test_an_unknown_feed_is_a_bad_request(connect, schema): + c = synced(connect, schema) + msg_id = c.send(schema.type_numbers["subscribe"], {schema.keys["feed"]: 99}, with_id=True) + assert error_name(schema, reply_to(c, msg_id)) == "badRequest" + + +def test_memory_status_arrives_within_the_status_period(connect, schema): + """The device sends its status every 30 s.""" + c = synced(connect, schema) + f = c.recv_until(lambda f: f["t"] == schema.type_numbers["memory"], timeout=40)[-1] + memory = schema.expand(f["d"]) + assert memory["freeHeap"] > 0 + assert memory["uptime"] > 0 diff --git a/test/e2e/test_survival.py b/test/e2e/test_survival.py new file mode 100644 index 0000000..357c6d1 --- /dev/null +++ b/test/e2e/test_survival.py @@ -0,0 +1,68 @@ +"""The device must survive the loads that ran its heap out +(docs/adr/0005-one-websocket-protocol.md): every client reconnecting at once, +and a browser loading the page.""" +import threading +import urllib.request + +import pytest + +from proto_client import ProtoClient, heap, run_sync + +# What the heap may drop to under load. Below it the next allocation in the +# web server is close to failing, which aborts. +HEAP_FLOOR = 20 * 1024 +MAX_CLIENTS = 5 + + +@pytest.fixture +def watch_heap(host): + """Fails the test when the device rebooted or its heap fell under the floor.""" + before = heap(host) + yield before + after = heap(host) + assert after["uptime"] > before["uptime"], "the device rebooted" + assert after["min"] >= HEAP_FLOOR or after["min"] == before["min"], ( + "the heap fell to %u bytes (floor %u)" % (after["min"], HEAP_FLOOR)) + + +def test_every_client_reconnecting_at_once(connect, schema, watch_heap): + clients = [connect() for _ in range(MAX_CLIENTS)] + errors = [] + + def sync(c): + try: + run_sync(c, timeout=60) + except Exception as e: # reported below + errors.append(e) + threads = [threading.Thread(target=sync, args=(c,)) for c in clients] + [t.start() for t in threads] + [t.join() for t in threads] + assert not errors, errors + + +def test_one_client_too_many_is_closed(connect, schema): + clients = [connect(schema=schema.hash) for _ in range(MAX_CLIENTS)] + for c in clients: + run_sync(c) + extra = connect(schema=schema.hash) + assert extra.closed.wait(5), "a sixth client stayed connected" + + +def test_page_load(host, token, connect, watch_heap): + """The static files in parallel, as a browser fetches them, then the socket.""" + files = ["/", "/main.css", "/widgets.css", "/icons.css", "/index.js", "/icon.png"] + errors = [] + + def get(path): + try: + req = urllib.request.Request(f"http://{host}{path}", headers={"Accept-Encoding": "gzip"}) + with urllib.request.urlopen(req, timeout=60) as r: + r.read() + except Exception as e: # reported below + errors.append((path, e)) + threads = [threading.Thread(target=get, args=(p,)) for p in files] + [t.start() for t in threads] + c = connect() + run_sync(c, timeout=60) + [t.join() for t in threads] + assert not errors, errors diff --git a/test/e2e/test_sync.py b/test/e2e/test_sync.py new file mode 100644 index 0000000..5ce6841 --- /dev/null +++ b/test/e2e/test_sync.py @@ -0,0 +1,124 @@ +"""hello, the schema and the sync (docs/PROTOCOL.md).""" +import json +import os + +from proto_client import Schema, http_json, run_sync + +FIXTURES = os.path.join(os.path.dirname(__file__), "..", "fixtures", "proto2") + + +def items(frames, schema, section): + key = schema.keys[section] + out = [] + for f in frames: + v = f["d"].get(key) + if isinstance(v, list): + out.extend(v) + elif v is not None: + out.append(v) + return out + + +def test_hello_names_the_firmware_and_schema(connect): + c = connect() + assert c.status == 101 + hello = c.recv() + assert hello["v"] == 2 + assert hello["t"] == 1 + assert hello["d"]["fw"] + assert len(hello["d"]["sh"]) == 8 + + +def test_first_sync_carries_the_schema(connect): + c = connect() + schema = Schema() + hello, frames, _ = run_sync(c, schema) + assert schema.complete() + assert schema.type_numbers["hello"] == 1 + assert schema.keys["shadeId"] == "s" + for f in frames: + assert len(json.dumps(f, separators=(",", ":"))) < 4096 + + +def test_a_client_with_the_schema_gets_none(connect, schema): + c = connect(schema=schema.hash) + _, frames, _ = run_sync(c) + for f in frames: + for k in ("T", "F", "E", "Q"): + assert k not in f["d"] + + +def test_an_old_schema_hash_gets_the_schema_again(connect): + c = connect(schema="00000000") + schema = Schema() + run_sync(c, schema) + assert schema.complete() + + +def test_sync_holds_the_same_model_as_controller(connect, schema, host, token): + c = connect(schema=schema.hash) + _, frames, _ = run_sync(c) + headers = {"apikey": token} if token else {} + controller = http_json(host, "/controller", headers=headers) + shades = [schema.expand(s) for s in items(frames, schema, "shades")] + groups = [schema.expand(g) for g in items(frames, schema, "groups")] + rooms = items(frames, schema, "rooms") + assert sorted(s["shadeId"] for s in shades) == sorted(s["shadeId"] for s in controller["shades"]) + assert sorted(g["groupId"] for g in groups) == sorted(g["groupId"] for g in controller["groups"]) + assert len(rooms) == len(controller["rooms"]) + by_id = {s["shadeId"]: s for s in controller["shades"]} + for s in shades: + assert s.get("name", "") == by_id[s["shadeId"]]["name"] + assert s.get("remoteAddress", 0) == by_id[s["shadeId"]]["remoteAddress"] + for g in groups: + linked = sorted(x["shadeId"] for x in next(cg for cg in controller["groups"] if cg["groupId"] == g["groupId"])["linkedShades"]) + assert sorted(g.get("linkedShades", [])) == linked + assert len(items(frames, schema, "header")) == 1 + assert len(items(frames, schema, "status")) == 1 + + +def test_sync_waits_for_next(connect): + c = connect() + c.recv() # hello + first = c.recv() + assert first["t"] == 4 + # Without next nothing more comes, unless the whole sync fit one frame. + try: + more = c.recv(timeout=2) + except TimeoutError: + return + assert more["t"] == 5, "a second sync frame came without next" + + +def test_resync_sends_it_all_again(connect, schema): + c = connect(schema=schema.hash) + _, first, _ = run_sync(c) + c.send(schema.type_numbers["sync"]) + _, again, _ = run_sync(c, hello=False) + assert len(items(again, schema, "shades")) == len(items(first, schema, "shades")) + for f in again: + assert "T" not in f["d"] + + +def test_record_fixtures(connect, recording): + """With --record, saves hello and a whole first sync for the other suites.""" + if not recording: + return + c = connect() + schema = Schema() + run_sync(c, schema) + os.makedirs(FIXTURES, exist_ok=True) + hello = json.loads(c.raw[0]) + with open(os.path.join(FIXTURES, "hello.json"), "w", encoding="utf-8") as f: + json.dump(hello, f, indent=1) + with open(os.path.join(FIXTURES, "sync.json"), "w", encoding="utf-8") as f: + json.dump([json.loads(r) for r in c.raw[1:]], f, indent=1) + expanded = { + "hash": hello["d"]["sh"], + "types": {str(k): v for k, v in sorted(schema.types.items())}, + "fields": {k: {"name": n, "kind": kind, "default": d} for k, (n, kind, d) in sorted(schema.fields.items())}, + "errors": {str(k): v for k, v in sorted(schema.errors.items())}, + "feeds": {str(k): v for k, v in sorted(schema.feeds.items())}, + } + with open(os.path.join(FIXTURES, "schema.json"), "w", encoding="utf-8") as f: + json.dump(expanded, f, indent=1)