Get current status of all processes in the group.
(self)
| 430 | return completed_timings |
| 431 | |
| 432 | def get_status(self) -> GroupStatus: |
| 433 | """Get current status of all processes in the group.""" |
| 434 | process_statuses = [] |
| 435 | completed = 0 |
| 436 | failed = 0 |
| 437 | |
| 438 | for process in self.processes: |
| 439 | start_time = self._process_start_times.get(process) |
| 440 | if start_time is None: |
| 441 | # Process hasn't started yet, use current time |
| 442 | start_time = datetime.now() |
| 443 | running_duration = timedelta(0) |
| 444 | else: |
| 445 | running_duration = datetime.now() - start_time |
| 446 | |
| 447 | # Get last output line from process |
| 448 | last_output = self._get_last_output_line(process) |
| 449 | |
| 450 | # Get process name - use command friendly name if available |
| 451 | process_name = getattr(process, "name", None) |
| 452 | if not process_name: |
| 453 | if hasattr(process, "command"): |
| 454 | process_name = _get_friendly_test_name(process.command) |
| 455 | else: |
| 456 | process_name = f"Process-{id(process)}" |
| 457 | |
| 458 | status = ProcessStatus( |
| 459 | name=process_name, |
| 460 | is_alive=not process.finished, |
| 461 | is_completed=process.finished, |
| 462 | start_time=start_time, |
| 463 | running_duration=running_duration, |
| 464 | last_output_line=last_output, |
| 465 | return_value=process.returncode, |
| 466 | ) |
| 467 | |
| 468 | process_statuses.append(status) |
| 469 | |
| 470 | if status.is_completed: |
| 471 | completed += 1 |
| 472 | if status.return_value != 0: |
| 473 | failed += 1 |
| 474 | |
| 475 | return GroupStatus( |
| 476 | group_name=self.name, |
| 477 | processes=process_statuses, |
| 478 | total_processes=len(self.processes), |
| 479 | completed_processes=completed, |
| 480 | failed_processes=failed, |
| 481 | ) |
| 482 | |
| 483 | def _get_last_output_line(self, process: RunningProcess) -> Optional[str]: |
| 484 | """Extract the last line of output from a process.""" |
no test coverage detected