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 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()


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── 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)