| 626 | |
| 627 | |
| 628 | class MultiCapture(Generic[AnyStr]): |
| 629 | _state = None |
| 630 | _in_suspended = False |
| 631 | |
| 632 | def __init__( |
| 633 | self, |
| 634 | in_: CaptureBase[AnyStr] | None, |
| 635 | out: CaptureBase[AnyStr] | None, |
| 636 | err: CaptureBase[AnyStr] | None, |
| 637 | ) -> None: |
| 638 | self.in_: CaptureBase[AnyStr] | None = in_ |
| 639 | self.out: CaptureBase[AnyStr] | None = out |
| 640 | self.err: CaptureBase[AnyStr] | None = err |
| 641 | |
| 642 | def __repr__(self) -> str: |
| 643 | return ( |
| 644 | f"<MultiCapture out={self.out!r} err={self.err!r} in_={self.in_!r} " |
| 645 | f"_state={self._state!r} _in_suspended={self._in_suspended!r}>" |
| 646 | ) |
| 647 | |
| 648 | def start_capturing(self) -> None: |
| 649 | self._state = "started" |
| 650 | if self.in_: |
| 651 | self.in_.start() |
| 652 | if self.out: |
| 653 | self.out.start() |
| 654 | if self.err: |
| 655 | self.err.start() |
| 656 | |
| 657 | def pop_outerr_to_orig(self) -> tuple[AnyStr, AnyStr]: |
| 658 | """Pop current snapshot out/err capture and flush to orig streams.""" |
| 659 | out, err = self.readouterr() |
| 660 | if out: |
| 661 | assert self.out is not None |
| 662 | self.out.writeorg(out) |
| 663 | if err: |
| 664 | assert self.err is not None |
| 665 | self.err.writeorg(err) |
| 666 | return out, err |
| 667 | |
| 668 | def suspend_capturing(self, in_: bool = False) -> None: |
| 669 | self._state = "suspended" |
| 670 | if self.out: |
| 671 | self.out.suspend() |
| 672 | if self.err: |
| 673 | self.err.suspend() |
| 674 | if in_ and self.in_: |
| 675 | self.in_.suspend() |
| 676 | self._in_suspended = True |
| 677 | |
| 678 | def resume_capturing(self) -> None: |
| 679 | self._state = "started" |
| 680 | if self.out: |
| 681 | self.out.resume() |
| 682 | if self.err: |
| 683 | self.err.resume() |
| 684 | if self._in_suspended: |
| 685 | assert self.in_ is not None |
no outgoing calls
no test coverage detected