feat: tools/rundgang.py -- 24 Pruefungen gegen den laufenden Dienst
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.
Kein 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. Nur
lesend -- kein Upload, kein Loeschen, keine Einstellung wird angefasst.
ZUERST kommt, was OHNE ANMELDUNG DICHT SEIN MUSS (acht Wege, alle 401). Das ist
die Sorte Fehler, die man nicht sieht: eine Route, die versehentlich vor dem
Waechter registriert wird, funktioniert praechtig -- fuer alle. Genau das ist
hier schon zweimal passiert.
Dann Bestand, /f/ mit WebP-Aushandlung an derselben Adresse und Vary: Accept,
beide CSV-Exporte auf nackte LF, Papierkorb-Deckel, Sicherungsziele und die
Statusseite.
ZWEI FEHLSCHLAEGE BEIM BAUEN, beide im Test und nicht im Dienst -- sie stehen
als Kommentar drin, damit der naechste sie nicht wiederholt:
1. Ohne Filter zog er vehicles/asbo.webp und verglich WebP mit WebP. Gleiche
Groesse, Test rot, Dienst richtig.
2. Der Filter hiess dann "q=" statt "query=". Unbekannte Parameter ignoriert
der Dienst und liefert 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.
Gemessen: 24 in Ordnung, 0 nicht.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,183 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user