cthuloid / blender-addon / tests / test_connect_switch.py

"""The Connect switch is the ONE state that decides whether Blender talks to lights.

Run: <blender> -b --factory-startup --python blender-addon/tests/test_connect_switch.py

Uses the real sender and worker threads. The rig's fixtures are repointed at
local fake fixtures, so no real lamp is touched; one is left at a .local name
that never resolves, as an unplugged unit would be.
"""

import socket
import sys
import threading
import time

import addon_utils
import bpy

MOD = "bl_ext.user_default.cthuloid_lights"
BLEND = "/home/fleeky/cthuloid/LightArm/blender/light-test.blend"
DEAD = "cthuloid-test-nonexistent.local"

addon_utils.enable(MOD, default_set=False)
net = sys.modules[MOD + ".net"]
S = net.SENDER


class Fake:
    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 = []
        self.open_conns = 0
        threading.Thread(target=self._accept, daemon=True).start()

    def _accept(self):
        while True:
            c, _ = self.srv.accept()
            self.open_conns += 1
            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:
                d = b""
            if not d:
                self.open_conns -= 1
                return
            buf += d
            while b"\n" in buf:
                line, buf = buf.split(b"\n", 1)
                self.lines.append(line.decode())


def wait(pred, timeout=3.0):
    end = time.monotonic() + timeout
    while time.monotonic() < end:
        if pred():
            return True
        time.sleep(0.02)
    return pred()


failures = []


def check(name, ok):
    print(("PASS " if ok else "FAIL ") + name)
    if not ok:
        failures.append(name)


bpy.ops.wm.open_mainfile(filepath=BLEND)
scene = bpy.context.scene
props = scene.cthuloid

# 1. The file was saved with Live on; opening it must NOT connect.
check("opens disconnected", props.live is False)
check("no connections after load", wait(lambda: not S._targets, 1.0))

fakes = []
fx = list(props.fixtures)
for f in fx[:-1]:
    fk = Fake()
    fakes.append(fk)
    f.host, f.port = "127.0.0.1", fk.port
fx[-1].host = DEAD                    # an unplugged unit
fx[1].enabled = False                 # an unticked unit
time.sleep(0.3)
check("still nothing sent while disconnected", all(not fk.lines for fk in fakes))

# 2. Connect: ticked fixtures connect and get the current scene immediately.
props.live = True
check("connect pushes to ticked fixtures",
      wait(lambda: fakes[0].lines and fakes[2].lines))
check("unticked fixture untouched", not fakes[1].lines)

# 6. Ticking a fixture while connected brings it up right away.
fx[1].enabled = True
check("ticking while connected pushes to it", wait(lambda: fakes[1].lines))

# 5. Resend / push settings are available only while connected.
check("send_now available while connected", bpy.ops.cthuloid.send_now.poll())

# 3. Disconnect closes every connection and stops every retry.
props.live = False
check("disconnect drops every target", wait(lambda: not S._targets))
check("disconnect closes sockets", wait(lambda: all(fk.open_conns == 0 for fk in fakes)))
check("send_now unavailable while disconnected", not bpy.ops.cthuloid.send_now.poll())
check("push_settings unavailable while disconnected", not bpy.ops.cthuloid.push_settings.poll())

# Edits while disconnected send nothing.
before = [len(fk.lines) for fk in fakes]
bpy.data.objects["bgr-1-studiolight.local"].data.color = (0.2, 0.4, 0.6)
bpy.context.view_layer.update()
time.sleep(0.4)
check("edits while disconnected send nothing", [len(fk.lines) for fk in fakes] == before)

# 4. Blackout works while disconnected, then lets go of the connection.
bpy.ops.cthuloid.blackout()
check("blackout reaches fixtures while disconnected",
      wait(lambda: any(l == "stopfade" for l in fakes[0].lines)))
check("blackout releases its connections", wait(lambda: not any(
    k[0] == "127.0.0.1" for k in S._targets), 3.0))

addon_utils.disable(MOD, default_set=False)
print("ALL PASS" if not failures else "FAILED: %s" % failures)
sys.exit(1 if failures else 0)