diff --git a/src/ublue_update/cli.py b/src/ublue_update/cli.py index 4607f3d..29ea38c 100644 --- a/src/ublue_update/cli.py +++ b/src/ublue_update/cli.py @@ -11,7 +11,7 @@ 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_xdg_runtime_dir, get_active_sessions +from ublue_update.session import get_active_users from ublue_update.filelock import acquire_lock, release_lock @@ -33,7 +33,7 @@ def notify(title: str, body: str, actions: list = [], urgency: str = "normal"): if process_uid == 0: users = [] try: - users = get_active_sessions() + users = get_active_users() except KeyError as e: log.error("failed to get active logind session info", e) for user in users: @@ -43,11 +43,12 @@ def notify(title: str, body: str, actions: list = [], urgency: str = "normal"): log.error(f"failed to get xdg_runtime_dir for user: {user['Name']}", e) return user_args = [ - "/usr/bin/sudo", - "-u", - f"{user['Name']}", - "DISPLAY=:0", - f"DBUS_SESSION_BUS_ADDRESS=unix:path={xdg_runtime_dir}/bus", + "/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) @@ -108,7 +109,7 @@ def run_updates(system, system_update_available): ) users = [] try: - users = get_active_sessions() + users = get_active_users() except KeyError as e: log.error("failed to get active logind session info", e) @@ -133,23 +134,18 @@ def run_updates(system, system_update_available): """Users""" for user in users: - try: - xdg_runtime_dir = get_xdg_runtime_dir(user["User"]) - except KeyError as e: - log.error(f"failed to get xdg_runtime_dir for user: {user['Name']}", e) - break log.info( - f"""Running update for user: '{user['Name']}'""" - ) - + f"""Running update for user: '{user[1]}'""" + ) # magic number, corresponds to username (see session.py) out = subprocess.run( [ - "/usr/bin/sudo", - "-u", - f"{user['Name']}", - "DISPLAY=:0", - f"XDG_RUNTIME_DIR={xdg_runtime_dir}", - f"DBUS_SESSION_BUS_ADDRESS=unix:path={xdg_runtime_dir}/bus", + "/usr/bin/systemd-run", + "--setenv=TOPGRADE_SKIP_BRKC_NOTIFY=true", + "--user", + "--machine", + f"{user[1]}@", + "--pipe", + "--quiet", "/usr/bin/topgrade", "--config", "/usr/share/ublue-update/topgrade-user.toml", diff --git a/src/ublue_update/session.py b/src/ublue_update/session.py index f02001a..a54ce23 100644 --- a/src/ublue_update/session.py +++ b/src/ublue_update/session.py @@ -2,41 +2,22 @@ import json -def get_xdg_runtime_dir(uid): +def get_active_users(): out = subprocess.run( - ["/usr/bin/loginctl", "show-user", f"{uid}"], + [ + "/usr/bin/busctl", + "--system", + "-j", + "call", + "org.freedesktop.login1", + "/org/freedesktop/login1", + "org.freedesktop.login1.Manager", + "ListUsers", + ], capture_output=True, ) - loginctl_output = { - line.split("=")[0]: line.split("=")[-1] - for line in out.stdout.decode("utf-8").splitlines() - } - return loginctl_output["RuntimePath"] - - -def get_active_sessions(): - out = subprocess.run( - ["/usr/bin/loginctl", "list-sessions", "--output=json"], - capture_output=True, - ) - sessions = json.loads(out.stdout.decode("utf-8")) - session_properties = [] - active_sessions = [] - for session in sessions: - args = [ - "/usr/bin/loginctl", - "show-session", - f"{session['session']}", - ] - out = subprocess.run(args, capture_output=True) - if out.returncode == 0: - loginctl_output = { - line.split("=")[0]: line.split("=")[-1] - for line in out.stdout.decode("utf-8").splitlines() - } - session_properties.append(loginctl_output) - for session_info in session_properties: - graphical = session_info["Type"] == "x11" or session_info["Type"] == "wayland" - if graphical and session_info["Active"] == "yes": - active_sessions.append(session_info) - return active_sessions + # https://www.freedesktop.org/software/systemd/man/latest/org.freedesktop.login1.html + # ListUsers() returns an array of all currently logged in users. The structures in the array consist of the following fields: user id, user name, user object path. + 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] diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py new file mode 100644 index 0000000..a681c41 --- /dev/null +++ b/tests/unit/test_cli.py @@ -0,0 +1,305 @@ +import pytest +import sys +import os +from unittest.mock import patch, MagicMock + +# Add the src directory to the sys.path +sys.path.insert( + 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../src")) +) + +from ublue_update.cli import ( + notify, + ask_for_updates, + inhibitor_checks_failed, + run_updates, +) + + +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.log") +@patch("ublue_update.cli.subprocess.run") +def test_notify_no_dbus_notify(mock_run, mock_log, mock_os, mock_cfg): + mock_cfg.dbus_notify = False + assert notify("test_title", "test_body") is None + + +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.log") +@patch("ublue_update.cli.subprocess.run") +def test_notify_uid_user(mock_run, mock_log, mock_os, mock_cfg): + title = "test title" + body = "test body" + mock_cfg.dbus_notify = True + mock_os.getuid.return_value = 1001 + notify(title, body) + mock_run.assert_called_once_with( + [ + "/usr/bin/notify-send", + title, + body, + "--app-name=Universal Blue Updater", + "--icon=software-update-available-symbolic", + "--urgency=normal", + ], + capture_output=True, + ) + +@patch("ublue_update.cli.cfg") +def test_ask_for_updates_no_dbus_notify(mock_cfg): + mock_cfg.dbus_notify = False + assert ask_for_updates(True) is None + + +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.notify") +def test_ask_for_updates_notify_none(mock_notify, mock_cfg): + mock_cfg.dbus_notify = True + mock_notify.return_value = None + assert ask_for_updates(True) is None + mock_notify.assert_called_once_with( + "System Updater", + "Update available, but system checks failed. Update now?", + ["universal-blue-update-confirm=Confirm"], + "critical", + ) + + +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.notify") +@patch("ublue_update.cli.run_updates") +def test_ask_for_updates_system(mock_run_updates, mock_notify, mock_cfg): + mock_cfg.dbus_notify = True + mock_notify.return_value = MagicMock(stdout=b"universal-blue-update-confirm") + system = True + ask_for_updates(system) + mock_notify.assert_called_once_with( + "System Updater", + "Update available, but system checks failed. Update now?", + ["universal-blue-update-confirm=Confirm"], + "critical", + ) + mock_run_updates.assert_called_once_with(system, True) + + +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.notify") +@patch("ublue_update.cli.run_updates") +def test_ask_for_updates_user(mock_run_updates, mock_notify, mock_cfg): + mock_cfg.dbus_notify = True + mock_notify.return_value = MagicMock(stdout=b"universal-blue-update-confirm") + system = False + ask_for_updates(system) + mock_notify.assert_called_once_with( + "System Updater", + "Update available, but system checks failed. Update now?", + ["universal-blue-update-confirm=Confirm"], + "critical", + ) + mock_run_updates.assert_called_once_with(system, True) + + +def test_inhibitor_checks_failed(): + failure_message1 = "Failure 1" + failure_message2 = "Failure 2" + with pytest.raises(Exception, match=f"{failure_message1}\n - {failure_message2}"): + inhibitor_checks_failed([failure_message1, failure_message2], True, True, True) + + +@patch("ublue_update.cli.ask_for_updates") +@patch("ublue_update.cli.log") +def test_inhibitor_checks_failed_no_hw_check(mock_log, mock_ask_for_updates): + failure_message1 = "Failure 1" + failure_message2 = "Failure 2" + with pytest.raises(Exception, match=f"{failure_message1}\n - {failure_message2}"): + inhibitor_checks_failed([failure_message1, failure_message2], False, True, True) + mock_log.assert_called_once_with( + "Precondition checks failed, but update is available" + ) + mock_ask_for_updates.assert_called_once() + + +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.acquire_lock") +def test_run_updates_user_in_progress(mock_acquire_lock, mock_os): + mock_os.getuid.return_value = 1001 + mock_os.environ.get.return_value = "/path/to" + mock_os.path.isdir.return_value = True + mock_acquire_lock.return_value = None + with pytest.raises(Exception, match="updates are already running for this user"): + run_updates(False, True) + + +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.acquire_lock") +@patch("ublue_update.cli.transaction_wait") +def test_run_updates_user_system(mock_transaction_wait, mock_acquire_lock, mock_os): + mock_os.getuid.return_value = 1001 + mock_acquire_lock.return_value = 3 + mock_os.path.isdir.return_value = False + with pytest.raises( + Exception, + match="ublue-update needs to be run as root to perform system updates!", + ): + run_updates(True, True) + + +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.acquire_lock") +@patch("ublue_update.cli.transaction_wait") +@patch("ublue_update.cli.release_lock") +def test_run_updates_user_no_system( + mock_release_lock, mock_transaction_wait, mock_acquire_lock, mock_os +): + fd = 3 + mock_os.getuid.return_value = 1001 + mock_acquire_lock.return_value = fd + mock_os.path.isdir.return_value = False + run_updates(False, True) + mock_release_lock.assert_called_once_with(fd) + + +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.get_active_users") +@patch("ublue_update.cli.acquire_lock") +@patch("ublue_update.cli.transaction_wait") +@patch("ublue_update.cli.subprocess.run") +@patch("ublue_update.cli.log") +@patch("ublue_update.cli.pending_deployment_check") +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.release_lock") +@patch("ublue_update.cli.notify") +def test_run_updates_system( + mock_notify, + mock_release_lock, + mock_cfg, + mock_pending_deployment_check, + mock_log, + mock_run, + mock_transaction_wait, + mock_acquire_lock, + mock_get_active_sesions, + mock_os, +): + mock_os.getuid.return_value = 0 + mock_acquire_lock.return_value = 3 + output = MagicMock(stdout=b"test log") + output.returncode = 1 + mock_run.return_value = output + mock_pending_deployment_check.return_value = True + mock_cfg.dbus_notify.return_value = True + run_updates(True, True) + mock_notify.assert_any_call( + "System Updater", + "System passed checks, updating ...", + ) + mock_run.assert_any_call( + [ + "/usr/bin/topgrade", + "--config", + "/usr/share/ublue-update/topgrade-system.toml", + ], + capture_output=True, + ) + mock_notify.assert_any_call( + "System Updater", + "System update complete, pending changes will take effect after reboot. Reboot now?", + ["universal-blue-update-reboot=Reboot Now"], + ) + + +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.get_active_users") +@patch("ublue_update.cli.acquire_lock") +@patch("ublue_update.cli.transaction_wait") +@patch("ublue_update.cli.subprocess.run") +@patch("ublue_update.cli.log") +@patch("ublue_update.cli.pending_deployment_check") +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.release_lock") +@patch("ublue_update.cli.notify") +def test_run_updates_without_image_update( + mock_notify, + mock_release_lock, + mock_cfg, + mock_pending_deployment_check, + mock_log, + mock_run, + mock_transaction_wait, + mock_acquire_lock, + mock_get_active_sesions, + mock_os, +): + mock_os.getuid.return_value = 0 + mock_acquire_lock.return_value = 3 + output = MagicMock(stdout=b"test log") + output.returncode = 1 + mock_run.return_value = output + mock_pending_deployment_check.return_value = True + mock_cfg.dbus_notify.return_value = True + # System Update, but no Image Update Available + run_updates(True, False) + mock_notify.assert_not_called() + mock_run.assert_any_call( + [ + "/usr/bin/topgrade", + "--config", + "/usr/share/ublue-update/topgrade-system.toml", + ], + capture_output=True, + ) + mock_notify.assert_not_called() + + +@patch("ublue_update.cli.os") +@patch("ublue_update.cli.get_active_users") +@patch("ublue_update.cli.acquire_lock") +@patch("ublue_update.cli.transaction_wait") +@patch("ublue_update.cli.subprocess.run") +@patch("ublue_update.cli.log") +@patch("ublue_update.cli.pending_deployment_check") +@patch("ublue_update.cli.cfg") +@patch("ublue_update.cli.release_lock") +@patch("ublue_update.cli.notify") +def test_run_updates_system_reboot( + mock_notify, + mock_release_lock, + mock_cfg, + mock_pending_deployment_check, + mock_log, + mock_run, + mock_transaction_wait, + mock_acquire_lock, + mock_get_active_sesions, + mock_os, +): + mock_os.getuid.return_value = 0 + mock_acquire_lock.return_value = 3 + output = MagicMock(stdout=b"test log") + output.returncode = 1 + mock_run.return_value = output + mock_pending_deployment_check.return_value = True + mock_cfg.dbus_notify.return_value = True + reboot = MagicMock(stdout=b"universal-blue-update-reboot") + mock_notify.side_effect = [None, reboot] + run_updates(True, True) + mock_notify.assert_any_call( + "System Updater", + "System passed checks, updating ...", + ) + mock_run.assert_any_call( + [ + "/usr/bin/topgrade", + "--config", + "/usr/share/ublue-update/topgrade-system.toml", + ], + capture_output=True, + ) + mock_notify.assert_any_call( + "System Updater", + "System update complete, pending changes will take effect after reboot. Reboot now?", + ["universal-blue-update-reboot=Reboot Now"], + ) + mock_run.assert_any_call(["systemctl", "reboot"]) diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py new file mode 100644 index 0000000..e097334 --- /dev/null +++ b/tests/unit/test_session.py @@ -0,0 +1,39 @@ +import sys +import os +from unittest.mock import patch, MagicMock, mock_open + +# Add the src directory to the sys.path +sys.path.insert( + 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../src")) +) + +from ublue_update.session import get_active_users + +busctl_json_output = b"""{"type":"a(uso)","data":[[[1000,"user","/org/freedesktop/login1/user/_1000"]]]}""" + + +@patch("ublue_update.session.subprocess.run") +def test_get_active_users(mock_run): + mock_run.side_effect = [ + MagicMock(stdout=busctl_json_output), + ] + assert get_active_users() == [ + [ + 1000, + "user", + "/org/freedesktop/login1/user/_1000", + ] + ] + mock_run.assert_any_call( + [ + "/usr/bin/busctl", + "--system", + "-j", + "call", + "org.freedesktop.login1", + "/org/freedesktop/login1", + "org.freedesktop.login1.Manager", + "ListUsers", + ], + capture_output=True, + ) diff --git a/ublue-update.spec b/ublue-update.spec index 76b9b1b..09e92a3 100644 --- a/ublue-update.spec +++ b/ublue-update.spec @@ -29,7 +29,7 @@ BuildRequires: python-setuptools_scm BuildRequires: python-wheel Requires: skopeo Requires: libnotify -Requires: sudo +Requires: systemd %global sub_name %{lua:t=string.gsub(rpm.expand("%{NAME}"), "^ublue%-", ""); print(t)}