#!/usr/bin/env python3 """E2E: Projekt speichern/laden/loeschen/autostart. Startet ECHTEN Server als Subprozess, macht einen NEUSTART und beweist, dass das Autostart-Projekt beim Programmstart geladen wird. """ import json import os import shutil import subprocess import sys import time import urllib.error import urllib.parse import urllib.request ROOT = "/a0/usr/workdir/hms-mediaengine" PORT = 8094 BASE = "http://localhost:%d" % PORT LOG = "/tmp/hms_proj.log" server_proc = None def start_server() -> bool: global server_proc server_proc = subprocess.Popen( ["/usr/bin/python3", ROOT + "/run.py", "--port", str(PORT)], cwd=ROOT, stdout=open(LOG, "ab"), stderr=subprocess.STDOUT) deadline = time.time() + 25 while time.time() < deadline: try: with urllib.request.urlopen(BASE + "/api/health", timeout=2) as r: if json.loads(r.read().decode()).get("status") == "ok": return True except Exception: time.sleep(0.3) return False def stop_server(): global server_proc if server_proc is not None: server_proc.terminate() try: server_proc.wait(timeout=6) except subprocess.TimeoutExpired: server_proc.kill() server_proc = None time.sleep(0.5) 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) # ---------- Reset Runtime-Daten ---------- for d in ("media", "thumbs", "projects"): shutil.rmtree(os.path.join(ROOT, d), ignore_errors=True) try: os.remove(os.path.join(ROOT, "config.json")) except FileNotFoundError: pass try: # ---------- Serverstart 1 ---------- check("Server startet", start_server()) # Testvideo erzeugen + hochladen 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_proj.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_proj.mp4", "rb").read() r = post("/api/media/upload", raw=data, headers={"X-Filename": urllib.parse.quote("e2e_clip.mp4")}) check("Upload", bool(r.get("saved")), r) # Showzustand aufbauen: Layer + FX + Position + Master r = post("/api/layers/add", {"name": "e2e_clip.mp4"}) check("Layer hinzugefuegt", "layer" in r, r) r = post("/api/layers/update", {"id": 1, "alpha": 0.7, "x": 50, "y": 20, "fx1": "blur", "fx1_intensity": 0.8}) check("FX/Alpha/Pos gesetzt", r.get("layer", {}).get("fx1") == "blur", r) post("/api/master", {"intensity": 0.9}) # Projekt speichern r = post("/api/projects/save", {"name": "Show A"}) check("Projekt gespeichert", r.get("saved") == "Show A", r) # Pfad-Traversal im Projektnamen wird sanitizet r = post("/api/projects/save", {"name": "../../evil"}) check("Projektname sanitizet", r.get("saved") == "evil", r) pr = get("/api/projects") names = [p["name"] for p in pr["items"]] check("Liste enthaelt beide", {"Show A", "evil"} <= set(names), names) showa = next(p for p in pr["items"] if p["name"] == "Show A") check("Layerzahl in Liste", showa["layers"] == 1, showa) # Zustand zerstoeren post("/api/layers/remove", {"id": 1}) post("/api/master", {"intensity": 0.1}) st = get("/api/status") check("Zustand geleert", len(st["layers"]) == 0 and st["master"] < 0.2, "master=%s" % st["master"]) # Projekt laden -> vollstaendig wiederhergestellt r = post("/api/projects/load", {"name": "Show A"}) check("Projekt geladen", r.get("loaded") is True, r) st = get("/api/status") check("Layer wiederhergestellt", len(st["layers"]) == 1, "%d Layer" % len(st["layers"])) l1 = st["layers"][0] check("Alpha wiederhergestellt", abs(l1["alpha"] - 0.7) < 0.02, l1["alpha"]) check("FX1 wiederhergestellt", l1["fx1"] == "blur" and abs(l1["fx1_intensity"] - 0.8) < 0.03, "%s %s" % (l1["fx1"], l1["fx1_intensity"])) check("Position wiederhergestellt", l1["x"] == 50 and l1["y"] == 20, "x=%s y=%s" % (l1["x"], l1["y"])) check("Master wiederhergestellt", abs(st["master"] - 0.9) < 0.02, st["master"]) check("Engine rendert nach Load", st["running"] is True, "err=%s" % st["error"]) # Autostart setzen r = post("/api/projects/autostart", {"name": "Show A"}) check("Autostart gesetzt", r.get("autostart") == "Show A", r) # ---------- ECHTER SERVER-NEUSTART: Autostart-Beweis ---------- stop_server() check("Server neu gestartet", start_server()) st = get("/api/status") check("Autostart: Layer geladen", len(st["layers"]) == 1 and st["layers"][0]["fx1"] == "blur", "%d Layer fx1=%s" % (len(st["layers"]), st["layers"][0]["fx1"] if st["layers"] else "-")) time.sleep(3) st = get("/api/status") check("Autostart: Engine rendert", st["frames_rendered"] > 0, "frames=%d" % st["frames_rendered"]) cfg = json.load(open(os.path.join(ROOT, "config.json"))) check("config.json Autostart", cfg["engine"]["autostart_project"] == "Show A") # Nicht vorhandenes Projekt -> 404 r = post("/api/projects/load", {"name": "gibts_nicht"}) check("404 bei fehlendem Projekt", "error" in r, r) # Loeschen des Autostart-Projekts raeumt Autostart weg r = post("/api/projects/delete", {"name": "Show A"}) check("Projekt geloescht", r.get("deleted") is True, r) pr = get("/api/projects") check("Autostart automatisch geleert", pr.get("autostart") is None, pr) cfg = json.load(open(os.path.join(ROOT, "config.json"))) check("config.json Autostart geleert", cfg["engine"]["autostart_project"] is None) print("") print("ALLE PROJEKT-E2E-TESTS BESTANDEN") finally: stop_server()