feat: report tmux prompt injection failures
This commit is contained in:
@@ -113,6 +113,14 @@ class TmuxManager:
|
||||
"""Return the visible text of the named window's pane."""
|
||||
return await self.runner.capture_pane(window_name)
|
||||
|
||||
async def send_text(self, window_name: str, text: str) -> bool:
|
||||
"""Send literal text to a harness window."""
|
||||
return await self.runner.send_text(window_name, text)
|
||||
|
||||
async def send_enter(self, window_name: str) -> bool:
|
||||
"""Send Enter to a harness window."""
|
||||
return await self.runner.send_enter(window_name)
|
||||
|
||||
async def set_window_label(self, window_name: str, label: str) -> None:
|
||||
"""Set the status-bar label for a window (status symbol + name)."""
|
||||
await self.runner.set_window_label(window_name, label)
|
||||
|
||||
+22
-4
@@ -594,11 +594,29 @@ class TmuxRunner:
|
||||
for name in alive_map
|
||||
}
|
||||
|
||||
async def send_text(self, window_name: str, text: str) -> None:
|
||||
await self._exec("send-keys", "-t", self._target(window_name), "-l", "--", text)
|
||||
async def send_text(self, window_name: str, text: str) -> bool:
|
||||
rc, _, err = await self._exec(
|
||||
"send-keys",
|
||||
"-t",
|
||||
self._target(window_name),
|
||||
"-l",
|
||||
"--",
|
||||
text,
|
||||
)
|
||||
if rc != 0:
|
||||
log.error("send_text failed for %s: %s", window_name, err)
|
||||
return rc == 0
|
||||
|
||||
async def send_enter(self, window_name: str) -> None:
|
||||
await self._exec("send-keys", "-t", self._target(window_name), "Enter")
|
||||
async def send_enter(self, window_name: str) -> bool:
|
||||
rc, _, err = await self._exec(
|
||||
"send-keys",
|
||||
"-t",
|
||||
self._target(window_name),
|
||||
"Enter",
|
||||
)
|
||||
if rc != 0:
|
||||
log.error("send_enter failed for %s: %s", window_name, err)
|
||||
return rc == 0
|
||||
|
||||
async def capture_pane(self, window_name: str) -> str:
|
||||
_, stdout, _ = await self._exec(
|
||||
|
||||
@@ -697,6 +697,46 @@ async def test_respawn_pane_no_env(runner):
|
||||
assert "-e" not in call_args
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_text_returns_true_on_success(runner):
|
||||
runner._exec.return_value = (0, "", "")
|
||||
|
||||
assert await runner.send_text("hqt-1", "hello") is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_text_returns_false_on_failure(runner):
|
||||
runner._exec.return_value = (1, "", "no such target")
|
||||
|
||||
assert await runner.send_text("hqt-1", "hello") is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_enter_returns_true_on_success(runner):
|
||||
runner._exec.return_value = (0, "", "")
|
||||
|
||||
assert await runner.send_enter("hqt-1") is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_enter_returns_false_on_failure(runner):
|
||||
runner._exec.return_value = (1, "", "no such target")
|
||||
|
||||
assert await runner.send_enter("hqt-1") is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_manager_send_delegates(runner):
|
||||
mgr = TmuxManager(runner)
|
||||
runner.send_text = AsyncMock(return_value=True)
|
||||
runner.send_enter = AsyncMock(return_value=True)
|
||||
|
||||
assert await mgr.send_text("hqt-1", "continue") is True
|
||||
assert await mgr.send_enter("hqt-1") is True
|
||||
runner.send_text.assert_awaited_once_with("hqt-1", "continue")
|
||||
runner.send_enter.assert_awaited_once_with("hqt-1")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spawn_verify_dead_empty_text_returns_fallback_error(runner):
|
||||
"""When verify_window_alive returns (False, ''), spawn returns the fallback error message."""
|
||||
|
||||
Reference in New Issue
Block a user