Deploymentcenter/tests/e2e/harness.py

#!/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=3.0):
    try:
        infos = _ECHTES_GETADDRINFO(host, port, familie, socket.SOCK_STREAM)
    except socket.gaierror:
        return False
    for info in infos:
        s = socket.socket(info[0], socket.SOCK_STREAM)
        s.settimeout(timeout)
        try:
            s.connect(info[4])
            return True
        except OSError:
            continue
        finally:
            s.close()
    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')
        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