mirrors/kitty
1
0
Fork 0
mirror of https://github.com/kovidgoyal/kitty synced 2026-07-26 02:02:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

Simplify loop used to process input from child in test pty

Browse Source 
This might surface better errors than Timeout on the occassional test
failure.
This commit is contained in:
Kovid Goyal
2024-05-10 09:29:27 +05:30
parent 38799f3173
commit ab2a4f7de6
1 changed files with 6 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
kitty_tests/__init__.py
View File

@@ -354,23 +354,13 @@ class PTY:
def process_input_from_child(self, timeout=10): def process_input_from_child(self, timeout=10):
rd, wd, _ = select.select([self.master_fd], [self.master_fd] if self.write_buf else [], [], max(0, timeout)) rd, wd, _ = select.select([self.master_fd], [self.master_fd] if self.write_buf else [], [], max(0, timeout))
while wd: if wd:
try: n = os.write(self.master_fd, self.write_buf)
n = os.write(self.master_fd, self.write_buf)
except (BlockingIOError, OSError):
n = 0
if not n:
break
self.write_buf = self.write_buf[n:] self.write_buf = self.write_buf[n:]
bytes_read = 0 bytes_read = 0
while rd: if rd:
try: data = os.read(self.master_fd, io.DEFAULT_BUFFER_SIZE)
data = os.read(self.master_fd, io.DEFAULT_BUFFER_SIZE)
except (BlockingIOError, OSError):
data = b''
if not data:
break
bytes_read += len(data) bytes_read += len(data)
self.received_bytes += data self.received_bytes += data
parse_bytes(self.screen, data) parse_bytes(self.screen, data)
@@ -395,7 +385,8 @@ class PTY:
f'Child exited with exit status: {status} code: {ec} != {require_exit_code}.' f'Child exited with exit status: {status} code: {ec} != {require_exit_code}.'
f' Screen contents:\n{self.screen_contents()}') f' Screen contents:\n{self.screen_contents()}')
return status return status
self.process_input_from_child(timeout=0.02) with suppress(OSError):
self.process_input_from_child(timeout=0.02)
raise AssertionError(f'Child did not exit in {timeout} seconds. Screen contents:\n{self.screen_contents()}') raise AssertionError(f'Child did not exit in {timeout} seconds. Screen contents:\n{self.screen_contents()}')
def set_window_size(self, rows=25, columns=80, send_signal=True): def set_window_size(self, rows=25, columns=80, send_signal=True):