Projektwissen/scripts/run_tests.py

#!/usr/bin/env python3
"""Führt die unittest-Suiten der sn_*-Plugins im Mock-Modus (ohne QGIS) aus.

Die Plugins werden unter ihrem Paketnamen (sn_basis, sn_plan41, ...) in ein
temporäres Verzeichnis verlinkt, damit die Importe wie im QGIS-Plugin-Ordner
funktionieren. Läuft auf Linux, macOS und Windows mit jedem Python >= 3.11.

Aufrufe (aus dem Workspace-Root):
    python scripts/run_tests.py                       # alle Pakete mit Tests
    python scripts/run_tests.py sn_basis              # nur ein Paket
    python scripts/run_tests.py sn_basis --verbose    # volle unittest-Ausgabe
    python scripts/run_tests.py --baseline            # Vergleich mit docs/status/testbaseline.json
    python scripts/run_tests.py --update-baseline     # Baseline neu schreiben (nur nach Absprache)

Exit-Code: 0 = alle Suiten grün (bzw. mit --baseline: keine neuen roten Tests), sonst 1.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import tomllib
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
MANIFEST = ROOT / "workspace.toml"
BASELINE_DEFAULT = ROOT / "docs" / "status" / "testbaseline.json"


def lade_manifest() -> dict:
    with MANIFEST.open("rb") as f:
        return tomllib.load(f)


def verlinke(quelle: Path, ziel: Path) -> str:
    """Verknüpft quelle als ziel (Symlink, Windows-Junction oder Kopie)."""
    try:
        os.symlink(quelle, ziel, target_is_directory=True)
        return "symlink"
    except (OSError, NotImplementedError):
        pass
    if os.name == "nt":
        try:
            import _winapi  # type: ignore[attr-defined]

            _winapi.CreateJunction(str(quelle), str(ziel))
            return "junction"
        except Exception:
            pass
    shutil.copytree(
        quelle, ziel,
        ignore=shutil.ignore_patterns(".git", "__pycache__", "*.pyc"),
    )
    return "kopie"


def parse_ergebnis(ausgabe: str) -> dict:
    m = re.search(r"^Ran (\d+) tests?", ausgabe, re.M)
    gesamt = int(m.group(1)) if m else 0
    fail = err = skip = 0
    m = re.search(r"^FAILED \((.*)\)", ausgabe, re.M)
    if m:
        for teil in m.group(1).split(","):
            k, _, v = teil.strip().partition("=")
            if k == "failures":
                fail = int(v)
            elif k == "errors":
                err = int(v)
            elif k == "skipped":
                skip = int(v)
    else:
        m = re.search(r"^OK(?: \(skipped=(\d+)\))?", ausgabe, re.M)
        if m and m.group(1):
            skip = int(m.group(1))
    rot = sorted(set(re.findall(r"^(?:FAIL|ERROR): (\S+ \([\w.]+\))", ausgabe, re.M)))
    return {
        "gesamt": gesamt, "fehlschlaege": fail, "fehler": err,
        "uebersprungen": skip, "gruen": gesamt - fail - err - skip, "rot": rot,
    }


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("pakete", nargs="*", help="Paketnamen (Standard: alle mit tests=true)")
    ap.add_argument("--verbose", "-v", action="store_true", help="volle unittest-Ausgabe anzeigen")
    ap.add_argument("--baseline", action="store_true", help="mit Baseline vergleichen")
    ap.add_argument("--update-baseline", action="store_true", help="Baseline aus diesem Lauf schreiben")
    ap.add_argument("--baseline-datei", type=Path, default=BASELINE_DEFAULT)
    ap.add_argument("--json", type=Path, help="Ergebnis als JSON in diese Datei schreiben")
    args = ap.parse_args()

    manifest = lade_manifest()
    repos = {r["paket"]: r for r in manifest["repos"] if r.get("paket")}
    testbare = [p for p, r in repos.items() if r.get("tests")]
    gewuenscht = args.pakete or testbare
    unbekannt = [p for p in gewuenscht if p not in repos]
    if unbekannt:
        print(f"Unbekannte Pakete: {unbekannt}. Bekannt: {sorted(repos)}")
        return 2

    # Abhängigkeiten immer mit verlinken (sn_basis wird von allen importiert;
    # sn_basis.DataGrabber importiert sn_plan41 lazy).
    zu_verlinken: set[str] = set()
    for p in gewuenscht:
        zu_verlinken.add(p)
        zu_verlinken.update(repos[p].get("abhaengigkeiten", []))
    zu_verlinken.update(p for p in testbare)

    fehlend = [p for p in zu_verlinken if not (ROOT / repos[p]["name"]).is_dir()]
    if fehlend:
        print(f"Repository-Ordner fehlen: {[repos[p]['name'] for p in fehlend]} "
              f"-> zuerst 'python scripts/setup_workspace.py' ausführen.")
        return 2

    ergebnisse: dict[str, dict] = {}
    rc = 0
    with tempfile.TemporaryDirectory(prefix="sn_tests_") as tmp:
        tmp_pfad = Path(tmp)
        for p in sorted(zu_verlinken):
            verlinke(ROOT / repos[p]["name"], tmp_pfad / p)

        env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", PYTHONIOENCODING="utf-8", PYTHONPATH=str(tmp_pfad))
        for p in gewuenscht:
            if not repos[p].get("tests"):
                print(f"[{p}] keine Tests vorhanden – übersprungen")
                continue
            cmd = [sys.executable, "-m", "unittest", "discover", "-s", f"{p}/tests", "-t", ".", "-p", "test_*.py"]
            proz = subprocess.run(cmd, cwd=tmp_pfad, env=env, capture_output=True, text=True, errors="replace")
            ausgabe = proz.stdout + proz.stderr
            erg = parse_ergebnis(ausgabe)
            ergebnisse[p] = erg
            if args.verbose:
                print(f"\n{'=' * 70}\n[{p}] {' '.join(cmd)}\n{'=' * 70}\n{ausgabe}")
            status = "GRÜN" if not erg["rot"] and erg["gesamt"] else "ROT"
            print(f"[{p}] {status}: {erg['gruen']}/{erg['gesamt']} bestanden, "
                  f"{erg['fehlschlaege']} Fehlschläge, {erg['fehler']} Fehler, {erg['uebersprungen']} übersprungen")
            if erg["gesamt"] == 0:
                print(f"    Keine Tests gefunden oder Import-Fehler. Mit --verbose prüfen.")
                rc = 1

    # Baseline-Vergleich: verhindert, dass bereits rote Tests "vermehrt" werden.
    if args.baseline or args.update_baseline:
        alt: dict = {}
        baseline_vorhanden = args.baseline_datei.exists()
        if baseline_vorhanden:
            alt = json.loads(args.baseline_datei.read_text(encoding="utf-8"))
        elif not args.update_baseline:
            print(f"Keine Baseline unter {args.baseline_datei.relative_to(ROOT)} – alle roten Tests gelten als neu.")
        neu_rot_gesamt = 0
        for p, erg in (ergebnisse.items() if (baseline_vorhanden or not args.update_baseline) else []):
            alt_rot = set(alt.get(p, {}).get("rot", []))
            neu_rot = sorted(set(erg["rot"]) - alt_rot)
            neu_gruen = sorted(alt_rot - set(erg["rot"]))
            if neu_rot:
                neu_rot_gesamt += len(neu_rot)
                print(f"[{p}] NEU ROT ({len(neu_rot)}):")
                for t in neu_rot:
                    print(f"    - {t}")
            if neu_gruen:
                print(f"[{p}] neu grün ({len(neu_gruen)}): {', '.join(neu_gruen)}")
            if not neu_rot and not neu_gruen:
                print(f"[{p}] unverändert gegenüber Baseline")
        if args.update_baseline:
            alt.update({p: {k: v for k, v in erg.items()} for p, erg in ergebnisse.items()})
            args.baseline_datei.parent.mkdir(parents=True, exist_ok=True)
            args.baseline_datei.write_text(json.dumps(alt, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
            print(f"Baseline geschrieben: {args.baseline_datei.relative_to(ROOT)}")
        rc = 1 if neu_rot_gesamt else rc
    else:
        for erg in ergebnisse.values():
            if erg["rot"]:
                rc = 1
        if rc:
            print("\nRote Tests (Kurzliste, Details mit --verbose):")
            for p, erg in ergebnisses_sortiert(ergebnisse):
                for t in erg["rot"]:
                    print(f"  [{p}] {t}")

    if args.json:
        args.json.write_text(json.dumps(ergebnisse, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    return rc


def ergebnisses_sortiert(ergebnisse: dict[str, dict]):
    return sorted(ergebnisse.items())


if __name__ == "__main__":
    raise SystemExit(main())