172 lines
6.1 KiB
Python
172 lines
6.1 KiB
Python
|
#!/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", 0x5000)
|
|||
|
|
+ struct.pack(">H", 14) + bytes([9]) + b"\x00"
|
|||
|
|
+ struct.pack("<H", 0) + 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")
|