#!/usr/bin/env python3 """ Gemeinsamer Unterbau der Testlaeufe gegen die Testinstanz. Bewusst ohne pytest und ohne requests: Der Testlauf soll auf jedem Rechner und auf der Test-VM ohne vorherige Installation starten koennen. Sobald ein Testwerkzeug erst eingerichtet werden muss, wird es im Zweifel nicht ausgefuehrt - und eine Testsuite, die niemand startet, findet nichts. Aufbau: Client - HTTP-Aufrufe gegen die Instanz, mit Basic-Auth und Token Ctx - was ein Testfall an den naechsten weitergibt (IDs, Tokens) @test - Registrierung; die Reihenfolge der Registrierung ist die Reihenfolge der Ausfuehrung run - Ausfuehrung, Ausgabe, Bericht """ import base64 import json import os import socket import ssl import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path PROJECT_DIR = Path(__file__).resolve().parents[2] DEV_CONFIG = PROJECT_DIR / 'scripts' / 'deploy_dev_config.json' # --------------------------------------------------------------------------- # Adressfamilie # # Die Hoster-Adressen haben einen AAAA-Eintrag. Ist von einem Arbeitsplatz aus # keine IPv6-Route vorhanden, laeuft jeder einzelne Aufruf zuerst 21 Sekunden # in einen TCP-Timeout und faellt erst dann auf IPv4 zurueck - aus einem # Testlauf von einer Minute werden 40. curl faellt sofort zurueck (Happy # Eyeballs), Python nicht; deshalb steht das hier und nicht im Netzwerk. # # Einmal am Anfang geprueft, danach steht die Entscheidung fuer den ganzen # Lauf. Sie betrifft nur, wie der Testlauf die Instanz erreicht - nicht, was # er prueft. # --------------------------------------------------------------------------- _ECHTES_GETADDRINFO = socket.getaddrinfo def erreichbar(host, port, familie, timeout=4.0): """ Prueft, ob ueber diese Adressfamilie wirklich gearbeitet werden kann. Ein blosses connect() genuegt dafuer nicht - und das war der Fehler der ersten Fassung. Auf dieser Strecke gelingt der TCP-Verbindungsaufbau ueber IPv6 zeitweise, waehrend danach nichts mehr fliesst. Die Pruefung meldete dann "erreichbar", der Lauf blieb auf IPv6, und jede einzelne Anfrage lief 21 Sekunden in einen Timeout: 129 Testfaelle statt in elf Sekunden in ueber einer Stunde. Schlimmer noch, es war launisch - derselbe Lauf ging mal so und mal so aus, was die Ursache lange verdeckt hat. """ try: infos = _ECHTES_GETADDRINFO(host, port, familie, socket.SOCK_STREAM) except socket.gaierror: return False kontext = ssl.create_default_context() for info in infos: s = socket.socket(info[0], socket.SOCK_STREAM) s.settimeout(timeout) try: s.connect(info[4]) # Bis hierher kam die erste Fassung. Der Rest ist der Unterschied # zwischen "eine Verbindung steht" und "es kommt etwas zurueck": # TLS-Handschlag und eine echte Antwort. with kontext.wrap_socket(s, server_hostname=host) as tls: tls.settimeout(timeout) tls.sendall( b'HEAD /api/health HTTP/1.1\r\nHost: ' + host.encode() + b'\r\nConnection: close\r\nUser-Agent: dc-e2e\r\n\r\n') return tls.recv(16).startswith(b'HTTP/') except OSError: continue finally: try: s.close() except OSError: pass return False def waehle_adressfamilie(host, port=443, erzwinge=None): """ Liefert eine kurze Beschreibung fuer die Ausgabe und schaltet notfalls auf IPv4 um. """ if erzwinge == 'ipv4' or (erzwinge is None and not erreichbar(host, port, socket.AF_INET6)): def nur_ipv4(*a, **kw): args = list(a) if len(args) >= 3: args[2] = socket.AF_INET else: kw['family'] = socket.AF_INET return _ECHTES_GETADDRINFO(*args, **kw) socket.getaddrinfo = nur_ipv4 return 'IPv4 (IPv6 nicht erreichbar)' if erzwinge is None else 'IPv4 (erzwungen)' return 'IPv6/IPv4' # --------------------------------------------------------------------------- # Ausgabe # --------------------------------------------------------------------------- class Farbe: """ANSI-Farben, aber nur wenn die Ausgabe wirklich ein Terminal ist.""" aktiv = sys.stdout.isatty() and os.name != 'nt' or os.environ.get('DC_FARBE') == '1' @classmethod def _f(cls, code, text): return "\033[%sm%s\033[0m" % (code, text) if cls.aktiv else text @classmethod def gruen(cls, t): return cls._f('32', t) @classmethod def rot(cls, t): return cls._f('31', t) @classmethod def gelb(cls, t): return cls._f('33', t) @classmethod def grau(cls, t): return cls._f('90', t) # --------------------------------------------------------------------------- # Fehlerarten # # Der Unterschied zwischen "die Anwendung ist kaputt" und "dieser Testfall # laesst sich hier nicht pruefen" muss im Bericht sichtbar bleiben. Sonst # gewoehnt man sich an rote Zeilen, die keine sind. # --------------------------------------------------------------------------- class Fehlschlag(AssertionError): """Die Anwendung hat sich anders verhalten als erwartet.""" class Uebersprungen(Exception): """Der Testfall ist unter diesen Bedingungen nicht pruefbar.""" # --------------------------------------------------------------------------- # HTTP # --------------------------------------------------------------------------- class Antwort: def __init__(self, status, headers, body, ms, url, method): self.status = status self.headers = {k.lower(): v for k, v in (headers or {}).items()} self.body = body self.ms = ms self.url = url self.method = method try: self.json = json.loads(body) if body else None except Exception: self.json = None @property def ok(self): return isinstance(self.json, dict) and self.json.get('status') == 'success' @property def fehlercode(self): if isinstance(self.json, dict) and isinstance(self.json.get('error'), dict): return self.json['error'].get('code') return None def feld(self, *pfad, default=None): """Wert aus verschachtelter Antwort holen: resp.feld('state', 'tables').""" wert = self.json for teil in pfad: if isinstance(wert, dict) and teil in wert: wert = wert[teil] elif isinstance(wert, list) and isinstance(teil, int) and -len(wert) <= teil < len(wert): wert = wert[teil] else: return default return wert def kurz(self, n=400): text = self.body or '' return text[:n] + (' ...' if len(text) > n else '') def __repr__(self): return "<%s %s -> %d>" % (self.method, self.url, self.status) class Client: """ HTTP-Client fuer die Testinstanz. Zur Aufteilung der beiden Berechtigungen: Die Basic-Auth der Testinstanz geht in den Authorization-Header, das Token der Anwendung in X-Agent-Token. Beide in denselben Header zu packen ginge nicht - und die Anwendung akzeptiert X-Agent-Token ausdruecklich (Http::bearerToken). Wer den Bearer-Weg pruefen will, ruft mit basic=False und bearer=True auf; dann greift die Ausnahme der .htaccess fuer /api/. """ def __init__(self, base_url, basic_user=None, basic_pass=None, token=None, timeout=45): self.base_url = base_url.rstrip('/') self.basic_user = basic_user self.basic_pass = basic_pass self.token = token self.timeout = timeout self.aufrufe = 0 # Die Testinstanz laeuft hinter einem regulaeren Zertifikat; ein # eigener Kontext nur, damit ein Hoster-Zwischenzertifikat den # Testlauf nicht zum Erliegen bringt. self.ssl_ctx = ssl.create_default_context() def basic_header(self): roh = ("%s:%s" % (self.basic_user, self.basic_pass)).encode('utf-8') return 'Basic ' + base64.b64encode(roh).decode('ascii') def request(self, method, pfad, body=None, query=None, token='standard', bearer=False, basic=True, headers=None, roh_body=None): url = self.base_url + pfad if query: url += ('&' if '?' in url else '?') + urllib.parse.urlencode(query) kopf = {'Accept': 'application/json', 'User-Agent': 'dc-e2e/1.0'} if basic and self.basic_user: kopf['Authorization'] = self.basic_header() # token='standard' -> das Token des Laufs; None -> keins; sonst der Wert wert = self.token if token == 'standard' else token if wert: if bearer: kopf['Authorization'] = 'Bearer ' + wert else: kopf['X-Agent-Token'] = wert daten = None if roh_body is not None: daten = roh_body if isinstance(roh_body, bytes) else roh_body.encode('utf-8') kopf.setdefault('Content-Type', 'application/json') elif body is not None: daten = json.dumps(body).encode('utf-8') kopf['Content-Type'] = 'application/json' kopf.update(headers or {}) req = urllib.request.Request(url, data=daten, headers=kopf, method=method) start = time.time() # Ein einziger Wiederholungsversuch, und nur wenn die Verbindung # selbst gescheitert ist - nie bei einer HTTP-Antwort. Ein # zurueckgesetztes TCP sagt nichts ueber die Anwendung aus; ein # HTTP 500 dagegen schon, und das darf kein Wiederholungsversuch # verschlucken. for versuch in (1, 2): try: with urllib.request.urlopen(req, timeout=self.timeout, context=self.ssl_ctx) as r: inhalt = r.read().decode('utf-8', 'replace') antwort = Antwort(r.status, dict(r.headers), inhalt, int((time.time() - start) * 1000), url, method) break except urllib.error.HTTPError as e: inhalt = e.read().decode('utf-8', 'replace') antwort = Antwort(e.code, dict(e.headers), inhalt, int((time.time() - start) * 1000), url, method) break except urllib.error.URLError as e: if versuch == 2: raise Fehlschlag("%s %s nicht erreichbar: %s" % (method, url, e.reason)) time.sleep(0.5) self.aufrufe += 1 return antwort def get(self, pfad, **kw): return self.request('GET', pfad, **kw) def post(self, pfad, body=None, **kw): return self.request('POST', pfad, body=body, **kw) # --------------------------------------------------------------------------- # Pruefungen # --------------------------------------------------------------------------- def erwarte(bedingung, meldung): if not bedingung: raise Fehlschlag(meldung) def erwarte_status(resp, *codes): if resp.status not in codes: raise Fehlschlag("%s %s: HTTP %d erwartet %s | %s" % ( resp.method, resp.url, resp.status, '/'.join(str(c) for c in codes), resp.kurz(300))) return resp def erwarte_erfolg(resp): erwarte_status(resp, 200, 201) if not resp.ok: raise Fehlschlag("%s %s: Antwort ohne status=success | %s" % ( resp.method, resp.url, resp.kurz(300))) return resp def erwarte_fehler(resp, http=None, code=None): if http is not None: erwarte_status(resp, *(http if isinstance(http, (list, tuple)) else [http])) erwarte(isinstance(resp.json, dict) and resp.json.get('status') == 'error', "%s %s: Fehlerantwort erwartet, kam: %s" % (resp.method, resp.url, resp.kurz(300))) if code is not None: codes = code if isinstance(code, (list, tuple)) else [code] erwarte(resp.fehlercode in codes, "%s %s: error.code %s erwartet, war %r" % ( resp.method, resp.url, '/'.join(codes), resp.fehlercode)) return resp def ueberspringen(grund): raise Uebersprungen(grund) # --------------------------------------------------------------------------- # Registrierung # --------------------------------------------------------------------------- class Fall: def __init__(self, gruppe, name, fn, doc): self.gruppe = gruppe self.name = name self.fn = fn self.doc = doc FAELLE = [] def test(gruppe, name): """Registriert einen Testfall. Reihenfolge der Registrierung = Ablauf.""" def deco(fn): FAELLE.append(Fall(gruppe, name, fn, (fn.__doc__ or '').strip())) return fn return deco class Ctx: """ Der Zustand, den Testfaelle teilen. Ein Bugtracker-Item muss angelegt sein, bevor es sich uebernehmen laesst; ein Sub-Token muss existieren, bevor sich pruefen laesst, was es nicht darf. Solche Ketten hier abzubilden ist ehrlicher, als jedem Testfall ein eigenes Item unterzuschieben - geprueft wird schliesslich der Ablauf, den ein Agent tatsaechlich nimmt. """ def __init__(self, cfg, client): self.cfg = cfg self.client = client self.daten = {} def __getitem__(self, k): if k not in self.daten: ueberspringen("Voraussetzung fehlt: %s (vorheriger Testfall nicht durchgelaufen)" % k) return self.daten[k] def __setitem__(self, k, v): self.daten[k] = v def get(self, k, default=None): return self.daten.get(k, default) def hat(self, k): return k in self.daten # -- Steuerung der Instanz ------------------------------------------ def testctl(self, action, **params): body = dict(params) body['action'] = action return self.client.post('/testctl.php', body=body, token=self.cfg['keys']['shared_key']) # --------------------------------------------------------------------------- # Konfiguration # --------------------------------------------------------------------------- def lade_bekannte_befunde(): """ Testfaelle, die einen gemeldeten, noch offenen Befund beschreiben. Siehe die Begruendung in bekannte_befunde.json selbst: Ein dauerhaft roter Lauf ist genauso wertlos wie ein geloeschter Testfall - beim einen sieht niemand mehr hin, beim anderen ist der Befund vergessen. """ p = Path(__file__).resolve().parent / 'bekannte_befunde.json' if not p.exists(): return {} with open(p, 'r', encoding='utf-8') as f: roh = json.load(f) return {k: v for k, v in roh.items() if not k.startswith('_')} def lade_config(pfad=None): p = Path(pfad) if pfad else DEV_CONFIG if not p.exists(): print("Konfiguration fehlt: %s" % p) print("Vorlage: scripts/deploy_dev_config.example.json") sys.exit(2) with open(p, 'r', encoding='utf-8') as f: return json.load(f) # --------------------------------------------------------------------------- # Ausfuehrung # --------------------------------------------------------------------------- def run(cfg, gruppen=None, muster=None, reset=True, bericht_datei=None, stoppen=False, adressfamilie=None): host = urllib.parse.urlparse(cfg['base_url']).hostname weg = waehle_adressfamilie(host, 443, adressfamilie) client = Client(cfg['base_url'], cfg['basic_auth']['user'], cfg['basic_auth']['pass']) ctx = Ctx(cfg, client) faelle = FAELLE if gruppen: faelle = [f for f in faelle if f.gruppe in gruppen] if muster: m = muster.lower() faelle = [f for f in faelle if m in f.name.lower() or m in f.gruppe.lower()] if not faelle: print("Kein Testfall passt zur Auswahl.") return 2 print("Testinstanz : %s (%s)" % (cfg['base_url'], weg)) print("Testfaelle : %d in %d Gruppe(n)" % ( len(faelle), len(set(f.gruppe for f in faelle)))) if reset: print("") print("Ausgangszustand herstellen ...") r = ctx.testctl('reset', confirm=cfg['db']['name'], admin_user=cfg['admin']['user'], admin_pass=cfg['admin']['pass']) if not r.ok: print(Farbe.rot(" Zuruecksetzen fehlgeschlagen: %s" % r.kurz(500))) return 2 ctx['master_token'] = r.feld('master', 'raw_token') ctx['admin'] = r.feld('admin') # Das Installationskonto. Seit 2.9.3 nimmt /api/setup/v1/login nur # noch die Rolle "installer" an; die Setup-Faelle brauchen es. ctx['installer'] = r.feld('installer') client.token = ctx['master_token'] print(" %d Migration(en), %d Tabellen, Master-Token bereit." % ( len(r.feld('migration', 'applied', default=[])), r.feld('state', 'tables', default=0))) else: if not cfg.get('master_token'): print(Farbe.gelb(" Ohne --reset wird ein master_token in der Konfiguration erwartet.")) ctx['master_token'] = cfg.get('master_token') client.token = ctx['master_token'] ergebnisse = [] gruppe_aktiv = None bekannte = lade_bekannte_befunde() start_gesamt = time.time() for fall in faelle: if fall.gruppe != gruppe_aktiv: gruppe_aktiv = fall.gruppe print("") print(Farbe.grau("--- %s %s" % (gruppe_aktiv, '-' * max(0, 58 - len(gruppe_aktiv))))) start = time.time() zustand, meldung = 'ok', '' try: fall.fn(ctx) except Uebersprungen as e: zustand, meldung = 'skip', str(e) except Fehlschlag as e: zustand, meldung = 'fail', str(e) except Exception as e: # unerwarteter Fehler im Testfall selbst zustand = 'fail' meldung = "%s: %s" % (type(e).__name__, e) ms = int((time.time() - start) * 1000) # Bekannter, gemeldeter Befund? Dann ist ein Fehlschlag erwartet und # blockiert nichts - und ein unerwartetes Bestehen ist die Nachricht. bekannt = bekannte.get("%s/%s" % (fall.gruppe, fall.name)) item = bekannt.get('item') if bekannt else None if bekannt and zustand == 'fail': zustand = 'bekannt' elif bekannt and zustand == 'ok': zustand = 'behoben' ergebnisse.append({'gruppe': fall.gruppe, 'name': fall.name, 'zustand': zustand, 'ms': ms, 'meldung': meldung, 'zweck': fall.doc, 'item': item}) if zustand == 'ok': print(" %s %-52s %s" % (Farbe.gruen('OK '), fall.name, Farbe.grau('%4d ms' % ms))) elif zustand == 'bekannt': print(" %s %-52s %s" % (Farbe.gelb('BEK '), fall.name, Farbe.grau('bekannter Befund, Item #%s' % item))) elif zustand == 'behoben': print(" %s %-52s %s" % (Farbe.gruen('NEU!'), fall.name, Farbe.gruen('besteht wieder - Item #%s schliessen ' 'und Eintrag aus bekannte_befunde.json ' 'entfernen' % item))) elif zustand == 'skip': print(" %s %-52s %s" % (Farbe.gelb('UEBR'), fall.name, Farbe.grau(meldung[:60]))) else: print(" %s %-52s %s" % (Farbe.rot('FEHL'), fall.name, Farbe.grau('%4d ms' % ms))) for zeile in meldung.splitlines(): print(" %s" % Farbe.rot(zeile)) if stoppen: break dauer = time.time() - start_gesamt n_ok = sum(1 for e in ergebnisse if e['zustand'] == 'ok') n_fail = sum(1 for e in ergebnisse if e['zustand'] == 'fail') n_skip = sum(1 for e in ergebnisse if e['zustand'] == 'skip') n_bekannt = sum(1 for e in ergebnisse if e['zustand'] == 'bekannt') n_behoben = sum(1 for e in ergebnisse if e['zustand'] == 'behoben') # Ein Lauf, in dem fast nichts lief, ist kein bestandener Lauf. # # Uebersprungene Faelle zaehlten bisher nirgends hinein. Damit endete # `run.py --nur roadmap` mit "2 bestanden, 0 fehlgeschlagen, 5 # uebersprungen" und Rueckgabewert 0 - gruen, obwohl die Voraussetzung # (ctx['freies_item'] aus der Gruppe bugtracker) vom Filter # ausgeschlossen war und fuenf von sieben Faellen gar nicht liefen. # # Das Tor vor der Produktion war davon nie betroffen, dort laeuft alles. # Betroffen war der Moment, in dem ein Mensch einem einzelnen Ergebnis # glauben soll - und genau dort muss das Signal ehrlich sein. # # Kein Fehlschlag: Nichts ist kaputt. Aber auch kein Erfolg, denn geprueft # wurde es nicht. Dafuer gibt es den dritten Rueckgabewert, den run.py # ohnehin schon kennt: 2 = konnte nicht ordentlich laufen. unbrauchbar = n_fail == 0 and n_skip > (n_ok + n_behoben + n_bekannt) print("") print("=" * 70) zusammenfassung = "%d bestanden, %d fehlgeschlagen, %d uebersprungen" % ( n_ok + n_behoben, n_fail, n_skip) if n_bekannt: zusammenfassung += ", %d bekannte Befunde" % n_bekannt zusammenfassung += " (%.1f s, %d Aufrufe)" % (dauer, client.aufrufe) if n_fail: print(Farbe.rot(zusammenfassung)) elif unbrauchbar: print(Farbe.gelb(zusammenfassung)) else: print(Farbe.gruen(zusammenfassung)) if unbrauchbar: print("") print(Farbe.gelb( "Mehr uebersprungen als geprueft - dieser Lauf sagt nichts aus.")) print(Farbe.grau( " Uebersprungen wird, wenn eine Voraussetzung aus einem frueheren")) print(Farbe.grau( " Testfall fehlt. Bei --gruppe/--nur ist der meist mit weggefiltert:")) print(Farbe.grau( " ctx['freies_item'] etwa entsteht in der Gruppe bugtracker.")) print(Farbe.grau( " Abhilfe: die Gruppe mit aufnehmen, oder ohne Filter laufen lassen.")) if n_fail: print("") print("Fehlgeschlagen:") for e in ergebnisse: if e['zustand'] == 'fail': print(" - %s / %s" % (e['gruppe'], e['name'])) if n_bekannt: print("") print("Bekannte Befunde (gemeldet, noch offen - blockieren nichts):") for e in ergebnisse: if e['zustand'] == 'bekannt': print(" - Item #%s %s / %s" % (e['item'], e['gruppe'], e['name'])) if n_behoben: print("") print(Farbe.gruen("Wieder bestanden - Eintrag aus bekannte_befunde.json entfernen:")) for e in ergebnisse: if e['zustand'] == 'behoben': print(" - Item #%s %s / %s" % (e['item'], e['gruppe'], e['name'])) if bericht_datei: bericht = { 'instanz': cfg['base_url'], 'zeit_utc': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'dauer_s': round(dauer, 2), 'bestanden': n_ok + n_behoben, 'fehlgeschlagen': n_fail, 'uebersprungen': n_skip, 'bekannte_befunde': n_bekannt, 'wieder_bestanden': n_behoben, # Damit ein ausgewerteter Bericht dieselbe Auskunft gibt wie der # Rueckgabewert - sonst laese ein Werkzeug "0 fehlgeschlagen" und # hielte einen Lauf fuer bestanden, der nichts geprueft hat. 'aussagekraeftig': not unbrauchbar, 'faelle': ergebnisse, } Path(bericht_datei).write_text( json.dumps(bericht, indent=2, ensure_ascii=False), encoding='utf-8') print("") print("Bericht: %s" % bericht_datei) if n_fail: return 1 return 2 if unbrauchbar else 0