hms-mediaengine/tests/e2e_app_test.py

#!/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", 0x5000)
       + struct.pack(">H", 14) + bytes([1]) + 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")
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", 0x5000)
        + struct.pack(">H", 14) + bytes([2]) + b"\x00"
        + struct.pack("<H", 0) + 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")