diff --git a/routes/shell_routes.py b/routes/shell_routes.py index 165c9a9..dcd7823 100644 --- a/routes/shell_routes.py +++ b/routes/shell_routes.py @@ -14,9 +14,12 @@ from typing import Dict, Any try: import fcntl import pty -except ImportError: +except ImportError as exc: fcntl = None pty = None + _PTY_IMPORT_ERROR = exc +else: + _PTY_IMPORT_ERROR = None from fastapi import APIRouter, Request, HTTPException from fastapi.responses import StreamingResponse @@ -43,6 +46,8 @@ def _require_admin(request: Request): logger = logging.getLogger(__name__) +PTY_SUPPORTED = pty is not None and fcntl is not None and hasattr(os, "setsid") + def _find_line_break(buf): """Find next line terminator in buffer. Returns (index, separator_length) or (-1, 0).""" @@ -63,6 +68,7 @@ EXEC_TIMEOUT = 30 # seconds — shorter than agent's 60s STREAM_TIMEOUT = 120 # default for short commands MAX_OUTPUT = 200_000 # truncate limit TMUX_LOG_DIR = Path(tempfile.gettempdir()) / "odysseus-tmux" +PTY_UNSUPPORTED_ERROR = "pty_unsupported" class ShellExecRequest(BaseModel): @@ -102,9 +108,12 @@ async def _exec_shell(command: str, timeout: int = EXEC_TIMEOUT) -> Dict[str, An async def _generate_pty(cmd: str, timeout: int, request: Request): """Run command in a pseudo-TTY so tqdm/progress bars work natively.""" - if pty is None or fcntl is None: - yield f"data: {json.dumps({'stream': 'stderr', 'data': 'PTY streaming is not available on Windows'})}\n\n" - yield f"data: {json.dumps({'exit_code': -1})}\n\n" + if not PTY_SUPPORTED: + msg = "PTY streaming is not supported on this platform" + if _PTY_IMPORT_ERROR: + msg += f": {_PTY_IMPORT_ERROR}" + yield f"data: {json.dumps({'stream': 'stderr', 'data': msg, 'error': PTY_UNSUPPORTED_ERROR})}\n\n" + yield f"data: {json.dumps({'exit_code': -1, 'error': PTY_UNSUPPORTED_ERROR})}\n\n" return loop = asyncio.get_event_loop() diff --git a/tests/test_shell_routes.py b/tests/test_shell_routes.py index 30dd73e..4833ef3 100644 --- a/tests/test_shell_routes.py +++ b/tests/test_shell_routes.py @@ -1,9 +1,66 @@ -"""Tests for shell_routes.py — _find_line_break helper. -Imports the function directly since it has no app dependencies.""" +"""Tests for shell_routes.py helpers.""" + +import builtins +import importlib.util +import json +import sys +from pathlib import Path +from types import SimpleNamespace from routes.shell_routes import _find_line_break +def test_shell_routes_import_without_posix_pty_modules(monkeypatch): + """Native Windows has no fcntl/termios; importing routes must still work.""" + real_import = builtins.__import__ + + def fake_import(name, globals=None, locals=None, fromlist=(), level=0): + if name in {"fcntl", "pty"}: + raise ImportError(f"No module named {name!r}") + return real_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", fake_import) + cached_modules = {name: sys.modules.pop(name, None) for name in ("fcntl", "pty")} + + module_path = Path(__file__).resolve().parents[1] / "routes" / "shell_routes.py" + spec = importlib.util.spec_from_file_location("_shell_routes_without_pty", module_path) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + try: + spec.loader.exec_module(module) + finally: + sys.modules.pop(spec.name, None) + for name, cached_module in cached_modules.items(): + if cached_module is not None: + sys.modules[name] = cached_module + + assert module.PTY_SUPPORTED is False + assert module._find_line_break(b"ok\n") == (2, 1) + + +async def test_generate_pty_reports_explicit_unsupported_error(monkeypatch): + """Clients can distinguish unsupported PTY mode from process failures.""" + import routes.shell_routes as shell_routes + + monkeypatch.setattr(shell_routes, "PTY_SUPPORTED", False) + monkeypatch.setattr(shell_routes, "_PTY_IMPORT_ERROR", ImportError("No module named 'termios'")) + + request = SimpleNamespace(is_disconnected=lambda: False) + events = [ + json.loads(chunk.removeprefix("data: ").strip()) + async for chunk in shell_routes._generate_pty("echo hi", 5, request) + ] + + assert events == [ + { + "stream": "stderr", + "data": "PTY streaming is not supported on this platform: No module named 'termios'", + "error": shell_routes.PTY_UNSUPPORTED_ERROR, + }, + {"exit_code": -1, "error": shell_routes.PTY_UNSUPPORTED_ERROR}, + ] + + class TestFindLineBreak: """Test line-break detection in byte buffers."""