Function cmd_output_p

pre_commit/util.py:162–200

(
            *cmd: str,
            check: bool = True,
            **kwargs: Any,
    )

            self.close_r()

    def cmd_output_p(
            *cmd: str,
            check: bool = True,
            **kwargs: Any,
    ) -> tuple[int, bytes, bytes | None]:
        assert check is False
        assert kwargs['stderr'] == subprocess.STDOUT, kwargs['stderr']
        _setdefault_kwargs(kwargs)

        try:
            cmd = parse_shebang.normalize_cmd(cmd)
        except parse_shebang.ExecutableNotFoundError as e:
            return e.to_output()

        with open(os.devnull) as devnull, Pty() as pty:
            assert pty.r is not None
            kwargs.update({'stdin': devnull, 'stdout': pty.w, 'stderr': pty.w})
            try:
                proc = subprocess.Popen(cmd, **kwargs)
            except OSError as e:
                return _oserror_to_output(e)

            pty.close_w()

            buf = b''
            while True:
                try:
                    bts = os.read(pty.r, 4096)
                except OSError as e:
                    if e.errno == errno.EIO:
                        bts = b''
                    else:
                        raise
                else:
                    buf += bts
                if not bts:
                    break

        return proc.wait(), buf, None
else:  # pragma: no cover
    cmd_output_p = cmd_output_b