hms-mediaengine/_entwicklung/tests/e2e_projects_test.py

#!/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()