diff --git a/tests/test_cli/test_display.py b/tests/test_cli/test_display.py index 5105980..408e266 100644 --- a/tests/test_cli/test_display.py +++ b/tests/test_cli/test_display.py @@ -806,13 +806,13 @@ async def test_pretty_pty_write_and_keys(capsys: pytest.CaptureFixture[str]) -> await render_events( _aiter( [ - _pretty_call("write_pty_keys", '{"session_id": 3, "data": "ctrl+c"}'), - _result("session_id=3\nalive=true\nexit_code=\nwrote=1"), + _pretty_call("write_pty_keys", '{"session_id": 3, "data": ""}'), + _result("session_id=3\nalive=true\nexit_code=\nwrote=2"), ] ) ) out = capsys.readouterr().out - assert "* PTY Keys 3 (1 char)" in out + assert "* PTY Keys 3 '' (2 chars)" in out async def test_pretty_ask_into_pty_never_shows_the_answer(capsys: pytest.CaptureFixture[str]) -> None: diff --git a/tests/test_tools/test_process.py b/tests/test_tools/test_process.py index 349ab62..a64e4a9 100644 --- a/tests/test_tools/test_process.py +++ b/tests/test_tools/test_process.py @@ -4,6 +4,8 @@ import asyncio import sys from pathlib import Path +import pytest + from plyngent.tools.process import ( ask_into_pty, close_pty, @@ -514,13 +516,37 @@ def test_decode_write_data_escapes() -> None: from plyngent.tools.process.pty_terminal import decode_write_data assert decode_write_data(r"\x0f") == "\x0f" - assert decode_write_data("ctrl+x") == "\x18" - assert decode_write_data("CTRL+O") == "\x0f" assert decode_write_data(r"\e") == "\x1b" - assert decode_write_data("key=esc") == "\x1b" - assert decode_write_data("key=enter") == "\r" assert decode_write_data(r"a\nb") == "a\nb" assert decode_write_data("plain") == "plain" + # Markers: names, modifiers, repeats, and a literal `<`. + assert decode_write_data("") == "\x1b" + assert decode_write_data("") == "\x1b" + assert decode_write_data("") == "\r" + assert decode_write_data("") == "\x1b[Z" + assert decode_write_data("") == "\x18" + assert decode_write_data("") == "\x0f" + assert decode_write_data("") == "\x1b" + assert decode_write_data("") == "\x1bb" + assert decode_write_data("") == "\x1b\x1b[A" + assert decode_write_data("") == "\x1b[A" * 3 + assert decode_write_data("<") == "" + assert decode_write_data("") == "\x1b[H" + assert decode_write_data("") == "\x7f" + assert decode_write_data("") == "\x1b[5~" + assert decode_write_data("") == "\x1b[15~" + # A `<` that does not open a marker stays literal, spaces included. + assert decode_write_data("sort < file") == "sort < file" + assert decode_write_data("a < b > c") == "a < b > c" + + +def test_decode_write_data_refuses_an_unknown_key() -> None: + """A typo must be an error, not text typed into the terminal.""" + from plyngent.tools.process.pty_terminal import KeySyntaxError, decode_write_data + + for data in ("", "", "", "", "", ""): + with pytest.raises(KeySyntaxError): + _ = decode_write_data(data) def test_sanitize_pty_output_escapes_csi() -> None: @@ -567,7 +593,7 @@ async def test_write_pty_keys_ctrl_escape(workspace: object) -> None: ), ) session_id = _session_id(opened) - written = await call_async(write_pty_keys, session_id, "ctrl+c") + written = await call_async(write_pty_keys, session_id, "") assert "wrote=1" in written text = await call_async(read_pty, session_id, timeout=2.0) assert "error" not in text.lower() or "alive=" in text @@ -576,13 +602,26 @@ async def test_write_pty_keys_ctrl_escape(workspace: object) -> None: PtyManager.close_all() +async def test_write_pty_keys_refuses_an_unknown_marker(workspace: object) -> None: + """The typo comes back as an error and never reaches the session.""" + del workspace + try: + opened = await call_async(open_pty, _py("print('up')")) + session_id = _session_id(opened) + result = await call_async(write_pty_keys, session_id, "") + assert result.startswith("error: unknown key ") + _ = await call_async(close_pty, session_id) + finally: + PtyManager.close_all() + + def test_write_pty_is_literal() -> None: """Plain write_pty must not call the keys decoder.""" import inspect from plyngent.tools.process.pty_terminal import decode_write_data - assert decode_write_data("press ctrl+c to cancel") == "press " + chr(3) + " to cancel" + assert decode_write_data("press to cancel") == "press " + chr(3) + " to cancel" src = inspect.getsource(write_pty.handler) assert "decode_write_data" not in src doc = write_pty.description or write_pty.handler.__doc__ or ""