diff --git a/docs/revup.md b/docs/revup.md index a9ecbf9..d9dd228 100644 --- a/docs/revup.md +++ b/docs/revup.md @@ -27,6 +27,14 @@ full command-line of any subprocesses that are run along with their full output, as well as the full input and output of any graphql requests to the forge. +**--log-to-file** +: Writes debug details to a log file. Logs are written to +`$XDG_STATE_HOME/revup/logs` (or `~/.local/state/revup/logs` if that +isn't set) and named after the start time and process id. The 20 newest +logs are kept and older ones are deleted. + +Can be combined with --verbose to print debug info to both terminal and file. + **--help, -h** : Show this help page. diff --git a/revup/logs.py b/revup/logs.py index 7a09222..392b77a 100644 --- a/revup/logs.py +++ b/revup/logs.py @@ -1,11 +1,18 @@ import logging +import os import re -from typing import Dict +import sys +import time +from typing import Dict, List, Optional, Type, TypeVar from rich._log_render import LogRender from rich.logging import RichHandler from rich.text import Text +from revup.version import REVUP_VERSION + +HandlerType = TypeVar("HandlerType", bound=logging.Handler) + class RedactingFilter(logging.Filter): redactions: Dict[str, str] @@ -53,21 +60,47 @@ def set_render(self, log_render: LogRender) -> None: self._log_render = log_render -def configure_logger(debug: bool, redactions: Dict[str, str]) -> None: - root_logger = logging.getLogger() - root_logger.setLevel(logging.DEBUG if debug else logging.INFO) - - for h in root_logger.handlers: - if isinstance(h, RichHandler): - for k, v in redactions.items(): - for f in h.filters: - if isinstance(f, RedactingFilter): - f.redact(k, v) - return - - log_filter = RedactingFilter() - for k, v in redactions.items(): - log_filter.redact(k, v) +def find_handler(handler_type: Type[HandlerType]) -> Optional[HandlerType]: + """ + Find an already installed handler of the given type. + """ + for h in logging.getLogger().handlers: + if isinstance(h, handler_type): + return h + return None + + +def get_log_filter() -> RedactingFilter: + """ + The filter shared by all handlers, so redactions added at any point apply to all + backends. Must be called after configure_logger(). + """ + for h in logging.getLogger().handlers: + for f in h.filters: + if isinstance(f, RedactingFilter): + return f + raise AssertionError("Logger must be configured before redacting") + + +def prune_old_logs(log_dir: str, keep: int) -> None: + """ + Delete all but the newest `keep` log files. Names sort chronologically so we can + prune without stat-ing every file. + """ + try: + names: List[str] = sorted(n for n in os.listdir(log_dir) if n.endswith(".log")) + except OSError: + return + + for name in names[: max(len(names) - keep, 0)]: + try: + os.remove(os.path.join(log_dir, name)) + except OSError: + # Another revup may have pruned it already, or we may not own it. + pass + + +def make_console_handler(log_filter: logging.Filter) -> RichHandler: handler = RevupRichHandler(keywords=[]) handler.addFilter(log_filter) handler.set_render( @@ -79,12 +112,74 @@ def configure_logger(debug: bool, redactions: Dict[str, str]) -> None: ) ) handler.highlighter = None # type: ignore + return handler + + +def make_file_handler(log_filter: logging.Filter) -> Optional[logging.FileHandler]: + """ + Create a handler that writes debug logs for this invocation to a new file. Returns + None if the file couldn't be created, since logging is never worth failing over. + """ + # Follow the XDG base directory spec for state that should persist between restarts + # but isn't precious. + state_home = os.environ.get("XDG_STATE_HOME") + if not state_home: + state_home = os.path.join(os.path.expanduser("~"), ".local", "state") + log_dir = os.path.join(state_home, "revup", "logs") + + try: + os.makedirs(log_dir, exist_ok=True) + # Keep 19 old logs plus the one we're about to add, for 20 runs of history. + prune_old_logs(log_dir, 19) + path = os.path.join( + log_dir, "{}-{}.log".format(time.strftime("%Y%m%dT%H%M%S"), os.getpid()) + ) + handler = logging.FileHandler(path, encoding="utf-8") + except OSError as e: + logging.warning(f"Couldn't write verbose logs to {log_dir}: {e}") + return None + + handler.setLevel(logging.DEBUG) + handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)-7s %(message)s")) + handler.addFilter(log_filter) + return handler + + +def configure_logger(debug: bool, redactions: Dict[str, str], log_to_file: bool = False) -> None: + """ + Set up logging backends. Can be called more than once to update options, which is + needed since some logging happens before config and args are fully parsed. + + Args: + debug: Show debug logs on the console. + redactions: Strings that must never appear in any log, and their replacements. + log_to_file: Additionally write debug logs to a file in the state directory, + regardless of what the console shows. + """ + root_logger = logging.getLogger() + # The root level must allow debug logs through if any handler wants them; each + # handler then decides for itself what to show. + root_logger.setLevel(logging.DEBUG if debug or log_to_file else logging.INFO) + + console_handler = find_handler(RichHandler) + if console_handler is None: + console_handler = make_console_handler(RedactingFilter()) + root_logger.addHandler(console_handler) + console_handler.setLevel(logging.DEBUG if debug else logging.INFO) + + log_filter = get_log_filter() + for k, v in redactions.items(): + log_filter.redact(k, v) - root_logger.addHandler(handler) + if log_to_file and find_handler(logging.FileHandler) is None: + file_handler = make_file_handler(log_filter) + if file_handler is not None: + root_logger.addHandler(file_handler) + # Record what was run, since the file outlives the terminal it came from. + logging.debug("revup {} : {}".format(REVUP_VERSION, " ".join(sys.argv[1:]))) def redact(redactions: Dict[str, str]) -> None: - log_filter = logging.getLogger().handlers[0].filters[0] - assert isinstance(log_filter, RedactingFilter) + log_filter = get_log_filter() for k, v in redactions.items(): log_filter.redact(k, v) diff --git a/revup/revup.py b/revup/revup.py index dec2504..1885fe5 100755 --- a/revup/revup.py +++ b/revup/revup.py @@ -56,6 +56,7 @@ def make_toplevel_parser() -> RevupArgParser: revup_parser.add_argument("--fork-name", default="") revup_parser.add_argument("--editor") revup_parser.add_argument("--verbose", "-v", action="store_true") + revup_parser.add_argument("--log-to-file", action="store_true") revup_parser.add_argument("--keep-temp", "-k", action="store_true") revup_parser.add_argument("--git-path", default="") revup_parser.add_argument("--main-branch", default="main") @@ -110,7 +111,7 @@ async def get_config() -> config.Config: def dump_args(args: argparse.Namespace) -> None: - if args.verbose: + if logging.getLogger().isEnabledFor(logging.DEBUG): import json logging.debug(json.dumps(vars(args), default=str, indent=2)) @@ -338,6 +339,7 @@ async def main(revup_parser: RevupArgParser, all_parsers: List[RevupArgParser]) logs.configure_logger( debug=args.verbose, redactions={args.forge_oauth: ""} if args.forge_oauth else {}, + log_to_file=args.log_to_file, ) if args.cmd != "toolkit": diff --git a/revup/shell.py b/revup/shell.py index 49cad6f..3015ddc 100644 --- a/revup/shell.py +++ b/revup/shell.py @@ -22,6 +22,14 @@ _HANDLE = Union[None, int, IO[Any]] +def debug_enabled() -> bool: + """ + Whether any logging backend consumes debug logs. Verbose messages can be expensive + to format (decoding entire command outputs), so skip them if nobody will see them. + """ + return logging.getLogger().isEnabledFor(logging.DEBUG) + + def log_command(args: Sequence[str]) -> None: """ Given a command, print it in a both machine and human readable way. @@ -126,10 +134,10 @@ def __init__( Args: cwd: Current working directory of the shell. Pass None to initialize to the current cwd of the current process. - quiet: If True, suppress printing out the command executed - by the shell. By default, we print out commands for ease - of debugging. Quiet is most useful for non-mutating - shell commands. + quiet: If True, don't ask subprocesses to be verbose about what + they're doing, since that output goes straight to the user's + terminal. Note this doesn't affect debug logging of commands + and their output, which is governed by the log level. """ self.quiet = quiet self.cwd = cwd if cwd else os.getcwd() @@ -208,7 +216,7 @@ async def sh( and return it raiseonerror: whether to raise an error if return value is not 0 """ - if not self.quiet: + if debug_enabled(): log_command(args) start_time = time.time() tasks = await self.create_sh_task( @@ -224,7 +232,7 @@ async def sh( _, out, err, ret = await asyncio.gather(*tasks) ret = self.handle_sh_results(ret, out, err, stdout, raiseonerror, quiet, *args) - if not self.quiet: + if debug_enabled(): logging.debug("Took {}s".format(time.time() - start_time)) return ret @@ -245,7 +253,7 @@ async def piped_sh( start_time = time.time() read, write = os.pipe() log_args = args1 + ["|"] + args2 - if not self.quiet: + if debug_enabled(): log_command(log_args) tasks = await asyncio.gather( @@ -268,7 +276,7 @@ async def piped_sh( ret2 if ret1 == 0 else ret1, out2, err1 + err2, stdout, raiseonerror, quiet, *log_args ) os.close(read) - if not self.quiet: + if debug_enabled(): logging.debug("Took {}s".format(time.time() - start_time)) return ret @@ -284,9 +292,9 @@ def handle_sh_results( ) -> Tuple[int, str]: if returncode and err: logging.warning(err.decode(errors="backslashreplace")) - elif not (quiet or self.quiet) and err: + elif not quiet and debug_enabled() and err: logging.debug("# stderr:\n{}".format(err.decode(errors="backslashreplace"))) - if not (quiet or self.quiet) and out: + if not quiet and debug_enabled() and out: logging.debug( "{}{}".format( ("# stdout:\n" if err else ""), diff --git a/tests/test_logs.py b/tests/test_logs.py new file mode 100644 index 0000000..a2951ae --- /dev/null +++ b/tests/test_logs.py @@ -0,0 +1,190 @@ +"""Tests for logging backends, especially verbose logs to file.""" + +import logging +import os + +import pytest +from rich.logging import RichHandler + +from revup import logs, shell +from revup.version import REVUP_VERSION + + +@pytest.fixture(autouse=True) +def clean_logger(): + """Save and restore global logging state, since configure_logger mutates it.""" + root_logger = logging.getLogger() + old_handlers = list(root_logger.handlers) + old_level = root_logger.level + root_logger.handlers = [] + yield + for h in root_logger.handlers: + h.close() + root_logger.handlers = old_handlers + root_logger.setLevel(old_level) + + +@pytest.fixture +def log_dir(tmp_path, monkeypatch): + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path)) + return tmp_path / "revup" / "logs" + + +def get_log_file_path(): + handler = logs.find_handler(logging.FileHandler) + return handler.baseFilename if handler is not None else None + + +def read_log_file(): + path = get_log_file_path() + assert path is not None, "No log file was created" + with open(path, encoding="utf-8") as f: + return f.read() + + +def get_console_handler(): + handler = logs.find_handler(RichHandler) + assert handler is not None, "No console handler was installed" + return handler + + +class TestFileLoggingDisabled: + def test_no_file_created_by_default(self, log_dir): + logs.configure_logger(debug=False, redactions={}) + logging.debug("shouldn't be logged anywhere") + + assert get_log_file_path() is None + assert not log_dir.exists() + + def test_debug_logs_dropped_by_default(self, log_dir): + logs.configure_logger(debug=False, redactions={}) + assert not logging.getLogger().isEnabledFor(logging.DEBUG) + assert not shell.debug_enabled() + + def test_verbose_alone_writes_no_file(self, log_dir): + logs.configure_logger(debug=True, redactions={}) + assert get_log_file_path() is None + assert not log_dir.exists() + + +class TestFileLoggingEnabled: + def test_debug_logs_go_to_file(self, log_dir): + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + logging.debug("a verbose detail") + logging.info("something normal") + + contents = read_log_file() + assert "a verbose detail" in contents + assert "something normal" in contents + + def test_console_still_hides_debug(self, log_dir): + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + + # The root logger must pass debug records for the file backend to see them, + # but the console must not show them. + assert logging.getLogger().isEnabledFor(logging.DEBUG) + assert shell.debug_enabled() + assert get_console_handler().level == logging.INFO + + def test_verbose_shows_debug_on_console(self, log_dir): + logs.configure_logger(debug=True, redactions={}, log_to_file=True) + assert get_console_handler().level == logging.DEBUG + + def test_file_is_in_xdg_state_home(self, log_dir): + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + path = get_log_file_path() + assert path is not None + assert os.path.dirname(path) == str(log_dir) + assert path.endswith(".log") + + def test_falls_back_to_local_state(self, tmp_path, monkeypatch): + monkeypatch.delenv("XDG_STATE_HOME", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + + assert os.path.dirname(get_log_file_path()) == str( + tmp_path / ".local" / "state" / "revup" / "logs" + ) + + def test_command_line_is_recorded(self, log_dir): + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + assert "revup {}".format(REVUP_VERSION) in read_log_file() + + def test_redactions_apply_to_file(self, log_dir): + logs.configure_logger(debug=False, redactions={"secret_token": ""}, log_to_file=True) + logging.debug("using secret_token to connect") + + contents = read_log_file() + assert "secret_token" not in contents + assert "" in contents + + def test_later_redactions_apply_to_file(self, log_dir): + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + logs.redact({"found_later": ""}) + logging.debug("using found_later to connect") + + contents = read_log_file() + assert "found_later" not in contents + assert "" in contents + + def test_reconfigure_doesnt_add_second_file(self, log_dir): + logs.configure_logger(debug=False, redactions={}) + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + + assert len(list(log_dir.iterdir())) == 1 + assert len([h for h in logging.getLogger().handlers if isinstance(h, RichHandler)]) == 1 + + def test_unwritable_dir_doesnt_raise(self, log_dir): + # A file where the log directory should be makes makedirs fail. + log_dir.parent.mkdir(parents=True) + log_dir.write_text("not a directory") + + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + + assert get_log_file_path() is None + logging.info("still works") + + +class TestPruning: + def make_logs(self, log_dir, count): + """Create `count` log files whose names sort oldest to newest.""" + log_dir.mkdir(parents=True, exist_ok=True) + for i in range(1, count + 1): + (log_dir / "202601{:02d}T000000-1.log".format(i)).write_text("old") + + def test_prunes_to_keep_count(self, log_dir): + self.make_logs(log_dir, 5) + logs.prune_old_logs(str(log_dir), 2) + + names = sorted(f.name for f in log_dir.iterdir()) + assert names == ["20260104T000000-1.log", "20260105T000000-1.log"] + + def test_keeps_all_when_under_limit(self, log_dir): + self.make_logs(log_dir, 2) + logs.prune_old_logs(str(log_dir), 5) + assert len(list(log_dir.iterdir())) == 2 + + def test_ignores_other_files(self, log_dir): + self.make_logs(log_dir, 3) + (log_dir / "notes.txt").write_text("keep me") + logs.prune_old_logs(str(log_dir), 1) + + names = sorted(f.name for f in log_dir.iterdir()) + assert names == ["20260103T000000-1.log", "notes.txt"] + + def test_missing_dir_doesnt_raise(self, log_dir): + logs.prune_old_logs(str(log_dir / "nonexistent"), 1) + + def test_configure_keeps_twenty_runs_of_history(self, log_dir): + self.make_logs(log_dir, 25) + + logs.configure_logger(debug=False, redactions={}, log_to_file=True) + + names = sorted(f.name for f in log_dir.iterdir()) + assert len(names) == 20 + # The oldest were deleted, and this run's file is one of the 20. + assert names[0] == "20260107T000000-1.log" + assert os.path.basename(get_log_file_path()) in names