#!/usr/bin/env python3 """E2E-Test: Effekte (FX1/FX2) in der Live-Pipeline. Voraussetzung: Server laeuft auf http://localhost:8093 """ import json import socket import struct import subprocess import sys import time import urllib.error import urllib.parse import urllib.request BASE = "http://localhost:8093" def get(path): with urllib.request.urlopen(BASE + path, timeout=15) as r: return json.loads(r.read().decode()) def post(path, body=None): data = json.dumps(body or {}).encode() req = urllib.request.Request(BASE + path, data=data, method="POST") req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=120) as r: return json.loads(r.read().decode()) except urllib.error.HTTPError as e: return json.loads(e.read().decode()) def check(label, ok, detail=""): print(("PASS" if ok else "FAIL") + " " + label + " " + str(detail)) if not ok: sys.exit(1) # 1) Testvideo erzeugen falls noetig if not get("/api/health").get("status") == "ok": check("Health", False) print("PASS Health") # 2) FX-Katalog abrufbar und gefuellt fx = get("/api/fx") check("FX-Katalog >= 20 Effekte", len(fx) >= 20, "%d Effekte" % len(fx)) fx_ids = [f["id"] for f in fx] for must in ("blur", "sharpen", "color", "gamma", "pixelate", "vertigo"): check("FX enthalten: " + must, must in fx_ids) # 3) Medien pruefen/bereitstellen (Upload falls fehlt) lib = get("/api/media")["items"] names = [m["name"] for m in lib] if "e2e_clip.mp4" not in names: subprocess.run([ "/usr/bin/python3", "-c", "import gi; gi.require_version('Gst','1.0')\n" "from gi.repository import Gst; Gst.init(None)\n" "p = Gst.parse_launch('videotestsrc num-buffers=60 pattern=18 ! " "video/x-raw,width=320,height=180 ! x264enc tune=zerolatency ! " "mp4mux ! filesink location=/tmp/e2e_fx.mp4')\n" "p.set_state(Gst.State.PLAYING)\n" "p.get_bus().timed_pop_filtered(Gst.CLOCK_TIME_NONE, " "Gst.MessageType.EOS)\n" "p.set_state(Gst.State.NULL)\n", ], check=True) data = open("/tmp/e2e_fx.mp4", "rb").read() req = urllib.request.Request(BASE + "/api/media/upload", data=data, method="POST") req.add_header("Content-Type", "application/octet-stream") req.add_header("X-Filename", urllib.parse.quote("e2e_clip.mp4")) with urllib.request.urlopen(req, timeout=120) as r: json.loads(r.read().decode()) # 4) Layer mit Video st = get("/api/status") if not st["layers"]: r = post("/api/layers/add", {"name": "e2e_clip.mp4"}) check("Layer hinzugefuegt", "layer" in r, r) else: print("PASS Layer hinzugefuegt (bereits aktiv)") time.sleep(2) # 5) FX1 = blur setzen -> Struktur-Rebuild r = post("/api/layers/update", {"id": 1, "fx1": "blur"}) check("FX1 blur gesetzt", r.get("layer", {}).get("fx1") == "blur", r) time.sleep(2) st = get("/api/status") check("Engine laeuft mit FX1", st["running"] is True and st["error"] is None, "frames=%d err=%s" % (st["frames_rendered"], st["error"])) # 6) FX1-Intensitaet live aendern (KEIN Rebuild – Property-Update) frames_before = get("/api/status")["frames_rendered"] r = post("/api/layers/update", {"id": 1, "fx1_intensity": 1.0}) check("FX1-Intensitaet live", r.get("layer", {}).get("fx1_intensity") == 1.0, r) time.sleep(1.5) frames_after = get("/api/status")["frames_rendered"] check("Frames laufen weiter (live)", frames_after > frames_before, "%d -> %d" % (frames_before, frames_after)) # 7) FX2 = color dazu -> Rebuild mit 2 Effekten r = post("/api/layers/update", {"id": 1, "fx2": "color"}) check("FX2 color gesetzt", r.get("layer", {}).get("fx2") == "color", r) time.sleep(2) st = get("/api/status") check("Engine laeuft mit FX1+FX2", st["running"] is True, "frames=%d err=%s" % (st["frames_rendered"], st["error"])) # 8) Bild-Layer mit Pixelate + Vertigo (Bild-Effekt-Pfad) lib = get("/api/media")["items"] if any(m["name"] == "e2e_bild.png" for m in lib): r = post("/api/layers/add", {"name": "e2e_bild.png"}) if "layer" in r: lid = r["layer"]["id"] r2 = post("/api/layers/update", {"id": lid, "fx1": "pixelate"}) check("Bild-Layer mit Pixelate", r2.get("layer", {}).get("fx1") == "pixelate", r2) time.sleep(2) st = get("/api/status") check("Engine laeuft (Bild+FX)", st["running"] is True, "frames=%d" % st["frames_rendered"]) else: print("PASS Bild-Layer mit Pixelate (Layer-Limit, uebersprungen)") # 9) Effekt entfernen (fx1 = null) r = post("/api/layers/update", {"id": 1, "fx1": None}) check("FX1 entfernt", r.get("layer", {}).get("fx1") is None, r) time.sleep(2) st = get("/api/status") check("Engine laeuft ohne FX1", st["running"] is True, "frames=%d err=%s" % (st["frames_rendered"], st["error"])) # 10) DMX steuert FX1-Intensitaet (Kanal konfigurieren + Paket senden) r = post("/api/settings", {"artnet": { "enabled": True, "listen_port": 6455, "universes": [0], "master_channel": 10, "layer_start_channel": 2, "playback_channel": 6, "retrigger_channel": 7, "blackout_channel": 8, "fx1_channel": 20, "fx2_channel": 21, "signal_loss_policy": "hold", "signal_loss_timeout_s": 10}}) check("fx1_channel konfiguriert", bool(r.get("saved")), r) post("/api/layers/update", {"id": 1, "fx1": "blur"}) time.sleep(2) dmx = bytearray(512) dmx[9] = 255 # Master-Kanal 10 dmx[19] = 255 # fx1_channel 20 -> Intensitaet 1.0 dmx[20] = 128 # fx2_channel 21 -> Intensitaet ~0.5 pkt = (b"Art-Net\x00" + struct.pack("H", 14) + bytes([9]) + b"\x00" + struct.pack("H", 512) + bytes(dmx)) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.sendto(pkt, ("127.0.0.1", 6455)) s.close() time.sleep(1) st = get("/api/status") l1 = st["layers"][0] check("DMX steuert FX1-Intensitaet", l1["fx1_intensity"] == 1.0, "fx1_intensity=%s" % l1["fx1_intensity"]) # 11) Ungueltiger Effekt abgewiesen r = post("/api/layers/update", {"id": 1, "fx1": "gibt_es_nicht"}) check("Ungueltiger FX abgewiesen", r.get("layer", {}).get("fx1") is None, r) print("") print("ALLE FX-E2E-TESTS BESTANDEN")