#!/usr/bin/env python3
"""End-to-end test of command surface v3 against a REAL unit.
python3 test/e2e_v3.py 192.168.50.89
Two clients (alice, bob) exercise grants, signed frees, timers, legacy pwm
skipping held slots, stopfade, state, and epochfloor. Values stay dim. The
lamp's colours are read at the start and written back at the end (red clamped
to the 45000 cap the firmware enforces on input).
Needs PyNaCl for signature checks.
"""
import socket
import sys
import threading
import time
from nacl.signing import VerifyKey
HOST = sys.argv[1] if len(sys.argv) > 1 else "192.168.50.89"
PORT = 1337
failures = []
def check(name, ok, detail=""):
print(("PASS " if ok else "FAIL ") + name + ("" if ok else " " + detail))
if not ok:
failures.append(name)
class Conn:
def __init__(self, label):
self.label = label
self.sock = socket.create_connection((HOST, PORT), timeout=5)
self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
self.lines = []
self.lock = threading.Lock()
self.alive = True
threading.Thread(target=self._read, daemon=True).start()
def _read(self):
buf = b""
while self.alive:
try:
d = self.sock.recv(4096)
except OSError:
break
if not d:
break
buf += d
while b"\n" in buf:
line, buf = buf.split(b"\n", 1)
line = line.decode(errors="replace").strip()
if line:
with self.lock:
self.lines.append((time.monotonic(), line))
def send(self, line):
self.sock.sendall((line + "\n").encode())
def mark(self):
with self.lock:
return len(self.lines)
def wait(self, pred, since=0, timeout=4.0):
end = time.monotonic() + timeout
while time.monotonic() < end:
with self.lock:
for _, line in self.lines[since:]:
if pred(line):
return line
time.sleep(0.02)
return None
def close(self):
# shutdown() first: a bare close() while the reader thread sits in recv()
# doesn't actually close the socket on Linux, so no FIN reaches the unit.
self.alive = False
try:
self.sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
self.sock.close()
def starts(prefix):
return lambda l: l.startswith(prefix)
def verify_free(line, unit, pk_hex):
# free <s> <epoch> <n> <b> <g> <r> <why> <holder> <sig>
p = line.split()
msg = "cthuloid/free/1 %s %s %s %s %s %s %s" % (unit, p[1], p[2], p[3], p[4], p[5], p[6])
try:
VerifyKey(bytes.fromhex(pk_hex)).verify(msg.encode(), bytes.fromhex(p[9]))
return True
except Exception:
return False
a = Conn("alice")
b = Conn("bob")
hello = a.wait(starts("hello "))
check("greeting line 1 unchanged", a.wait(lambda l: l.endswith(" ready")) is not None)
check("hello line carries proto 3", hello is not None and "proto=3" in hello, str(hello))
fields = dict(kv.split("=", 1) for kv in hello.split()[1:]) if hello else {}
UNIT, EPOCH, PK = fields.get("unit"), int(fields.get("epoch", "0")), fields.get("pk", "")
print("unit=%s epoch=%d pk=%s…" % (UNIT, EPOCH, PK[:16]))
# Remember the look to put back afterwards.
m = a.mark()
a.send("state")
st = a.wait(starts("state "), m)
before = [int(x) for x in st.split()[6:]] if st else None
check("state reply", before is not None and len(before) == 12, str(st))
stats = a.wait(starts("stats "), m)
check("state ends with stats (epoch, pk)", stats is not None and ("epoch=%d" % EPOCH) in stats and PK in stats, str(stats))
# --- first come, first served -------------------------------------------------
ma, mb = a.mark(), b.mark()
a.send("grant 0 alice")
g = a.wait(starts("granted 0 "), ma)
check("alice granted slot 0", g is not None, str(g))
n0 = int(g.split()[3]) if g else 0
check("bob sees held broadcast", b.wait(lambda l: l == "held 0 alice %d %d" % (EPOCH, n0), mb) is not None)
mb = b.mark()
b.send("grant 0 bob")
check("bob refused: busy", b.wait(lambda l: l == "busy 0 alice", mb) is not None)
b.send("slot 0 %d %d 1 1 1" % (EPOCH, n0))
check("bob can't write alice's slot", b.wait(lambda l: l == "ERR notheld 0", mb) is not None)
# --- holder writes; legacy pwm skips the held slot -----------------------------
ma, mb = a.mark(), b.mark()
a.send("slot 0 %d %d 3000 0 0" % (EPOCH, n0))
b.send("pwm 1 1 1 2000 0 0 0 0 0 0 0 0")
vals = b.wait(lambda l: l.startswith("values ") and l.split()[1:4] == ["3000", "0", "0"]
and l.split()[4] == "2000", mb)
check("values broadcast: holder's slot kept, legacy applied elsewhere", vals is not None, str(vals))
check("slot write is silent on success", a.wait(lambda l: l.startswith("ERR"), ma, timeout=0.5) is None)
# --- release -> signed free --------------------------------------------------------
mb = b.mark()
a.send("release 0")
fr = b.wait(starts("free 0 "), mb)
check("release broadcasts free with final value", fr is not None and fr.split()[4:9] == ["3000", "0", "0", "release", "alice"], str(fr))
check("free signature verifies", fr is not None and verify_free(fr, UNIT, PK))
if fr:
bad = fr.split()
bad[4] = "3001"
check("tampered free fails verification", not verify_free(" ".join(bad), UNIT, PK))
# --- idle expiry (2 s) ---------------------------------------------------------------
ma, mb = a.mark(), b.mark()
a.send("grant 1 alice")
g1 = a.wait(starts("granted 1 "), ma)
t0 = time.monotonic()
fr = b.wait(lambda l: l.startswith("free 1 ") and " idle " in l, mb, timeout=4)
dt = time.monotonic() - t0
check("idle grant freed after ~2 s", fr is not None and 1.8 < dt < 2.8, "dt=%.2f %s" % (dt, fr))
# --- disconnect frees ------------------------------------------------------------------
c = Conn("carol")
c.wait(starts("hello "))
mb = b.mark()
c.send("grant 2 carol")
b.wait(starts("held 2 carol"), mb)
c.close()
t0 = time.monotonic()
fr = b.wait(lambda l: l.startswith("free 2 ") and " disconnect " in l, mb, timeout=6)
dt = time.monotonic() - t0
check("holder disconnect frees its slot at once (not via idle)", fr is not None and dt < 0.5, "dt=%.2f %s" % (dt, fr))
# --- grant all is atomic; stopfade frees everything --------------------------------------
ma, mb = a.mark(), b.mark()
b.send("grant 3 bob")
b.wait(starts("granted 3 "), mb)
a.send("grant all alice")
check("grant all refused while bob holds slot 3", a.wait(lambda l: l == "busy 3 bob", ma) is not None)
mb = b.mark()
a.send("stopfade")
check("stopfade frees bob's slot (blackout)", b.wait(lambda l: l.startswith("free 3 ") and " blackout " in l, mb) is not None)
ma = a.mark()
a.send("grant all alice")
granted = [a.wait(starts("granted %d " % s), ma) for s in range(4)]
check("grant all succeeds when free", all(granted), str(granted))
ma = a.mark()
a.send("state")
helds = [a.wait(starts("held %d alice" % s), ma) for s in range(4)]
check("state lists every held slot", all(helds))
a.send("release all")
# --- epochfloor -------------------------------------------------------------------------
ma = a.mark()
a.send("epochfloor %d" % (EPOCH + 5))
ep = a.wait(starts("epoch "), ma)
check("epochfloor raises the epoch past the floor", ep == "epoch %d" % (EPOCH + 6), str(ep))
ma = a.mark()
a.send("grant 0 alice")
g = a.wait(starts("granted 0 "), ma)
check("new grants use the new epoch, numbering restarted", g == "granted 0 %d 1" % (EPOCH + 6), str(g))
a.send("release 0")
ma = a.mark()
a.send("epochfloor 1")
check("epochfloor never lowers", a.wait(starts("epoch "), ma) == "epoch %d" % (EPOCH + 6))
# --- put the look back ----------------------------------------------------------------
time.sleep(0.3)
if before:
restore = [min(v, 45000) if i % 3 == 2 else v for i, v in enumerate(before)]
a.send("pwm " + " ".join(str(v) for v in restore))
time.sleep(0.5)
a.close()
b.close()
print("\n%d failure(s)" % len(failures) if failures else "\nALL PASS")
sys.exit(1 if failures else 0)