Aether/scripts/zeichnung-selbsttest.py

#!/usr/bin/env python3
"""Selbsttest der Zeichnungen (Excalidraw).

    python3 scripts/zeichnung-selbsttest.py [BASIS-URL] [BENUTZER] [PASSWORT]

Eine Zeichnung ist bei uns **eine Datei** – das ist die ganze Entscheidung
hinter der Umsetzung. Dieser Test prueft genau das nach: Wenn sie wirklich
eine Datei ist, muss sie alles koennen, was Dateien koennen – Versionen,
Freigabelinks, Gruppenordner, Papierkorb. Und das ist nicht selbstverstaendlich,
sondern der Gewinn, den man sonst nur behauptet.

Dazu die eine Stelle, die ohne gleichzeitiges Arbeiten wehtut: Zwei Leute
speichern nacheinander, und der Zweite darf den Ersten nicht spurlos
ueberschreiben.
"""
import hashlib
import json
import struct
import zlib
import sys
import urllib.error
import urllib.request
import uuid
import http.cookiejar

BASE = (sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8090").rstrip("/")
USER = sys.argv[2] if len(sys.argv) > 2 else "admin"
PASSWORT = sys.argv[3] if len(sys.argv) > 3 else "admin123"

ok_count, fail = 0, []


def check(name, cond, detail=""):
    global ok_count
    if cond:
        ok_count += 1
        print(f"  \033[32mok\033[0m   {name}")
    else:
        fail.append(name)
        print(f"  \033[31mFEHL\033[0m {name}   {detail}")


def sitzung():
    return urllib.request.build_opener(
        urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))


def ruf(opener, method, path, body=None, headers=None, roh=False):
    hdrs = dict(headers or {})
    data = None
    if body is not None:
        data = json.dumps(body).encode()
        hdrs["Content-Type"] = "application/json"
    req = urllib.request.Request(BASE + path, data=data, headers=hdrs, method=method)
    try:
        with (opener or urllib.request.build_opener()).open(req) as r:
            payload = r.read()
            if roh:
                return r.status, payload
            text = payload.decode()
            return r.status, (json.loads(text).get("data") if text else None)
    except urllib.error.HTTPError as e:
        return e.code, None


def anmelden(user, passwort):
    o = sitzung()
    st, d = ruf(o, "POST", "/api/v1/auth/login", {"username": user, "password": passwort})
    assert st == 200, f"Anmeldung {user} fehlgeschlagen: {st}"
    return o, {"X-CSRF-Token": d["csrf_token"]}


def hochladen(opener, csrf, name, inhalt: bytes, folder_id=None):
    grenze = "----aether" + uuid.uuid4().hex
    teile = []
    if folder_id:
        teile.append(f'--{grenze}\r\nContent-Disposition: form-data; name="folder_id"\r\n\r\n{folder_id}\r\n')
    kopf = (f'--{grenze}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\n'
            f'Content-Type: application/json\r\n\r\n')
    leib = b"".join(t.encode() for t in teile) + kopf.encode() + inhalt + f"\r\n--{grenze}--\r\n".encode()
    req = urllib.request.Request(BASE + "/api/v1/files", data=leib, method="POST",
                                 headers={"Content-Type": f"multipart/form-data; boundary={grenze}", **csrf})
    with opener.open(req) as r:
        return json.loads(r.read().decode())["data"]


def zeichnung(elemente=0):
    return json.dumps({
        "type": "excalidraw", "version": 2, "source": "aether",
        "elements": [{"id": f"e{i}", "type": "rectangle", "x": i * 10, "y": 0,
                      "width": 80, "height": 40} for i in range(elemente)],
        "appState": {}, "files": {},
    }).encode()


def winziges_png() -> bytes:
    """Ein gueltiges 1x1-PNG, von Hand gebaut – ohne Abhaengigkeit."""
    def chunk(typ, daten):
        c = typ + daten
        return struct.pack(">I", len(daten)) + c + struct.pack(">I", zlib.crc32(c))
    ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0)
    return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr)
            + chunk(b"IDAT", zlib.compress(b"\x00\xff\x00\x00")) + chunk(b"IEND", b""))


def roh_senden(opener, path, daten: bytes, typ: str, csrf):
    req = urllib.request.Request(BASE + path, data=daten, method="POST",
                                 headers={"Content-Type": typ, **csrf})
    try:
        with opener.open(req) as r:
            return r.status
    except urllib.error.HTTPError as e:
        return e.code


print(f"Zeichnungs-Selbsttest gegen {BASE}\n")
admin, CSRF = anmelden(USER, PASSWORT)
_, ich = ruf(admin, "GET", "/api/v1/auth/me")
CSRF = {"X-CSRF-Token": ich["csrf_token"]}

name = f"selbsttest-{uuid.uuid4().hex[:6]}.excalidraw"
angelegt, links, gruppen = [], [], []


def aufraeumen():
    for lid in links:
        ruf(admin, "DELETE", f"/api/v1/shares/{lid}", None, CSRF)
    for fid in angelegt:
        ruf(admin, "DELETE", f"/api/v1/files/{fid}", None, CSRF)
    _, muell = ruf(admin, "GET", "/api/v1/trash")
    for f in muell or []:
        if f.get("name", "").startswith("selbsttest-"):
            ruf(admin, "DELETE", f"/api/v1/trash/{f['id']}", None, CSRF)
    for gid in gruppen:
        ruf(admin, "DELETE", f"/api/v1/admin/groups/{gid}", None, CSRF)


try:
    print("── Als Datei anlegen ──")
    inhalt = zeichnung(3)
    datei = hochladen(admin, CSRF, name, inhalt)
    angelegt.append(datei["id"])
    check("angelegt", datei["size"] == len(inhalt), datei["size"])
    check("Pruefsumme stimmt", datei["checksum"] == hashlib.sha256(inhalt).hexdigest())
    check("als JSON erkannt", datei["mime_type"] == "application/json", datei["mime_type"])

    st, roh = ruf(admin, "GET", f"/api/v1/files/{datei['id']}/download?inline=1", None, None, True)
    check("Inhalt kommt unveraendert zurueck", st == 200 and roh == inhalt, st)

    print("\n── Aendern erzeugt eine Version ──")
    # Speichern heisst: gleicher Name, gleicher Ordner, neuer Upload.
    zweite = hochladen(admin, CSRF, name, zeichnung(7))
    check("gleiche Datei, neue Fassung", zweite["id"] == datei["id"] and zweite["version"] > datei["version"],
          f"{datei['version']} -> {zweite['version']}")
    st, versionen = ruf(admin, "GET", f"/api/v1/files/{datei['id']}/versions")
    check("die alte Fassung liegt in der Historie", st == 200 and len(versionen or []) >= 1,
          len(versionen or []))

    print("\n── Ueberschreiben faellt auf ──")
    # Ohne gleichzeitiges Arbeiten ist die Fassungsnummer der einzige Hinweis
    # darauf, dass jemand anders gespeichert hat. Sie MUSS sich aendern.
    dritte = hochladen(admin, CSRF, name, zeichnung(9))
    check("jede Speicherung hebt die Fassung", dritte["version"] > zweite["version"],
          f"{zweite['version']} -> {dritte['version']}")
    _, liste = ruf(admin, "GET", "/api/v1/files")
    inliste = next((f for f in (liste or {}).get("files", []) if f["id"] == datei["id"]), None)
    check("und die Liste zeigt sie – darauf beruht der Abgleich im Editor",
          inliste is not None and inliste["version"] == dritte["version"],
          inliste and inliste["version"])

    print("\n── Erbt alles, was Dateien koennen ──")
    st, link = ruf(admin, "POST", f"/api/v1/files/{datei['id']}/share", {}, CSRF)
    check("Freigabelink darauf (201)", st == 201, st)
    if st == 201:
        links.append(link["id"])
        st, geholt = ruf(None, "POST", f"/share/{link['token']}/download", None, None, True)
        check("oeffentlich abrufbar, Inhalt stimmt", st == 200 and geholt == zeichnung(9), st)

    gname = "selbsttest-zeichnung-" + uuid.uuid4().hex[:6]
    st, gruppe = ruf(admin, "POST", "/api/v1/admin/groups", {"name": gname, "description": ""}, CSRF)
    if st in (200, 201):
        gruppen.append(gruppe["id"])
        ruf(admin, "POST", f"/api/v1/admin/groups/{gruppe['id']}/members",
            {"user_id": ich["user"]["id"]}, CSRF)
        _, wurzeln = ruf(admin, "GET", "/api/v1/files")
        gordner = next((f for f in (wurzeln or {}).get("folders", []) if f.get("group_id")), None)
        if gordner:
            gz = hochladen(admin, CSRF, "selbsttest-gruppe.excalidraw", zeichnung(2), gordner["id"])
            angelegt.append(gz["id"])
            check("Zeichnung im Gruppenordner", gz["folder_id"] == gordner["id"])

    print("\n── Vorschaubild ──")
    # Der Server kann es nicht selbst rechnen: In der Datei steht JSON. Der
    # Browser, der gerade zeichnet, kann es – deshalb darf er es schicken.
    check("frische Zeichnung hat noch keines", not datei.get("has_thumbnail"))
    st = roh_senden(admin, f"/api/v1/files/{datei['id']}/thumbnail", winziges_png(), "image/png", CSRF)
    check("Vorschaubild setzen (204)", st == 204, st)
    _, liste2 = ruf(admin, "GET", "/api/v1/files")
    jetzt = next((f for f in (liste2 or {}).get("files", []) if f["id"] == datei["id"]), None)
    check("und die Liste weiss davon", jetzt and jetzt.get("has_thumbnail"), jetzt)
    st, _ = ruf(admin, "GET", f"/api/v1/files/{datei['id']}/thumbnail", None, None, True)
    check("es laesst sich abrufen", st == 200, st)

    # **Den Bytes wird nicht geglaubt.** Sie gehen durch denselben Weg wie
    # jedes andere Vorschaubild – dekodieren, verkleinern, neu kodieren.
    # Sonst liesse sich Beliebiges in den thumbs-Bereich legen.
    st = roh_senden(admin, f"/api/v1/files/{datei['id']}/thumbnail", b"<html>kein bild</html>",
                    "image/png", CSRF)
    check("was kein Bild ist, wird abgelehnt (400)", st == 400, st)
    st = roh_senden(admin, f"/api/v1/files/{datei['id']}/thumbnail", winziges_png(),
                    "text/html", CSRF)
    check("fremder Typ wird abgelehnt (415)", st == 415, st)

    print("\n── Schriften werden selbst ausgeliefert ──")
    # Ohne das laedt Excalidraw sie von aussen – und die Sicherheitsrichtlinie
    # verbietet genau das. Man saehe eine Ersatzschrift und suchte den Grund
    # im Zeichenbrett statt in der Richtlinie.
    st, _ = ruf(None, "GET", "/excalidraw/fonts/Excalifont/"
                "Excalifont-Regular-349fac6ca4700ffec595a7150a0d1e1d.woff2", None, None, True)
    check("Excalifont liegt auf dem eigenen Server", st == 200, st)

finally:
    aufraeumen()

print(f"\n{ok_count} bestanden, {len(fail)} fehlgeschlagen")
if fail:
    print("Fehlgeschlagen: " + ", ".join(fail))
sys.exit(1 if fail else 0)