"""The addon's sender against proto-3 units (shared control).
Run: python3 blender-addon/tests/test_net_v3.py
net.py and protocol.py import nothing from Blender, so they are exercised
directly against a fake unit that referees grants like the firmware.
"""
import importlib.util
import os
import sys
import time
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
from fake_unit3 import FakeUnit3, wait # noqa: E402
PKG = os.path.join(HERE, "..", "cthuloid_lights")
sys.path.insert(0, PKG) # net.py falls back to a top-level `import protocol`
import protocol # noqa: E402
import net # noqa: E402
failures = []
def check(name, ok, detail=""):
print(("PASS " if ok else "FAIL ") + name + ("" if ok else " " + str(detail)))
if not ok:
failures.append(name)
def frame(slots, fade=0):
"""slots: list of 4 (b, g, r) tuples."""
vals = []
for s in slots:
vals.extend(s)
return protocol.Frame(tuple(vals), fade)
DARK = [(0, 0, 0)] * 4
def sender():
s = net.Sender()
s.min_interval = 0.0
s.holder = "blender-test"
s.start()
return s
def test_grants_and_slot_frames():
fake = FakeUnit3()
s = sender()
s.send("127.0.0.1", fake.port, frame([(1, 2, 3), (4, 5, 6), (0, 0, 0), (0, 0, 0)]), force=True)
check("connect + force: every slot grabbed and set",
wait(lambda: fake.values[:6] == [1, 2, 3, 4, 5, 6] and len(fake.received("grant ")) == 4))
check("frames carry the grant", all(l.split()[2] == "5" for l in fake.received("slot ")))
n_before = len(fake.received("slot "))
s.send("127.0.0.1", fake.port, frame([(9, 9, 9), (4, 5, 6), (0, 0, 0), (0, 0, 0)]))
check("a later change sends only the slot that changed",
wait(lambda: fake.values[:3] == [9, 9, 9]) and
[l.split()[1] for l in fake.received("slot ")[n_before:]] == ["0"],
fake.received("slot ")[n_before:])
check("no whole-unit pwm on a proto-3 unit", not fake.received("pwm"))
s.stop()
fake.close()
def test_release_after_quiet():
fake = FakeUnit3()
s = sender()
s.send("127.0.0.1", fake.port, frame([(7, 7, 7)] + DARK[1:]), force=True)
wait(lambda: fake.values[:3] == [7, 7, 7])
t0 = time.monotonic()
ok = wait(lambda: len(fake.received("release ")) >= 4, timeout=4)
dt = time.monotonic() - t0
check("held slots are let go after ~1.5 s without change (inside the unit's 2 s idle)",
ok and 1.2 < dt < 2.2, "dt=%.2f %s" % (dt, fake.received("release ")))
s.stop()
fake.close()
def test_busy_slot_is_skipped_and_reported():
fake = FakeUnit3()
web = fake.external()
web.sendall(b"grant 1 alice\n")
wait(lambda: fake.holds[1] is not None)
s = sender()
s.send("127.0.0.1", fake.port, frame([(1, 1, 1), (2, 2, 2), (3, 3, 3), (4, 4, 4)]), force=True)
check("other slots still set", wait(lambda: fake.values[6:9] == [3, 3, 3]))
check("alice's slot untouched", fake.values[3:6] == [0, 0, 0])
status, note = s.status_of("127.0.0.1", fake.port)
check("fixture reports who holds it", status == "connected" and "held by alice" in note, (status, note))
# Alice lets go after setting her own colour: Blender must NOT stomp on it
# until Blender's own scene changes.
web.sendall(b"slot 1 5 1 50 60 70 0\nrelease 1\n")
wait(lambda: fake.holds[1] is None)
time.sleep(0.6)
check("after alice lets go, her colour stays (Blender only sends on its own change)",
fake.values[3:6] == [50, 60, 70], fake.values[3:6])
s.send("127.0.0.1", fake.port, frame([(1, 1, 1), (8, 8, 8), (3, 3, 3), (4, 4, 4)]))
check("Blender's next change to that slot applies", wait(lambda: fake.values[3:6] == [8, 8, 8]))
check("note cleared once nothing is held by others", "held by" not in s.status_of("127.0.0.1", fake.port)[1])
web.close()
s.stop()
fake.close()
def test_proto2_unchanged():
fake = FakeUnit3(proto=2)
s = sender()
s.send("127.0.0.1", fake.port, frame([(1, 2, 3)] + DARK[1:]), force=True)
check("proto-2 unit gets a whole-unit pwm, no grants",
wait(lambda: fake.values[:3] == [1, 2, 3]) and not fake.received("grant") and fake.received("pwm 1 2 3"))
s.send("127.0.0.1", fake.port, frame([(4, 5, 6)] + DARK[1:], fade=500))
check("fade becomes pwm2", wait(lambda: any(l.startswith("pwm2 1 500 4 5 6") for _, l in fake.lines)))
s.stop()
fake.close()
def test_blackout_frees_and_regrabs():
fake = FakeUnit3()
s = sender()
s.send("127.0.0.1", fake.port, frame([(5, 5, 5)] + DARK[1:]), force=True)
wait(lambda: fake.values[:3] == [5, 5, 5])
s.send("127.0.0.1", fake.port, protocol.blackout_command(), force=True)
check("blackout reaches the unit", wait(lambda: fake.values[:3] == [0, 0, 0] and fake.holds == [None] * 4))
grants = len(fake.received("grant "))
s.send("127.0.0.1", fake.port, frame([(6, 6, 6)] + DARK[1:]))
check("next change grabs again and applies",
wait(lambda: fake.values[:3] == [6, 6, 6]) and len(fake.received("grant ")) > grants and
not fake.received("ERR"), fake.received("ERR"))
s.stop()
fake.close()
for t in (test_grants_and_slot_frames, test_release_after_quiet, test_busy_slot_is_skipped_and_reported,
test_proto2_unchanged, test_blackout_frees_and_regrabs):
t()
print("\n%d failure(s)" % len(failures) if failures else "\nALL PASS")
sys.exit(1 if failures else 0)