Aether/scripts/rescan-selbsttest.py

#!/usr/bin/env python3
"""Selbsttest des Abgleichs zwischen Platte und Datenbank.

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

Der Abgleich (Admin → Speicher → „Rescan storage") übernimmt Dateien, die von
außen dazugekommen sind, und **entfernt Einträge**, deren Datei fort ist. Das
zweite ist der Grund für diesen Test: Ein Fehler darin löscht Daten, die
niemand vermisst, bis er sie braucht.

Der Test schreibt selbst Dateien in das Datenverzeichnis – er muss also auf
dem Rechner laufen, auf dem die Daten liegen, und räumt am Ende auf.
"""
import hashlib
import http.cookiejar
import json
import os
import shutil
import sys
import urllib.error
import urllib.request
import uuid

BASE = (sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8090").rstrip("/")
DATEN = sys.argv[2] if len(sys.argv) > 2 else os.path.expanduser("~/aether-data")
USER = sys.argv[3] if len(sys.argv) > 3 else "admin"
PASSWORT = sys.argv[4] if len(sys.argv) > 4 else "admin123"

ok_count = 0
fail = []


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}")


jar = http.cookiejar.CookieJar()
OPENER = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))


def api(method, path, body=None, headers=None, opener=None, roh=False):
    data = json.dumps(body).encode() if body is not None else None
    hdrs = dict(headers or {})
    hdrs["Content-Type"] = "application/json"
    req = urllib.request.Request(BASE + path, data=data, headers=hdrs, method=method)
    try:
        with (opener or 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


st, login = api("POST", "/api/v1/auth/login", {"username": USER, "password": PASSWORT})
assert st == 200, f"Anmeldung fehlgeschlagen: {st}"
CSRF = {"X-CSRF-Token": login["csrf_token"]}

HOME = os.path.join(DATEN, "users", USER, "home")
assert os.path.isdir(HOME), f"Datenverzeichnis nicht gefunden: {HOME}"

MARKE = "selbsttest-" + uuid.uuid4().hex[:8]
ORDNER = os.path.join(HOME, MARKE)
angelegt = []


def aufraeumen():
    shutil.rmtree(ORDNER, ignore_errors=True)
    # Was der Abgleich angelegt hat, wieder wegräumen – über die
    # Schnittstelle, damit auch die Einträge verschwinden.
    _, lst = api("GET", "/api/v1/files")
    for f in (lst or {}).get("folders", []):
        if f["name"] == MARKE:
            api("DELETE", f"/api/v1/folders/{f['id']}", None, CSRF)
    api("POST", "/api/v1/admin/rescan?anwenden=true", {}, CSRF)


def schreiben(name, inhalt: bytes):
    os.makedirs(ORDNER, exist_ok=True)
    pfad = os.path.join(ORDNER, name)
    with open(pfad, "wb") as f:
        f.write(inhalt)
    return pfad


def rescan(anwenden=False):
    st, r = api("POST", f"/api/v1/admin/rescan{'?anwenden=true' if anwenden else ''}", {}, CSRF)
    assert st == 200, f"Abgleich fehlgeschlagen: {st}"
    return r


def eigene(r):
    return [e for e in (r.get("samples") or [])
            if MARKE in e["path"] or MARKE in e["storage_key"]]


print(f"Abgleich-Selbsttest gegen {BASE}\nDaten: {DATEN}\n")

try:
    print("── Prüfen ändert nichts ──")
    INHALT = b"aus der Sicherung zurueckgeholt\n"
    schreiben("beleg.txt", INHALT)

    vorher = rescan()
    meine = eigene(vorher)
    check("neue Datei wird gefunden",
          len(meine) == 1 and meine[0]["kind"] == "new", meine)

    _, lst = api("GET", "/api/v1/files")
    check("aber noch nicht sichtbar",
          all(f["name"] != MARKE for f in lst["folders"]), lst["folders"])

    nochmal = rescan()
    check("zweimal prüfen liefert dasselbe",
          len(eigene(nochmal)) == 1, eigene(nochmal))

    print("\n── Übernehmen ──")
    r = rescan(anwenden=True)
    check("meldet eine Übernahme", r["adopted"] >= 1 and r.get("errors", 0) == 0,
          {k: v for k, v in r.items() if k != "samples"})

    _, lst = api("GET", "/api/v1/files")
    ordner = {f["name"]: f for f in lst["folders"]}
    check("Ordner ist angelegt", MARKE in ordner, list(ordner))
    _, drin = api("GET", f"/api/v1/files?folder_id={ordner[MARKE]['id']}")
    datei = next((f for f in drin["files"] if f["name"] == "beleg.txt"), None)
    check("Datei ist sichtbar", datei is not None, drin["files"])
    # Ohne richtige Prüfsumme merkt niemand, wenn die Datei später kippt –
    # genau die Zusage, die Aether sonst überall einhält.
    check("Prüfsumme stimmt",
          datei and datei["checksum"] == hashlib.sha256(INHALT).hexdigest(), datei)
    check("Größe stimmt", datei and datei["size"] == len(INHALT), datei)
    # Beim Hochladen schickt der Browser den Typ mit; hier gibt es niemanden,
    # der ihn kennt – er muss aus Endung oder Inhalt kommen.
    check("Dateityp erkannt", datei and datei["mime_type"] == "text/plain",
          datei["mime_type"] if datei else None)

    st, inhalt = api("GET", f"/api/v1/files/{datei['id']}/download", None, None, roh=True)
    check("Herunterladen liefert die Datei", st == 200 and inhalt == INHALT, inhalt[:40])

    check("danach ist nichts mehr offen", len(eigene(rescan())) == 0, eigene(rescan()))

    print("\n── Von außen ersetzt ──")
    NEU = b"jemand hat sie per SFTP ueberschrieben\n"
    schreiben("beleg.txt", NEU)
    r = rescan()
    meine = eigene(r)
    check("Änderung wird bemerkt", len(meine) == 1 and meine[0]["kind"] == "changed", meine)

    rescan(anwenden=True)
    _, drin = api("GET", f"/api/v1/files?folder_id={ordner[MARKE]['id']}")
    datei = next(f for f in drin["files"] if f["name"] == "beleg.txt")
    check("Größe berichtigt", datei["size"] == len(NEU), datei["size"])
    check("Prüfsumme berichtigt",
          datei["checksum"] == hashlib.sha256(NEU).hexdigest(), datei["checksum"])

    print("\n── Datei ist fort ──")
    os.remove(os.path.join(ORDNER, "beleg.txt"))
    r = rescan()
    meine = eigene(r)
    check("verwaister Eintrag wird gefunden",
          len(meine) == 1 and meine[0]["kind"] == "orphaned", meine)

    _, drin = api("GET", f"/api/v1/files?folder_id={ordner[MARKE]['id']}")
    check("solange nur geprüft wird, bleibt der Eintrag",
          any(f["name"] == "beleg.txt" for f in drin["files"]), drin["files"])

    rescan(anwenden=True)
    _, drin = api("GET", f"/api/v1/files?folder_id={ordner[MARKE]['id']}")
    check("nach dem Anwenden ist er weg",
          all(f["name"] != "beleg.txt" for f in drin["files"]), drin["files"])

    print("\n── Was der Abgleich **nicht** anfassen darf ──")
    # Papierkorb und frühere Fassungen liegen im selben Baum. Würden sie
    # übernommen, tauchten gelöschte Dateien wieder auf und jede frühere
    # Fassung würde zu einer eigenen Datei.
    trash = os.path.join(DATEN, "users", USER, ".trash")
    versions = os.path.join(DATEN, "users", USER, ".versions")
    os.makedirs(trash, exist_ok=True)
    os.makedirs(versions, exist_ok=True)
    t1 = os.path.join(trash, MARKE + "-papierkorb.txt")
    v1 = os.path.join(versions, MARKE + "-fassung")
    open(t1, "wb").write(b"im Papierkorb\n")
    open(v1, "wb").write(b"fruehere Fassung\n")
    r = rescan()
    check("Papierkorb bleibt unangetastet",
          not any("papierkorb" in e["storage_key"] for e in r["samples"]), r["samples"][:3])
    check("frühere Fassungen bleiben unangetastet",
          not any(".versions" in e["storage_key"] for e in r["samples"]), r["samples"][:3])
    os.remove(t1)
    os.remove(v1)

    print("\n── Nur für Verwalter ──")
    fremd_name = "rescan-selbsttest-fremd"
    fremd_pw = "Fremd!" + uuid.uuid4().hex[:12]
    st, fremd = api("POST", "/api/v1/admin/users",
                    {"username": fremd_name, "display_name": "Fremd",
                     "password": fremd_pw, "is_admin": False, "quota_bytes": 0}, CSRF)
    if st in (200, 201):
        jar2 = http.cookiejar.CookieJar()
        op2 = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar2))
        _, l2 = api("POST", "/api/v1/auth/login",
                    {"username": fremd_name, "password": fremd_pw}, None, op2)
        st, _ = api("POST", "/api/v1/admin/rescan", {},
                    {"X-CSRF-Token": l2["csrf_token"]}, op2)
        check("gewöhnliches Konto darf nicht", st == 403, st)
        api("DELETE", f"/api/v1/admin/users/{fremd['id']}", None, CSRF)

finally:
    aufraeumen()

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