update

src/ublue_update/cli.py

@@ -11,7 +11,8 @@
 from ublue_update.update_inhibitors.hardware import check_hardware_inhibitors
 from ublue_update.update_inhibitors.custom import check_custom_inhibitors
 from ublue_update.config import cfg
-from ublue_update.session import get_active_users
+from ublue_update.session import get_active_users, run_uid
+from ublue_update.update_drivers.brew import brew_update
 from ublue_update.filelock import acquire_lock, release_lock
 
 
@@ -37,16 +38,7 @@ def notify(title: str, body: str, actions: list = [], urgency: str = "normal"):
         except KeyError as e:
             log.error("failed to get active logind session info", e)
         for user in users:
-            user_args = [
-                "/usr/bin/systemd-run",
-                "--user",
-                "--machine",
-                f"{user[1]}@",  # magic number, corresponds to user name in ListUsers (see session.py)
-                "--pipe",
-                "--quiet",
-            ]
-            user_args += args
-            out = subprocess.run(user_args, capture_output=True)
+            out = run_uid(user[0], args)
             if actions != []:
                 return out
         return
@@ -142,31 +134,22 @@ def run_updates(system, system_update_available, dry_run):
         log.debug(out.stdout.decode("utf-8"))
 
         if out.returncode != 0:
-            print(f"topgrade returned code {out.returncode}, program output:")
-            print(out.stdout.decode("utf-8"))
+            log.error(f"topgrade returned code {out.returncode}, program output:")
+            log.error(out.stdout.decode("utf-8"))
             os._exit(out.returncode)
 
         """Users"""
         for user in users:
+
             log.info(
                 f"""Running update for user: '{user[1]}'"""
             )  # magic number, corresponds to username (see session.py)
 
-            args = [
-                "/usr/bin/systemd-run",
-                "--setenv=TOPGRADE_SKIP_BRKC_NOTIFY=true",
-                "--user",
-                "--machine",
-                f"{user[1]}@",
-                "--pipe",
-                "--quiet",
-            ] + topgrade_user
-
-            out = subprocess.run(
-                args,
-                capture_output=True,
-            )
+            out = run_uid(
+                user[0], ["--setenv=TOPGRADE_SKIP_BRKC_NOTIFY=true"] + topgrade_user
+            )  # uid for user (session.py)
             log.debug(out.stdout.decode("utf-8"))
+        brew_update(dry_run)
         log.info("System update complete")
         if pending_deployment_check() and system_update_available and cfg.dbus_notify:
             out = notify(
@@ -236,7 +219,10 @@ def main():
     cfg.load_config(cli_args.config)
 
     if cli_args.dry_run:
-        # run system updates
+        # "dry run" the hardware tests as well
+        _, _ = check_hardware_inhibitors()
+        _, _ = check_custom_inhibitors()
+        # run the update function with "dry run" set to true
         run_updates(False, True, True)
         os._exit(0)
 

src/ublue_update/session.py

@@ -1,5 +1,8 @@
 import subprocess
 import json
+import logging
+
+log = logging.getLogger(__name__)
 
 
 def get_active_users():
@@ -21,3 +24,16 @@ def get_active_users():
     users = json.loads(out.stdout.decode("utf-8"))
     # sample output: {'type': 'a(uso)', 'data': [[[1000, 'user', '/org/freedesktop/login1/user/_1000']]]
     return users["data"][0]
+
+
+def run_uid(uid: int, args):
+    run_args = [
+        "/usr/bin/systemd-run",
+        "--user",
+        "--machine",
+        f"{uid}@",
+        "--pipe",
+        "--quiet",
+    ]
+
+    return subprocess.run(run_args + args, capture_output=True)

src/ublue_update/update_drivers/brew.py

@@ -0,0 +1,40 @@
+import os
+from ublue_update.session import run_uid
+import logging
+
+log = logging.getLogger(__name__)
+
+brew_prefix = "/home/linuxbrew/.linuxbrew"
+brew_cellar = f"{brew_prefix}/Cellar"
+brew_repo = f"{brew_prefix}/Homebrew"
+
+
+def detect_user():
+    if not os.exists(brew_prefix):
+        return -1
+    return os.stat(brew_prefix).st_uid
+
+
+def brew_update(dry_run):
+    uid = detect_user()
+    if uid == -1 or dry_run:
+        return
+    log.info(f"running brew updates for uid: {uid}")
+    path = f"{os.environ["PATH"]}:{brew_prefix}/bin:{brew_prefix}/sbin"
+    args = [
+        f"--E=HOMEBREW_PREFIX='{brew_prefix}'",
+        f"--E=HOMEBREW_CELLAR='{brew_cellar}'",
+        f"--E=HOMEBREW_REPOSITORY='{brew_repo}'",
+        f"--E=PATH='{path}'",
+    ]
+    out = run_uid(args + ["brew", "update"])
+    if out.returncode != 0:
+        log.error(f"brew update failed, returned code {out.returncode}, program output:")
+        log.error(out.stdout.decode("utf-8"))
+        return
+    out = run_uid(args + ["brew", "upgrade"])
+    if out.returncode != 0:
+        log.error(f"brew upgrade failed, returned code {out.returncode}, program output:")
+        log.error(out.stdout.decode("utf-8"))
+        return
+    log.info("brew updates completed")