57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
|
import fcntl
|
||
|
|
import os
|
||
|
|
import pty
|
||
|
|
import struct
|
||
|
|
import termios
|
||
|
|
|
||
|
|
from tamq.ptytap import PtyTap, copy_winsize, write_all
|
||
|
|
|
||
|
|
|
||
|
|
class RecordingBroker:
|
||
|
|
def __init__(self):
|
||
|
|
self.lines = []
|
||
|
|
|
||
|
|
def inspect_line(self, line):
|
||
|
|
self.lines.append(line)
|
||
|
|
return line.startswith("@")
|
||
|
|
|
||
|
|
|
||
|
|
def test_input_observer_accepts_raw_terminal_carriage_returns():
|
||
|
|
broker = RecordingBroker()
|
||
|
|
observed = []
|
||
|
|
tap = PtyTap(["true"], broker, on_line=observed.append)
|
||
|
|
buffer = bytearray()
|
||
|
|
|
||
|
|
tap._observe_input(buffer, b"@activity-core: hel")
|
||
|
|
tap._observe_input(buffer, b"lo\rplain line\n")
|
||
|
|
|
||
|
|
assert broker.lines == ["@activity-core: hello", "plain line"]
|
||
|
|
assert observed == ["@activity-core: hello"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_copy_winsize_preserves_rows_columns_and_pixels():
|
||
|
|
source_master, source_slave = pty.openpty()
|
||
|
|
target_master, target_slave = pty.openpty()
|
||
|
|
expected = struct.pack("HHHH", 41, 103, 900, 1600)
|
||
|
|
try:
|
||
|
|
fcntl.ioctl(source_slave, termios.TIOCSWINSZ, expected)
|
||
|
|
assert copy_winsize(source_slave, target_master) is True
|
||
|
|
actual = fcntl.ioctl(target_slave, termios.TIOCGWINSZ, b"\0" * 8)
|
||
|
|
assert actual == expected
|
||
|
|
finally:
|
||
|
|
for fd in (source_master, source_slave, target_master, target_slave):
|
||
|
|
os.close(fd)
|
||
|
|
|
||
|
|
|
||
|
|
def test_write_all_retries_partial_writes(monkeypatch):
|
||
|
|
writes = []
|
||
|
|
|
||
|
|
def partial_write(fd, data):
|
||
|
|
chunk = bytes(data[:2])
|
||
|
|
writes.append((fd, chunk))
|
||
|
|
return len(chunk)
|
||
|
|
|
||
|
|
monkeypatch.setattr(os, "write", partial_write)
|
||
|
|
write_all(9, b"abcde")
|
||
|
|
assert writes == [(9, b"ab"), (9, b"cd"), (9, b"e")]
|