#!/usr/bin/env python3 """E2E-Test: Upload, Thumbnails, Layer, konfigurierbares Art-Net, Persistenz. 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, raw=None, headers=None): data = raw if raw is not None else json.dumps(body or {}).encode() req = urllib.request.Request(BASE + path, data=data, method="POST") req.add_header("Content-Type", "application/json" if raw is None else "application/octet-stream") for k, v in (headers or {}).items(): req.add_header(k, v) 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) Health h = get("/api/health") check("Health", h.get("status") == "ok") # 2) Testdateien erzeugen (Video + PNG via GStreamer) 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=30 pattern=18 ! " "video/x-raw,width=320,height=180 ! x264enc tune=zerolatency ! " "mp4mux ! filesink location=/tmp/e2e_clip.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" "p2 = Gst.parse_launch('videotestsrc num-buffers=1 pattern=3 ! " "video/x-raw,width=320,height=180 ! pngenc ! " "filesink location=/tmp/e2e_bild.png')\n" "p2.set_state(Gst.State.PLAYING)\n" "p2.get_bus().timed_pop_filtered(Gst.CLOCK_TIME_NONE, " "Gst.MessageType.EOS)\n" "p2.set_state(Gst.State.NULL)\n", ], check=True) # 3) Upload Video ueber HTTP (wie die Web-UI) video = open("/tmp/e2e_clip.mp4", "rb").read() r = post("/api/media/upload", raw=video, headers={"X-Filename": urllib.parse.quote("e2e_clip.mp4")}) check("Upload Video", bool(r.get("saved")), r) # 4) Upload Bild bild = open("/tmp/e2e_bild.png", "rb").read() r = post("/api/media/upload", raw=bild, headers={"X-Filename": urllib.parse.quote("e2e_bild.png")}) check("Upload Bild", bool(r.get("saved")), r) # 5) Bibliothek + Metadaten lib = get("/api/media")["items"] names = [m["name"] for m in lib] check("Bibliothek beide Dateien", {"e2e_clip.mp4", "e2e_bild.png"} <= set(names), names) clip = next(m for m in lib if m["name"] == "e2e_clip.mp4") check("Video-Dauer erkannt", clip.get("duration_s") is not None, "dauer=%ss" % clip.get("duration_s")) check("Bild als image", next(m for m in lib if m["name"] == "e2e_bild.png")["type"] == "image") # 6) Thumbnail abrufbar (echtes JPEG) req = urllib.request.Request(BASE + "/thumb/e2e_clip.mp4") with urllib.request.urlopen(req, timeout=15) as resp: thumb = resp.read() check("Thumbnail JPEG", thumb[:2] == b"\xff\xd8", "%d bytes" % len(thumb)) # 7) Pfad-Traversal abgewiesen try: urllib.request.urlopen(BASE + "/media/..%2Fconfig.json", timeout=5) check("Traversal-Schutz", False) except urllib.error.HTTPError as e: check("Traversal-Schutz", e.code == 404) # 8) Layer dynamisch: Video + Bild r = post("/api/layers/add", {"name": "e2e_clip.mp4"}) check("Layer #1 (Video)", "layer" in r, r) r = post("/api/layers/add", {"name": "e2e_bild.png"}) check("Layer #2 (Bild)", "layer" in r, r) time.sleep(3) st = get("/api/status") check("Engine rendert", st["frames_rendered"] > 0, "frames=%d layer=%d" % (st["frames_rendered"], len(st["layers"]))) # 9) Art-Net UMBERSTELLEN: Port 6455, Master-Kanal 10 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, "signal_loss_policy": "hold", "signal_loss_timeout_s": 10}}) check("Settings gespeichert", bool(r.get("saved")), r) time.sleep(2) # 10) DMX an NEUEN Port: Kanal 10 = 64 => Master ~0.25 dmx = bytearray(512) dmx[9] = 64 pkt = (b"Art-Net\x00" + struct.pack("H", 14) + bytes([1]) + 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") check("DMX auf neuem Port + Kanal", 0.2 < st["master"] < 0.3, "master=%s accepted=%s" % (st["master"], st["dmx"]["accepted"])) # 11) Kanal 11 (Layer 2 ab Start-Kanal 2) = 255 => Layer-Alpha dmx2 = bytearray(512) dmx2[9] = 255 dmx2[10] = 255 pkt2 = (b"Art-Net\x00" + struct.pack("H", 14) + bytes([2]) + b"\x00" + struct.pack("H", 512) + bytes(dmx2)) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.sendto(pkt2, ("127.0.0.1", 6455)) s.close() time.sleep(1) st = get("/api/status") l2 = st["layers"][1] check("Layer-Alpha via DMX", l2["alpha"] > 0.9, "l2.alpha=%s" % l2["alpha"]) # 12) config.json persistiert die Umstellung cfg = json.load(open("config.json")) check("config.json Port 6455", cfg["artnet"]["listen_port"] == 6455) check("config.json Master-Kanal 10", cfg["artnet"]["master_channel"] == 10) # 13) Layer entfernen => Engine laeuft weiter r = post("/api/layers/remove", {"id": 1}) check("Layer entfernt", bool(r.get("removed")), r) time.sleep(2) st = get("/api/status") check("Engine nach Remove ok", st["running"] is True, "layer=%d frames=%d" % (len(st["layers"]), st["frames_rendered"])) # 14) Ungueltige Settings abgewiesen (Port 99999) r = post("/api/settings", {"artnet": {"listen_port": 99999}}) check("Ungueltige Settings abgewiesen", "error" in r, r) print("") print("ALLE E2E-TESTS BESTANDEN")