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.
(test_path: Path)
| 263 | active_procs_lock = threading.Lock() |
| 264 | |
| 265 | def test_callback(test_path: Path) -> TestResult: |
| 266 | """Run a single test executable and return TestResult. |
| 267 | |
| 268 | Output is always captured silently (echo=False) so that the caller |
| 269 | can print it sequentially on the main thread, avoiding interleaved |
| 270 | output when tests run in parallel. |
| 271 | """ |
| 272 | try: |
| 273 | # FastLED #3011 — refuse to run missing or stale artifacts. |
| 274 | # The streaming compiler submits tests as soon as Ninja prints |
| 275 | # "Linking <X>" (a PRE-link announcement, not success). If |
| 276 | # the link then fails, what's at test_path is either nothing |
| 277 | # or yesterday's DLL — running it produces false pass / fail. |
| 278 | failure = validate_test_artifact(test_path, ctx.start_time) |
| 279 | if failure is not None: |
| 280 | return failure |
| 281 | |
| 282 | if test_path.suffix.lower() in (".dll", ".so", ".dylib"): |
| 283 | runner_suffix = ".exe" if os.name == "nt" else "" |
| 284 | if test_path.parent.name == "tests": |
| 285 | runner = ctx.build_dir / "tests" / f"runner{runner_suffix}" |
| 286 | else: |
| 287 | runner = ( |
| 288 | ctx.build_dir / "examples" / f"example_runner{runner_suffix}" |
| 289 | ) |
| 290 | if not runner.exists(): |
| 291 | return TestResult( |
| 292 | success=False, |
| 293 | output=f"[MESON] ⚠️ Runner not found: {runner}", |
| 294 | ) |
| 295 | cmd = [str(runner), str(test_path)] |
| 296 | else: |
| 297 | cmd = [str(test_path)] |
| 298 | |
| 299 | env = streaming_env |
| 300 | file_filter = getattr(test_callback, "_test_file_filter", None) |
| 301 | if file_filter: |
| 302 | env = dict(streaming_env) |
| 303 | env["FL_TEST_FILE_FILTER"] = file_filter |
| 304 | |
| 305 | proc = RunningProcess( |
| 306 | cmd, |
| 307 | cwd=ctx.source_dir, |
| 308 | timeout=600, |
| 309 | auto_run=True, |
| 310 | check=False, |
| 311 | env=env, |
| 312 | output_formatter=TimestampFormatter(), |
| 313 | ) |
| 314 | |
| 315 | with active_procs_lock: |
| 316 | active_procs.add(proc) |
| 317 | try: |
| 318 | returncode = proc.wait(echo=False) |
| 319 | finally: |
| 320 | with active_procs_lock: |
| 321 | active_procs.discard(proc) |
| 322 | captured = str(proc.stdout) |
no test coverage detected