diff --git a/docs/changes.rst b/docs/changes.rst index d2e15354af..52a87d9fa4 100644 --- a/docs/changes.rst +++ b/docs/changes.rst @@ -170,6 +170,10 @@ New features: - create/import-tar --json: report the deduplicated size of the new archive, #10335. It is also included in the archive_progress JSON output. +- extract/export-tar --list --log-json: output a file_status JSON object per listed item, + like create does. The text listing of export-tar has the same "+" prefix as extract's now. + prune/delete/undelete --list --log-json: output an archive_status JSON object per listed + archive, #9454. Fixes: @@ -186,6 +190,13 @@ Other changes: botocore (S3) service models borg needs, and build cryptography against the bundled OpenSSL in the Linux binaries instead of bundling a second OpenSSL with it, #10345. +- cockpit: process borg's --log-json output (progress, file list, log messages, prompts) + instead of parsing text lines, #9454. The display depends on the command: archive + statistics for create/import-tar/recreate/transfer (with the final statistics from + --json), a progress bar for extract/export-tar, the progress phases for the other + commands. Yes/no prompts are shown as a dialog. The cockpit exits with the exit code + of the borg command. +- docs: add a usage page for the cockpit TUI. Version 2.0.0b24 (2026-09-02) ----------------------------- diff --git a/docs/installation.rst b/docs/installation.rst index 513a2b89ca..52bc5ca3d7 100644 --- a/docs/installation.rst +++ b/docs/installation.rst @@ -193,7 +193,7 @@ development header files (sometimes in a separate `-dev` or `-devel` package). - borgstore[rest,blake3,sftp] ~= 0.6.1 (use `pip install borgbackup[sftp]`) * Optionally, if you wish to use rclone Backend: - borgstore[rest,blake3,rclone] ~= 0.6.1 (use `pip install borgbackup[rclone]`) -* Optionally, if you wish to use the TUI (``borg --cockpit``): +* Optionally, if you wish to use the cockpit TUI (``borg --cockpit``, see :ref:`cockpit`): - textual >= 6.8.0 (use `pip install borgbackup[cockpit]`) If you have troubles finding the right package names, have a look at the diff --git a/docs/internals/frontends.rst b/docs/internals/frontends.rst index 7beda3bf1b..a5ae452186 100644 --- a/docs/internals/frontends.rst +++ b/docs/internals/frontends.rst @@ -158,15 +158,47 @@ progress_percent Unix timestamp (float) file_status - This is only output by :ref:`borg_create`, :ref:`borg_import-tar` and :ref:`borg_recreate` if - ``--list`` is specified. The usual rules for the file listing applies, including the - ``--filter`` option. + One object per listed item, output by :ref:`borg_create`, :ref:`borg_import-tar`, + :ref:`borg_recreate`, :ref:`borg_extract` and :ref:`borg_export-tar` if ``--list`` is + specified. The usual rules for the file listing apply, including the ``--filter`` option. status - Single-character status as for regular list output + Single-character status as for regular list output: the item flags of :ref:`borg_create`, + or ``+`` (item extracted / exported) and ``-`` (item excluded) for :ref:`borg_extract` + and :ref:`borg_export-tar`. path Path of the file system object +archive_status + One object per listed archive, output by :ref:`borg_prune` (``--list``, ``--list-kept`` and + ``--list-pruned``), :ref:`borg_delete` and :ref:`borg_undelete` (``--list``). With ``--json``, + :ref:`borg_prune` outputs the archives on *stdout* instead. + + status + *kept* or *pruned* (:ref:`borg_prune`), *deleted* (:ref:`borg_delete`) or *undeleted* + (:ref:`borg_undelete`). With ``--dry-run``, this is what would be done. + name, archive + Name of the archive + id + Archive ID (hex) + time + Archive timestamp + message + The text line of the ``--list`` output, e.g. *Keeping archive (rule: daily #1): ...* + + :ref:`borg_prune` additionally gives the keys of the archive objects of ``borg prune --json``: + the keys requested via ``--format`` and + + group + Object mapping the ``--group-by`` keys to the values of this archive + kept + *true* if the archive is kept, *false* if it is pruned + keep_rule, kept_oldest, kept_archive_number + For a kept archive: the rule keeping it (e.g. *daily*), whether it is the oldest archive kept + by the rule, and its number within the rule (1 = the most recent one) + deleted_archive_number + For a pruned archive: its number among the pruned archives (1 = the first one pruned) + log_message Any regular log output invokes this type. Regular log options and filtering applies to these as well. @@ -210,6 +242,32 @@ See Prompts_ for the types used by prompts. {"type": "file_status", "status": "d", "path": "src"} {"time": 1787900398.686938, "type": "archive_progress", "finished": true} +:ref:`borg_extract` file listing, with ``--exclude src/linux/baz/file3``:: + + {"type": "file_status", "status": "+", "path": "src"} + {"type": "file_status", "status": "+", "path": "src/linux"} + {"type": "file_status", "status": "+", "path": "src/linux/baz"} + {"type": "file_status", "status": "+", "path": "src/linux/baz/file2"} + {"type": "file_status", "status": "-", "path": "src/linux/baz/file3"} + {"type": "file_status", "status": "+", "path": "src/linux/file1"} + +:ref:`borg_prune` archive listing, with ``--list --dry-run --keep-daily=1``:: + + {"name": "daily", "archive": "daily", "id": "2c77c68a...", "time": "2026-09-09T02:00:00.000000+02:00", + "group": {"name": "daily", "host": "host"}, "kept": true, "keep_rule": "daily", "kept_oldest": false, + "kept_archive_number": 1, "status": "kept", "type": "archive_status", + "message": "Keeping archive (rule: daily #1): daily Wed, 2026-09-09 02:00:00 +0200 [2c77c68a...]"} + {"name": "daily", "archive": "daily", "id": "99a5671a...", "time": "2026-09-08T02:00:00.000000+02:00", + "group": {"name": "daily", "host": "host"}, "kept": false, "deleted_archive_number": 1, "status": "pruned", + "type": "archive_status", + "message": "Would prune: daily Tue, 2026-09-08 02:00:00 +0200 [99a5671a...]"} + +:ref:`borg_delete` archive listing, with ``--list``:: + + {"name": "daily", "archive": "daily", "id": "99a5671a...", "time": "2026-09-08T02:00:00.000000+02:00", + "status": "deleted", "type": "archive_status", + "message": "Deleted archive: daily Tue, 2026-09-08 02:00:00 +0200 [99a5671a...] (1/1)"} + Saving the local cache at the end of :ref:`borg_create`:: {"message": "Saving files cache", "operation": 1, "msgid": "cache.close", "type": "progress_message", "finished": false, "time": 1787900398.719723} diff --git a/docs/usage.rst b/docs/usage.rst index 02a74f95e2..7a3c887060 100644 --- a/docs/usage.rst +++ b/docs/usage.rst @@ -35,6 +35,7 @@ Usage .. toctree:: usage/general + usage/cockpit usage/repo-create usage/repo-space diff --git a/docs/usage/cockpit.rst b/docs/usage/cockpit.rst new file mode 100644 index 0000000000..ca01119bd2 --- /dev/null +++ b/docs/usage/cockpit.rst @@ -0,0 +1,80 @@ +.. highlight:: none +.. _cockpit: + +Cockpit +------- + +The cockpit is a full-screen terminal user interface showing what a borg command does +while it runs: its progress, its statistics, the file list and the log messages, all +updated live. To use it, put ``--cockpit`` in front of the command:: + + $ borg --cockpit -r /path/to/repo create --list my-files ~/Documents + $ borg --cockpit -r /path/to/repo extract --list my-files + $ borg --cockpit -r /path/to/repo check --repair + +The cockpit needs the ``textual`` package: ``pip install borgbackup[cockpit]`` installs +it, the binary releases include it (see :ref:`installation`). It needs a terminal of at +least 80x24 characters, a taller terminal gives the log more room. + +How it works +~~~~~~~~~~~~ + +The cockpit runs the borg command as a subprocess with ``--log-json`` and ``--progress`` +added, and builds its display from the JSON output borg produces for frontends, see +:ref:`json_output`. Apart from that, the command runs exactly like it does without +``--cockpit``, with the options you gave it. + +.. note:: + + ``--progress`` makes ``extract`` and ``export-tar`` read the archive metadata once + more before they start, to determine the total amount of data for the progress bar, + so they start a bit later than usual, see :ref:`borg_extract`. + +The lower part of the screen is the log: borg's messages, warnings and errors, the file +list if you gave ``--list``, and everything else borg outputs. The panel in the upper +right depends on the command: + +``create``, ``import-tar``, ``recreate``, ``transfer`` + The statistics of the archive being created: the number of files, the original and + the deduplicated size, the counts of added, modified and unchanged files, the path + being processed and the throughput in files and bytes per second, with a history + graph. For ``create`` and ``import-tar``, the exact final statistics of the new + archive (what ``--stats`` prints) are shown in the panel and in the log when borg + has finished. + +``extract``, ``export-tar`` + A progress bar with the percentage and the estimated remaining time, the amount of + data extracted so far, the throughput and the counts of the ``--list`` lines. + +All other commands + The phases of the operation borg reports progress for, e.g. "Checking index" and + "Checking archives" for ``check``, each with a progress bar. For ``prune``, also the + numbers of kept and pruned archives. + +Every panel also shows the elapsed time, the number of warnings and errors and, when +borg has finished, its exit code. The cockpit stays on the screen until you press ``q``, +so you can have a look at the log and the numbers. It then exits with the exit code of +the borg command, see :ref:`return_codes`. + +Prompts and passphrases +~~~~~~~~~~~~~~~~~~~~~~~ + +When borg asks a yes/no question (e.g. ``check --repair`` asks whether you know what you +are doing), the cockpit shows a dialog: answer with the YES or NO button, or type another +answer into the input field. + +The cockpit can not enter a passphrase. Give it to borg via the environment, e.g. by +setting ``BORG_PASSPHRASE`` or ``BORG_PASSCOMMAND`` (see :ref:`env_vars`). Otherwise the +cockpit shows a hint that borg is waiting for a passphrase, and you have to quit and try +again. + +Keys +~~~~ + +``q`` (or Ctrl-C) + Quit. If borg is still running, it is asked to terminate (SIGTERM) and the cockpit + waits until it has exited. + +``t`` + Toggle the universal translator: the labels are shown in Borg speak. Resistance is + futile. diff --git a/docs/usage_general.rst.inc b/docs/usage_general.rst.inc index 1b8d661e93..6e65e8c2f7 100644 --- a/docs/usage_general.rst.inc +++ b/docs/usage_general.rst.inc @@ -14,6 +14,8 @@ .. include:: usage/general/logging.rst.inc +.. _return_codes: + .. include:: usage/general/return-codes.rst.inc .. include:: usage/general/config.rst.inc diff --git a/src/borg/archiver/__init__.py b/src/borg/archiver/__init__.py index caa320611a..8935bf6eb8 100644 --- a/src/borg/archiver/__init__.py +++ b/src/borg/archiver/__init__.py @@ -38,6 +38,7 @@ from ..helpers import add_warning, BorgWarning, BackupWarning from ..helpers import format_file_size from ..helpers import remove_surrogates, text_to_json + from ..helpers import bin_to_hex, OutputTimestamp, BorgJsonEncoder from ..helpers import DatetimeWrapper, replace_placeholders from ..helpers.argparsing import flatten_namespace, ArgumentTypeError, ArgumentParser, SUPPRESS from ..helpers import is_slow_msgpack, is_supported_msgpack, sysinfo @@ -140,6 +141,9 @@ def __init__(self, lock_wait=None, prog=None): self.lock_wait = lock_wait self.prog = prog self.start_backup = None + # for print_file_status(): the commands with a file listing set these from their options. + self.output_list = False + self.output_filter = None def print_warning(self, msg, *args, **kw): warning_code = kw.get("wc", EXIT_WARNING) # note: wc=None can be used to not influence exit code @@ -171,6 +175,27 @@ def print_file_status(self, status, path): else: logging.getLogger("borg.output.list").info("%1s %s", status, remove_surrogates(path)) + def print_archive_status(self, status, archive_info, message, data=None): + """ + List an archive a command processed, like print_file_status() lists the items of a file listing. + + With --log-json, an archive_status JSON object is printed: name, id and time of the archive, the + (e.g. "kept", "pruned", "deleted"), the (the text line) and the keys of . + Without it, the text line goes to the "borg.output.list" logger. The callers check the --list options. + """ + if self.log_json: + json_data = { + "name": archive_info.name, + "archive": archive_info.name, + "id": bin_to_hex(archive_info.id), + "time": OutputTimestamp(archive_info.ts), + } + json_data |= data or {} + json_data |= {"status": status, "type": "archive_status", "message": message} + print(json.dumps(json_data, cls=BorgJsonEncoder), file=sys.stderr) + else: + logging.getLogger("borg.output.list").info(message) + def preprocess_args(self, args): deprecations = [ # ('--old', '--new' or None, 'Warning: "--old" has been deprecated. Use "--new" instead.'), @@ -281,7 +306,12 @@ def build_parser(self): parser.add_argument( "-V", "--version", action="version", version="%(prog)s " + __version__, help="show version number and exit" ) - parser.add_argument("--cockpit", dest="cockpit", action="store_true", help="Start the Borg TUI") + parser.add_argument( + "--cockpit", + dest="cockpit", + action="store_true", + help="run the command in the cockpit TUI, a full-screen progress display", + ) parser.common_options.add_common_group(parser, provide_defaults=True) common_parser = ArgumentParser(prog=self.prog) @@ -655,10 +685,13 @@ def main(): # pragma: no cover print("Please install them using: pip install 'borgbackup[cockpit]'", file=sys.stderr) sys.exit(EXIT_ERROR) - app = BorgCockpitApp() - app.borg_args = [arg for arg in sys.argv[1:] if arg != "--cockpit"] + app = BorgCockpitApp( + borg_args=[arg for arg in sys.argv[1:] if arg != "--cockpit"], command=getattr(args, "subcommand", None) + ) app.run() - sys.exit(EXIT_SUCCESS) # borg subprocess RC was already shown on the TUI + # exit with the exit code of the borg subprocess (it was shown on the TUI); rc < 0: borg could not be run. + rc = app.session.rc + sys.exit(rc if rc is not None and rc >= 0 else EXIT_ERROR) # normal borg CLI operation try: diff --git a/src/borg/archiver/delete_cmd.py b/src/borg/archiver/delete_cmd.py index 22c94a1db3..6650a20447 100644 --- a/src/borg/archiver/delete_cmd.py +++ b/src/borg/archiver/delete_cmd.py @@ -1,5 +1,3 @@ -import logging - from ._common import with_repository, archive_match_patterns from ..constants import * # NOQA from ..helpers import format_archive, CommandError, bin_to_hex, archivename_validator @@ -38,7 +36,6 @@ def do_delete(self, args, repository): ) deleted = False - logger_list = logging.getLogger("borg.output.list") for i, archive_info in enumerate(archive_infos, 1): name, id, hex_id = archive_info.name, archive_info.id, bin_to_hex(archive_info.id) # format early before deletion of the archive @@ -54,7 +51,7 @@ def do_delete(self, args, repository): deleted = True if self.output_list: msg = "Would delete: {} ({}/{})" if dry_run else "Deleted archive: {} ({}/{})" - logger_list.info(msg.format(archive_formatted, i, count)) + self.print_archive_status("deleted", archive_info, msg.format(archive_formatted, i, count)) if dry_run: logger.info("Finished dry-run.") elif deleted: diff --git a/src/borg/archiver/extract_cmd.py b/src/borg/archiver/extract_cmd.py index 274caf8c35..4029e03b2a 100644 --- a/src/borg/archiver/extract_cmd.py +++ b/src/borg/archiver/extract_cmd.py @@ -39,6 +39,7 @@ def do_extract(self, args, repository, manifest, archive): progress = args.progress output_list = args.output_list + self.output_list = output_list # for print_file_status() dry_run = args.dry_run stdout = args.stdout sparse = args.sparse @@ -69,8 +70,7 @@ def do_extract(self, args, repository, manifest, archive): is_matched = matcher.match(orig_path) if output_list: - log_prefix = "+" if is_matched else "-" - logging.getLogger("borg.output.list").info(f"{log_prefix} {remove_surrogates(item.path)}") + self.print_file_status("+" if is_matched else "-", item.path) if is_matched: if not dry_run: diff --git a/src/borg/archiver/prune_cmd.py b/src/borg/archiver/prune_cmd.py index 7c258d2220..44a212f209 100644 --- a/src/borg/archiver/prune_cmd.py +++ b/src/borg/archiver/prune_cmd.py @@ -1,6 +1,5 @@ from typing import Callable, NamedTuple from datetime import datetime, timedelta -import logging import math from functools import partial, wraps import os @@ -245,7 +244,6 @@ def do_prune(self, args, repository, manifest): ) logger.info("Keeping %d archives, pruning %d archives.", len(keep), len(archives_to_prune)) - list_logger = logging.getLogger("borg.output.list") # set up counters for the progress display num_archives_deleted = 0 pi = ProgressIndicatorPercent(total=len(archives_to_prune), msg="Pruning archives %3.0f%%", msgid="prune") @@ -254,28 +252,30 @@ def do_prune(self, args, repository, manifest): break # get_item_data/format_item may internally load the archive from the repository, # so we must call it before deleting the archive. - if args.json: + if args.json or self.log_json: archive_data = formatter.get_item_data(archive_info, jsonline=True) archive_data["group"] = dict(zip(group_by, group_of[archive_info])) - else: + if not args.json: archive_formatted = formatter.format_item(archive_info, jsonline=False) if archive_info in archives_to_prune: if not args.json: pi.show() num_archives_deleted += 1 + status = "pruned" if args.dry_run: log_message = "Would prune:" else: log_message = f"Pruning archive ({num_archives_deleted}/{len(archives_to_prune)}):" manifest.archives.delete_by_id(archive_info.id) - if args.json: + if args.json or self.log_json: archive_data["kept"] = False archive_data["deleted_archive_number"] = num_archives_deleted else: result = keep[archive_info] result_message = f"{result.rule.key}{'[oldest]' if result.oldest else ''} #{result.idx + 1}" log_message = f"Keeping archive (rule: {result_message}):" - if args.json: + status = "kept" + if args.json or self.log_json: archive_data["kept"] = True archive_data["keep_rule"] = result.rule.key archive_data["kept_oldest"] = result.oldest @@ -293,7 +293,8 @@ def do_prune(self, args, repository, manifest): or (args.list_pruned and archive_info in archives_to_prune) or (args.list_kept and archive_info not in archives_to_prune) ): - list_logger.info(f"{log_message:<44} {archive_formatted}") + message = f"{log_message:<44} {archive_formatted}" + self.print_archive_status(status, archive_info, message, archive_data if self.log_json else None) if not args.json: pi.finish() if args.json: diff --git a/src/borg/archiver/tar_cmds.py b/src/borg/archiver/tar_cmds.py index be2d9801eb..cc02c5c6ee 100644 --- a/src/borg/archiver/tar_cmds.py +++ b/src/borg/archiver/tar_cmds.py @@ -396,6 +396,7 @@ def _export_tar(self, args, archive, tarstream): progress = args.progress output_list = args.output_list + self.output_list = output_list # for print_file_status() strip_components = args.strip_components hlm = HardLinkManager(id_type=bytes, info_type=str) # hlid -> path @@ -500,7 +501,7 @@ def sparsify_tarinfo(item, tarinfo): if args.tar_format in ("BORG", "PAX"): tarinfo.pax_headers = item_to_paxheaders(args.tar_format, item) if output_list: - logging.getLogger("borg.output.list").info(remove_surrogates(orig_path)) + self.print_file_status("+", orig_path) sparse_content = sparsify_tarinfo(item, tarinfo) if args.sparse and needs_content else None if sparse_content is not None: stream_plan, map_bytes = sparse_content diff --git a/src/borg/archiver/undelete_cmd.py b/src/borg/archiver/undelete_cmd.py index 38e54549ee..09ae6f9c15 100644 --- a/src/borg/archiver/undelete_cmd.py +++ b/src/borg/archiver/undelete_cmd.py @@ -1,5 +1,3 @@ -import logging - from ._common import with_repository, archive_match_patterns from ..constants import * # NOQA from ..helpers import format_archive, CommandError, bin_to_hex, archivename_validator @@ -35,7 +33,6 @@ def do_undelete(self, args, repository): raise CommandError("Aborting: if you really want to undelete all archives, please use -a 'sh:*'.") undeleted = False - logger_list = logging.getLogger("borg.output.list") for i, archive_info in enumerate(archive_infos, 1): name, id, hex_id = archive_info.name, archive_info.id, bin_to_hex(archive_info.id) try: @@ -47,7 +44,9 @@ def do_undelete(self, args, repository): undeleted = True if self.output_list: msg = "Would undelete: {} ({}/{})" if dry_run else "Undeleted archive: {} ({}/{})" - logger_list.info(msg.format(format_archive(archive_info), i, count)) + self.print_archive_status( + "undeleted", archive_info, msg.format(format_archive(archive_info), i, count) + ) if dry_run: logger.info("Finished dry-run.") elif undeleted: diff --git a/src/borg/cockpit/app.py b/src/borg/cockpit/app.py index e1ed1419ed..4d7c454538 100644 --- a/src/borg/cockpit/app.py +++ b/src/borg/cockpit/app.py @@ -3,12 +3,12 @@ """ import asyncio -import time -from textual.app import App, ComposeResult -from textual.widgets import Header, Footer -from textual.containers import Horizontal, Container +from textual.app import App +from textual.css.query import NoMatches +from .events import Question +from .session import Session from .theme import theme @@ -21,20 +21,33 @@ class BorgCockpitApp(App): CSS_PATH = "cockpit.tcss" BINDINGS = [("q", "quit", "Quit"), ("ctrl+c", "quit", "Quit"), ("t", "toggle_translator", "Toggle Translator")] - def compose(self) -> ComposeResult: - """Create child widgets for the app.""" - from .widgets import LogoPanel, StatusPanel, StandardLog - - yield Header(show_clock=True) - - with Container(id="main-grid"): - with Horizontal(id="top-row"): - yield LogoPanel(id="logopanel") - yield StatusPanel(id="status") - - yield StandardLog(id="standard-log") - - yield Footer() + SPEED_INTERVAL = 1.0 # seconds between two speed samples (one sparkline column each) + REFRESH_INTERVAL = 0.2 # seconds between two refreshes of the widgets from the session + # These commands output the statistics of the new archive as JSON on stdout when given --json. + FINAL_STATS_COMMANDS = ("create", "import-tar") + + def __init__(self, borg_args=None, command=None, runner_factory=None, **kwargs): + """ + :param borg_args: the borg command line to run, without --cockpit [borg --version]. + :param command: the borg subcommand in borg_args, e.g. "create"; it selects the screen [None: generic]. + :param runner_factory: callable(args, callback, json_stdout=...) giving a BorgRunner-like object, for tests. + """ + super().__init__(**kwargs) + self.borg_args = ["--version"] if borg_args is None else list(borg_args) + self.command = command + self.json_stdout = command in self.FINAL_STATS_COMMANDS + self.runner_factory = runner_factory + self.session = Session(command=command, capture_stdout=self.json_stdout) + self.main_screen = None + self.runner = None + self.runner_task = None + + def get_default_screen(self): + """The screen for the command that runs (Textual calls this when the app starts).""" + from .screens import screen_for_command + + self.main_screen = screen_for_command(self.command)() + return self.main_screen def get_theme_variable_defaults(self): # make these variables available to ALL themes @@ -53,9 +66,6 @@ def on_load(self) -> None: def on_mount(self) -> None: """Initialize components.""" - self.query_one("#logo").styles.animate("opacity", 1, duration=1) - self.query_one("#slogan").styles.animate("opacity", 1, duration=1) - # Delay runner start until after widgets are fully mounted self.call_after_refresh(self.start_runner) @@ -63,43 +73,58 @@ def start_runner(self) -> None: """Start the Borg runner after all widgets are mounted.""" from .runner import BorgRunner - # Speed tracking - self.total_lines_processed = 0 - self.last_lines_processed = 0 - self.speed_timer = self.set_interval(1.0, self.compute_speed) - - self.start_time = time.monotonic() - self.process_running = True - args = getattr(self, "borg_args", ["--version"]) # Default to safe command if none passed - self.runner = BorgRunner(args, self.handle_log_event) + factory = self.runner_factory or BorgRunner + self.runner = factory(self.borg_args, self.handle_event, json_stdout=self.json_stdout) self.runner_task = asyncio.create_task(self.runner.start()) - - def compute_speed(self) -> None: - """Calculate and update speed (lines per second).""" - current_lines = self.total_lines_processed - lines_per_second = float(current_lines - self.last_lines_processed) - self.last_lines_processed = current_lines - - status_panel = self.query_one("#status") - status_panel.update_speed(lines_per_second / 1000) - if self.process_running: - status_panel.elapsed_time = time.monotonic() - self.start_time + self.speed_timer = self.set_interval(self.SPEED_INTERVAL, self.sample_speed) + self.refresh_timer = self.set_interval(self.REFRESH_INTERVAL, self.refresh_from_session) + + @property + def process_running(self): + return self.session.running + + def handle_event(self, event) -> None: + """Process an event from the runner: the session does the bookkeeping, a prompt needs a dialog.""" + self.session.feed(event) + if isinstance(event, Question) and event.needs_answer: + from .prompt import PromptModal + + self.push_screen(PromptModal(event.message), callback=self.send_answer) + + def send_answer(self, answer) -> None: + """Send the answer given in the prompt dialog to borg.""" + if answer is not None and self.runner is not None: + self.run_worker(self.runner.answer(answer)) + + def sample_speed(self) -> None: + """Compute the current rates and show them.""" + self.session.sample() + try: + self.main_screen.sample_speed(self.session) + except NoMatches: + pass # the widgets are being torn down (the app exits), the timer still fires + + def refresh_from_session(self) -> None: + """Show the current state of the session in the widgets.""" + try: + self.main_screen.refresh_from_session(self.session) + except NoMatches: + pass # see sample_speed() async def on_unmount(self) -> None: """Cleanup resources on app shutdown.""" - if hasattr(self, "runner"): + if self.runner is not None: await self.runner.stop() async def action_quit(self) -> None: """Handle quit action.""" if hasattr(self, "speed_timer"): self.speed_timer.stop() - if hasattr(self, "runner"): + if self.runner is not None: await self.runner.stop() - if hasattr(self, "runner_task"): + if self.runner_task is not None: await self.runner_task - self.query_one("#logo").styles.animate("opacity", 0, duration=2) - self.query_one("#slogan").styles.animate("opacity", 0, duration=2) + self.main_screen.fade_out() await asyncio.sleep(2) # give the user a chance the see the borg RC self.exit() @@ -108,22 +133,4 @@ def action_toggle_translator(self) -> None: from .translator import TRANSLATOR TRANSLATOR.toggle() - # Refresh dynamic UI elements - self.query_one("#status").refresh_ui_labels() - self.query_one("#standard-log").update_title() - self.query_one("#slogan").update_slogan() - - def handle_log_event(self, data: dict): - """Process a event from BorgRunner.""" - msg_type = data.get("type", "log") - - if msg_type == "stream_line": - self.total_lines_processed += 1 - line = data.get("line", "") - widget = self.query_one("#standard-log") - widget.add_line(line) - - elif msg_type == "process_finished": - self.process_running = False - rc = data.get("rc", 0) - self.query_one("#status").rc = rc + self.main_screen.refresh_ui_labels() diff --git a/src/borg/cockpit/cockpit.tcss b/src/borg/cockpit/cockpit.tcss index 8b8f401649..379a257612 100644 --- a/src/borg/cockpit/cockpit.tcss +++ b/src/borg/cockpit/cockpit.tcss @@ -76,8 +76,8 @@ Footer { border: double $primary; /* If content grows too large, scroll rather than pushing the log off-screen */ overflow-y: auto; - /* Adjust this if status or logo panel shall get more/less height. */ - height: 16; + /* The screens set the height to what their status panel needs (see CockpitScreen). */ + height: 19; } #logopanel { @@ -173,7 +173,6 @@ Pulsar.dim { #speed-sparkline { width: 100%; height: 4; - margin-bottom: 1; } .status { @@ -199,3 +198,65 @@ Pulsar.dim { .rc-error { color: $error; } + +/* The dialog for borg's yes/no prompts */ +PromptModal { + align: center middle; +} + +#prompt-dialog { + width: 80%; + max-width: 100; + height: auto; + border: double $primary; + background: $surface; + padding: 1 2; +} + +#prompt-message { + height: auto; + margin-bottom: 1; +} + +#prompt-buttons { + height: auto; + align-horizontal: center; +} + +#prompt-buttons Button { + margin: 0 2; +} + +/* The progress bar of the extract screen: bar, percentage and ETA over the full panel width */ +#extract-bar { + width: 100%; +} + +#extract-bar Bar { + width: 1fr; +} + +#extract-bar Bar > .bar--bar { + color: $primary; + background: $panel; +} + +#extract-bar Bar > .bar--indeterminate { + color: $primary; + background: $panel; +} + +#extract-bar Bar > .bar--complete { + color: $success; + background: $panel; +} + +/* The lines of the status panels take only the height they need, what follows them comes right after. */ +#statuses { + height: auto; +} + +/* The phase list of the generic screen */ +#phases { + height: auto; +} diff --git a/src/borg/cockpit/events.py b/src/borg/cockpit/events.py new file mode 100644 index 0000000000..64b8368777 --- /dev/null +++ b/src/borg/cockpit/events.py @@ -0,0 +1,214 @@ +""" +Borg Cockpit - typed events. + +The runner turns the JSON lines borg writes to stderr with --log-json into the event objects defined +here (see docs/internals/frontends.rst for the JSON API), so that the rest of the cockpit never deals +with raw dicts. Everything else the runner observes (stdout lines, non-JSON stderr lines, the process +exit) is an event as well. +""" + +import json +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class Event: + """Base class of everything the runner hands to the application.""" + + +@dataclass(frozen=True) +class LogMessage(Event): + """log_message: regular log output (--info, --debug, warnings, errors).""" + + message: str + levelname: str = "INFO" + name: str = "" + msgid: str | None = None + time: float = 0.0 + + +@dataclass(frozen=True) +class ProgressMessage(Event): + """progress_message: what borg is working on, without a quantitative progress.""" + + operation: int + message: str + msgid: str | None = None + finished: bool = False + time: float = 0.0 + + +@dataclass(frozen=True) +class ProgressPercent(Event): + """progress_percent: progress with a current and a total value.""" + + operation: int + message: str + current: int | None = None + total: int | None = None + info: list | None = None + msgid: str | None = None + finished: bool = False + time: float = 0.0 + + +@dataclass(frozen=True) +class ArchiveProgress(Event): + """archive_progress: statistics while an archive is being created (create, import-tar, recreate, transfer).""" + + original_size: int = 0 + deduplicated_size: int = 0 + nfiles: int = 0 + hashing_time: float = 0.0 + chunking_time: float = 0.0 + files_stats: dict = field(default_factory=dict) + path: str | None = None + finished: bool = False + time: float = 0.0 + + +@dataclass(frozen=True) +class FileStatus(Event): + """file_status: one line of the --list output of create, import-tar and recreate.""" + + status: str + path: str + + +@dataclass(frozen=True) +class ArchiveStatus(Event): + """archive_status: one archive listed by prune, delete or undelete.""" + + name: str + status: str # kept, pruned, deleted, undeleted + message: str # the text line of the listing + data: dict = field(default_factory=dict) # the whole object, see the frontends docs for its keys + + +@dataclass(frozen=True) +class Question(Event): + """question_*: a yes/no prompt (kind "prompt" / "prompt_retry") or a message about how a prompt was answered.""" + + kind: str + message: str + msgid: str | None = None + env_var: str | None = None + + @property + def needs_answer(self): + """Is borg waiting for an answer on stdin?""" + return self.kind in ("prompt", "prompt_retry") + + +@dataclass(frozen=True) +class UnknownJson(Event): + """A JSON object with a type the cockpit does not know.""" + + data: dict + + +@dataclass(frozen=True) +class RawLine(Event): + """A line that is not a JSON object: stdout output, or stderr output written outside of --log-json.""" + + stream: str # "stdout" or "stderr" + line: str + partial: bool = False # True: not terminated by a newline (yet), e.g. a prompt waiting for input + + +@dataclass(frozen=True) +class ProcessFinished(Event): + """The borg process has exited (or could not be started: rc -1 and an error message).""" + + rc: int + error: str | None = None + + +def _opt_int(value): + """An int for JSON numbers, None for anything else (missing, null, ...).""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return int(value) + + +def _float(value): + return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else 0.0 + + +def _opt_str(value): + return value if isinstance(value, str) else None + + +def parse_json_line(line): + """ + Parse one line of borg's --log-json output into an Event. + + Returns None if the line is not a JSON object with a "type" key, so that the caller can pass it + on as a RawLine. Unknown types give an UnknownJson event, missing keys get defaults: the cockpit + must keep working with older and newer borg versions. + """ + try: + data = json.loads(line) + except ValueError: + return None + if not isinstance(data, dict) or not isinstance(data.get("type"), str): + return None + msg_type = data["type"] + message = _opt_str(data.get("message")) or "" + msgid = _opt_str(data.get("msgid")) + timestamp = _float(data.get("time")) + finished = bool(data.get("finished", False)) + if msg_type == "log_message": + return LogMessage( + message=message, + levelname=_opt_str(data.get("levelname")) or "INFO", + name=_opt_str(data.get("name")) or "", + msgid=msgid, + time=timestamp, + ) + if msg_type == "progress_message": + return ProgressMessage( + operation=_opt_int(data.get("operation")) or 0, + message=message, + msgid=msgid, + finished=finished, + time=timestamp, + ) + if msg_type == "progress_percent": + info = data.get("info") + return ProgressPercent( + operation=_opt_int(data.get("operation")) or 0, + message=message, + current=_opt_int(data.get("current")), + total=_opt_int(data.get("total")), + info=list(info) if isinstance(info, list) else None, + msgid=msgid, + finished=finished, + time=timestamp, + ) + if msg_type == "archive_progress": + files_stats = data.get("files_stats") + if not isinstance(files_stats, dict): + files_stats = {} + return ArchiveProgress( + original_size=_opt_int(data.get("original_size")) or 0, + deduplicated_size=_opt_int(data.get("deduplicated_size")) or 0, + nfiles=_opt_int(data.get("nfiles")) or 0, + hashing_time=_float(data.get("hashing_time")), + chunking_time=_float(data.get("chunking_time")), + files_stats={status: count for status, count in files_stats.items() if isinstance(count, int)}, + path=_opt_str(data.get("path")), + finished=finished, + time=timestamp, + ) + if msg_type == "file_status": + return FileStatus(status=_opt_str(data.get("status")) or "?", path=_opt_str(data.get("path")) or "") + if msg_type == "archive_status": + return ArchiveStatus( + name=_opt_str(data.get("name")) or "", status=_opt_str(data.get("status")) or "", message=message, data=data + ) + if msg_type.startswith("question_"): + return Question( + kind=msg_type[len("question_") :], message=message, msgid=msgid, env_var=_opt_str(data.get("env_var")) + ) + return UnknownJson(data) diff --git a/src/borg/cockpit/prompt.py b/src/borg/cockpit/prompt.py new file mode 100644 index 0000000000..eaf6ed38d8 --- /dev/null +++ b/src/borg/cockpit/prompt.py @@ -0,0 +1,35 @@ +""" +Borg Cockpit - modal dialog for borg's yes/no prompts. +""" + +from textual.app import ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Button, Input, Static + + +class PromptModal(ModalScreen[str]): + """ + Shows the message of a question_prompt and returns the answer to send to borg's stdin. + + The buttons send YES / NO, which borg accepts for all its prompts, including the "Type 'YES'" ones. + The input field is for anything else, e.g. an empty answer to select the default. + """ + + def __init__(self, message): + super().__init__() + self.message = message + + def compose(self) -> ComposeResult: + with Vertical(id="prompt-dialog"): + yield Static(self.message, id="prompt-message", markup=False) + yield Input(placeholder="other answer, Enter sends it", id="prompt-input") + with Horizontal(id="prompt-buttons"): + yield Button("YES", id="prompt-yes", variant="success") + yield Button("NO", id="prompt-no", variant="error") + + def on_button_pressed(self, event: Button.Pressed) -> None: + self.dismiss("YES" if event.button.id == "prompt-yes" else "NO") + + def on_input_submitted(self, event: Input.Submitted) -> None: + self.dismiss(event.value) diff --git a/src/borg/cockpit/runner.py b/src/borg/cockpit/runner.py index 010d39449a..3aea3b8b38 100644 --- a/src/borg/cockpit/runner.py +++ b/src/borg/cockpit/runner.py @@ -1,74 +1,151 @@ """ -Borg Runner - Manages Borg subprocess execution and output parsing. +Borg Runner - runs borg as a subprocess and turns its output into events. """ import asyncio import logging import os import sys -from collections.abc import Callable + +from ..platformflags import is_win32 +from .events import ProcessFinished, RawLine, parse_json_line + +# The options the cockpit needs for machine-readable output. They are common options, so they are +# valid in front of the subcommand; borg merges them with the options the user gave at any level. +INJECTED_OPTIONS = ("--log-json", "--progress") + + +def borg_command(args, executable=None, json_stdout=False): + """ + Build the command line to run borg with the given args and the cockpit's options injected. + + :param args: the borg command line (without the borg executable and without --cockpit). + :param executable: the command prefix starting borg [the interpreter / pyinstaller binary running this code]. + :param json_stdout: also add --json, so the command outputs its final results as JSON on stdout. + """ + if executable is None: + if getattr(sys, "frozen", False): + executable = [sys.executable] # sys.executable is the pyinstaller-made binary + else: + executable = [sys.executable, "-m", "borg"] + args = list(args) + if json_stdout and "--json" not in args: + # --json is an option of the subcommand, so it must come after it: at the end of the command line, + # or before a "--" end-of-options marker (as used by e.g. --paths-from-command). + args.insert(args.index("--") if "--" in args else len(args), "--json") + injected = [option for option in INJECTED_OPTIONS if option not in args] + return list(executable) + injected + args class BorgRunner: """ - Manages the execution of the borg subprocess and parses its JSON output. + Runs borg as a subprocess, parses its output into events and hands them to a callback, one at a time. + + stderr carries the --log-json stream, one JSON object per line, see docs/internals/frontends.rst. + Everything else (stdout lines, stderr lines that are not JSON) is passed on as RawLine events. + stdin is a pipe, so that the application can answer borg's yes/no prompts via answer(). + + On POSIX, borg runs in a new session and thus has no controlling terminal: nothing it does can + mess with the terminal the TUI runs on. A passphrase prompt then falls back to stderr/stdin, where + the cockpit sees it (see Session), instead of being painted over the TUI. """ - def __init__(self, command: list[str], log_callback: Callable[[dict], None]): - self.command = command - self.log_callback = log_callback - self.process: asyncio.subprocess.Process | None = None - self.logger = logging.getLogger(__name__) + READ_SIZE = 64 * 1024 + # An unterminated line (e.g. a prompt waiting for input) is passed on after this idle time [seconds]. + PARTIAL_LINE_TIMEOUT = 1.0 + # How long to wait for borg to finish after SIGTERM before killing it [seconds]. + TERMINATE_TIMEOUT = 10.0 - async def start(self): + def __init__(self, args, callback, *, executable=None, json_stdout=False): """ - Starts the Borg subprocess and processes its output. + :param args: the borg command line (without the borg executable and without --cockpit). + :param callback: called with each Event, the last one being ProcessFinished. + :param executable: see borg_command(), for tests. + :param json_stdout: see borg_command(). """ + self.args = list(args) + self.callback = callback + self.executable = executable + self.json_stdout = json_stdout + self.process = None + self.logger = logging.getLogger(__name__) + + async def start(self): + """Run borg to completion, handing all events to the callback.""" if self.process is not None: self.logger.warning("Borg process already running.") return - - if getattr(sys, "frozen", False): - cmd = [sys.executable] + self.command # executable == pyinstaller binary - else: - cmd = [sys.executable, "-m", "borg"] + self.command # executable == python interpreter - + cmd = borg_command(self.args, self.executable, self.json_stdout) self.logger.info(f"Starting Borg process: {cmd}") - env = os.environ.copy() env["PYTHONUNBUFFERED"] = "1" - + kwargs = {} if is_win32 else {"start_new_session": True} try: self.process = await asyncio.create_subprocess_exec( - *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env + *cmd, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + **kwargs, ) - - async def read_stream(stream, stream_name): - while line := await stream.readline(): - decoded_line = line.decode("utf-8", errors="replace").rstrip() - if decoded_line: - self.log_callback({"type": "stream_line", "stream": stream_name, "line": decoded_line}) - - # Read both streams concurrently - await asyncio.gather(read_stream(self.process.stdout, "stdout"), read_stream(self.process.stderr, "stderr")) - + await asyncio.gather(self._read(self.process.stdout, "stdout"), self._read(self.process.stderr, "stderr")) rc = await self.process.wait() - self.log_callback({"type": "process_finished", "rc": rc}) - + self.callback(ProcessFinished(rc=rc)) except Exception as e: self.logger.error(f"Failed to run Borg process: {e}") - self.log_callback({"type": "process_finished", "rc": -1, "error": str(e)}) + self.callback(ProcessFinished(rc=-1, error=str(e))) finally: self.process = None + async def _read(self, stream, name): + """Pass on the lines of ; an unterminated line is passed on as partial after some idle time.""" + buffer = b"" + while True: + try: + data = await asyncio.wait_for(stream.read(self.READ_SIZE), self.PARTIAL_LINE_TIMEOUT) + except TimeoutError: + if buffer: + self._line(name, buffer, partial=True) + buffer = b"" + continue + if not data: # EOF + if buffer: + self._line(name, buffer) + return + *lines, buffer = (buffer + data).split(b"\n") + for line in lines: + self._line(name, line) + + def _line(self, stream, raw, partial=False): + line = raw.decode("utf-8", errors="replace").rstrip("\r") + if not line.strip(): + return + event = parse_json_line(line) if stream == "stderr" and not partial else None + self.callback(event if event is not None else RawLine(stream=stream, line=line, partial=partial)) + + async def answer(self, text): + """Send the answer to a prompt to borg's stdin.""" + process = self.process + if process is None or process.stdin is None: + return + try: + process.stdin.write((text + "\n").encode("utf-8")) + await process.stdin.drain() + except (OSError, ValueError) as e: # borg is gone or its stdin is closed + self.logger.warning(f"Could not send the answer to borg: {e}") + async def stop(self): - """ - Stops the Borg subprocess if it is running. - """ - if self.process and self.process.returncode is None: + """Terminate borg if it is still running; borg handles SIGTERM by finishing in an orderly way.""" + process = self.process + if process is not None and process.returncode is None: self.logger.info("Terminating Borg process...") try: - self.process.terminate() - await self.process.wait() + process.terminate() + try: + await asyncio.wait_for(process.wait(), self.TERMINATE_TIMEOUT) + except TimeoutError: + process.kill() + await process.wait() except ProcessLookupError: - pass # Process already dead + pass # already dead diff --git a/src/borg/cockpit/screens.py b/src/borg/cockpit/screens.py new file mode 100644 index 0000000000..78c7eb7417 --- /dev/null +++ b/src/borg/cockpit/screens.py @@ -0,0 +1,91 @@ +""" +Borg Cockpit - the screens, one per kind of borg command. + +All screens have the same layout: the header, the logo panel next to a status panel (the part that +depends on the command), the log panel and the footer. +""" + +from textual.app import ComposeResult +from textual.containers import Container, Horizontal +from textual.screen import Screen +from textual.widgets import Footer, Header + +from .widgets import CreateStatusPanel, ExtractStatusPanel, GenericStatusPanel, LogoPanel, StandardLog + + +class CockpitScreen(Screen): + """The common layout; PANEL is the status panel class of the screen.""" + + PANEL = GenericStatusPanel + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + + with Container(id="main-grid"): + with Horizontal(id="top-row"): + yield LogoPanel(id="logopanel") + yield self.PANEL(id="status") + + yield StandardLog(id="standard-log") + + yield Footer() + + def on_mount(self) -> None: + # the top row is as high as the status panel needs, plus the border of the row. + self.query_one("#top-row").styles.height = self.PANEL.HEIGHT + 2 + self.query_one("#logo").styles.animate("opacity", 1, duration=1) + self.query_one("#slogan").styles.animate("opacity", 1, duration=1) + + def refresh_from_session(self, session) -> None: + """Show the current state of the session.""" + self.query_one("#status").update_from_session(session) + lines, dropped = session.drain() + self.query_one("#standard-log").add_lines(lines, dropped) + + def sample_speed(self, session) -> None: + """Called once per second, after Session.sample(): update the speed display.""" + self.query_one("#status").update_speed(session) + + def refresh_ui_labels(self) -> None: + """Redo the labels with the current translation.""" + self.query_one("#status").refresh_ui_labels() + self.query_one("#standard-log").update_title() + self.query_one("#slogan").update_slogan() + + def fade_out(self) -> None: + """Fade the logo out, for the exit.""" + self.query_one("#logo").styles.animate("opacity", 0, duration=2) + self.query_one("#slogan").styles.animate("opacity", 0, duration=2) + + +class CreateScreen(CockpitScreen): + """create, import-tar, recreate, transfer: the statistics of the archive being created.""" + + PANEL = CreateStatusPanel + + +class ExtractScreen(CockpitScreen): + """extract, export-tar: a progress bar over the bytes to extract.""" + + PANEL = ExtractStatusPanel + + +class GenericScreen(CockpitScreen): + """All other commands: the phases borg reports progress for.""" + + PANEL = GenericStatusPanel + + +SCREENS = { + "create": CreateScreen, + "import-tar": CreateScreen, + "recreate": CreateScreen, + "transfer": CreateScreen, + "extract": ExtractScreen, + "export-tar": ExtractScreen, +} + + +def screen_for_command(command): + """The screen class for a borg subcommand (None: unknown command).""" + return SCREENS.get(command, GenericScreen) diff --git a/src/borg/cockpit/session.py b/src/borg/cockpit/session.py new file mode 100644 index 0000000000..2307b72ecd --- /dev/null +++ b/src/borg/cockpit/session.py @@ -0,0 +1,378 @@ +""" +Borg Cockpit - the state of one borg run. + +A Session is fed with the events the runner produces and is read by the widgets. It has no Textual +dependency, so it can be tested by replaying recorded JSON lines. All counting happens here and the +buffers are bounded, so that a flood of events (e.g. --list over millions of files) costs little more +than the JSON parsing; the widgets render what is in here at their own pace. +""" + +import json +import time +from collections import Counter, deque +from dataclasses import dataclass +from datetime import timedelta + +from ..helpers import format_file_size, format_timedelta +from .events import ( + ArchiveProgress, + ArchiveStatus, + FileStatus, + LogMessage, + ProcessFinished, + ProgressMessage, + ProgressPercent, + Question, + RawLine, + UnknownJson, +) + +# Python's getpass() prints this when it can not use a terminal (the runner starts borg without one) +# and falls back to reading the passphrase from stdin, which the cockpit does not support (yet). +# (The names avoid the word "pass", which makes bandit see hardcoded passwords in these messages.) +NO_TERMINAL_WARNING = "Warning: Password input may be echoed." +NO_TERMINAL_HINT = ( + "borg waits for a passphrase, but the cockpit can not enter one. " + "Quit, set BORG_PASSPHRASE, BORG_PASSCOMMAND or BORG_PASSPHRASE_FD and start again." +) + + +@dataclass +class Line: + """One line for the log panel.""" + + text: str + kind: str # "status": a --list line, tag is the status char. "archive": tag is kept / pruned / deleted / ... + # "log": tag is the level name. "raw", "hint". + tag: str = "" + + +@dataclass +class Phase: + """One progress operation of borg (progress_percent / progress_message), identified by its operation id.""" + + operation: int + msgid: str | None + message: str = "" + current: int | None = None + total: int | None = None + finished: bool = False + rate: float = 0.0 # increase of current per second, see Session.sample() + sampled_current: int | None = None # current at the previous Session.sample() + + @property + def fraction(self): + """Progress as 0.0 .. 1.0, None if unknown.""" + if not self.total or self.current is None: + return None + return min(max(self.current / self.total, 0.0), 1.0) + + +class Session: + LINES_MAX = 200 # log lines buffered between two drains, older lines are dropped (and counted) + + def __init__(self, command=None, capture_stdout=False): + """ + :param command: the borg subcommand that runs, e.g. "create" [None: unknown]. + :param capture_stdout: collect stdout instead of logging it: it carries the --json output, see final_json. + """ + self.command = command + self.capture_stdout = capture_stdout + self.stdout_lines = [] # the captured stdout lines + self.final_json = None # the --json output of borg (create, import-tar), parsed when borg has finished + self.started = time.monotonic() + self.finished_at = None + self.rc = None # exit code of borg, None while it runs + self.error = None # why borg could not be run, if so + self.archive_progress = None # the latest ArchiveProgress carrying statistics + self.archive_finished = False + self.status_counts = Counter() # status char -> count, from the --list lines + self.archive_counts = Counter() # status -> count, the archives listed by prune / delete / undelete + self.phases = {} # operation id -> Phase, in order of appearance + self._active_phase = None # operation id of the phase updated last + self.progress_text = "" # what borg works on right now: the current path or progress message + self.pending_question = None # the Question borg waits for an answer to + self.passphrase_needed = False + self.warnings = 0 # WARNING log messages + self.errors = 0 # ERROR and CRITICAL log messages + self._lines = deque(maxlen=self.LINES_MAX) + self._dropped = 0 + self._sample = (self.started, 0, 0, 0) # time, nfiles, original_size, deduplicated_size + self.files_per_second = 0.0 + self.original_bytes_per_second = 0.0 + self.deduplicated_bytes_per_second = 0.0 + + # derived values for the widgets + + @property + def running(self): + return self.rc is None + + @property + def elapsed(self): + """Seconds since the start of the run, frozen when it has finished.""" + end = time.monotonic() if self.finished_at is None else self.finished_at + return end - self.started + + @property + def final_stats(self): + """The archive statistics from the --json output, None until borg has finished (and only for some commands).""" + if not isinstance(self.final_json, dict): + return None + archive = self.final_json.get("archive") + # create --dry-run creates no archive and has its (reduced) stats at the top level. + stats = archive.get("stats") if isinstance(archive, dict) else self.final_json.get("stats") + return stats if isinstance(stats, dict) else None + + def _final_archive(self, key): + archive = self.final_json.get("archive") if isinstance(self.final_json, dict) else None + return archive.get(key) if isinstance(archive, dict) else None + + @property + def archive_name(self): + """The name of the created archive, from the --json output.""" + name = self._final_archive("name") + return name if isinstance(name, str) else None + + @property + def archive_duration(self): + """The duration of the archive creation [seconds], from the --json output.""" + duration = self._final_archive("duration") + return float(duration) if isinstance(duration, (int, float)) else None + + def _final_stat(self, key, types=int): + stats = self.final_stats + value = stats.get(key) if stats else None + return value if isinstance(value, types) and not isinstance(value, bool) else None + + # The final statistics (exact) are preferred over the --list lines (exact, but subject to --filter), + # which are preferred over archive_progress: that one is rate limited and its final object carries no + # statistics, so its counts can be a little behind at the end of a run (and stay at zero for a run + # shorter than the update interval). + + @property + def nfiles(self): + """Number of regular files (final stats, archive_progress) or of all listed items (--list lines).""" + nfiles = self._final_stat("nfiles") + if nfiles is not None: + return nfiles + if self.status_counts: + return sum(self.status_counts.values()) + if self.archive_progress is not None: + return self.archive_progress.nfiles + return 0 + + @property + def files_stats(self): + """status char -> count, from the final stats, the --list lines or archive_progress.""" + files_stats = self._final_stat("files_stats", dict) + if files_stats is not None: + return dict(files_stats) + if self.status_counts: + return dict(self.status_counts) + if self.archive_progress is not None: + return dict(self.archive_progress.files_stats) + return {} + + def count(self, statuses): + """Number of items having one of the given status characters.""" + stats = self.files_stats + return sum(stats.get(status, 0) for status in statuses) + + @property + def original_size(self): + size = self._final_stat("original_size") + if size is not None: + return size + return None if self.archive_progress is None else self.archive_progress.original_size + + @property + def deduplicated_size(self): + size = self._final_stat("deduplicated_size") + if size is not None: + return size + return None if self.archive_progress is None else self.archive_progress.deduplicated_size + + @property + def active_phase(self): + """The unfinished phase that was updated last, None if there is none.""" + phase = self.phases.get(self._active_phase) + return None if phase is None or phase.finished else phase + + def phase(self, msgid): + """The phase with the given msgid (the last one, if there are several), None if there is none.""" + for phase in reversed(self.phases.values()): + if phase.msgid == msgid: + return phase + return None + + # feeding + + def feed(self, event): + """Update the state with one event from the runner.""" + match event: + case LogMessage(): + self._feed_log_message(event) + case FileStatus(): + self._add_status(event.status, event.path) + case ArchiveStatus(): + self.archive_counts[event.status] += 1 + self._add_line(Line(event.message, "archive", event.status)) + case ArchiveProgress(): + if event.finished: + # the final object carries no statistics, keep the previous ones. + self.archive_finished = True + self.progress_text = "" + else: + self.archive_progress = event + self.progress_text = event.path or "" + case ProgressPercent() | ProgressMessage(): + self._feed_phase(event) + case Question(): + self._feed_question(event) + case RawLine(): + self._feed_raw_line(event) + case UnknownJson(): + self._add_line(Line(json.dumps(event.data), "raw", "stderr")) + case ProcessFinished(): + self.rc = event.rc + self.error = event.error + self.finished_at = time.monotonic() + self.pending_question = None + self.progress_text = "" + if event.error: + self._add_line(Line(event.error, "log", "ERROR")) + if self.stdout_lines: + self._parse_stdout() + + def _feed_log_message(self, event): + if event.levelname == "WARNING": + self.warnings += 1 + elif event.levelname in ("ERROR", "CRITICAL"): + self.errors += 1 + self._add_line(Line(event.message, "log", event.levelname)) + + def _add_status(self, status, path): + self.status_counts[status] += 1 + self._add_line(Line(f"{status} {path}", "status", status)) + + def _feed_phase(self, event): + phase = self.phases.get(event.operation) + if phase is None: + phase = self.phases[event.operation] = Phase(event.operation, event.msgid) + phase.finished = event.finished + if not event.finished: + phase.message = event.message + if isinstance(event, ProgressPercent): + phase.current = event.current + phase.total = event.total + self._active_phase = event.operation + self.progress_text = "" if event.finished else event.message + + def _feed_question(self, event): + if event.needs_answer: + self.pending_question = event + self._add_line(Line(event.message, "log", "PROMPT")) + else: + self.pending_question = None + self._add_line(Line(event.message, "log", "INFO")) + + def _feed_raw_line(self, event): + if event.stream == "stdout" and self.capture_stdout: + self.stdout_lines.append(event.line) + return + self._add_line(Line(event.line, "raw", event.stream)) + if event.stream == "stderr" and event.line == NO_TERMINAL_WARNING: + self.passphrase_needed = True + self._add_line(Line(NO_TERMINAL_HINT, "hint")) + + def _parse_stdout(self): + """The captured stdout is the --json output: keep it and log its statistics like --stats would.""" + try: + data = json.loads("\n".join(self.stdout_lines)) + except ValueError: + data = None + if isinstance(data, dict): + self.final_json = data + for text in self.final_stats_lines(): + self._add_line(Line(text, "log", "STATS")) + else: # not what we expected, show it as it is + for line in self.stdout_lines: + self._add_line(Line(line, "raw", "stdout")) + + def final_stats_lines(self): + """The statistics from the --json output as text lines, like the --stats output of borg.""" + lines = [] + name, fingerprint = self.archive_name, self._final_archive("id") + if name is not None: + lines.append(f"Archive name: {name}") + if isinstance(fingerprint, str): + lines.append(f"Archive fingerprint: {fingerprint}") + duration = self.archive_duration + if duration is not None: + lines.append(f"Duration: {format_timedelta(timedelta(seconds=duration))}") + if isinstance(self.final_json, dict) and self.final_json.get("dry_run"): + lines.append("Dry run: no archive was created.") + nfiles = self._final_stat("nfiles") + if nfiles is not None: + lines.append(f"Number of files: {nfiles}") + for key, label in (("original_size", "Original size"), ("deduplicated_size", "Deduplicated size")): + size = self._final_stat(key) + if size is not None: + lines.append(f"{label}: {format_file_size(size)}") + for key, label in (("hashing_time", "Time spent in hashing"), ("chunking_time", "Time spent in chunking")): + seconds = self._final_stat(key, (int, float)) + if seconds is not None: + lines.append(f"{label}: {format_timedelta(timedelta(seconds=seconds))}") + files_stats = self._final_stat("files_stats", dict) + if files_stats is not None: + for status, label in ( + ("A", "Added files"), + ("U", "Unchanged files"), + ("M", "Modified files"), + ("E", "Error files"), + ("C", "Files changed while reading"), + ): + lines.append(f"{label}: {files_stats.get(status, 0)}") + store_stats = self._final_stat("store_stats", dict) + if store_stats: + from ..archive import format_store_stats + + lines.extend(format_store_stats(store_stats).splitlines()) + return lines + + def _add_line(self, line): + if len(self._lines) == self._lines.maxlen: + self._dropped += 1 + self._lines.append(line) + + def drain(self): + """Take the buffered log lines: (lines, number of older lines dropped since the previous drain).""" + lines, dropped = list(self._lines), self._dropped + self._lines.clear() + self._dropped = 0 + return lines, dropped + + def sample(self, now=None): + """Compute the rates (files/s, bytes/s, progress of the phases) since the previous sample() call.""" + now = time.monotonic() if now is None else now + then, nfiles, original_size, deduplicated_size = self._sample + dt = now - then + if dt <= 0: + return + if not self.running: # nothing moves anymore + self.files_per_second = self.original_bytes_per_second = self.deduplicated_bytes_per_second = 0.0 + for phase in self.phases.values(): + phase.rate = 0.0 + return + current = (self.nfiles, self.original_size or 0, self.deduplicated_size or 0) + self.files_per_second = max(current[0] - nfiles, 0) / dt + self.original_bytes_per_second = max(current[1] - original_size, 0) / dt + self.deduplicated_bytes_per_second = max(current[2] - deduplicated_size, 0) / dt + self._sample = (now, *current) + for phase in self.phases.values(): + if phase.finished or phase.current is None: + phase.rate = 0.0 + elif phase.sampled_current is not None: + phase.rate = max(phase.current - phase.sampled_current, 0) / dt + phase.sampled_current = phase.current diff --git a/src/borg/cockpit/translator.py b/src/borg/cockpit/translator.py index 2302d451c4..28fb51faa3 100644 --- a/src/borg/cockpit/translator.py +++ b/src/borg/cockpit/translator.py @@ -12,6 +12,18 @@ "Other: ": "Other: ", "Errors: ": "Escaped: ", "RC: ": "Termination Code: ", + "Speed: ": "Assimilation rate: ", + "Original: ": "Raw biomass: ", + "Deduplicated: ": "Assimilated biomass: ", + "Progress: ": "Assimilating: ", + "Warnings: ": "Anomalies: ", + "Archive: ": "Collective: ", + "Archives: ": "Collectives: ", + "Extracted: ": "Released: ", + "Items: ": "Entities: ", + "Included: ": "Selected: ", + "Excluded: ": "Irrelevant: ", + "Phases": "Stages", "Log": "Subspace Transmissions", } diff --git a/src/borg/cockpit/widgets.py b/src/borg/cockpit/widgets.py index 4aef72e88f..9fba471373 100644 --- a/src/borg/cockpit/widgets.py +++ b/src/borg/cockpit/widgets.py @@ -3,26 +3,126 @@ """ import random +import re import time +from datetime import timedelta from rich.markup import escape from textual.app import ComposeResult -from textual.reactive import reactive -from textual.widgets import Static, RichLog +from textual.widgets import ProgressBar, RichLog, Static from textual.containers import Vertical, Container -from ..helpers import classify_ec +from ..helpers import classify_ec, format_file_size, format_timedelta +from ..helpers.parseformat import ellipsis_truncate from .translator import T, TRANSLATOR -class StatusPanel(Static): - elapsed_time = reactive(0.0, init=False) - files_count = reactive(0, init=False) # unchanged + modified + added + other + error - unchanged_count = reactive(0, init=False) - modified_count = reactive(0, init=False) - added_count = reactive(0, init=False) - other_count = reactive(0, init=False) - error_count = reactive(0, init=False) - rc = reactive(None, init=False) +class StatusPanelBase(Static): + """ + Base class of the panels showing the numbers of a borg run, next to the logo. + + Subclasses compose their lines (Static widgets with an id) and implement show_session(), which + shows the state of a Session in them; HEIGHT is the number of lines they need, the screen sizes + the top row accordingly. A line is only updated when its text changes. + """ + + HEIGHT = 0 + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.session = None + self.shown = {} # widget id -> text currently shown + + @staticmethod + def _line(widget_id, label, value="", classes="status"): + """A Static for one "Label: value" line, for compose().""" + return Static(T(label) + value, classes=classes, id=widget_id) + + def show(self, widget_id, text): + """Show in the widget with , if it is not shown already.""" + if self.shown.get(widget_id) != text: + self.shown[widget_id] = text + self.query_one(f"#{widget_id}").update(text) + + def show_value(self, widget_id, label, value, truncate=False): + """Show a translated label and a value; long values can be truncated to the panel width, so they don't wrap.""" + label = T(label) + value = str(value) + if truncate and value: + space = (self.size.width or 60) - len(label) - 1 + value = ellipsis_truncate(value, space).rstrip() + self.show(widget_id, label + escape(value)) + + def update_from_session(self, session): + """Show the current state of the session.""" + self.session = session + self.show_session(session) + + def show_session(self, session): + raise NotImplementedError + + def update_speed(self, session): + """Called once per second, after Session.sample(): update the speed display, if the panel has one.""" + + def show_speed(self, session): + """Show the current rates, if the panel has a speed display.""" + + def refresh_ui_labels(self): + """Redo all lines with the current translation.""" + self.shown.clear() + if self.session is not None: + self.show_session(self.session) + self.show_speed(self.session) + + # lines most panels have + + @staticmethod + def _format_size(size): + return "-" if size is None else format_file_size(size) + + def show_elapsed(self, session): + if TRANSLATOR.enabled: + # There seems to be no official formula for stardates, so we make something up. + # When showing the stardate, it is an absolute time, not relative "elapsed time". + ut = time.time() + sd = (ut - 1735689600) / 60.0 # Minutes since 2025-01-01 00:00.00 UTC + self.show("status-elapsed", f"Stardate {sd:.1f}") + else: + seconds = int(session.elapsed) + days, seconds = divmod(seconds, 86400) + h, m, s = seconds // 3600, (seconds % 3600) // 60, seconds % 60 + self.show("status-elapsed", f"Elapsed: {days:02d}d {h:02d}:{m:02d}:{s:02d}") + + def show_count(self, widget_id, label, count): + """Show a count that is fine when zero and a warning otherwise.""" + widget = self.query_one(f"#{widget_id}") + widget.set_class(count == 0, "errors-ok") + widget.set_class(count != 0, "errors-warning") + self.show_value(widget_id, label, count) + + def show_warnings(self, session): + self.show_count("status-warnings", "Warnings: ", session.warnings + session.errors) + + def show_activity(self, session): + """What borg works on right now.""" + self.show_value("status-activity", "Progress: ", session.progress_text, truncate=True) + + def show_rc(self, session): + rc = session.rc + if rc is None: + self.show("status-rc", T("RC: ") + "RUNNING") + return + status = classify_ec(rc) + widget = self.query_one("#status-rc") + widget.set_class(status == "success", "rc-ok") + widget.set_class(status == "warning", "rc-warning") + widget.set_class(status not in ("success", "warning"), "rc-error") # error, signal + self.show("status-rc", T("RC: ") + str(rc)) + + +class CreateStatusPanel(StatusPanelBase): + """create, import-tar, recreate, transfer: the statistics of the archive being created.""" + + HEIGHT = 17 # sparkline (4), speed (1), 12 lines def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -31,154 +131,220 @@ def __init__(self, *args, **kwargs): def compose(self) -> ComposeResult: with Vertical(): yield SpeedSparkline(self.speed_history, id="speed-sparkline") - yield Static(T("Speed: 0/s"), id="status-speed") + yield self._line("status-speed", "Speed: ", "0 files/s", classes="") with Vertical(id="statuses"): - yield Static(T("Elapsed: 00d 00:00:00"), classes="status", id="status-elapsed") - yield Static(T("Files: 0"), classes="status", id="status-files") - yield Static(T("Unchanged: 0"), classes="status", id="status-unchanged") - yield Static(T("Modified: 0"), classes="status", id="status-modified") - yield Static(T("Added: 0"), classes="status", id="status-added") - yield Static(T("Other: 0"), classes="status", id="status-other") - yield Static(T("Errors: 0"), classes="status error-ok", id="status-errors") - yield Static(T("RC: RUNNING"), classes="status", id="status-rc") - - def update_speed(self, kfiles_per_second: float): - self.speed_history.append(kfiles_per_second) + yield self._line("status-elapsed", "Elapsed: ", "00d 00:00:00") + yield self._line("status-files", "Files: ", "0") + yield self._line("status-original", "Original: ", "-") + yield self._line("status-deduplicated", "Deduplicated: ", "-") + yield self._line("status-unchanged", "Unchanged: ", "0") + yield self._line("status-modified", "Modified: ", "0") + yield self._line("status-added", "Added: ", "0") + yield self._line("status-other", "Other: ", "0") + yield self._line("status-errors", "Errors: ", "0", classes="status errors-ok") + yield self._line("status-warnings", "Warnings: ", "0", classes="status errors-ok") + yield self._line("status-activity", "Progress: ") + yield self._line("status-rc", "RC: ", "RUNNING") + + def show_session(self, session): + self.show_elapsed(session) + self.show_value("status-files", "Files: ", session.nfiles) + original, deduplicated = session.original_size, session.deduplicated_size + self.show_value("status-original", "Original: ", self._format_size(original)) + ratio = f" ({deduplicated * 100 / original:.1f}%)" if deduplicated is not None and original else "" + self.show_value("status-deduplicated", "Deduplicated: ", self._format_size(deduplicated) + ratio) + self.show_value("status-unchanged", "Unchanged: ", session.count("U-")) + self.show_value("status-modified", "Modified: ", session.count("M")) + self.show_value("status-added", "Added: ", session.count("A+")) + self.show_value("status-other", "Other: ", sum(session.files_stats.values()) - session.count("U-MA+E")) + self.show_count("status-errors", "Errors: ", session.count("E")) + self.show_warnings(session) + if not session.running and session.archive_name is not None: + # the final --json output tells about the archive that was created. + duration = session.archive_duration + value = session.archive_name + if duration is not None: + value += f" ({format_timedelta(timedelta(seconds=duration))})" + self.show_value("status-activity", "Archive: ", value, truncate=True) + else: + self.show_activity(session) + self.show_rc(session) + + def update_speed(self, session): + self.speed_history.append(session.files_per_second) self.speed_history = self.speed_history[-SpeedSparkline.HISTORY_SIZE :] - # Use our custom update method self.query_one("#speed-sparkline").update_data(self.speed_history) - self.query_one("#status-speed").update(T(f"Speed: {int(kfiles_per_second * 1000)}/s")) + self.show_speed(session) - def watch_error_count(self, count: int) -> None: - sw = self.query_one("#status-errors") - if count == 0: - sw.remove_class("errors-warning") - sw.add_class("errors-ok") - else: - sw.remove_class("errors-ok") - sw.add_class("errors-warning") - sw.update(T(f"Errors: {count}")) + def show_speed(self, session): + rates = f"{session.files_per_second:.0f} files/s, {format_file_size(session.original_bytes_per_second)}/s" + self.show("status-speed", T("Speed: ") + rates) - def watch_files_count(self, count: int) -> None: - self.query_one("#status-files").update(T(f"Files: {count}")) - def watch_unchanged_count(self, count: int) -> None: - self.query_one("#status-unchanged").update(T(f"Unchanged: {count}")) +class ExtractStatusPanel(StatusPanelBase): + """extract, export-tar: a progress bar over the bytes to extract, and the counts of the --list lines.""" - def watch_modified_count(self, count: int) -> None: - self.query_one("#status-modified").update(T(f"Modified: {count}")) + HEIGHT = 14 # sparkline (4), speed (1), progress bar (1), 8 lines + PHASE = "extract" # the msgid of the progress operation the bar shows - def watch_added_count(self, count: int) -> None: - self.query_one("#status-added").update(T(f"Added: {count}")) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.speed_history = [0.0] * SpeedSparkline.HISTORY_SIZE - def watch_other_count(self, count: int) -> None: - self.query_one("#status-other").update(T(f"Other: {count}")) + def compose(self) -> ComposeResult: + with Vertical(): + yield SpeedSparkline(self.speed_history, id="speed-sparkline") + yield self._line("status-speed", "Speed: ", "0 B/s", classes="") + yield ProgressBar(id="extract-bar") # indeterminate until the total is known - def watch_rc(self, rc: int): - label = self.query_one("#status-rc") - if rc is None: - label.update(T("RC: RUNNING")) - return + with Vertical(id="statuses"): + yield self._line("status-extracted", "Extracted: ", "-") + yield self._line("status-elapsed", "Elapsed: ", "00d 00:00:00") + yield self._line("status-items", "Items: ", "0") + yield self._line("status-included", "Included: ", "0") + yield self._line("status-excluded", "Excluded: ", "0") + yield self._line("status-warnings", "Warnings: ", "0", classes="status errors-ok") + yield self._line("status-activity", "Progress: ") + yield self._line("status-rc", "RC: ", "RUNNING") + + def show_session(self, session): + phase = session.phase(self.PHASE) + bar = self.query_one("#extract-bar") + extracted = "-" + if phase is not None and phase.total: + current = phase.total if phase.finished else min(phase.current or 0, phase.total) + bar.update(total=phase.total, progress=current) + extracted = f"{format_file_size(current)} / {format_file_size(phase.total)}" + elif phase is not None and phase.finished: # there was nothing to extract + bar.update(total=1, progress=1) + self.show_value("status-extracted", "Extracted: ", extracted) + self.show_elapsed(session) + self.show_value("status-items", "Items: ", sum(session.status_counts.values())) + self.show_value("status-included", "Included: ", session.status_counts.get("+", 0)) + self.show_value("status-excluded", "Excluded: ", session.status_counts.get("-", 0)) + self.show_warnings(session) + self.show_activity(session) + self.show_rc(session) + + def _rate(self, session): + phase = session.phase(self.PHASE) + return 0.0 if phase is None else phase.rate + + def update_speed(self, session): + self.speed_history.append(self._rate(session)) + self.speed_history = self.speed_history[-SpeedSparkline.HISTORY_SIZE :] + self.query_one("#speed-sparkline").update_data(self.speed_history) + self.show_speed(session) - label.remove_class("rc-ok") - label.remove_class("rc-warning") - label.remove_class("rc-error") + def show_speed(self, session): + self.show("status-speed", T("Speed: ") + f"{format_file_size(self._rate(session))}/s") - status = classify_ec(rc) - if status == "success": - label.add_class("rc-ok") - elif status == "warning": - label.add_class("rc-warning") - else: # error, signal - label.add_class("rc-error") - label.update(T(f"RC: {rc}")) +class GenericStatusPanel(StatusPanelBase): + """All other commands: elapsed time, warnings, exit code and the phases borg reports progress for.""" - def watch_elapsed_time(self, elapsed: float) -> None: - if TRANSLATOR.enabled: - # There seems to be no official formula for stardates, so we make something up. - # When showing the stardate, it is an absolute time, not relative "elapsed time". - ut = time.time() - sd = (ut - 1735689600) / 60.0 # Minutes since 2025-01-01 00:00.00 UTC - msg = f"Stardate {sd:.1f}" - else: - seconds = int(elapsed) - days, seconds = divmod(seconds, 86400) - h, m, s = seconds // 3600, (seconds % 3600) // 60, seconds % 60 - msg = f"Elapsed: {days:02d}d {h:02d}:{m:02d}:{s:02d}" - self.query_one("#status-elapsed").update(msg) + HEIGHT = 17 # 4 lines, the title, PHASE_LINES + PHASE_LINES = 12 # the phases shown (the last ones, if there are more) + BAR_WIDTH = 10 + PERCENTAGE = re.compile(r"\s*\d+(\.\d+)?%$") # the percentage at the end of a progress message - def refresh_ui_labels(self): - """Update static UI labels with current translation.""" - self.watch_elapsed_time(self.elapsed_time) - self.query_one("#status-files").update(T(f"Files: {self.files_count}")) - self.query_one("#status-unchanged").update(T(f"Unchanged: {self.unchanged_count}")) - self.query_one("#status-modified").update(T(f"Modified: {self.modified_count}")) - self.query_one("#status-added").update(T(f"Added: {self.added_count}")) - self.query_one("#status-other").update(T(f"Other: {self.other_count}")) - self.query_one("#status-errors").update(T(f"Errors: {self.error_count}")) - - if self.rc is not None: - self.watch_rc(self.rc) - else: - self.query_one("#status-rc").update(T("RC: RUNNING")) + def compose(self) -> ComposeResult: + with Vertical(): + with Vertical(id="statuses"): + yield self._line("status-elapsed", "Elapsed: ", "00d 00:00:00") + yield self._line("status-warnings", "Warnings: ", "0", classes="status errors-ok") + yield self._line("status-archives", "Archives: ", "-") + yield self._line("status-rc", "RC: ", "RUNNING") + yield Static(T("Phases"), classes="panel-title", id="phases-title") + yield Static("", id="phases") + + def show_session(self, session): + self.show_elapsed(session) + self.show_warnings(session) + counts = session.archive_counts + statuses = [status for status in ("kept", "pruned", "deleted", "undeleted") if counts[status]] + statuses += [status for status in counts if status not in statuses] + parts = [f"{counts[status]} {status}" for status in statuses] + self.show_value("status-archives", "Archives: ", ", ".join(parts) if parts else "-") + self.show_rc(session) + self.show("phases-title", T("Phases")) + space = (self.size.width or 60) - self.BAR_WIDTH - 3 + lines = [] + for phase in list(session.phases.values())[-self.PHASE_LINES :]: + if phase.finished: + mark, filled, style = "✔", self.BAR_WIDTH, "green" + else: + fraction = phase.fraction + mark, filled, style = "▶", 0 if fraction is None else round(fraction * self.BAR_WIDTH), "bold white" + bar = "█" * filled + "░" * (self.BAR_WIDTH - filled) + message = phase.message or phase.msgid or "" + if phase.finished: # the last percentage borg reported before finishing is not the final one + message = self.PERCENTAGE.sub("", message) + text = ellipsis_truncate(message, space).rstrip() + lines.append(f"[{style}]{mark} {bar} {escape(text)}[/]") + self.show("phases", "\n".join(lines)) class StandardLog(Vertical): + """The log panel: log messages, --list lines and everything else borg outputs.""" + + # Styles for the --list status characters, see "Item flags" in the borg create help. + STATUS_STYLES = { + "E": "red", # error + "C": "yellow", # regular file, changed while reading + "?": "red", # missing status, a bug + "A": "white", # added regular file (cache miss, slow!) + "M": "white", # modified regular file (cache hit, but different, slow!) + "U": "green", # unchanged regular file (cache hit) + "-": "white", # excluded + "x": "white", # skipped (dataless) + } + DEFAULT_STATUS_STYLE = "green" # d, b, c, h, s, f, i: metadata only. +: included. + # Styles for the log levels (and the prompts and the final statistics). + LEVEL_STYLES = { + "DEBUG": "dim", + "WARNING": "yellow", + "ERROR": "red", + "CRITICAL": "bold red", + "PROMPT": "bold yellow", + "STATS": "bold", + } + MAX_LINES = 5000 # lines kept for scrolling back + def compose(self) -> ComposeResult: yield Static(T("Log"), classes="panel-title", id="standard-log-title") - yield RichLog(id="standard-log-content", highlight=False, markup=True, auto_scroll=True, max_lines=None) + yield RichLog( + id="standard-log-content", highlight=False, markup=True, auto_scroll=True, max_lines=self.MAX_LINES + ) def update_title(self): self.query_one("#standard-log-title").update(T("Log")) - def add_line(self, line: str): - # TODO: make this more generic, use json output from borg. - # currently, this is only really useful for borg create/extract --list - line = line.rstrip() - if len(line) == 0: + @classmethod + def style_for(cls, line): + """The rich style for a Line from the Session, None for plain text.""" + if line.kind == "status": + return cls.STATUS_STYLES.get(line.tag, cls.DEFAULT_STATUS_STYLE) + if line.kind == "archive": + return "green" if line.tag in ("kept", "undeleted") else "white" # the archive stays / is back + if line.kind == "log": + return cls.LEVEL_STYLES.get(line.tag) + if line.kind == "hint": + return "bold yellow" + return None + + def add_lines(self, lines, dropped=0): + """Append the lines taken from Session.drain(); dropped lines are only mentioned.""" + if not lines and not dropped: return - - markup_tag = None - if len(line) >= 2: - if line[1] == " " and line[0] in "EAMUdcbs+-": - # looks like from borg create/extract --list - status_panel = self.app.query_one("#status") - status_panel.files_count += 1 - status = line[0] - match status: - case "E": - status_panel.error_count += 1 - case "U" | "-": - status_panel.unchanged_count += 1 - case "M": - status_panel.modified_count += 1 - case "A" | "+": - status_panel.added_count += 1 - case "d" | "c" | "b" | "s": - status_panel.other_count += 1 - - markup_tag = { - "E": "red", # Error - "A": "white", # Added regular file (cache miss, slow!) - "M": "white", # Modified regular file (cache hit, but different, slow!) - "U": "green", # Updated regular file (cache hit) - "d": "green", # directory - "c": "green", # char device - "b": "green", # block device - "s": "green", # socket - "-": "white", # excluded - "+": "green", # included - }.get(status) - log_widget = self.query_one("#standard-log-content") - - safe_line = escape(line) - if markup_tag: - safe_line = f"[{markup_tag}]{safe_line}[/]" - - log_widget.write(safe_line) + if dropped: + log_widget.write(f"[dim]... {dropped} more lines not shown ...[/]") + for line in lines: + text = escape(line.text) + style = self.style_for(line) + log_widget.write(f"[{style}]{text}[/]" if style else text) class Starfield(Static): diff --git a/src/borg/testsuite/archiver/delete_cmd_test.py b/src/borg/testsuite/archiver/delete_cmd_test.py index e90d324e44..10bae135a6 100644 --- a/src/borg/testsuite/archiver/delete_cmd_test.py +++ b/src/borg/testsuite/archiver/delete_cmd_test.py @@ -2,6 +2,8 @@ from ...constants import * # NOQA from ...helpers import CommandError +import json + from . import cmd, create_regular_file, generate_archiver_tests, RK_ENCRYPTION pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,binary") # NOQA @@ -135,3 +137,24 @@ def test_delete_name_and_match_archives_are_combined(archivers, request, monkeyp cmd(archiver, "delete", "home", "-a", "host:host1") output = cmd(archiver, "repo-list", "--format={hostname}{NL}") assert output.strip() == "host2" + + +def test_delete_list_json(archivers, request): + archiver = request.getfixturevalue(archivers) + create_regular_file(archiver.input_path, "file1", size=1024 * 80) + cmd(archiver, "repo-create", RK_ENCRYPTION) + cmd(archiver, "create", "test1", "input") + cmd(archiver, "create", "test2", "input") + # with --log-json, the listing consists of archive_status objects (one per archive), no text lines + output = cmd(archiver, "delete", "--dry-run", "--list", "--log-json", "-a", "sh:test*") + messages = [json.loads(line) for line in output.splitlines()] + statuses = [msg for msg in messages if msg["type"] == "archive_status"] + assert {(msg["name"], msg["status"]) for msg in statuses} == {("test1", "deleted"), ("test2", "deleted")} + for msg in statuses: + assert msg["message"].startswith("Would delete: ") and msg["name"] in msg["message"] + assert msg["archive"] == msg["name"] and len(msg["id"]) == 64 and "T" in msg["time"] + output = cmd(archiver, "delete", "--list", "--log-json", "test1") + statuses = [msg for msg in map(json.loads, output.splitlines()) if msg["type"] == "archive_status"] + assert [(msg["name"], msg["status"]) for msg in statuses] == [("test1", "deleted")] + assert statuses[0]["message"].startswith("Deleted archive: ") and statuses[0]["message"].endswith("(1/1)") + assert "test1" not in cmd(archiver, "repo-list") diff --git a/src/borg/testsuite/archiver/extract_cmd_test.py b/src/borg/testsuite/archiver/extract_cmd_test.py index f40edea26e..a4cb2914d0 100644 --- a/src/borg/testsuite/archiver/extract_cmd_test.py +++ b/src/borg/testsuite/archiver/extract_cmd_test.py @@ -1,5 +1,6 @@ import errno import io +import json import os from pathlib import Path import shutil @@ -1145,3 +1146,21 @@ def close(self): out = _extract_with_raw_file_class(archiver, BadCloseRaw, BackupIOError) assert f"input/file1: close: [Errno {errno.EIO}] Input/output error" in out assert os.path.getsize("output/input/file1") == 1024 # the content was written completely + + +def test_extract_list_json(archivers, request): + archiver = request.getfixturevalue(archivers) + cmd(archiver, "repo-create", RK_ENCRYPTION) + create_regular_file(archiver.input_path, "file1", size=1024) + create_regular_file(archiver.input_path, "file2", size=1024) + cmd(archiver, "create", "test", "input") + + with changedir("output"): + output = cmd(archiver, "extract", "test", "--list", "--log-json", "-e", "input/file2") + # with --log-json, the listing consists of file_status objects (one per item), no text lines + messages = [json.loads(line) for line in output.splitlines()] + file_status = [msg for msg in messages if msg["type"] == "file_status"] + assert {"type": "file_status", "status": "+", "path": "input"} in file_status + assert {"type": "file_status", "status": "+", "path": "input/file1"} in file_status + assert {"type": "file_status", "status": "-", "path": "input/file2"} in file_status + assert len(file_status) == 3 diff --git a/src/borg/testsuite/archiver/prune_cmd_test.py b/src/borg/testsuite/archiver/prune_cmd_test.py index e101d4cd1a..d6a66d6f59 100644 --- a/src/borg/testsuite/archiver/prune_cmd_test.py +++ b/src/borg/testsuite/archiver/prune_cmd_test.py @@ -1074,3 +1074,32 @@ def test_prune_group_by_invalid_key(archivers, request): cmd(archiver, "repo-create", RK_ENCRYPTION) output = cmd(archiver, "prune", "--group-by", "bogus", "--keep-daily=1", exit_code=2) assert "Invalid group-by key: bogus" in output + + +def test_prune_list_json(archivers, request, backup_files): + archiver = request.getfixturevalue(archivers) + cmd(archiver, "repo-create", RK_ENCRYPTION) + cmd(archiver, "create", "test1", backup_files) + cmd(archiver, "create", "test2", backup_files) + # with --log-json, the listing consists of archive_status objects (one per listed archive), no text lines + output = prune_ungrouped(archiver, "--list", "--dry-run", "--keep-daily=1", "--log-json") + messages = [json.loads(line) for line in output.splitlines()] + statuses = {msg["name"]: msg for msg in messages if msg["type"] == "archive_status"} + assert set(statuses) == {"test1", "test2"} + pruned, kept = statuses["test1"], statuses["test2"] + assert pruned["status"] == "pruned" and kept["status"] == "kept" + assert pruned["kept"] is False and pruned["deleted_archive_number"] == 1 + assert pruned["message"].startswith("Would prune:") and "test1" in pruned["message"] + assert kept["kept"] is True and kept["keep_rule"] == "daily" and kept["kept_oldest"] is False + assert kept["kept_archive_number"] == 1 + assert kept["message"].startswith("Keeping archive (rule: daily #1):") and "test2" in kept["message"] + for status in statuses.values(): + assert status["archive"] == status["name"] and len(status["id"]) == 64 and "T" in status["time"] + assert status["group"] == {} # grouping is switched off by prune_ungrouped() + # --list-pruned lists the pruned archives only, and it really prunes without --dry-run + output = prune_ungrouped(archiver, "--list-pruned", "--keep-daily=1", "--log-json") + messages = [json.loads(line) for line in output.splitlines()] + statuses = [msg for msg in messages if msg["type"] == "archive_status"] + assert [(msg["name"], msg["status"], msg["kept"]) for msg in statuses] == [("test1", "pruned", False)] + assert statuses[0]["message"].startswith("Pruning archive (1/1):") + assert "test1" not in cmd(archiver, "repo-list") diff --git a/src/borg/testsuite/archiver/tar_cmds_test.py b/src/borg/testsuite/archiver/tar_cmds_test.py index 497fd96cf2..31a5edc665 100644 --- a/src/borg/testsuite/archiver/tar_cmds_test.py +++ b/src/borg/testsuite/archiver/tar_cmds_test.py @@ -808,3 +808,22 @@ def set_acl(path, access=None, default=None): assert "acl_default" in extracted_dir_acl assert extracted_dir_acl["acl_default"] == dir_acl["acl_default"] assert b"user:root:r--" in dir_acl["acl_default"] + + +def test_export_tar_list_json(archivers, request): + archiver = request.getfixturevalue(archivers) + create_test_files(archiver.input_path) + os.unlink("input/flagfile") + cmd(archiver, "repo-create", RK_ENCRYPTION) + cmd(archiver, "create", "test", "input") + # the text listing has the "+" prefix, like the listing of borg extract + output = cmd(archiver, "export-tar", "test", "simple.tar", "--list", "--tar-format=GNU") + assert "+ input/file1\n" in output + assert "+ input/dir2\n" in output + # with --log-json, the listing consists of file_status objects (one per item), no text lines + output = cmd(archiver, "export-tar", "test", "simple2.tar", "--list", "--log-json", "--tar-format=GNU") + messages = [json.loads(line) for line in output.splitlines()] + file_status = [msg for msg in messages if msg["type"] == "file_status"] + assert {"type": "file_status", "status": "+", "path": "input/file1"} in file_status + assert {"type": "file_status", "status": "+", "path": "input/dir2"} in file_status + assert all(msg["status"] == "+" for msg in file_status) diff --git a/src/borg/testsuite/archiver/undelete_cmd_test.py b/src/borg/testsuite/archiver/undelete_cmd_test.py index 431ed62b06..1051d674e3 100644 --- a/src/borg/testsuite/archiver/undelete_cmd_test.py +++ b/src/borg/testsuite/archiver/undelete_cmd_test.py @@ -2,6 +2,8 @@ from ...constants import * # NOQA from ...helpers import CommandError +import json + from . import cmd, create_regular_file, generate_archiver_tests, RK_ENCRYPTION pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,binary") # NOQA @@ -125,3 +127,26 @@ def test_undelete_multiple_run(archivers, request): assert "normal" in output assert "deleted1" in output assert "deleted2" in output + + +def test_undelete_list_json(archivers, request): + archiver = request.getfixturevalue(archivers) + create_regular_file(archiver.input_path, "file1", size=1024 * 80) + cmd(archiver, "repo-create", RK_ENCRYPTION) + cmd(archiver, "create", "deleted1", "input") + cmd(archiver, "create", "deleted2", "input") + cmd(archiver, "delete", "deleted1") + cmd(archiver, "delete", "deleted2") + # with --log-json, the listing consists of archive_status objects (one per archive), no text lines + output = cmd(archiver, "undelete", "--dry-run", "--list", "--log-json", "-a", "sh:deleted*") + messages = [json.loads(line) for line in output.splitlines()] + statuses = [msg for msg in messages if msg["type"] == "archive_status"] + assert {(msg["name"], msg["status"]) for msg in statuses} == {("deleted1", "undeleted"), ("deleted2", "undeleted")} + for msg in statuses: + assert msg["message"].startswith("Would undelete: ") and msg["name"] in msg["message"] + assert msg["archive"] == msg["name"] and len(msg["id"]) == 64 and "T" in msg["time"] + output = cmd(archiver, "undelete", "--list", "--log-json", "-a", "sh:deleted1") + statuses = [msg for msg in map(json.loads, output.splitlines()) if msg["type"] == "archive_status"] + assert [(msg["name"], msg["status"]) for msg in statuses] == [("deleted1", "undeleted")] + assert statuses[0]["message"].startswith("Undeleted archive: ") and statuses[0]["message"].endswith("(1/1)") + assert "deleted1" in cmd(archiver, "repo-list") diff --git a/src/borg/testsuite/cockpit_session_test.py b/src/borg/testsuite/cockpit_session_test.py new file mode 100644 index 0000000000..298c9109e0 --- /dev/null +++ b/src/borg/testsuite/cockpit_session_test.py @@ -0,0 +1,493 @@ +"""Tests for the cockpit's event parsing, session model and borg runner. They do not need Textual.""" + +import asyncio +import json +import sys +import time + +import pytest + +from borg.cockpit.events import ( + ArchiveProgress, + ArchiveStatus, + FileStatus, + LogMessage, + ProcessFinished, + ProgressMessage, + ProgressPercent, + Question, + RawLine, + UnknownJson, + parse_json_line, +) +from borg.cockpit.runner import INJECTED_OPTIONS, BorgRunner, borg_command +from borg.cockpit.session import NO_TERMINAL_WARNING, NO_TERMINAL_HINT, Session + +# JSON lines as documented in docs/internals/frontends.rst +ARCHIVE_PROGRESS = ( + '{"original_size": 250012, "deduplicated_size": 250012, "nfiles": 3, "hashing_time": 0.5, "chunking_time": 0.25, ' + '"files_stats": {"A": 3, "d": 3}, "store_stats": {}, "path": "src/linux/file1", "time": 1787900398.684961, ' + '"type": "archive_progress", "finished": false}' +) +ARCHIVE_PROGRESS_FINISHED = '{"time": 1787900398.686938, "type": "archive_progress", "finished": true}' +FILE_STATUS = '{"type": "file_status", "status": "A", "path": "src/linux/baz/file2"}' +PROGRESS_PERCENT = ( + '{"message": " 20.0% Extracting: src/linux/baz/file3", "current": 50012, "total": 250012, ' + '"info": ["src/linux/baz/file3"], "operation": 1, "msgid": "extract", "type": "progress_percent", ' + '"finished": false, "time": 1787900399.5558112}' +) +PROGRESS_PERCENT_FINISHED = ( + '{"message": "", "operation": 1, "msgid": "extract", "type": "progress_percent", "finished": true, ' + '"time": 1787900399.556339}' +) +PROGRESS_MESSAGE = ( + '{"message": "Saving files cache", "operation": 2, "msgid": "cache.close", "type": "progress_message", ' + '"finished": false, "time": 1787900398.719723}' +) +LOG_MESSAGE = ( + '{"type": "log_message", "time": 1787900383.5105972, "message": "Repository does not exist.", ' + '"levelname": "ERROR", "name": "borg.archiver", "msgid": "Repository.DoesNotExist"}' +) +QUESTION_PROMPT = ( + '{"type": "question_prompt", "msgid": "BORG_CHECK_I_KNOW_WHAT_I_AM_DOING", ' + '"message": "This is a potentially dangerous function.\\n' + "Type 'YES' if you understand this and want to continue: \"}" +) +QUESTION_ENV_ANSWER = ( + '{"env_var": "BORG_CHECK_I_KNOW_WHAT_I_AM_DOING", "type": "question_env_answer", ' + '"msgid": "BORG_CHECK_I_KNOW_WHAT_I_AM_DOING", "message": "NO (from BORG_CHECK_I_KNOW_WHAT_I_AM_DOING)"}' +) + + +def test_parse_log_message(): + event = parse_json_line(LOG_MESSAGE) + assert event == LogMessage( + message="Repository does not exist.", + levelname="ERROR", + name="borg.archiver", + msgid="Repository.DoesNotExist", + time=1787900383.5105972, + ) + + +def test_parse_progress_percent(): + event = parse_json_line(PROGRESS_PERCENT) + assert isinstance(event, ProgressPercent) + assert (event.operation, event.msgid, event.current, event.total) == (1, "extract", 50012, 250012) + assert event.info == ["src/linux/baz/file3"] + assert event.message.endswith("src/linux/baz/file3") + assert not event.finished + finished = parse_json_line(PROGRESS_PERCENT_FINISHED) + assert isinstance(finished, ProgressPercent) + assert finished.finished and finished.current is None and finished.total is None and finished.message == "" + + +def test_parse_progress_message(): + event = parse_json_line(PROGRESS_MESSAGE) + assert event == ProgressMessage( + operation=2, message="Saving files cache", msgid="cache.close", finished=False, time=1787900398.719723 + ) + + +def test_parse_archive_progress(): + event = parse_json_line(ARCHIVE_PROGRESS) + assert isinstance(event, ArchiveProgress) + assert (event.original_size, event.deduplicated_size, event.nfiles) == (250012, 250012, 3) + assert (event.hashing_time, event.chunking_time) == (0.5, 0.25) + assert event.files_stats == {"A": 3, "d": 3} + assert event.path == "src/linux/file1" + assert not event.finished + finished = parse_json_line(ARCHIVE_PROGRESS_FINISHED) + assert isinstance(finished, ArchiveProgress) + assert finished.finished and finished.path is None and finished.nfiles == 0 + + +def test_parse_file_status(): + assert parse_json_line(FILE_STATUS) == FileStatus(status="A", path="src/linux/baz/file2") + + +def test_parse_question(): + prompt = parse_json_line(QUESTION_PROMPT) + assert isinstance(prompt, Question) + assert prompt.kind == "prompt" and prompt.needs_answer + assert prompt.msgid == "BORG_CHECK_I_KNOW_WHAT_I_AM_DOING" + assert prompt.message.startswith("This is a potentially dangerous function.\n") + env_answer = parse_json_line(QUESTION_ENV_ANSWER) + assert isinstance(env_answer, Question) + assert env_answer.kind == "env_answer" and not env_answer.needs_answer + assert env_answer.env_var == "BORG_CHECK_I_KNOW_WHAT_I_AM_DOING" + + +def test_parse_unknown_type_and_missing_keys(): + assert parse_json_line('{"type": "something_new", "x": 1}') == UnknownJson({"type": "something_new", "x": 1}) + # no crash on missing / odd keys, defaults are used + assert parse_json_line('{"type": "log_message"}') == LogMessage(message="") + assert parse_json_line('{"type": "archive_progress", "files_stats": null, "nfiles": "3"}') == ArchiveProgress() + assert parse_json_line('{"type": "progress_percent", "operation": 7, "current": null}') == ProgressPercent( + operation=7, message="" + ) + + +@pytest.mark.parametrize("line", ["", "not json", "42", "[1, 2]", '"text"', '{"no": "type"}', '{"type": 5}']) +def test_parse_not_an_event(line): + assert parse_json_line(line) is None + + +def feed_lines(session, lines): + for line in lines: + event = parse_json_line(line) + session.feed(event if event is not None else RawLine(stream="stderr", line=line)) + + +def test_session_archive_progress(): + session = Session() + assert session.running and session.nfiles == 0 and session.original_size is None + feed_lines(session, [ARCHIVE_PROGRESS]) + assert session.nfiles == 3 + assert session.original_size == 250012 and session.deduplicated_size == 250012 + assert session.files_stats == {"A": 3, "d": 3} + assert session.count("A") == 3 and session.count("dbcs") == 3 and session.count("U-") == 0 + assert session.progress_text == "src/linux/file1" + feed_lines(session, [ARCHIVE_PROGRESS_FINISHED]) + # the final object carries no statistics, the previous ones stay + assert session.archive_finished + assert session.nfiles == 3 and session.original_size == 250012 + assert session.progress_text == "" + + +def test_session_counts_list_lines(): + session = Session() + session.feed(FileStatus(status="A", path="a")) + session.feed(FileStatus(status="A", path="b")) + session.feed(FileStatus(status="d", path="dir")) + session.feed(FileStatus(status="E", path="broken")) + assert session.nfiles == 4 # all listed items + assert session.files_stats == {"A": 2, "d": 1, "E": 1} + lines, dropped = session.drain() + assert dropped == 0 + assert [(line.kind, line.tag, line.text) for line in lines] == [ + ("status", "A", "A a"), + ("status", "A", "A b"), + ("status", "d", "d dir"), + ("status", "E", "E broken"), + ] + # the list lines stay the source of the counts, archive_progress (rate limited, can lag behind) only gives the sizes + session.feed(ArchiveProgress(nfiles=3, original_size=1000, deduplicated_size=10, files_stats={"A": 1, "d": 1})) + assert session.nfiles == 4 and session.files_stats == {"A": 2, "d": 1, "E": 1} + assert session.original_size == 1000 and session.deduplicated_size == 10 + + +def test_session_archive_status(): + session = Session() + session.feed(ArchiveStatus(name="a1", status="pruned", message="Would prune: a1", data={"kept": False})) + session.feed(ArchiveStatus(name="a2", status="kept", message="Keeping archive (rule: daily #1): a2")) + session.feed(ArchiveStatus(name="a3", status="kept", message="Keeping archive (rule: daily #2): a3")) + session.feed(ArchiveStatus(name="a4", status="deleted", message="Deleted archive: a4 (1/1)")) + assert session.archive_counts == {"kept": 2, "pruned": 1, "deleted": 1} + assert session.files_stats == {} # archives are not items of a file listing + lines, _ = session.drain() + assert [(line.kind, line.tag, line.text) for line in lines] == [ + ("archive", "pruned", "Would prune: a1"), + ("archive", "kept", "Keeping archive (rule: daily #1): a2"), + ("archive", "kept", "Keeping archive (rule: daily #2): a3"), + ("archive", "deleted", "Deleted archive: a4 (1/1)"), + ] + + +def test_session_phases(): + session = Session() + feed_lines(session, [PROGRESS_PERCENT, PROGRESS_MESSAGE]) + assert list(session.phases) == [1, 2] + extract = session.phases[1] + assert (extract.msgid, extract.current, extract.total, extract.finished) == ("extract", 50012, 250012, False) + assert extract.fraction == pytest.approx(0.2, abs=0.001) + assert session.phases[2].fraction is None + assert session.progress_text == "Saving files cache" + feed_lines(session, [PROGRESS_PERCENT_FINISHED]) + assert extract.finished and extract.message.endswith("file3") # the message of the last update stays + assert session.progress_text == "" + # extract first reports a total of 0 while it computes the total, that must not crash the fraction + session.feed(ProgressPercent(operation=3, message="Calculating total archive size...", current=0, total=0)) + assert session.phases[3].fraction is None + + +def test_session_questions(): + session = Session() + feed_lines(session, [QUESTION_PROMPT]) + assert session.pending_question is not None and session.pending_question.needs_answer + lines, _ = session.drain() + assert lines[0].kind == "log" and lines[0].tag == "PROMPT" + feed_lines(session, [QUESTION_ENV_ANSWER]) + assert session.pending_question is None + + +def test_session_drain_is_bounded(): + session = Session() + n = 2 * Session.LINES_MAX + 5 + for i in range(n): + session.feed(RawLine(stream="stdout", line=f"line {i}")) + lines, dropped = session.drain() + assert len(lines) == Session.LINES_MAX + assert dropped == n - Session.LINES_MAX + assert lines[0].text == f"line {n - Session.LINES_MAX}" and lines[-1].text == f"line {n - 1}" + assert lines[0].kind == "raw" and lines[0].tag == "stdout" + assert session.drain() == ([], 0) + + +def test_session_sample_rates(): + session = Session() + session.feed(ArchiveProgress(nfiles=10, original_size=1000, deduplicated_size=100)) + session.sample(now=session.started + 2.0) + assert session.files_per_second == 5.0 + assert session.original_bytes_per_second == 500.0 + assert session.deduplicated_bytes_per_second == 50.0 + session.feed(ArchiveProgress(nfiles=10, original_size=1000, deduplicated_size=100)) + session.sample(now=session.started + 3.0) + assert session.files_per_second == 0.0 + session.sample(now=session.started + 3.0) # no time passed: keep the rates + assert session.files_per_second == 0.0 + + +def test_session_passphrase_hint(): + session = Session() + session.feed(RawLine(stream="stderr", line=NO_TERMINAL_WARNING)) + session.feed(RawLine(stream="stderr", line="Enter passphrase for key /repo: ", partial=True)) + assert session.passphrase_needed + lines, _ = session.drain() + assert [(line.kind, line.text) for line in lines] == [ + ("raw", NO_TERMINAL_WARNING), + ("hint", NO_TERMINAL_HINT), + ("raw", "Enter passphrase for key /repo: "), + ] + + +def test_session_process_finished(): + session = Session() + feed_lines(session, [QUESTION_PROMPT, PROGRESS_MESSAGE]) + session.feed(ProcessFinished(rc=2, error="boom")) + assert not session.running and session.rc == 2 and session.error == "boom" + assert session.pending_question is None and session.progress_text == "" + elapsed = session.elapsed + time.sleep(0.01) + assert session.elapsed == elapsed # frozen + lines, _ = session.drain() + assert lines[-1].kind == "log" and lines[-1].tag == "ERROR" and lines[-1].text == "boom" + + +def test_borg_command(): + assert borg_command(["create", "arch", "src"], executable=["borg"]) == [ + "borg", + "--log-json", + "--progress", + "create", + "arch", + "src", + ] + # no duplicates when the user gave them already + assert borg_command(["--progress", "create"], executable=["borg"]) == ["borg", "--log-json", "--progress", "create"] + assert borg_command(["create", "--log-json"], executable=["borg"]) == ["borg", "--progress", "create", "--log-json"] + # the default runs the interpreter that runs the cockpit + assert borg_command(["--version"])[: -len(INJECTED_OPTIONS) - 1] == [sys.executable, "-m", "borg"] + + +FAKE_BORG = """ +import json, sys + +def err(obj): + sys.stderr.write(json.dumps(obj) + "\\n") + sys.stderr.flush() + +err({"type": "log_message", "levelname": "INFO", "name": "borg.test", "message": "hello"}) +err({"type": "file_status", "status": "A", "path": "a/b"}) +sys.stderr.write("plain text\\n") +sys.stderr.flush() +print("stdout line", flush=True) +sys.stderr.write("Enter something: ") # a prompt: no newline, waits for stdin +sys.stderr.flush() +answer = sys.stdin.readline().strip() +print("got " + answer, flush=True) +sys.exit(2) +""" + + +def test_runner(): + events = [] + runner = BorgRunner(["--whatever"], events.append, executable=[sys.executable, "-c", FAKE_BORG]) + runner.PARTIAL_LINE_TIMEOUT = 0.2 + + async def run(): + task = asyncio.create_task(runner.start()) + deadline = time.monotonic() + 30 + while not any(isinstance(e, RawLine) and e.partial for e in events): + assert time.monotonic() < deadline, f"no partial line seen, events: {events}" + await asyncio.sleep(0.02) + await runner.answer("YES") + await asyncio.wait_for(task, 30) + + asyncio.run(run()) + assert LogMessage(message="hello", levelname="INFO", name="borg.test") in events + assert FileStatus(status="A", path="a/b") in events + assert RawLine(stream="stderr", line="plain text") in events + assert RawLine(stream="stdout", line="stdout line") in events + assert RawLine(stream="stderr", line="Enter something: ", partial=True) in events + assert RawLine(stream="stdout", line="got YES") in events + assert events[-1] == ProcessFinished(rc=2) + assert runner.process is None + + +def test_runner_start_failure(): + events = [] + runner = BorgRunner([], events.append, executable=["/nonexistent/borg-binary"]) + asyncio.run(runner.start()) + assert len(events) == 1 + assert isinstance(events[0], ProcessFinished) and events[0].rc == -1 and events[0].error + + +def test_borg_command_json_stdout(): + assert borg_command(["create", "x"], executable=["borg"], json_stdout=True) == [ + "borg", + "--log-json", + "--progress", + "create", + "x", + "--json", + ] + # --json goes before a "--" end-of-options marker, and is not duplicated + assert borg_command(["create", "x", "--", "-p"], executable=["borg"], json_stdout=True)[-4:] == [ + "x", + "--json", + "--", + "-p", + ] + assert borg_command(["create", "--json", "x"], executable=["borg"], json_stdout=True).count("--json") == 1 + + +FINAL_JSON = """ +{ + "archive": { + "duration": 90.5, + "id": "0123abcd0123abcd", + "name": "test", + "stats": { + "chunking_time": 0.25, + "deduplicated_size": 300, + "files_stats": {"A": 2, "M": 1, "d": 1}, + "hashing_time": 0.5, + "nfiles": 3, + "original_size": 3000, + "store_stats": {"store_calls": 7, "store_volume": 2048} + } + }, + "repository": {"id": "abcd", "location": "/repo"} +} +""" + + +def test_session_final_stats_from_stdout(): + session = Session(command="create", capture_stdout=True) + session.feed(ArchiveProgress(nfiles=2, original_size=2000, deduplicated_size=200, files_stats={"A": 2})) + session.feed(FileStatus(status="A", path="a")) + for line in FINAL_JSON.splitlines(): + session.feed(RawLine(stream="stdout", line=line)) + assert session.final_json is None and session.stdout_lines # only parsed at the end + lines, _ = session.drain() + assert [line.kind for line in lines] == ["status"] # the captured stdout is not logged + session.feed(ProcessFinished(rc=0)) + assert session.final_json["archive"]["name"] == "test" + assert session.archive_name == "test" and session.archive_duration == 90.5 + # the final statistics win over the --list lines and archive_progress + assert session.nfiles == 3 and session.files_stats == {"A": 2, "M": 1, "d": 1} + assert session.original_size == 3000 and session.deduplicated_size == 300 + lines, _ = session.drain() + assert all(line.kind == "log" and line.tag == "STATS" for line in lines) + assert [line.text for line in lines] == [ + "Archive name: test", + "Archive fingerprint: 0123abcd0123abcd", + "Duration: 1 minutes 30.500 seconds", + "Number of files: 3", + "Original size: 3.00 kB", + "Deduplicated size: 300 B", + "Time spent in hashing: 0.500 seconds", + "Time spent in chunking: 0.250 seconds", + "Added files: 2", + "Unchanged files: 0", + "Modified files: 1", + "Error files: 0", + "Files changed while reading: 0", + "Store store calls: 7", + "Store store volume: 2.05 kB", + ] + + +def test_session_final_stats_dry_run(): + session = Session(command="create", capture_stdout=True) + for line in '{"dry_run": true, "stats": {"nfiles": 5, "original_size": 1234}, "repository": {}}'.splitlines(): + session.feed(RawLine(stream="stdout", line=line)) + session.feed(ProcessFinished(rc=0)) + assert session.archive_name is None + assert session.nfiles == 5 and session.original_size == 1234 and session.deduplicated_size is None + lines, _ = session.drain() + assert [line.text for line in lines] == [ + "Dry run: no archive was created.", + "Number of files: 5", + "Original size: 1.23 kB", + ] + + +def test_session_stdout_that_is_not_json(): + session = Session(command="create", capture_stdout=True) + session.feed(RawLine(stream="stdout", line="just text")) + session.feed(ProcessFinished(rc=2)) + assert session.final_json is None and session.final_stats is None + lines, _ = session.drain() + assert [(line.kind, line.text) for line in lines] == [("raw", "just text")] + + +def test_session_counts_warnings(): + session = Session() + session.feed(LogMessage(message="w", levelname="WARNING")) + session.feed(LogMessage(message="e", levelname="ERROR")) + session.feed(LogMessage(message="c", levelname="CRITICAL")) + session.feed(LogMessage(message="i", levelname="INFO")) + assert (session.warnings, session.errors) == (1, 2) + + +def test_session_phase_lookup_and_rates(): + session = Session() + session.feed(ProgressPercent(operation=1, msgid="extract", message="", current=0, total=1000)) + session.feed(ProgressMessage(operation=2, msgid="cache.close", message="Saving")) + assert session.phase("extract").operation == 1 and session.phase("nope") is None + assert session.active_phase.operation == 2 + session.sample(now=session.started + 1.0) + session.feed(ProgressPercent(operation=1, msgid="extract", message="", current=500, total=1000)) + session.sample(now=session.started + 2.0) + assert session.phase("extract").rate == 500.0 + assert session.active_phase.operation == 1 + session.feed(ProgressPercent(operation=1, msgid="extract", finished=True, message="")) + assert session.active_phase is None + session.sample(now=session.started + 3.0) + assert session.phase("extract").rate == 0.0 + # after the run, the rates are zero + session.feed(ArchiveProgress(nfiles=100)) + session.feed(ProcessFinished(rc=0)) + session.sample(now=session.started + 4.0) + assert session.files_per_second == 0.0 + + +def test_parse_archive_status(): + line = ( + '{"name": "daily-2026-09-09", "archive": "daily-2026-09-09", "id": "ab12", ' + '"time": "2026-09-09T02:00:00+02:00", ' + '"group": {"name": "daily"}, "kept": true, "keep_rule": "daily", "kept_oldest": false, ' + '"kept_archive_number": 1, "status": "kept", "type": "archive_status", ' + '"message": "Keeping archive (rule: daily #1): ..."}' + ) + event = parse_json_line(line) + assert isinstance(event, ArchiveStatus) + assert (event.name, event.status) == ("daily-2026-09-09", "kept") + assert event.message.startswith("Keeping archive") + assert event.data["keep_rule"] == "daily" and event.data["group"] == {"name": "daily"} + line = '{"type": "archive_status", "name": "old", "status": "deleted", "message": "Deleted archive: old (1/1)"}' + assert parse_json_line(line) == ArchiveStatus( + name="old", status="deleted", message="Deleted archive: old (1/1)", data=json.loads(line) + ) diff --git a/src/borg/testsuite/cockpit_test.py b/src/borg/testsuite/cockpit_test.py index d2eeba240e..5d2a4d5566 100644 --- a/src/borg/testsuite/cockpit_test.py +++ b/src/borg/testsuite/cockpit_test.py @@ -1,12 +1,29 @@ +"""Tests for the cockpit application. They need Textual; the borg process is faked, except in the slow test.""" + import asyncio +import json import subprocess +import time import pytest +from borg.cockpit.events import ( + ArchiveProgress, + ArchiveStatus, + FileStatus, + LogMessage, + ProcessFinished, + ProgressMessage, + ProgressPercent, + Question, + RawLine, +) from borg.platformflags import is_freebsd, is_win32 try: from borg.cockpit.app import BorgCockpitApp + from borg.cockpit.prompt import PromptModal + from borg.cockpit.screens import CreateScreen, ExtractScreen, GenericScreen, screen_for_command have_cockpit = True except ImportError: @@ -15,6 +32,224 @@ pytestmark = pytest.mark.skipif(not have_cockpit, reason="can not import BorgCockpitApp, is textual installed?") +class FakeRunner: + """Replays events instead of running borg. After a prompt, it waits for the answer.""" + + def __init__(self, args, callback, events=(), rc=0, json_stdout=False): + self.args = list(args) + self.callback = callback + self.events = list(events) + self.rc = rc + self.json_stdout = json_stdout + self.answers = [] + self.answered = asyncio.Event() + + async def start(self): + for event in self.events: + self.callback(event) + if isinstance(event, Question) and event.needs_answer: + await self.answered.wait() + self.answered.clear() + await asyncio.sleep(0) + self.callback(ProcessFinished(rc=self.rc)) + + async def answer(self, text): + self.answers.append(text) + self.answered.set() + + async def stop(self): + pass + + +def make_runner_factory(events, rc=0): + """A runner_factory for BorgCockpitApp, remembering the FakeRunner it created in the returned list.""" + created = [] + + def factory(args, callback, **kwargs): + runner = FakeRunner(args, callback, events=events, rc=rc, **kwargs) + created.append(runner) + return runner + + return factory, created + + +async def wait_until(pilot, predicate, timeout=10.0): + deadline = time.monotonic() + timeout + while not predicate(): + assert time.monotonic() < deadline, "timeout while waiting for the app" + await pilot.pause(0.05) + + +async def run_to_the_end(app, inspect=None): + """ + Run the app until the (fake) borg has finished and the widgets show the final state. + + Returns the texts shown by the status panel, the log text and the result of inspect(app), if given + (the widgets can only be inspected while the app runs). + """ + async with app.run_test(size=(100, 30)) as pilot: + await wait_until(pilot, lambda: not app.session.running) + await pilot.pause(0.5) # let the refresh timer show the final state + check_layout(app) + return app.query_one("#status").shown, log_text(app), inspect(app) if inspect else None + + +def log_text(app): + return "\n".join(strip.text for strip in app.query_one("#standard-log-content").lines) + + +def check_layout(app): + """The status panel must fit into the top row, next to the logo, with the log panel below.""" + top_row, status, log = app.query_one("#top-row"), app.query_one("#status"), app.query_one("#standard-log") + assert top_row.size.height == status.HEIGHT # size is the content area, without the border + assert status.region.bottom <= top_row.region.bottom + rc_line = app.query_one("#status-rc") + assert rc_line.region.height == 1 and rc_line.region.bottom <= status.region.bottom + assert log.region.y >= top_row.region.bottom and log.size.height >= 5 + + +FINAL_JSON = { + "archive": { + "name": "test", + "id": "0123abcd" * 8, + "duration": 1.5, + "stats": { + "nfiles": 3, + "original_size": 3000, + "deduplicated_size": 300, + "hashing_time": 0.1, + "chunking_time": 0.2, + "files_stats": {"A": 2, "M": 1, "d": 1}, + "store_stats": {"store_calls": 7}, + }, + }, + "repository": {"id": "ab" * 32, "location": "/repo"}, +} + + +def test_screen_for_command(): + assert screen_for_command("create") is CreateScreen + assert screen_for_command("import-tar") is CreateScreen + assert screen_for_command("extract") is ExtractScreen + assert screen_for_command("export-tar") is ExtractScreen + assert screen_for_command("check") is GenericScreen + assert screen_for_command(None) is GenericScreen + + +def test_app_create_screen(): + events = [ + LogMessage(message="Creating archive", levelname="INFO"), + LogMessage(message="something is odd", levelname="WARNING"), + ArchiveProgress( + original_size=1000, deduplicated_size=100, nfiles=2, files_stats={"A": 1, "M": 1}, path="src/a" + ), + FileStatus(status="A", path="src/a"), + FileStatus(status="M", path="src/b"), + FileStatus(status="d", path="src"), + ArchiveProgress( + original_size=2900, deduplicated_size=290, nfiles=3, files_stats={"A": 2, "M": 1, "d": 1}, path="src/c" + ), + ArchiveProgress(finished=True), + ] + events += [RawLine(stream="stdout", line=line) for line in json.dumps(FINAL_JSON, indent=4).splitlines()] + factory, runners = make_runner_factory(events, rc=1) + app = BorgCockpitApp(borg_args=["create", "test", "src"], command="create", runner_factory=factory) + shown, text, _ = asyncio.run(run_to_the_end(app)) + assert isinstance(app.main_screen, CreateScreen) + assert runners[0].args == ["create", "test", "src"] and runners[0].json_stdout + # the final numbers come from the --json output + assert shown["status-files"] == "Files: 3" + assert shown["status-original"] == "Original: 3.00 kB" + assert shown["status-deduplicated"] == "Deduplicated: 300 B (10.0%)" + assert shown["status-added"] == "Added: 2" and shown["status-modified"] == "Modified: 1" + assert shown["status-other"] == "Other: 1" and shown["status-errors"] == "Errors: 0" + assert shown["status-warnings"] == "Warnings: 1" + assert shown["status-activity"].startswith("Archive: test (1.") + assert shown["status-rc"] == "RC: 1" + assert "Creating archive" in text and "something is odd" in text + assert "A src/a" in text and "M src/b" in text and "d src" in text + assert "Archive name: test" in text and "Number of files: 3" in text and "Store store calls: 7" in text + + +def test_app_extract_screen(): + events = [ + ProgressPercent(operation=1, msgid="extract", message="Calculating total archive size", current=0, total=0), + ProgressPercent(operation=1, msgid="extract", message=" 25.0% Extracting: a", current=250, total=1000), + FileStatus(status="+", path="a"), + FileStatus(status="-", path="b"), + ProgressPercent(operation=1, msgid="extract", message=" 75.0% Extracting: c", current=750, total=1000), + ProgressPercent(operation=1, msgid="extract", finished=True, message=""), + ProgressPercent(operation=2, msgid="extract.permissions", message="Setting directory permissions 50%"), + ] + factory, runners = make_runner_factory(events) + app = BorgCockpitApp(borg_args=["extract", "--list", "test"], command="extract", runner_factory=factory) + shown, text, bar = asyncio.run( + run_to_the_end(app, lambda app: (app.query_one("#extract-bar").total, app.query_one("#extract-bar").progress)) + ) + assert isinstance(app.main_screen, ExtractScreen) + assert not runners[0].json_stdout + assert bar == (1000, 1000) # finished: complete + assert shown["status-extracted"] == "Extracted: 1.00 kB / 1.00 kB" + assert shown["status-items"] == "Items: 2" + assert shown["status-included"] == "Included: 1" and shown["status-excluded"] == "Excluded: 1" + assert shown["status-rc"] == "RC: 0" + assert "+ a" in text and "- b" in text + + +def test_app_generic_screen(): + events = [ + ProgressPercent(operation=1, msgid="check.index", message="Checking index 50%", current=50, total=100), + ProgressMessage(operation=2, msgid="cache.close", message="Saving files cache"), + ProgressPercent(operation=1, msgid="check.index", finished=True, message=""), + LogMessage(message="Archive consistency check complete, no problems found.", levelname="INFO"), + ArchiveStatus(name="old", status="pruned", message="Would prune: old"), + ArchiveStatus(name="new", status="kept", message="Keeping archive (rule: daily #1): new"), + ] + factory, runners = make_runner_factory(events) + app = BorgCockpitApp(borg_args=["check"], command="check", runner_factory=factory) + shown, text, _ = asyncio.run(run_to_the_end(app)) + assert isinstance(app.main_screen, GenericScreen) + assert shown["phases-title"] == "Phases" + assert shown["phases"].splitlines() == [ + "[green]✔ ██████████ Checking index[/]", # finished: without the last percentage + "[bold white]▶ ░░░░░░░░░░ Saving files cache[/]", + ] + assert shown["status-warnings"] == "Warnings: 0" and shown["status-rc"] == "RC: 0" + assert shown["status-archives"] == "Archives: 1 kept, 1 pruned" + assert "no problems found" in text and "Would prune: old" in text + + +def test_app_answers_prompt(): + events = [ + Question(kind="prompt", message="Do something dangerous? [yN]: ", msgid="BORG_TEST_PROMPT"), + Question(kind="accepted_true", message="Doing it."), + ] + factory, runners = make_runner_factory(events) + + async def run(): + app = BorgCockpitApp(borg_args=["check", "--repair"], command="check", runner_factory=factory) + async with app.run_test() as pilot: + await wait_until(pilot, lambda: isinstance(app.screen, PromptModal)) + assert app.session.pending_question is not None + + # the dialog must be composed and laid out before it can be clicked + def dialog_ready(): + buttons = app.screen.query("#prompt-yes") + return bool(buttons) and buttons.first().region.width > 0 + + await wait_until(pilot, dialog_ready) + await pilot.pause(0.1) + await pilot.click("#prompt-yes") + await wait_until(pilot, lambda: not app.session.running) + assert runners[0].answers == ["YES"] + assert app.session.pending_question is None + await pilot.pause(0.5) + assert app.query_one("#status").shown["status-rc"] == "RC: 0" + assert "Doing it." in log_text(app) + + asyncio.run(run()) + + def test_cockpit_app_create_archive(tmp_path): if not (is_freebsd or is_win32): pytest.skip("this slow test shall only run on FreeBSD and Windows") @@ -27,21 +262,23 @@ def test_cockpit_app_create_archive(tmp_path): subprocess.run(["borg", "-r", str(repo_path), "repo-create", "--encryption", "none-sha256"], check=True) async def run(): - app = BorgCockpitApp() - app.borg_args = ["-r", str(repo_path), "create", "--list", "test", str(input_path)] + app = BorgCockpitApp( + borg_args=["-r", str(repo_path), "create", "--list", "test", str(input_path)], command="create" + ) async with app.run_test() as pilot: assert "BorgBackup" in app.TITLE assert app.is_running # Wait for process to finish - while getattr(app, "process_running", True): + while app.session.running: await pilot.pause(0.1) + await pilot.pause(0.5) # let the refresh timer show the final state - status_panel = pilot.app.query_one("#status") - assert status_panel.rc == 0 - - assert app.total_lines_processed > 0 + assert app.session.rc == 0 + assert app.session.count("A") == 5000 + assert app.session.archive_name == "test" # from the --json output + assert app.query_one("#status").shown["status-rc"] == "RC: 0" await pilot.press("q") # quit app