Developer
49180e4c6d
- Make the host PTY master non-blocking and wait for write readiness. - Retry partial writes so a large bracketed paste always delivers its closing marker instead of leaving pi in paste mode. - Add regression coverage for partial PTY writes and resolve diagnostics. Quality gates: ruff, mypy, focused pytest (3 passed)
98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
"""Unit tests for TerminalSession PTY handling."""
|
|
|
|
import uuid
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest # pyright: ignore[reportMissingImports]
|
|
|
|
from src.services.terminal.terminal_session import TerminalSession
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_start_passes_container_user_to_docker_exec() -> None:
|
|
"""When container_user is set, docker exec receives --user <user>."""
|
|
session = TerminalSession(
|
|
session_id=str(uuid.uuid4()),
|
|
instance_id=uuid.uuid4(),
|
|
container_id="container-123",
|
|
container_user="dev",
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"src.services.terminal.terminal_session.pty.openpty",
|
|
return_value=(1, 2),
|
|
),
|
|
patch("src.services.terminal.terminal_session.os.set_blocking") as set_blocking,
|
|
patch("src.services.terminal.terminal_session.tty.setraw"),
|
|
patch(
|
|
"src.services.terminal.terminal_session.asyncio.create_subprocess_exec",
|
|
new=AsyncMock(),
|
|
) as mock_exec,
|
|
patch("src.services.terminal.terminal_session.os.close"),
|
|
):
|
|
await session.start()
|
|
|
|
args, _kwargs = mock_exec.call_args
|
|
assert "docker" in args
|
|
assert "exec" in args
|
|
assert "--user" in args
|
|
user_index = args.index("--user")
|
|
assert args[user_index + 1] == "dev"
|
|
assert "container-123" in args
|
|
set_blocking.assert_called_once_with(1, False)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_start_omits_user_when_not_configured() -> None:
|
|
"""Without container_user, docker exec does not receive --user."""
|
|
session = TerminalSession(
|
|
session_id=str(uuid.uuid4()),
|
|
instance_id=uuid.uuid4(),
|
|
container_id="container-123",
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"src.services.terminal.terminal_session.pty.openpty",
|
|
return_value=(1, 2),
|
|
),
|
|
patch("src.services.terminal.terminal_session.os.set_blocking"),
|
|
patch("src.services.terminal.terminal_session.tty.setraw"),
|
|
patch(
|
|
"src.services.terminal.terminal_session.asyncio.create_subprocess_exec",
|
|
new=AsyncMock(),
|
|
) as mock_exec,
|
|
patch("src.services.terminal.terminal_session.os.close"),
|
|
):
|
|
await session.start()
|
|
|
|
args, _kwargs = mock_exec.call_args
|
|
assert "--user" not in args
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_write_input_retries_partial_pty_writes() -> None:
|
|
"""A large paste is fully written even when the PTY accepts it in chunks."""
|
|
session = TerminalSession(
|
|
session_id=str(uuid.uuid4()),
|
|
instance_id=uuid.uuid4(),
|
|
container_id="container-123",
|
|
)
|
|
session._master_fd = 42
|
|
|
|
with patch(
|
|
"src.services.terminal.terminal_session.os.write",
|
|
side_effect=[2, 2, 1],
|
|
) as write:
|
|
await session.write_input(b"hello")
|
|
|
|
assert [bytes(call.args[1]) for call in write.call_args_list] == [
|
|
b"hello",
|
|
b"llo",
|
|
b"o",
|
|
]
|