Run multiple test processes using RunningProcessGroup Args: processes: List of RunningProcess objects to execute parallel: Whether to run processes in parallel or sequentially (ignored if NO_PARALLEL is set) verbose: Whether to show all output Returns:
(
processes: list[RunningProcess], parallel: bool = True, verbose: bool = False
)
| 1370 | |
| 1371 | |
| 1372 | def run_test_processes( |
| 1373 | processes: list[RunningProcess], parallel: bool = True, verbose: bool = False |
| 1374 | ) -> list[ProcessTiming]: |
| 1375 | """ |
| 1376 | Run multiple test processes using RunningProcessGroup |
| 1377 | |
| 1378 | Args: |
| 1379 | processes: List of RunningProcess objects to execute |
| 1380 | parallel: Whether to run processes in parallel or sequentially (ignored if NO_PARALLEL is set) |
| 1381 | verbose: Whether to show all output |
| 1382 | |
| 1383 | Returns: |
| 1384 | List of ProcessTiming objects with execution times |
| 1385 | """ |
| 1386 | from ci.util.running_process_group import ( |
| 1387 | ExecutionMode, |
| 1388 | ProcessExecutionConfig, |
| 1389 | RunningProcessGroup, |
| 1390 | ) |
| 1391 | |
| 1392 | start_time = time.time() |
| 1393 | |
| 1394 | # Force sequential execution if NO_PARALLEL is set |
| 1395 | if os.environ.get("NO_PARALLEL"): |
| 1396 | parallel = False |
| 1397 | |
| 1398 | if not processes: |
| 1399 | ts_print("\n✓ No tests to run") |
| 1400 | return [] |
| 1401 | |
| 1402 | try: |
| 1403 | # Configure execution mode |
| 1404 | execution_mode = ( |
| 1405 | ExecutionMode.PARALLEL if parallel else ExecutionMode.SEQUENTIAL |
| 1406 | ) |
| 1407 | |
| 1408 | config = ProcessExecutionConfig( |
| 1409 | execution_mode=execution_mode, |
| 1410 | verbose=verbose, |
| 1411 | max_failures_before_abort=MAX_FAILURES_BEFORE_ABORT, |
| 1412 | enable_stuck_detection=True, |
| 1413 | stuck_timeout_seconds=_GLOBAL_TIMEOUT, |
| 1414 | live_updates=True, # Enable real-time display |
| 1415 | display_type="auto", # Auto-detect best display format |
| 1416 | ) |
| 1417 | |
| 1418 | # Create and run process group |
| 1419 | group = RunningProcessGroup( |
| 1420 | processes=processes, config=config, name="TestProcesses" |
| 1421 | ) |
| 1422 | |
| 1423 | # Start real-time display if we have processes and live updates enabled |
| 1424 | display_thread = None |
| 1425 | if len(processes) > 0 and config.live_updates and not verbose: |
| 1426 | # Only show live display if not in verbose mode (verbose already shows all output) |
| 1427 | try: |
| 1428 | from ci.util.process_status_display import display_process_status |
| 1429 |
no test coverage detected