tegwick
8186550c9a
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
101 lines
3 KiB
Python
101 lines
3 KiB
Python
import os
|
|
import pty
|
|
import select
|
|
import termios
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from tamq.composer import ComposerError, compose_message, recipient_order
|
|
|
|
|
|
def _read_until(master: int, marker: bytes, timeout: float = 2) -> bytes:
|
|
output = bytearray()
|
|
deadline = time.monotonic() + timeout
|
|
while marker not in output and time.monotonic() < deadline:
|
|
readable, _, _ = select.select([master], [], [], 0.05)
|
|
if readable:
|
|
output.extend(os.read(master, 4096))
|
|
return bytes(output)
|
|
|
|
|
|
def _compose_in_pty(recipients, typed):
|
|
master, slave = pty.openpty()
|
|
original = termios.tcgetattr(slave)
|
|
result = {}
|
|
|
|
def run():
|
|
try:
|
|
result["value"] = compose_message(
|
|
recipients, input_fd=slave, output_fd=slave
|
|
)
|
|
except Exception as exc: # surfaced in the test thread
|
|
result["error"] = exc
|
|
|
|
thread = threading.Thread(target=run)
|
|
try:
|
|
thread.start()
|
|
output = bytearray(_read_until(master, b": "))
|
|
os.write(master, typed)
|
|
thread.join(timeout=2)
|
|
assert not thread.is_alive()
|
|
output.extend(_read_until(master, b"\r\n", timeout=0.2))
|
|
assert "error" not in result
|
|
assert termios.tcgetattr(slave) == original
|
|
return result["value"], bytes(output)
|
|
finally:
|
|
os.close(master)
|
|
os.close(slave)
|
|
|
|
|
|
def test_recipient_order_prefers_latest_then_session_peers():
|
|
assert recipient_order(
|
|
"audit-core",
|
|
"flex-auth",
|
|
["audit-core", "railiance-platform", "flex-auth", "railiance-platform"],
|
|
) == ("flex-auth", "railiance-platform")
|
|
assert recipient_order(
|
|
"audit-core", None, ["audit-core", "railiance-platform"]
|
|
) == ("railiance-platform",)
|
|
assert recipient_order("audit-core", "bad\x1btarget", ["../bad"]) == ()
|
|
|
|
|
|
def test_composer_accepts_literal_prose_and_cycles_recipient_with_tab():
|
|
result, output = _compose_in_pty(
|
|
("flex-auth", "railiance-platform"),
|
|
b"What's up?\t Really?\r",
|
|
)
|
|
|
|
assert result == ("railiance-platform", "What's up? Really?")
|
|
assert b"@flex-auth: " in output
|
|
assert b"@railiance-platform: What's up?" in output
|
|
|
|
|
|
def test_composer_supports_unicode_backspace_and_ctrl_u():
|
|
result, _ = _compose_in_pty(
|
|
("flex-auth",),
|
|
"caf\u00e9x\b!\x15final\r".encode(),
|
|
)
|
|
assert result == ("flex-auth", "final")
|
|
|
|
|
|
def test_composer_ctrl_c_cancels_and_restores_terminal():
|
|
result, output = _compose_in_pty(("flex-auth",), b"draft\x03")
|
|
assert result is None
|
|
assert b"^C\r\n" in output
|
|
|
|
|
|
def test_composer_requires_a_terminal():
|
|
read_fd, write_fd = os.pipe()
|
|
try:
|
|
with pytest.raises(ComposerError, match="requires a terminal"):
|
|
compose_message(("flex-auth",), input_fd=read_fd, output_fd=write_fd)
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(write_fd)
|
|
|
|
|
|
def test_composer_requires_a_recipient():
|
|
with pytest.raises(ComposerError, match="no recipient"):
|
|
compose_message(())
|