cthuloid / blender-addon / tests / test_net_isolation.py

"""A dead fixture must not delay a live one.

Run: python3 blender-addon/tests/test_net_isolation.py

net.py imports nothing from Blender, so it is exercised directly. A local TCP
server stands in for a live fixture; a .local name that does not exist stands
in for a powered-off one. Resolving such a name blocks getaddrinfo for ~5 s
(nss-mdns timeout), which is exactly what a real rig with unplugged units does.
"""

import importlib.util
import os
import socket
import sys
import threading
import time

HERE = os.path.dirname(os.path.abspath(__file__))
NET_PY = os.path.join(HERE, "..", "cthuloid_lights", "net.py")
sys.path.insert(0, os.path.dirname(NET_PY))   # net.py imports protocol

spec = importlib.util.spec_from_file_location("cthuloid_net", NET_PY)
net = importlib.util.module_from_spec(spec)
spec.loader.exec_module(net)

DEAD_HOST = "cthuloid-test-nonexistent.local"
BUDGET_S = 0.5      # a live fixture must see each update within this


class FakeFixture:
    """Accepts connections, greets like the firmware, records each line's arrival."""

    def __init__(self):
        self.srv = socket.socket()
        self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.srv.bind(("127.0.0.1", 0))
        self.srv.listen(4)
        self.port = self.srv.getsockname()[1]
        self.lines = []          # (monotonic arrival, line)
        self.lock = threading.Lock()
        threading.Thread(target=self._accept, daemon=True).start()

    def _accept(self):
        while True:
            try:
                c, _ = self.srv.accept()
            except OSError:
                return
            c.sendall(b"fake ready\r\n")
            threading.Thread(target=self._read, args=(c,), daemon=True).start()

    def _read(self, c):
        buf = b""
        while True:
            try:
                d = c.recv(4096)
            except OSError:
                return
            if not d:
                return
            buf += d
            while b"\n" in buf:
                line, buf = buf.split(b"\n", 1)
                with self.lock:
                    self.lines.append((time.monotonic(), line.decode()))

    def wait_for(self, text, timeout):
        end = time.monotonic() + timeout
        while time.monotonic() < end:
            with self.lock:
                for t, line in self.lines:
                    if line == text:
                        return t
            time.sleep(0.01)
        return None


def main():
    fx = FakeFixture()
    s = net.Sender()
    s.min_interval = 0.0
    s.start()
    failures = []
    try:
        # Bring the live fixture up first so its connect is not what we time.
        s.send("127.0.0.1", fx.port, "pwm 0\n")
        if fx.wait_for("pwm 0", 3.0) is None:
            print("FAIL: live fixture never received its first frame")
            return 1

        # Now a dead fixture joins, as an unplugged unit in the scene does.
        s.send(DEAD_HOST, 1337, "pwm 1\n")
        time.sleep(0.3)     # let the worker start resolving it

        for i in range(2, 7):
            payload = "pwm %d" % i
            t0 = time.monotonic()
            s.send("127.0.0.1", fx.port, payload + "\n")
            t1 = fx.wait_for(payload, 15.0)
            lat = None if t1 is None else t1 - t0
            print("update %d -> live fixture: %s" % (
                i, "never" if lat is None else "%.3f s" % lat))
            if lat is None or lat > BUDGET_S:
                failures.append(i)
            time.sleep(0.5)
    finally:
        s.stop()

    if failures:
        print("FAIL: %d/5 updates exceeded %.1f s" % (len(failures), BUDGET_S))
        return 1
    print("PASS: every update reached the live fixture within %.1f s" % BUDGET_S)
    return 0


if __name__ == "__main__":
    sys.exit(main())