(self, data: str | bytes)
| 120 | self._pty_master_ref = None |
| 121 | |
| 122 | async def send(self, data: str | bytes): |
| 123 | if self._proc is None: |
| 124 | raise RuntimeError("TTYSpawn is not started") |
| 125 | if not _IS_WIN: |
| 126 | master = ( |
| 127 | self._pty_master_ref.get("fd") |
| 128 | if self._pty_master_ref is not None |
| 129 | else self._pty_master |
| 130 | ) |
| 131 | if master is None: |
| 132 | raise RuntimeError("TTYSpawn PTY is closed") |
| 133 | if getattr(self._proc, "returncode", None) is not None: |
| 134 | raise RuntimeError("TTYSpawn process has exited") |
| 135 | if isinstance(data, str): |
| 136 | data = data.encode(self.encoding) |
| 137 | try: |
| 138 | self._proc.stdin.write(data) # type: ignore |
| 139 | await self._proc.stdin.drain() # type: ignore |
| 140 | except OSError as e: |
| 141 | if e.errno in (errno.EBADF, errno.EIO, errno.EINVAL): |
| 142 | self._release_pty_master() |
| 143 | raise RuntimeError("TTYSpawn PTY is closed") from e |
| 144 | raise |
| 145 | |
| 146 | async def sendline(self, line: str): |
| 147 | await self.send(line + "\n") |
no test coverage detected