feat(ci): redesign quarantine report for readability and error visibility

mergify_cli/ci/junit_processing/cli.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import sys
+import typing
 
 import click
 import opentelemetry.trace
@@ -10,6 +11,13 @@
 from mergify_cli.ci.junit_processing import upload
 
 
+if typing.TYPE_CHECKING:
+    from opentelemetry.sdk.trace import ReadableSpan
+
+SEPARATOR = "══════════════════════════════════════════"
+SEPARATOR_LIGHT = "──────────────────────────────────────────"
+
+
 async def process_junit_files(
     *,
     api_url: str,
@@ -20,30 +28,35 @@ async def process_junit_files(
     tests_target_branch: str,
     files: tuple[str, ...],
 ) -> None:
-    click.echo("🚀 CI Insights · Upload JUnit")
-    click.echo("────────────────────────────")
-    click.echo(f"📂 Discovered reports: {len(files)}")
+    # ── Header ──
+    click.echo(SEPARATOR)
+    click.echo("  🚀 CI Insights")
+    click.echo("")
+    click.echo(
+        "  Uploads JUnit test results to Mergify CI Insights and evaluates\n"
+        "  quarantine status for failing tests. This step determines the\n"
+        "  final CI status — quarantined failures are ignored.\n"
+        "  Learn more: https://docs.mergify.com/ci-insights/quarantine",
+    )
+    click.echo(SEPARATOR)
 
     try:
-        spans = await junit.files_to_spans(
+        run_id, spans = await junit.files_to_spans(
             files,
             test_language=test_language,
             test_framework=test_framework,
         )
     except junit.InvalidJunitXMLError as e:
-        click.echo(
-            click.style(
-                f"❌ Error converting JUnit XML file to spans: {e.details}",
-                fg="red",
-            ),
-            err=True,
+        _print_early_exit_error(
+            f"Failed to parse JUnit XML: {e.details}",
+            "Check that your test framework is generating valid JUnit XML output.",
         )
         sys.exit(1)
 
     if not spans:
-        click.echo(
-            click.style("❌ No spans found in the JUnit files", fg="red"),
-            err=True,
+        _print_early_exit_error(
+            "No spans found in the JUnit files",
+            "Check that the JUnit XML files are not empty.",
         )
         sys.exit(1)
 
@@ -54,9 +67,9 @@ async def process_junit_files(
     ]
 
     if not tests_cases:
-        click.echo(
-            click.style("❌ No test cases found in the JUnit files", fg="red"),
-            err=True,
+        _print_early_exit_error(
+            "No test cases found in the JUnit files",
+            "Check that your test step ran successfully before this step.",
         )
         sys.exit(1)
 
@@ -67,55 +80,63 @@ async def process_junit_files(
             if span.status.status_code == opentelemetry.trace.StatusCode.ERROR
         ],
     )
-    nb_success_spans = len(
-        [
-            span
-            for span in tests_cases
-            if span.status.status_code == opentelemetry.trace.StatusCode.OK
-        ],
-    )
-    click.echo(
-        f"🧪 Parsed tests: {len(tests_cases)} (✅ passed: {nb_success_spans} | ❌ failed: {nb_failing_spans})",
-    )
+    click.echo("")
+    click.echo(f"  Run ID: {run_id}")
 
     # NOTE: Check quarantine before uploading in order to properly modify the
     # "cicd.test.quarantined" attribute for the required spans.
     quarantine_final_failure_message: str | None = None
+    quarantine_error: str | None = None
     try:
-        failing_tests_not_quarantined_count = (
-            await quarantine.check_and_update_failing_spans(
-                api_url,
-                token,
-                repository,
-                tests_target_branch,
-                spans,
-            )
+        result = await quarantine.check_and_update_failing_spans(
+            api_url,
+            token,
+            repository,
+            tests_target_branch,
+            spans,
         )
     except quarantine.QuarantineFailedError as exc:
-        click.echo(click.style(exc.message, fg="red"), err=True)
-        click.echo(
-            click.style(quarantine.QUARANTINE_INFO_ERROR_MSG, fg="red"),
-            err=True,
-        )
+        quarantine_error = exc.message
         quarantine_final_failure_message = (
-            "Unable to determine quarantined failures due to above error"
-        )
-    except Exception as exc:
-        msg = (
-            f"❌ An unexpected error occurred when checking quarantined tests: {exc!s}"
+            f"Treating {nb_failing_spans}/{nb_failing_spans} failures as blocking"
         )
-        click.echo(click.style(msg, fg="red"), err=True)
-        click.echo(
-            click.style(quarantine.QUARANTINE_INFO_ERROR_MSG, fg="red"),
-            err=True,
+        failing_spans = [
+            span
+            for span in tests_cases
+            if span.status.status_code == opentelemetry.trace.StatusCode.ERROR
+        ]
+        result = quarantine.QuarantineResult(
+            failing_spans=failing_spans,
+            quarantined_spans=[],
+            non_quarantined_spans=failing_spans,
+            failing_tests_not_quarantined_count=len(failing_spans),
         )
+    except Exception as exc:
+        quarantine_error = str(exc)
         quarantine_final_failure_message = (
-            "Unable to determine quarantined failures due to above error"
+            f"Treating {nb_failing_spans}/{nb_failing_spans} failures as blocking"
+        )
+        failing_spans = [
+            span
+            for span in tests_cases
+            if span.status.status_code == opentelemetry.trace.StatusCode.ERROR
+        ]
+        result = quarantine.QuarantineResult(
+            failing_spans=failing_spans,
+            quarantined_spans=[],
+            non_quarantined_spans=failing_spans,
+            failing_tests_not_quarantined_count=len(failing_spans),
         )
     else:
-        if failing_tests_not_quarantined_count > 0:
-            quarantine_final_failure_message = f"{failing_tests_not_quarantined_count} unquarantined failures detected ({nb_failing_spans - failing_tests_not_quarantined_count} quarantined)"
+        if result.failing_tests_not_quarantined_count > 0:
+            count = result.failing_tests_not_quarantined_count
+            total = len(result.failing_spans)
+            quarantined = total - count
+            quarantine_final_failure_message = (
+                f"{quarantined}/{total} failures quarantined"
+            )
 
+    upload_error: str | None = None
     try:
         upload.upload(
             api_url=api_url,
@@ -124,23 +145,142 @@ async def process_junit_files(
             spans=spans,
         )
     except Exception as e:
-        click.echo(
-            click.style(f"❌ Error uploading JUnit XML reports: {e}", fg="red"),
-            err=True,
-        )
+        upload_error = str(e)
+
+    reports_label = f"{len(files)} {'report' if len(files) == 1 else 'reports'}"
+    failures_label = (
+        f"{nb_failing_spans} {'failure' if nb_failing_spans == 1 else 'failures'}"
+    )
+    if upload_error is not None:
+        click.echo(f"      ☁️ {reports_label} not uploaded")
+    else:
+        click.echo(f"      ☁️ {reports_label} uploaded")
+    click.echo(f"      🧪 {len(tests_cases)} tests ({failures_label})")
+
+    # ── Upload error ──
+    if upload_error is not None:
+        click.echo("\n  ⚠️ Failed to upload test results")
+        click.echo("    Mergify CI Insights won't process these test results.")
+        click.echo("    Quarantine status and CI outcome are unaffected.")
+        click.echo("")
+        click.echo("      ┌ Details")
+        for line in upload_error.splitlines():
+            click.echo(f"      │  {line}")
+        click.echo("      └─")
 
+    # ── Quarantine ──
+    if _print_quarantine_section(result, error=quarantine_error):
+        pass
+
+    # ── Verdict ──
+    nb_quarantined_failures = len(result.failing_spans) if result is not None else 0
+    click.echo("")
+    click.echo(SEPARATOR)
     if quarantine_final_failure_message is None:
-        click.echo("\n🎉 Verdict")
-        click.echo(
-            f"• Status: ✅ OK — all {nb_failing_spans} failures are quarantined (ignored for CI status)",
-        )
+        if nb_quarantined_failures == 0:
+            click.echo("✅ OK — all tests passed, no quarantine needed")
+        else:
+            click.echo(
+                f"✅ OK — {nb_quarantined_failures}/{nb_quarantined_failures}"
+                " failures quarantined, CI status unaffected",
+            )
         quarantine_exit_error_code = 0
     else:
-        click.echo("\n❌ Verdict")
         click.echo(
-            f"• Status: 🔴 FAIL — {quarantine_final_failure_message}",
+            f"❌ FAIL — {quarantine_final_failure_message}",
         )
         quarantine_exit_error_code = 1
 
-    click.echo(f"• Exit code: {quarantine_exit_error_code}")
+    click.echo(f"  Exit code: {quarantine_exit_error_code}")
+    click.echo(SEPARATOR)
     sys.exit(quarantine_exit_error_code)
+
+
+def _print_quarantine_section(
+    result: quarantine.QuarantineResult | None,
+    *,
+    error: str | None = None,
+) -> bool:
+    """Print the quarantine section. Returns True if anything was printed."""
+    if result is None and error is None:
+        return False
+
+    if result is not None and not result.failing_spans and error is None:
+        return False
+
+    click.echo("")
+    click.echo(SEPARATOR_LIGHT)
+    click.echo("")
+    click.echo("🛡️ Quarantine")
+
+    if error is not None:
+        click.echo("")
+        click.echo("  ⚠️ Failed to check quarantine status")
+        click.echo("    Contact Mergify support if this error persists.")
+        click.echo("")
+        click.echo("      ┌ Details")
+        for line in error.splitlines():
+            click.echo(f"      │  {line}")
+        click.echo("      └─")
+
+    if result is not None and result.quarantined_spans:
+        click.echo(f"\n  🔒 Quarantined ({len(result.quarantined_spans)}):")
+        for span in result.quarantined_spans:
+            click.echo(f"      · {span.name}")
+
+    if result is not None and result.non_quarantined_spans:
+        label = (
+            "Could not verify quarantine status"
+            if error is not None
+            else "Unquarantined"
+        )
+        click.echo(f"\n  ❌ {label} ({len(result.non_quarantined_spans)}):")
+        for span in result.non_quarantined_spans:
+            _print_failure_block(span)
+
+    return True
+
+
+def _print_failure_block(span: ReadableSpan) -> None:
+    """Print a box-drawn failure block for a single failing test span."""
+    click.echo(f"\n      ┌ {span.name}")
+
+    if span.attributes is None:
+        click.echo("      │")
+        click.echo("      │  (no error details in JUnit report)")
+        click.echo("      └─")
+        return
+
+    exc_type = span.attributes.get("exception.type")
+    exc_message = span.attributes.get("exception.message")
+    exc_stacktrace = span.attributes.get("exception.stacktrace")
+
+    if not exc_type and not exc_message and not exc_stacktrace:
+        click.echo("      │")
+        click.echo("      │  (no error details in JUnit report)")
+        click.echo("      └─")
+        return
+
+    if exc_type or exc_message:
+        parts = []
+        if exc_type:
+            parts.append(str(exc_type))
+        if exc_message:
+            parts.append(str(exc_message))
+        click.echo("      │")
+        click.echo(f"      │  {': '.join(parts)}")
+
+    if exc_stacktrace:
+        click.echo("      │")
+        for line in str(exc_stacktrace).splitlines():
+            click.echo(f"      │  {line}")
+
+    click.echo("      └─")
+
+
+def _print_early_exit_error(message: str, hint: str) -> None:
+    """Print an error section for early exits (invalid XML, no tests, etc.)."""
+    click.echo(f"❌ FAIL — {message}")
+    click.echo(f"  {hint}")
+    click.echo("  Exit code: 1")
+    click.echo(SEPARATOR)

mergify_cli/ci/junit_processing/junit.py

@@ -18,7 +18,6 @@
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 import opentelemetry.trace.span
 
-from mergify_cli import console
 from mergify_cli.ci import detector
 
 
@@ -37,11 +36,10 @@ async def files_to_spans(
     files: tuple[str, ...],
     test_language: str | None = None,
     test_framework: str | None = None,
-) -> list[ReadableSpan]:
+) -> tuple[str, list[ReadableSpan]]:
     spans = []
 
     run_id = ID_GENERATOR.generate_span_id().to_bytes(8, "big").hex()
-    console.print(f"🛠️ MERGIFY_TEST_RUN_ID={run_id}")
 
     for filename in files:
         spans.extend(
@@ -53,7 +51,7 @@ async def files_to_spans(
             ),
         )
 
-    return spans
+    return run_id, spans
 
 
 async def junit_to_spans(