test: start the e2e suite for socket protocol 2

A pytest suite run by hand against a device (test/e2e/README.md): hello, the schema and its hash, the sync with next and resync, replies to bad frames, feed subscriptions, the status event, every client syncing at once, a sixth client refused, a page load, and the socket under the device's security mode. It never transmits, changes settings or reboots. --record saves hello, the sync and the schema as fixtures.

PROTOCOL.md now describes the protocol as built.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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.

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")

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"]}

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()

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

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

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)