oto-enterprise-os-dtp/05_deliverables_mvp/qa/regression/reglib/discovery.py

"""Découverte des suites de tests + preuve de couverture (faits, pas déclaration).

Les suites de la matrice sont DÉRIVÉES du CI Gitea (`parse_ci`), jamais listées à
la main : le périmètre « exhaustif » est donc auto-prouvé — tout job de test
ajouté au CI entre dans la matrice, toute suite retirée en sort. Pour chaque suite
on lit ensuite des FAITS de disque (fichiers `test_*.py`, méthodes `def test_`) —
aucun compteur de résultat n'est inventé ici (#6).

La couverture est PROUVÉE : chaque suite gated doit (a) exister sur disque avec un
répertoire `tests/`, (b) alimenter le job d'agrégat `gate`. Le harnais lui-même
(`self_module`) est exclu — séparation des pouvoirs (ISA 315), et il ne
s'auto-exécute pas (évite la récursion).
"""

from __future__ import annotations

import os
import re

from .deps import DELIVERABLES_ROOT, parse_ci

_TEST_FILE_RE = re.compile(r"^test_.*\.py$")
_TEST_METHOD_RE = re.compile(r"^\s*def\s+(test_\w+)\s*\(")


def _slug(path: str) -> str:
    """Identifiant stable d'une suite depuis son chemin (déterministe)."""
    return path.replace("/", ".")


def count_tests(tests_dir: str) -> tuple[int, int]:
    """(nb fichiers test_*.py, nb méthodes def test_*) dans `tests_dir`.

    Faits de disque purs — base du recensement exhaustif. Zéro exécution."""
    if not os.path.isdir(tests_dir):
        return (0, 0)
    files = 0
    methods = 0
    for name in os.listdir(tests_dir):
        if not _TEST_FILE_RE.match(name):
            continue
        files += 1
        with open(os.path.join(tests_dir, name), encoding="utf-8") as fh:
            for line in fh:
                if _TEST_METHOD_RE.match(line):
                    methods += 1
    return (files, methods)


def disk_test_modules() -> list[str]:
    """Chemins de modules (relatifs à DELIVERABLES_ROOT, POSIX) portant un
    répertoire `tests/` sur DISQUE — indépendamment du CI.

    Sert à prouver le sens INVERSE de la couverture : tout module qui gagne des
    tests sur disque DOIT être câblé au CI, sinon ses tests ne tournent jamais et
    n'entrent jamais dans la matrice (dérive silencieuse disque→CI). Faits de
    disque purs, ordonnés → déterministe."""
    mods: set[str] = set()
    for dirpath, dirnames, _ in os.walk(DELIVERABLES_ROOT):
        # On n'explore pas l'intérieur des caches/tests eux-mêmes.
        dirnames[:] = [d for d in dirnames if d != "__pycache__"]
        if os.path.basename(dirpath) == "tests":
            module_abs = os.path.dirname(dirpath)
            rel = os.path.relpath(module_abs, DELIVERABLES_ROOT)
            mods.add(rel.replace(os.sep, "/"))
    return sorted(mods)


def discover_suites(spec: dict) -> list[dict]:
    """Liste ordonnée des suites gated (hors self), enrichie des faits de disque.

    Ordonné par `path` → sortie déterministe et diffable."""
    self_module = spec["self_module"]
    ci = parse_ci()
    job_to_path = ci["job_to_path"]
    gate_needs = set(ci["gate_needs"])

    path_to_jobs: dict[str, list[str]] = {}
    for job, path in job_to_path.items():
        path_to_jobs.setdefault(path, []).append(job)

    suites: list[dict] = []
    for path in sorted(p for p in job_to_path.values() if p != self_module):
        jobs = sorted(path_to_jobs[path])
        abs_path = os.path.join(DELIVERABLES_ROOT, path)
        tests_dir = os.path.join(abs_path, "tests")
        files, methods = count_tests(tests_dir)
        suites.append({
            "id": _slug(path),
            "path": path,
            "jobs": jobs,
            "in_gate": all(j in gate_needs for j in jobs),
            "has_tests_dir": os.path.isdir(tests_dir),
            "test_files": files,
            "test_methods": methods,
        })
    return suites


def coverage_report(spec: dict, suites: list[dict]) -> dict:
    """Diagnostic de couverture sérialisable. `ok=True` ssi chaque suite gated a
    un répertoire `tests/` sur disque ET alimente le gate, et que le harnais
    lui-même est bien exclu du périmètre."""
    self_module = spec["self_module"]
    ci = parse_ci()
    ci_paths = set(ci["job_to_path"].values())

    missing_tests_dir = sorted(s["path"] for s in suites if not s["has_tests_dir"])
    not_in_gate = sorted(s["path"] for s in suites if not s["in_gate"])
    self_present = any(s["path"] == self_module for s in suites)

    # Sens inverse : tout module portant un `tests/` sur disque DOIT être gated
    # (sinon ses tests ne tournent jamais → dérive silencieuse disque→CI).
    orphan_tests_dirs = sorted(set(disk_test_modules()) - ci_paths)

    ok = (
        not missing_tests_dir
        and not not_in_gate
        and not self_present
        and not orphan_tests_dirs
        and self_module in ci_paths   # le harnais DOIT être gated lui aussi
    )
    return {
        "ok": ok,
        "suites_count": len(suites),
        "gated_in_ci": len(ci_paths),
        "self_module_excluded": self_module,
        "self_module_gated": self_module in ci_paths,
        "missing_tests_dir": missing_tests_dir,
        "not_in_gate": not_in_gate,
        "orphan_tests_dirs": orphan_tests_dirs,
    }