Function run_streaming_path

ci/meson/streaming_runner.py:253–426

Execute the streaming compile + test path and return the aggregated result.

(ctx: StreamingContext)

def run_streaming_path(ctx: StreamingContext) -> MesonTestResult:
    """Execute the streaming compile + test path and return the aggregated result."""
    if ctx.verbose:
        _ts_print("[MESON] Using streaming execution (compile + test in parallel)")

    failed_test_outputs: dict[str, str] = {}
    streaming_env = _make_streaming_env(ctx.source_dir, ctx.build_dir)
    failed_outputs_lock = threading.Lock()

    active_procs: set[RunningProcess] = set()
    active_procs_lock = threading.Lock()

    def test_callback(test_path: Path) -> TestResult:
        """Run a single test executable and return TestResult.

        Output is always captured silently (echo=False) so that the caller
        can print it sequentially on the main thread, avoiding interleaved
        output when tests run in parallel.
        """
        try:
            # FastLED #3011 — refuse to run missing or stale artifacts.
            # The streaming compiler submits tests as soon as Ninja prints
            # "Linking <X>" (a PRE-link announcement, not success). If
            # the link then fails, what's at test_path is either nothing
            # or yesterday's DLL — running it produces false pass / fail.
            failure = validate_test_artifact(test_path, ctx.start_time)
            if failure is not None:
                return failure

            if test_path.suffix.lower() in (".dll", ".so", ".dylib"):
                runner_suffix = ".exe" if os.name == "nt" else ""
                if test_path.parent.name == "tests":
                    runner = ctx.build_dir / "tests" / f"runner{runner_suffix}"
                else:
                    runner = (
                        ctx.build_dir / "examples" / f"example_runner{runner_suffix}"
                    )
                if not runner.exists():
                    return TestResult(
                        success=False,
                        output=f"[MESON] ⚠️  Runner not found: {runner}",
                    )
                cmd = [str(runner), str(test_path)]
            else:
                cmd = [str(test_path)]

            env = streaming_env
            file_filter = getattr(test_callback, "_test_file_filter", None)
            if file_filter:
                env = dict(streaming_env)
                env["FL_TEST_FILE_FILTER"] = file_filter

            proc = RunningProcess(
                cmd,
                cwd=ctx.source_dir,
                timeout=600,
                auto_run=True,
                check=False,