d4rk_media/tools/rundgang.py

"""Ein Rundgang durch den LAUFENDEN Dienst.

    python tools/rundgang.py                      <- gegen PUBLIC_URL aus .env
    python tools/rundgang.py http://localhost:8096

WOZU, wo doch jede Aenderung einzeln gemessen wird: weil sie einzeln gemessen
wird. An einem Tag mit sechs Neustarts fasst niemand mehr alles an, und die
Sachen, die keiner mehr anfasst, sind genau die, die kaputtgehen -- ein WebP,
das nach einem Umbau nicht mehr ausgehandelt wird, meldet sich nicht.

WAS ER NICHT IST: ein Ersatz fuer die Tests. Er prueft nicht, ob eine Funktion
das Richtige rechnet, sondern ob der Dienst, so wie er gerade laeuft, die
Sachen noch tut, die er koennen muss. Deshalb geht er ueber HTTP und nicht am
Code vorbei.

Er braucht das Admin-Passwort aus der .env, liest sie aber nur -- und laeuft
NUR LESEND: kein Upload, kein Loeschen, keine Einstellung wird angefasst.
"""

import json
import os
import re
import subprocess
import sys
import tempfile

WURZEL = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ENV = dict(re.findall(r'^([A-Z_]+)=(.*)$',
                      open(os.path.join(WURZEL, '.env'), encoding='utf-8').read(), re.M))
BASIS = (sys.argv[1] if len(sys.argv) > 1 else ENV.get('PUBLIC_URL', '')).rstrip('/')
if not BASIS:
    sys.exit('Keine Adresse: weder als Argument noch als PUBLIC_URL in der .env.')

ABLAGE = tempfile.mkdtemp(prefix='rundgang-')
KEKS = os.path.join(ABLAGE, 'keks.txt')
RUMPF = os.path.join(ABLAGE, 'rumpf.bin')

CRLF = b'\r\n'

gut, schlecht = 0, 0


def curl(*args, keks=True):
    """Der Rumpf geht in eine DATEI und nicht durch die Standardausgabe: ein PNG
    durch einen Textkanal zu ziehen, endet unter Windows an cp1252 -- und der
    Rundgang stirbt dann an der Datei, die er gerade erfolgreich geholt hat."""
    befehl = ['curl', '-s', '-o', RUMPF, '-w', '%{http_code}|%{size_download}|%{content_type}']
    if keks:
        befehl += ['-b', KEKS]
    r = subprocess.run(befehl + list(args), capture_output=True, text=True, timeout=90)
    code, groesse, typ = (r.stdout.split('|') + ['', '', ''])[:3]
    try:
        text = open(RUMPF, 'rb').read().decode('utf-8')
    except (UnicodeDecodeError, FileNotFoundError):
        text = ''
    return code, int(groesse or 0), typ, text


def pruefe(name, bedingung, gemessen):
    global gut, schlecht
    if bedingung:
        gut += 1
        print(f'  ok    {name:<44} {gemessen}')
    else:
        schlecht += 1
        print(f'  NEIN  {name:<44} {gemessen}')


print(f'Rundgang gegen {BASIS}\n')

# -- Anmelden -----------------------------------------------------------------
r = subprocess.run(
    ['curl', '-s', '-c', KEKS, '-X', 'POST', f'{BASIS}/api/dash/auth/login',
     '-H', 'content-type: application/json', '--data-binary', '@-',
     '-o', os.devnull, '-w', '%{http_code}'],
    input=json.dumps({'username': ENV.get('ADMIN_USERNAME', 'admin'),
                      'password': ENV['ADMIN_PASSWORD']}),
    capture_output=True, text=True,
)
pruefe('Anmelden', r.stdout == '200', f'HTTP {r.stdout}')
if r.stdout != '200':
    sys.exit('Ohne Anmeldung hat der Rest keinen Sinn.')

# -- Wer nicht angemeldet ist, kommt nicht rein --------------------------------
#
# Steht ZUERST, weil es die Sorte Fehler ist, die man nicht sieht: eine Route,
# die versehentlich vor dem Waechter registriert wird, funktioniert praechtig --
# fuer alle. Genau das ist hier schon zweimal passiert.
print('\n-- Was ohne Anmeldung dicht sein muss ' + '-' * 32)
for weg in ('/api/dash/media', '/api/dash/tokens', '/api/dash/verlauf/export',
            '/api/dash/papierkorb', '/api/dash/benutzer', '/api/dash/settings',
            '/api/dash/einstellungen/export', '/api/dash/verwaltung'):
    code, *_ = curl(f'{BASIS}{weg}', keks=False)
    pruefe(weg, code == '401', f'HTTP {code}')

print('\n-- Was ohne Anmeldung offen sein muss ' + '-' * 32)
for weg in ('/health', '/status'):
    code, *_ = curl(f'{BASIS}{weg}', keks=False)
    pruefe(weg, code == '200', f'HTTP {code}')

# -- Der Bestand ---------------------------------------------------------------
print('\n-- Der Bestand ' + '-' * 55)
code, _, _, rumpf = curl(f'{BASIS}/api/dash/uebersicht')
u = json.loads(rumpf) if code == '200' else {}
pruefe('Uebersicht', code == '200' and u.get('files', 0) > 0, f"{u.get('files')} Dateien")

code, _, _, rumpf = curl(f'{BASIS}/api/dash/media?limit=1')
erstes = ((json.loads(rumpf).get('items') if code == '200' else None) or [{}])[0]
pruefe('Mediathek blaettert', bool(erstes.get('path')), erstes.get('path', '-'))

# AUSDRUECKLICH EIN PNG, und das ist zweimal danebengegangen:
#
#   1. Ohne Filter kam "vehicles/asbo.webp" -- da vergleicht der Rundgang WebP
#      mit WebP, findet gleiche Groessen und meldet einen Fehler, den der Dienst
#      nicht hat.
#   2. Der Filter hiess dann "q=" statt "query=". Unbekannte Parameter
#      ignoriert der Dienst, liefert also ALLES -- was wie ein kaputter Filter
#      aussieht und ein kaputter Aufruf ist.
#
# Ein Test, der das Falsche misst, ist schlimmer als keiner: er wird geglaubt.
code, _, _, rumpf = curl(f'{BASIS}/api/dash/media?query=.png&limit=5')
png = next((i for i in (json.loads(rumpf).get('items') or [])
            if i['path'].lower().endswith('.png')), {})
pruefe('Ein PNG zum Vergleichen', bool(png.get('path')), png.get('path', 'keines im Bestand'))

# -- Ausliefern ----------------------------------------------------------------
print('\n-- Eine Datei wirklich ausliefern ' + '-' * 36)
if png.get('path'):
    ziel = f"{BASIS}/f/{png['path']}"
    code, groesse, typ, _ = curl(ziel, keks=False)
    pruefe('/f/ liefert aus', code == '200' and groesse > 0, f'{groesse} B, {typ}')

    # WebP an DERSELBEN Adresse -- der Kern der Aushandlung, und die Stelle, an
    # der der Zwischenspeicher des Proxys schon zweimal danebengelegen hat.
    code, gw, tw, _ = curl('-H', 'Accept: image/webp', ziel, keks=False)
    pruefe('WebP an derselben Adresse', code == '200' and tw == 'image/webp', f'{gw} B, {tw}')
    pruefe('WebP ist kleiner', 0 < gw < groesse, f'{gw} B statt {groesse} B')

    r = subprocess.run(['curl', '-s', '-D', '-', '-o', os.devnull,
                        '-H', 'Accept: image/webp', ziel], capture_output=True, text=True)
    hat = 'vary: accept' in r.stdout.lower()
    pruefe('Vary: Accept gesetzt', hat, 'ja' if hat else 'FEHLT -- der Proxy mischt die Fassungen')

# -- Verlauf als CSV -----------------------------------------------------------
print('\n-- Der Verlauf als CSV ' + '-' * 47)
for anhang, name in (('', 'dateien'), ('?was=verwaltung', 'verwaltung')):
    datei = os.path.join(ABLAGE, f'{name}.csv')
    r = subprocess.run(['curl', '-s', '-b', KEKS, '-D', '-', '-o', datei,
                        f'{BASIS}/api/dash/verlauf/export{anhang}'],
                       capture_output=True, text=True)
    roh = open(datei, 'rb').read()
    nackte = roh.count(b'\n') - roh.count(b'\r\n')
    # Nackte LF sind der Punkt: Excel packt bei ihnen die halbe Datei in eine
    # Zelle, und das faellt erst dem auf, der die Datei braucht.
    pruefe(f'CSV {name}', name in r.stdout and nackte == 0,
           f'{roh.count(CRLF)} Zeilen, {nackte} nackte LF')

# -- Papierkorb und Sicherung --------------------------------------------------
print('\n-- Papierkorb und Sicherung ' + '-' * 42)
code, _, _, rumpf = curl(f'{BASIS}/api/dash/papierkorb')
pk = json.loads(rumpf)
pruefe('Deckel steht in der Antwort', pk.get('maxBytes', 0) > 0,
       f"{round(pk.get('maxBytes', 0) / 1024 ** 3, 1)} GB")

code, _, _, rumpf = curl(f'{BASIS}/api/dash/settings')
einst = json.loads(rumpf)
sic, letzte = einst['sicherung'], (einst['sicherung'].get('letzte') or {})
pruefe('Zweites Ziel eingetragen', bool(sic.get('zweitziel')),
       sic.get('zweitziel') or 'keines -- die Sicherung liegt auf EINER Platte')
pruefe('Letzte Sicherung gut', letzte.get('ok') is True,
       f"ok={letzte.get('ok')}, zweit={letzte.get('zweit')}, hoch={letzte.get('hoch')}")

code, _, _, rumpf = curl(f'{BASIS}/status?streng=1', keks=False)
st = json.loads(rumpf)
schief = [p['was'] for p in st['pruefungen'] if not p['ok']]
pruefe('Alle Pruefungen der Statusseite gruen', not schief, ', '.join(schief) or 'alle')
for p in st['pruefungen']:
    if p.get('hinweis'):
        print(f"        {p['was']}: {p['hinweis']}")

print('\n' + '=' * 70)
print(f'  {gut} in Ordnung, {schlecht} nicht.')
sys.exit(1 if schlecht else 0)