02.11.05 — Plugins de rapport#

Dossier source : src/ocarina/opinionated/plugins/reports/
pretty_print_results, results_to_json, generate_docx_proof, timing. Tous opt-in. Exécutables séquentiellement ou en parallèle via run_plugins.

`pretty_print_results`#

# src/ocarina/opinionated/plugins/reports/pretty_print_results.py
def pretty_print_results(results: TestCycleResults, *, with_colors: bool = False) -> None:
    """Display the full test results."""

Campaign
• Suite
  > Test case 1
    » PASSED
  > Test case 2
    » FAILED
      → Error message
        ⫸ At step 3
  > Test case 3
    » SKIPPED

Test results:
1 FAILED | 1 PASSED | 1 SKIPPED

Aspect	Détail
Hiérarchie	3 niveaux : campagne `•`, suite `>`, cas `»`
Niveaux	`PASSED` (vert), `FAILED` (rouge), `SKIPPED` (gris).
Erreur	Message + step number où ça a fail (« At step 3 »)
Summary	Comptage agrégé en bas

`results_to_json`#

def generate_json_results(*, results: TestCycleResults, output_dir: Path, logger: ILogger) -> None:
    """Write test results to a JSON file in results_dir."""
    output_dir.mkdir(parents=True, exist_ok=True)
    max_stem_length = 8
    max_attempts = 500
    for _ in range(max_attempts):
        filename = f"{uuid.uuid4().hex[:max_stem_length]}.json"
        file_path = output_dir / filename
        if not file_path.exists():
            break
    else:
        raise RuntimeError(f"Can't generate unique JSON filename in: {output_dir}.")

{
  "Campaign": {
    "Suite": {
      "test_case": [
        {"status": "success" | "fail", "error": "..."},
        counter,
        metadata
      ]
    }
  }
}

def _result_to_serializable(v: Any) -> dict[str, str]:
    if is_ok(v):
        return {"status": "success"}
    if is_fail(v):
        return {"status": "fail", "error": str(v.error)}
    raise TypeError(f"Expected Ok or Fail instances, but got: {type(v)}")

Le filename est un UUID hex de 8 chars → 16⁸ ≈ 4 milliards d’options. Pas de collision dans la pratique. 500 retries de sécurité.

`generate_docx_proof`#

def generate_docx_proof(
    *,
    logs_root: Path,
    output_root: Path,
    logger: ILogger,
    max_workers: int = 1,
    format_date: Callable[[datetime], str] = _default_format_date,
) -> None:
    """Transform text log files from automated tests into formatted Word documents,
    including screenshots."""

Crée un sous-répertoire unique sous output_root (UUID hex 4 chars).
Parcourt l’arbre de logs (logs_root/campaign/suite/test.log).
Pour chaque .log : crée un .docx.
Dans chaque .docx :
- Heading 1 : nom de campagne.
- Heading 2 : nom de suite.
- Heading 3 : nom de cas.
- Body : lignes des logs, avec dates UTC adaptées à l’heure locale.
- Quand la ligne contient "Screenshot: <path>" : insère l’image.

Génération parallélisée (depuis 1.1.9)#

Chaque cas constitue une opération disque indépendante : il lit son log, construit son Document et écrit dans un fichier de sortie qui lui est propre. Les cas se parallélisent donc sans accroc. Depuis Ocarina 1.1.9, la génération peut être parallélisée à la demande, via max_workers :

max_workers <= 1 (par défaut) : le traitement séquentiel d’origine, inchangé — aucune liste matérialisée, aucun pool, aucun thread.
max_workers > 1 : les cas sont répartis sur un ThreadPoolExecutor dont le nombre de workers est borné au nombre de cas. Le seul objet partagé est le logger, dont les écritures peuvent s’entrelacer — sans conséquence pour un reporter best-effort.

Le gain est réel mais se prête mal à un multiplicateur unique : il dépend du nombre de cas, de la taille des logs et du poids des images. Sur la CI e2e d’ocarina-with-playwright-example (50 DOCX), l’activer divise presque par deux la durée de génération.

_DEFAULT_UTC_DATE_REGEX = re.compile(r"\[UTC_DATE::([^]]+)]")

def _default_format_date(dt: datetime) -> str:
    return dt.strftime("[%m/%d/%Y | %Hh%M:%S.%f]")

def _replace_utc_date(line: str, *, utc_date_regex, format_date) -> str:
    def _repl(m: re.Match[str]) -> str:
        with suppress(Exception):
            local_dt = datetime.fromisoformat(m.group(1).replace("Z", "+00:00")).astimezone()
            return format_date(local_dt)
        return m.group(0)
    return utc_date_regex.sub(_repl, line)

→ [UTC_DATE::2026-05-18T09:42:31.123456+00:00] devient [05/18/2026 | 11h42:31.123456] (heure locale).

Formateur de date personnalisé (depuis 1.1.10)#

Le format américain par défaut ci-dessus (_default_format_date) n’est plus figé dans le code. Depuis Ocarina 1.1.10, generate_docx_proof accepte un callable format_date : il reçoit la date du marqueur déjà ramenée à l’heure locale et renvoie le texte de remplacement complet, crochets compris. C’est donc à l’utilisateur de décider de la mise en forme.

generate_docx_proof(
    logs_root=...,
    output_root=...,
    logger=...,
    format_date=lambda dt: dt.strftime("FR[%d/%m/%Y à %Hh%M:%S]"),
)

→ le même marqueur s’affiche désormais FR[18/05/2026 à 11h42:31]. Sans argument, le format par défaut demeure l’affichage américain : le code appelant existant n’a rien à changer.

Génération durcie (depuis 1.1.10)#

1.1.10 met aussi la génération à l’abri des courses entre workers et la rend plus honnête sur les échecs partiels :

Réservation atomique du chemin : le repli sur un nom raccourci (quand un nom est trop long) ne procède plus en deux temps (vérifier l’existence puis créer). Le fichier est réservé en un seul appel système — os.open(new_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) — si bien que deux workers parallélisés ne peuvent plus revendiquer en même temps le même nom dérivé d’un UUID. Une tentative qui échoue supprime au passage son fichier témoin via docx_path.unlink().
Statistiques de génération : le compteur interne devient un _GenerationStats(attempted, succeeded) plutôt qu’un simple int, ce qui permet de distinguer les cas découverts des fichiers réellement écrits. Le reporter affiche alors le ratio — Generated 1/2 DOCX en cas d’échec partiel — et signale every DOCX generation failed au lieu du trompeur no test case found lorsque rien n’a été écrit.
Unicité des noms insensible à la casse : les noms de campagne, de suite et de cas de test sont désormais comparés via la clé unicodedata.normalize("NFC", name).casefold(), si bien que Login et login sont reconnus comme un doublon, comme il se doit (names must be unique (case-insensitive)). Les noms d’origine restent intacts dans le modèle : seule la clé de comparaison subit le casefold.

Détection des screenshots#

_DEFAULT_SCREENSHOT_NEEDLE = "Screenshot: "

Quand une ligne contient cette needle, on extrait le path, on insère l’image dans le doc. Le path est tronqué proprement (UUID) si trop long.

`timing`#

from contextlib import contextmanager

@contextmanager
def timing(*, prefix: str = "Duration:", seconds_label: str = "seconds"):
    start = time.perf_counter()
    interrupted = False
    try:
        yield
    except (KeyboardInterrupt, WarmupTimeoutError):
        interrupted = True
        raise
    finally:
        if not interrupted:
            end = time.perf_counter()
            elapsed = end - start
            human_readable = format_elapsed(elapsed)
            p = f"{prefix} " if prefix else ""
            print(f"\n{p}{human_readable} ({elapsed:.2f} {seconds_label})")

with timing(prefix="Tests duration:"):
    bootstrap(...)

→ À la sortie du with :

Tests duration: 5 minutes and 23 seconds (323.45 seconds)

Notes :

interrupted = True si KeyboardInterrupt ou WarmupTimeoutError : on ne pollue pas la sortie avec un temps de run interrompu.
format_elapsed : format human-readable (5 minutes and 23 seconds).

`format_elapsed`#

def format_elapsed(seconds: float) -> str:
    secs = int(seconds)
    days, secs = divmod(secs, 86400)
    hours, secs = divmod(secs, 3600)
    minutes, secs = divmod(secs, 60)

    parts = []
    if days: parts.append(f"{days} day{'s' if days > 1 else ''}")
    if hours: parts.append(f"{hours} hour{'s' if hours > 1 else ''}")
    if minutes: parts.append(f"{minutes} minute{'s' if minutes > 1 else ''}")
    if secs or not parts:
        parts.append(f"{secs} second{'s' if secs > 1 else ''}")

    if len(parts) > 1:
        return ", ".join(parts[:-1]) + " and " + parts[-1]
    return parts[0]

Singulier/pluriel automatique. Format Oxford comma-ish (« 2 hours, 5 minutes and 3 seconds »).

Exécution parallèle avec `run_plugins`#

def run_plugins(*plugins: Effect, exceptions_logger: ILogger) -> None:
    if not plugins:
        return

    def _run_plugin(plugin: Effect, exceptions_logger: ILogger) -> None:
        try:
            plugin()
        except Exception as exc:
            exceptions_logger.exception("The plugin failed.", exc=exc)

    if len(plugins) == 1:
        _run_plugin(plugins[0], exceptions_logger)
        return

    with ThreadPoolExecutor(max_workers=len(plugins)) as executor:
        futures = [executor.submit(_run_plugin, plugin, exceptions_logger) for plugin in plugins]
        for f in futures:
            f.result()

1 plugin → séquentiel (pas la peine de spawn un thread).
N plugins → ThreadPoolExecutor(max_workers=N) : tous en parallèle.
Aucun plugin ne tue les autres : chaque plugin est wrappé dans _run_plugin qui catch + log via exceptions_logger.

generate_docx_proof est le plugin le plus coûteux (parse de logs, génération Word, insertion d’images) ; generate_json_results prend 0.1s. Lancés ensemble, le temps réel écoulé est celui du plugin le plus lent, pas leur somme. Et depuis 1.1.9, ce goulot se parallélise lui aussi en interne (voir la note sur la génération parallélisée plus haut) : il ne pèse plus autant qu’avant.

Pourquoi `run_plugins` prend `results` en argument#

def bootstrap[T](
    *,
    test_cycle: TestCycle[T],
    run_plugins: Callable[[TestCycleResults], None],
    post_exec: Callable[[TestCycleResults], None] | None = None,
    saturate_workers: bool = True,
) -> None:
    results = test_cycle.run_all(saturate_workers=saturate_workers)
    run_plugins(results)
    if post_exec:
        post_exec(results)

L’utilisateur fournit un run_plugins: Callable[[TestCycleResults], None] :

run_plugins=lambda results: run_plugins(
    lambda: generate_docx_proof(logs_root=..., output_root=..., logger=...),
    lambda: generate_json_results(results=results, output_dir=..., logger=...),
    exceptions_logger=...,
),

→ run_plugins (le outer, lambda) capture results. Le inner run_plugins (la fonction du framework) exécute les plugins.

Cas important : generate_docx_proof n’a pas besoin de results (il lit l’arbre de logs sur le disque). Mais generate_json_results a besoin de results. La lambda (results) -> ... permet à chaque plugin de capturer ce dont il a besoin.

02.11.05 — Plugins de rapport#

pretty_print_results#

results_to_json#

generate_docx_proof#