diff --git a/owilix/core/manager/ui.py b/owilix/core/manager/ui.py index eb6f6b6..642fd49 100644 --- a/owilix/core/manager/ui.py +++ b/owilix/core/manager/ui.py @@ -745,6 +745,22 @@ def currentItemProgress(): TextColumn("{task.fields[current_item]}"), # Display the current item ) +def _stdin_is_interactive() -> bool: + """Whether stdin can be put into cbreak mode to read a keypress. + + ``isatty()`` is the question that matters, but it can itself raise on a + closed or replaced stdin (pytest's capture, a daemonised server), and a + prompt helper is the wrong place to raise from -- so anything unexpected + counts as "not interactive". + """ + import sys + + try: + return bool(sys.stdin) and sys.stdin.isatty() + except Exception: + return False + + def ask_yes_no(console, question, default='y', auto_confirm=False, **console_kwargs): """ Prompt the user for a yes/no answer and return True/False, with enhancements by Rich. @@ -759,7 +775,22 @@ def ask_yes_no(console, question, default='y', auto_confirm=False, **console_kwa # Auto-confirm if flag is set if auto_confirm: return True - + + # Without a terminal there is nobody to answer. + # + # The prompt below puts stdin into cbreak mode to read a single keypress, + # and termios.tcgetattr raises "Inappropriate ioctl for device" when stdin + # is not a tty -- so any non-interactive caller that omitted --yes got a + # traceback instead of an answer. Declining is the safe reading: it cancels + # rather than proceeding with something the caller never confirmed. + if not _stdin_is_interactive(): + console.print( + f"{question} [yellow]-- declined: no terminal to prompt on. " + "Pass --yes (or auto_confirm) to confirm non-interactively.[/yellow]", + **console_kwargs, + ) + return False + valid_responses = { "y": True, "yes": True, "n": False, "no": False, diff --git a/owilix/core/tasks/remote.py b/owilix/core/tasks/remote.py index e0abe44..d5ca5c7 100644 --- a/owilix/core/tasks/remote.py +++ b/owilix/core/tasks/remote.py @@ -5,6 +5,7 @@ Migrated from owilix/cmd/remote.py. import asyncio import inspect import gzip +import logging import os import json import fnmatch @@ -34,6 +35,8 @@ from owilix.core.sync import ( ) from owilix.core.exceptions import AuthenticationError from owilix.core.types import CommandResult, ErrorType, ExitCode + +logger = logging.getLogger("owilix") from owilix.core.db.models import OWIlixSQLQuery from owilix.core.db.duckdb_executor import OWIDuckDBSelectExecutor from owilix.core.tasks.query_utils import extract_domain_components, parse_files_pattern @@ -2850,17 +2853,21 @@ def remote_readme( has_stats = False has_hosts_cache = False + # `except Exception`, not a bare `except`: a bare clause also catches + # KeyboardInterrupt and SystemExit, so Ctrl-C during a long summarize + # run over many datasets was swallowed and the loop carried on. + # "The file is not there" is the expected case, hence debug, not error. try: readme_content = d.repository.readlines(d, "README.md") has_readme = readme_content is not None and len(readme_content) > 0 - except: - pass + except Exception as e: + logger.debug(f"Could not read README.md for {d.path}: {e}") try: stats_content = d.repository.readlines(d, "stats.json") has_stats = stats_content is not None and len(stats_content) > 0 - except: - pass + except Exception as e: + logger.debug(f"Could not read stats.json for {d.path}: {e}") cache_path = _hosts_cache_path(d, cache_dir=hosts_cache_dir) has_hosts_cache = os.path.exists(cache_path) diff --git a/tests/owilix/core/test_ask_yes_no_tty.py b/tests/owilix/core/test_ask_yes_no_tty.py new file mode 100644 index 0000000..fd35403 --- /dev/null +++ b/tests/owilix/core/test_ask_yes_no_tty.py @@ -0,0 +1,92 @@ +"""F6 -- `ask_yes_no` assumed a terminal. + +The prompt reads a single keypress, which needs stdin in cbreak mode, and +``termios.tcgetattr(sys.stdin)`` raises ``Inappropriate ioctl for device`` when +stdin is not a tty. Any non-interactive caller that omitted ``--yes`` therefore +got a traceback rather than an answer -- reachable from cron, from a container, +and from the HTTP service. +""" +import io +from unittest.mock import MagicMock + +from owilix.core.manager.ui import ask_yes_no + + +class _FakeStdin(io.StringIO): + def __init__(self, interactive: bool, keypress: str = "y"): + super().__init__(keypress) + self._interactive = interactive + + def isatty(self) -> bool: + return self._interactive + + def fileno(self) -> int: + # A StringIO has no descriptor; the prompt path calls fileno() to put + # the terminal into cbreak mode, and the termios calls are patched out. + return 0 + + +class TestAskYesNoWithoutATerminal: + def test_declines_instead_of_raising(self, monkeypatch): + monkeypatch.setattr("sys.stdin", _FakeStdin(interactive=False)) + console = MagicMock() + + # Must not raise "Inappropriate ioctl for device". + assert ask_yes_no(console, "Pull 5 datasets?") is False + + def test_says_why_and_how_to_proceed(self, monkeypatch): + monkeypatch.setattr("sys.stdin", _FakeStdin(interactive=False)) + console = MagicMock() + + ask_yes_no(console, "Pull 5 datasets?") + + printed = " ".join(str(call) for call in console.print.call_args_list) + assert "--yes" in printed, "an operator needs to be told the way through" + + def test_declining_is_not_confirming(self, monkeypatch): + """The safe reading: cancel rather than proceed unconfirmed. + + `default='y'` describes what Enter means for a human at a prompt, not + consent from a caller that was never asked. + """ + monkeypatch.setattr("sys.stdin", _FakeStdin(interactive=False)) + + assert ask_yes_no(MagicMock(), "Remove 584 datasets?", default="y") is False + + def test_auto_confirm_still_wins(self, monkeypatch): + """--yes must keep working non-interactively; that is the whole point.""" + monkeypatch.setattr("sys.stdin", _FakeStdin(interactive=False)) + + assert ask_yes_no(MagicMock(), "Pull 5 datasets?", auto_confirm=True) is True + + def test_a_broken_stdin_is_treated_as_non_interactive(self, monkeypatch): + """isatty() itself can raise on a closed or replaced stdin.""" + + class ExplodingStdin: + def isatty(self): + raise ValueError("I/O operation on closed file") + + monkeypatch.setattr("sys.stdin", ExplodingStdin()) + + assert ask_yes_no(MagicMock(), "Pull 5 datasets?") is False + + +class TestAskYesNoWithATerminal: + def test_a_real_terminal_still_prompts(self, monkeypatch): + """The interactive path must be untouched.""" + monkeypatch.setattr("sys.stdin", _FakeStdin(interactive=True)) + + called = {} + + def fake_tcgetattr(_fd): + called["prompted"] = True + return [] + + monkeypatch.setattr("termios.tcgetattr", fake_tcgetattr) + monkeypatch.setattr("termios.tcsetattr", lambda *a, **k: None) + monkeypatch.setattr("tty.setcbreak", lambda *a, **k: None) + + result = ask_yes_no(MagicMock(), "Pull 5 datasets?") + + assert called.get("prompted") is True + assert result is True