fix(cli): improve command execution handling

Allow callers to define successful exit codes instead of relying on
subprocess exceptions, making command handling more flexible and
explicit.

Use shell-safe quoting when logging commands and keep failure
diagnostics consistent with the CLI output helpers.

1 files changed, 17 insertions, 15 deletions

diff --git a/prometheus_borgmatic_exporter/__init__.py b/prometheus_borgmatic_exporter/__init__.py
index b53caac..f1ac3ad 100644
--- a/prometheus_borgmatic_exporter/__init__.py
+++ b/prometheus_borgmatic_exporter/__init__.py
@@ -6,6 +6,7 @@
 
 import json
 import os
+import shlex
 import subprocess
 from datetime import datetime
 from pathlib import Path
@@ -80,27 +81,29 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.""")
 
 @click.pass_context
 def run_command(
-    ctx: click.Context, command: list[str]
+    ctx: click.Context, command: list[str], accepted_exit_codes: tuple[int, ...] = (0,)
 ) -> subprocess.CompletedProcess[str]:
-    """Run a shell command and return the result.
+    """Run a command and return the result.
 
     Logs the command at debug level before execution. Exits the application
     with the command's return code if it fails.
 
     :param ctx: Click context injected by :func:`click.pass_context`.
     :param command: Command and its arguments.
+    :param accepted_exit_codes: Exit codes considered successful.
     :return: Completed process with stdout and stderr captured.
     """
-    debug("Running command", " ".join(command), err=True)
-    try:
-        return subprocess.run(
-            command, capture_output=True, encoding="utf-8", check=True
+    command_string = shlex.join(command)
+    debug("Running command", command_string, err=True)
+    result = subprocess.run(command, capture_output=True, encoding="utf-8", check=False)
+    if result.returncode not in accepted_exit_codes:
+        error(
+            f"Command ({command_string}) failed with exit code:", str(result.returncode)
         )
-    except subprocess.CalledProcessError as e:
-        error(f"Command ({' '.join(command)}) failed with exit code {e.returncode}")
-        if e.stderr:
-            error("stderr:", e.stderr.strip())
-        ctx.exit(e.returncode)
+        if result.stderr:
+            error("stderr:", result.stderr.strip())
+        ctx.exit(result.returncode)
+    return result
 
 
 @click.pass_context
@@ -179,8 +182,7 @@ def fetch_repositories(
         debug(f"Fetched {len(repositories)} repositories", err=True)
         return repositories, extract_archive_counts(repositories)
 
-    info_command += ["--archive", "latest"]
-    borgmatic_info = run_command(info_command)
+    borgmatic_info = run_command(info_command + ["--archive", "latest"])
     repositories = cast(
         list[types.BorgInfoReturnType], load_json(borgmatic_info.stdout)
     )
@@ -356,7 +358,7 @@ def collect_archive_metrics(
     "quiet_flag",
     is_flag=True,
     default=False,
-    help="Suppress all non-error output.",
+    help="Suppress all non-error output, including verbose output.",
 )
 @click.option(
     "-v",
@@ -386,7 +388,7 @@ def prometheus_borgmatic_exporter(
     all_archives_flag: bool,
     quiet_flag: bool,
     verbose_flag: bool,
-):
+) -> None:
     """Collect borgmatic repository metrics and export them for Prometheus.
 
     Queries borgmatic for repository and archive statistics and writes them to