HMS MediaEngine Agent
362e089be0
Der Nutzer hat recht: Der Ordner war voller Entwicklungs-Muell. Jetzt ist sauber getrennt: ROOT (was der Nutzer sieht und braucht): - run.py = das Programm - hms_app/ = der Anwendungscode - HMS MediaEngine.app = macOS Doppelklick-Starter - HMS-Start.vbs = Windows Doppelklick-Starter - HMS-Install.vbs = Windows Erst-Installation - HMS-Mac-Install.command = macOS Homebrew-Installation - HMS-Portable-Install.command = macOS Portable-Installation (16GB-Fix) - installer_gui.py = grafischer Installer - launcher.pyw + launcher_core.py = interne Start-Logik - LIESMICH.txt = 10-Zeilen-Kurzanleitung - .gitignore _entwicklung/ (alles andere, NICHT benoetigt): - packages/ apps/ native/ plugins/ tools/ schemas/ tests/ docs/ build/ fixture_profiles/ - PLAN.md STATUS.md ERRORS.md TEST_REPORT.md CHANGELOG.md README.md - pyproject.toml uv.lock setup_*.sh/ps1 make_mac_app.py Diese Trennung gilt ab sofort fuer alle Commits. Der Nutzer kann _entwicklung/ loeschen wenn er Platz braucht - die App laeuft ohne. Verifiziert: App startet nach Aufraeumen unveraendert (Health 200).
175 lines
6.1 KiB
Python
175 lines
6.1 KiB
Python
#!/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")
|