diff --git a/openspace/entrypoints/cli/main.py b/openspace/entrypoints/cli/main.py index 3dfcb26..16672fd 100644 --- a/openspace/entrypoints/cli/main.py +++ b/openspace/entrypoints/cli/main.py @@ -449,7 +449,8 @@ async def main(): ) startup_console_log_handlers = [] else: - await interactive_mode(openspace, ui_manager) + if not await interactive_mode(openspace, ui_manager): + return 1 except KeyboardInterrupt: print("\n\nInterrupt signal detected") diff --git a/openspace/entrypoints/cli/text_loop.py b/openspace/entrypoints/cli/text_loop.py index 76868df..6f5727b 100644 --- a/openspace/entrypoints/cli/text_loop.py +++ b/openspace/entrypoints/cli/text_loop.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import logging +import sys from typing import Optional from openspace.runtime import ExecutionRequest, ExecutionResult @@ -94,7 +95,7 @@ async def _execute_task(openspace: OpenSpace, query: str, ui_manager: UIManager) return result -async def interactive_mode(openspace: OpenSpace, ui_manager: UIManager): +async def interactive_mode(openspace: OpenSpace, ui_manager: UIManager) -> bool: CLIDisplay.print_interactive_header() while True: @@ -123,10 +124,19 @@ async def interactive_mode(openspace: OpenSpace, ui_manager: UIManager): except KeyboardInterrupt: print("\n\nInterrupt signal detected, exiting...") break + except EOFError: + print( + "\nInput stream closed; exiting interactive mode. " + "Use --query for non-interactive execution.", + file=sys.stderr, + ) + return False except Exception as e: logger.error(f"Error: {e}", exc_info=True) print(f"\nError: {e}") + return True + async def single_query_mode(openspace: OpenSpace, query: str, ui_manager: UIManager): CLIDisplay.print_task_header(query, title="▶ Single Query Execution") diff --git a/tests/entrypoints/cli/test_text_loop.py b/tests/entrypoints/cli/test_text_loop.py new file mode 100644 index 0000000..f130207 --- /dev/null +++ b/tests/entrypoints/cli/test_text_loop.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +from unittest.mock import Mock + +import pytest + +from openspace.entrypoints.cli.text_loop import interactive_mode + + +@pytest.mark.asyncio +async def test_interactive_mode_exits_after_stdin_eof(monkeypatch, capsys) -> None: + calls = 0 + + def closed_stdin(_prompt: str) -> str: + nonlocal calls + calls += 1 + raise EOFError + + monkeypatch.setattr("builtins.input", closed_stdin) + openspace = Mock() + ui_manager = Mock() + + completed = await interactive_mode(openspace, ui_manager) + + assert completed is False + assert calls == 1 + openspace.execute.assert_not_called() + assert "Input stream closed; exiting interactive mode" in capsys.readouterr().err