tegwick
17b6bd7ae7
Some checks failed
tamq-ci / test (push) Failing after 6s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc
149 lines
4.5 KiB
Python
149 lines
4.5 KiB
Python
import fcntl
|
||
import os
|
||
import pty
|
||
import struct
|
||
import termios
|
||
|
||
from tamq.ptytap import (
|
||
PtyTap,
|
||
TerminalOutputObserver,
|
||
copy_winsize,
|
||
observed_line,
|
||
write_all,
|
||
)
|
||
|
||
|
||
class RecordingBroker:
|
||
def __init__(self):
|
||
self.lines = []
|
||
|
||
def inspect_line(self, line):
|
||
self.lines.append(line)
|
||
return line.startswith("To:")
|
||
|
||
|
||
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"To:activity-core: hel")
|
||
tap._observe_input(buffer, b"lo\rplain line\n")
|
||
|
||
assert broker.lines == ["To:activity-core: hello", "plain line"]
|
||
assert observed == ["To:activity-core: hello"]
|
||
|
||
|
||
def test_input_observer_accepts_enhanced_terminal_enter_sequences():
|
||
broker = RecordingBroker()
|
||
observed = []
|
||
tap = PtyTap(["true"], broker, on_line=observed.append)
|
||
buffer = bytearray()
|
||
|
||
tap._observe_input(buffer, b"To:activity-core: CSI-u\x1b[13;1")
|
||
assert broker.lines == []
|
||
tap._observe_input(buffer, b":1uTo:other: modify\x1b[27;1;13~")
|
||
tap._observe_input(buffer, b"plain\x1bOM")
|
||
|
||
assert broker.lines == [
|
||
"To:activity-core: CSI-u",
|
||
"To:other: modify",
|
||
"plain",
|
||
]
|
||
assert observed == ["To:activity-core: CSI-u", "To:other: modify"]
|
||
|
||
|
||
def test_input_observer_ignores_terminal_replies_before_typed_line():
|
||
broker = RecordingBroker()
|
||
observed = []
|
||
tap = PtyTap(["true"], broker, on_line=observed.append)
|
||
buffer = bytearray()
|
||
|
||
tap._observe_input(buffer, b"\x1b[9;1R\x1b[?1;2;4c")
|
||
tap._observe_input(buffer, b"To:activity-core: Helol\x7f\x7flo\r")
|
||
|
||
assert broker.lines == ["To:activity-core: Hello"]
|
||
assert observed == ["To:activity-core: Hello"]
|
||
|
||
|
||
def test_observed_line_removes_common_terminal_protocol_wrappers():
|
||
raw = (
|
||
b"\x1b[200~To:target: pasted\x1b[201~"
|
||
b"\x1b]10;rgb:ffff/ffff/ffff\x07"
|
||
b"\x1bP1$r0m\x1b\\"
|
||
)
|
||
|
||
assert observed_line(raw) == "To:target: pasted"
|
||
|
||
|
||
def test_output_observer_normalizes_ansi_and_deduplicates_redraws():
|
||
lines = []
|
||
observer = TerminalOutputObserver(lines.append)
|
||
observer.feed(b"plain\r\n\x1b[31mTo:target: worker\x1b[0m\r")
|
||
observer.feed(b"To:target: worker\r")
|
||
assert lines == ["plain", "To:target: worker"]
|
||
|
||
|
||
def test_output_observer_fails_closed_on_recent_operator_echo():
|
||
lines = []
|
||
observer = TerminalOutputObserver(lines.append)
|
||
observer.note_operator_line("To:target: operator")
|
||
observer.feed(b"To:target: operator\r\nTo:target: worker\r\n")
|
||
assert lines == ["To:target: worker"]
|
||
|
||
|
||
def test_output_observer_preserves_full_screen_program_gutters():
|
||
lines = []
|
||
observer = TerminalOutputObserver(lines.append)
|
||
observer.feed(
|
||
b"\x1b[4;1H\xe2\x80\xba To:target: operator echo"
|
||
b"\x1b[5;1H\xe2\x80\xa2 tO:target: worker reply"
|
||
b"\x1b[6;1H"
|
||
)
|
||
assert lines == ["› To:target: operator echo", "• tO:target: worker reply"]
|
||
|
||
|
||
def test_output_observer_emits_one_empty_line_for_crlf_terminator():
|
||
lines = []
|
||
observer = TerminalOutputObserver(lines.append)
|
||
observer.feed(b"To:target: first\r\nfollowup\r\n\r\nordinary\r\n")
|
||
assert lines == ["To:target: first", "followup", "", "ordinary"]
|
||
|
||
|
||
def test_output_observer_infers_empty_full_screen_row_from_cursor_gap():
|
||
lines = []
|
||
observer = TerminalOutputObserver(lines.append)
|
||
observer.feed(
|
||
b"\x1b[4;1H\xe2\x80\xa2 To:target: first"
|
||
b"\x1b[5;1H followup"
|
||
b"\x1b[7;1Hprompt"
|
||
)
|
||
assert lines == ["• To:target: first", " followup", ""]
|
||
|
||
|
||
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")]
|