diff --git a/.github/workflows/unit.yml b/.github/workflows/unit.yml new file mode 100644 index 0000000..f78d2ea --- /dev/null +++ b/.github/workflows/unit.yml @@ -0,0 +1,27 @@ +name: Run Pytest +on: + push: + branches: + - main + pull_request: + branches: + - main + workflow_dispatch: +jobs: + unit-test: + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v3 + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.11' + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install pytest psutil + - name: Run Pytest + run: | + pytest -v + diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py new file mode 100644 index 0000000..a04ef4a --- /dev/null +++ b/tests/unit/test_cli.py @@ -0,0 +1,263 @@ +import pytest +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.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") == 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", + f"--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) == 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) == 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_sessions") +@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_sessions") +@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_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..fcd9d45 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,129 @@ +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.config import find_default_config_file, load_value, Config + +toml_example = b""" +[app] +name = "MyApplication" +version = "1.0.0" +description = "A simple example application" +""" + + +@patch("ublue_update.config.log") +@patch("ublue_update.config.os.path.isfile") +def test_find_default_config_file_success_first(mock_isfile, mock_log): + test_path = "/etc/ublue-update/ublue-update.toml" + mock_isfile.return_value = True + + assert find_default_config_file() == test_path + mock_isfile.assert_called_once_with(test_path) + + +@patch("ublue_update.config.log") +@patch("ublue_update.config.os.path.isfile") +def test_find_default_config_file_success_second(mock_isfile, mock_log): + test_path = "/usr/etc/ublue-update/ublue-update.toml" + mock_isfile.side_effect = [False, True] + + assert find_default_config_file() == test_path + mock_isfile.call_count == 2 + + +def test_load_value_success(): + dct = {"key": "val"} + + assert load_value(dct, "key") == "val" + + +def test_load_value_fail(): + dct = {"key": "val"} + + assert load_value(dct, "key2") == None + + +@patch("builtins.open", new_callable=mock_open, read_data=toml_example) +@patch("ublue_update.config.log.debug") +@patch("ublue_update.config.os.path.abspath") +@patch("ublue_update.config.Config.load_values") +def test_load_config(mock_load_values, mock_abspath, mock_debug, mock_open): + config_path = "/path/to/config.toml" + mock_abspath.return_value = config_path + + instance = Config() + instance.load_config(config_path) + + mock_load_values.assert_called_once_with( + { + "app": { + "name": "MyApplication", + "version": "1.0.0", + "description": "A simple example application", + } + } + ) + mock_open.assert_called_once_with(config_path, "rb") + mock_abspath.assert_called_once_with(config_path) + mock_debug.assert_called_once_with(f"Configuration loaded from {config_path}") + + +@patch("ublue_update.config.find_default_config_file") +@patch("builtins.open", new_callable=mock_open, read_data=toml_example) +@patch("ublue_update.config.log.debug") +@patch("ublue_update.config.os.path.abspath") +@patch("ublue_update.config.Config.load_values") +def test_load_config_default( + mock_load_values, mock_abspath, mock_debug, mock_open, mock_find_default_config_file +): + config_path = "/etc/ublue-update/ublue-update.toml" + mock_find_default_config_file.return_value = config_path + mock_abspath.return_value = config_path + + instance = Config() + instance.load_config() + + mock_load_values.assert_called_once_with( + { + "app": { + "name": "MyApplication", + "version": "1.0.0", + "description": "A simple example application", + } + } + ) + mock_open.assert_called_once_with(config_path, "rb") + mock_abspath.assert_called_once_with(config_path) + mock_debug.assert_called_once_with(f"Configuration loaded from {config_path}") + + +@patch("ublue_update.config.load_value") +def test_load_values(mock_load_value): + config = {"key": "val"} + mock_load_value.side_effect = [ + False, + None, + None, + None, + None, + [], + ] + + instance = Config() + instance.load_values(config) + + mock_load_value.call_count == 6 + assert instance.dbus_notify == False + assert instance.custom_check_scripts == [] + mock_load_value.assert_any_call(config, "notify", "dbus_notify") + mock_load_value.assert_any_call(config, "checks", "network_not_metered") + mock_load_value.assert_any_call(config, "checks", "min_battery_percent") + mock_load_value.assert_any_call(config, "checks", "max_cpu_load_percent") + mock_load_value.assert_any_call(config, "checks", "max_mem_percent") + mock_load_value.assert_any_call(config, "checks", "scripts") diff --git a/tests/unit/test_filelock.py b/tests/unit/test_filelock.py new file mode 100644 index 0000000..30e2afb --- /dev/null +++ b/tests/unit/test_filelock.py @@ -0,0 +1,65 @@ +import sys +import os +import fcntl +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.filelock import acquire_lock, release_lock + + +@patch("ublue_update.filelock.os.open") +@patch("ublue_update.filelock.fcntl.flock") +@patch("ublue_update.filelock.os.close") +@patch("ublue_update.filelock.time.sleep", return_value=None) +@patch("ublue_update.filelock.log") +def test_acquire_lock_success(mock_log, mock_sleep, mock_close, mock_flock, mock_open): + mock_fd = 3 + mock_open.return_value = mock_fd + mock_flock.return_value = None + + result_fd = acquire_lock("test_lock_file") + + assert result_fd == mock_fd + mock_open.assert_called_once_with( + "test_lock_file", os.O_RDWR | os.O_CREAT | os.O_TRUNC + ) + mock_flock.assert_called_once_with(mock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + mock_close.assert_not_called() + mock_log.info.assert_not_called() + + +@patch("ublue_update.filelock.os.open") +@patch("ublue_update.filelock.fcntl.flock") +@patch("ublue_update.filelock.os.close") +@patch("ublue_update.filelock.time.sleep", return_value=None) +@patch("ublue_update.filelock.log") +def test_acquire_lock_timeout(mock_log, mock_sleep, mock_close, mock_flock, mock_open): + mock_fd = 3 + mock_open.return_value = mock_fd + mock_flock.side_effect = IOError + + result_fd = acquire_lock("test_lock_file") + + assert result_fd is None + mock_open.assert_called_once_with( + "test_lock_file", os.O_RDWR | os.O_CREAT | os.O_TRUNC + ) + assert mock_flock.call_count >= 5 + mock_close.assert_called_once_with(mock_fd) + mock_log.info.assert_called() + + +@patch("ublue_update.filelock.fcntl.flock") +@patch("ublue_update.filelock.os.close") +def test_release_lock(mock_close, mock_flock): + mock_fd = 3 + + result = release_lock(mock_fd) + + mock_flock.assert_called_once_with(mock_fd, fcntl.LOCK_UN) + mock_close.assert_called_once_with(mock_fd) + assert result is None diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py new file mode 100644 index 0000000..156def5 --- /dev/null +++ b/tests/unit/test_session.py @@ -0,0 +1,152 @@ +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_sessions, get_xdg_runtime_dir + +loginctl_output = b""" +UID=1001 +GID=1001 +Name=test +Timestamp=Thu 2024-08-15 18:18:08 UTC +TimestampMonotonic=293807858 +RuntimePath=/run/user/1001 +Service=user@1001.service +Slice=user-1001.slice +Display=c3 +State=active +Sessions=c4 c3 c1 3 +IdleHint=no +IdleSinceHint=0 +IdleSinceHintMonotonic=0 +Linger=yes +""" + +loginctl_json_output = b""" +[ + { + "session" : "3", + "uid" : 1001, + "user" : "test", + "seat" : null, + "leader" : 6205, + "class" : "manager", + "tty" : null, + "idle" : false, + "since" : null + }, + { + "session" : "c1", + "uid" : 1001, + "user" : "test", + "seat" : null, + "leader" : 6230, + "class" : "manager", + "tty" : null, + "idle" : false, + "since" : null + } +] +""" + +session_info = [ + b""" +Id=3 +User=1001 +Name=test +Timestamp=Thu 2024-08-15 18:18:08 UTC +TimestampMonotonic=293993628 +VTNr=0 +Remote=no +Service=systemd-user +Leader=6205 +Audit=3 +Type=wayland +Class=manager +Active=yes +State=active +IdleHint=no +IdleSinceHint=0 +IdleSinceHintMonotonic=0 +LockedHint=no +""", + b""" +Id=c1 +User=1001 +Name=test +Timestamp=Thu 2024-08-15 18:18:09 UTC +TimestampMonotonic=295100128 +VTNr=0 +Remote=no +Service=systemd-user +Leader=6230 +Audit=3 +Type=unspecified +Class=manager +Active=yes +State=active +IdleHint=no +IdleSinceHint=0 +IdleSinceHintMonotonic=0 +LockedHint=no +""", +] + + +@patch("ublue_update.session.subprocess.run") +def test_get_xdg_runtime_dir(mock_run): + mock_run.return_value = MagicMock(stdout=loginctl_output) + assert get_xdg_runtime_dir(1001) == "/run/user/1001" + mock_run.assert_called_once_with( + ["/usr/bin/loginctl", "show-user", "1001"], capture_output=True + ) + + +@patch("ublue_update.session.subprocess.run") +def test_get_active_sessions(mock_run): + mock_session1 = MagicMock(stdout=session_info[0]) + mock_session2 = MagicMock(stdout=session_info[1]) + mock_session1.returncode = 0 + mock_session2.returncode = 0 + mock_run.side_effect = [ + MagicMock(stdout=loginctl_json_output), + mock_session1, + mock_session2, + ] + assert get_active_sessions() == [ + { + "": "", + "Id": "3", + "User": "1001", + "Name": "test", + "Timestamp": "Thu 2024-08-15 18:18:08 UTC", + "TimestampMonotonic": "293993628", + "VTNr": "0", + "Remote": "no", + "Service": "systemd-user", + "Leader": "6205", + "Audit": "3", + "Type": "wayland", + "Class": "manager", + "Active": "yes", + "State": "active", + "IdleHint": "no", + "IdleSinceHint": "0", + "IdleSinceHintMonotonic": "0", + "LockedHint": "no", + } + ] + mock_run.assert_any_call( + ["/usr/bin/loginctl", "list-sessions", "--output=json"], capture_output=True + ) + mock_run.assert_any_call( + ["/usr/bin/loginctl", "show-session", "3"], capture_output=True + ) + mock_run.assert_any_call( + ["/usr/bin/loginctl", "show-session", "c1"], capture_output=True + ) diff --git a/tests/unit/update_checks/test_system.py b/tests/unit/update_checks/test_system.py new file mode 100644 index 0000000..e8314a6 --- /dev/null +++ b/tests/unit/update_checks/test_system.py @@ -0,0 +1,103 @@ +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.update_checks.system import ( + skopeo_inspect, + system_update_check, + pending_deployment_check, +) + + +@patch("ublue_update.update_checks.system.run") +def test_skopeo_inspect(mock_run): + test_input = "latest" + test_output = '{"Digest": "mocked_digest"}' + mock_skopeo_out = MagicMock() + mock_skopeo_out.stdout = test_output + + mock_run.return_value = mock_skopeo_out + assert skopeo_inspect(test_input) == "mocked_digest" + mock_run.assert_called_once_with( + ["skopeo", "inspect", test_input], capture_output=True + ) + + +@patch("ublue_update.update_checks.system.run") +@patch("ublue_update.update_checks.system.skopeo_inspect") +@patch("ublue_update.update_checks.system.log") +def test_system_update_check(mock_log, mock_skopeo_inspect, mock_run): + # Test when the system update is available + mock_run.return_value.stdout = '{"deployments": [{"base-commit-meta": {"ostree.manifest-digest": "digest1"}, "container-image-reference": "image:tag"}]}' + mock_skopeo_inspect.return_value = "digest2" + mock_log.info = MagicMock() + mock_log.error = MagicMock() + + assert system_update_check() + mock_log.info.assert_called_with("System update available.") + mock_run.assert_called_once_with( + ["rpm-ostree", "status", "--json"], capture_output=True + ) + mock_skopeo_inspect.assert_called_once_with("docker://tag") + + +@patch("ublue_update.update_checks.system.run") +@patch("ublue_update.update_checks.system.skopeo_inspect") +@patch("ublue_update.update_checks.system.log") +def test_system_update_check_no_update(mock_log, mock_skopeo_inspect, mock_run): + # Test when there is no update available + mock_run.return_value.stdout = '{"deployments": [{"base-commit-meta": {"ostree.manifest-digest": "digest1"}, "container-image-reference": "image:tag"}]}' + mock_skopeo_inspect.return_value = "digest1" + mock_log.info = MagicMock() + mock_log.error = MagicMock() + + assert not system_update_check() + mock_log.info.assert_called_with("No system update available.") + mock_run.assert_called_once_with( + ["rpm-ostree", "status", "--json"], capture_output=True + ) + mock_skopeo_inspect.assert_called_once_with("docker://tag") + + +@patch("ublue_update.update_checks.system.run") +@patch("ublue_update.update_checks.system.log") +def test_system_update_check_json_error(mock_log, mock_run): + # Test handling of JSON decoding errors + mock_run.return_value.stdout = "invalid json" + mock_log.info = MagicMock() + mock_log.error = MagicMock() + + assert not system_update_check() + mock_log.error.assert_called_with( + "update check failed, system isn't managed by rpm-ostree container native" + ) + mock_run.assert_called_once_with( + ["rpm-ostree", "status", "--json"], capture_output=True + ) + + +@patch("ublue_update.update_checks.system.run") +def test_pending_deployment_check(mock_run): + # Test when there is no pending deployment + mock_run.return_value.returncode = 77 + + assert pending_deployment_check() + mock_run.assert_called_once_with( + ["rpm-ostree", "status", "--pending-exit-77"], capture_output=True + ) + + +@patch("ublue_update.update_checks.system.run") +def test_pending_deployment_check_with_pending(mock_run): + # Test when there is a pending deployment + mock_run.return_value.returncode = 1 + + assert not pending_deployment_check() + mock_run.assert_called_once_with( + ["rpm-ostree", "status", "--pending-exit-77"], capture_output=True + ) diff --git a/tests/unit/update_checks/test_wait.py b/tests/unit/update_checks/test_wait.py new file mode 100644 index 0000000..ee40870 --- /dev/null +++ b/tests/unit/update_checks/test_wait.py @@ -0,0 +1,85 @@ +import sys +import os +from unittest.mock import patch, MagicMock +from ublue_update.update_checks.wait import transaction, transaction_wait + +# Add the src directory to the sys.path +sys.path.insert( + 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) +) + + +@patch("ublue_update.update_checks.wait.run") +@patch("ublue_update.update_checks.wait.log") +def test_transaction_success(mock_log, mock_run): + # Mock subprocess.run to return valid JSON output + mock_run.return_value = MagicMock(stdout='{"transaction": "some_transaction_data"}') + mock_log.error = MagicMock() + + result = transaction() + + assert result == "some_transaction_data" + mock_run.assert_called_once_with( + ["/usr/bin/rpm-ostree", "status", "--json"], capture_output=True + ) + mock_log.error.assert_not_called() + + +@patch("ublue_update.update_checks.wait.run") +@patch("ublue_update.update_checks.wait.log") +def test_transaction_json_decode_error(mock_log, mock_run): + # Mock subprocess.run to return invalid JSON output + mock_run.return_value = MagicMock(stdout="invalid json") + mock_log.error = MagicMock() + + result = transaction() + + assert result is None + mock_run.assert_called_once_with( + ["/usr/bin/rpm-ostree", "status", "--json"], capture_output=True + ) + mock_log.error.assert_called_once_with( + "can't get transaction, system not managed with rpm-ostree container native" + ) + + +@patch("ublue_update.update_checks.wait.run") +@patch("ublue_update.update_checks.wait.log") +def test_transaction_key_error(mock_log, mock_run): + # Mock subprocess.run to return JSON with missing 'transaction' key + mock_run.return_value = MagicMock(stdout='{"some_key": "some_value"}') + mock_log.error = MagicMock() + + result = transaction() + + assert result is None + mock_run.assert_called_once_with( + ["/usr/bin/rpm-ostree", "status", "--json"], capture_output=True + ) + mock_log.error.assert_called_once_with( + "can't get transaction, system not managed with rpm-ostree container native" + ) + + +@patch("ublue_update.update_checks.wait.transaction") +@patch("ublue_update.update_checks.wait.sleep", return_value=None) +def test_transaction_wait(mock_sleep, mock_transaction): + # Mock transaction() to return None eventually + mock_transaction.side_effect = [1, 2, None] + + transaction_wait() + + assert mock_transaction.call_count == 3 + mock_sleep.assert_called_with(1) + + +@patch("ublue_update.update_checks.wait.transaction") +@patch("ublue_update.update_checks.wait.sleep", return_value=None) +def test_transaction_wait_no_sleep(mock_sleep, mock_transaction): + # Mock transaction() to return None immediately + mock_transaction.return_value = None + + transaction_wait() + + mock_transaction.assert_called_once() + mock_sleep.assert_not_called() diff --git a/tests/unit/update_inhibitors/test_custom.py b/tests/unit/update_inhibitors/test_custom.py new file mode 100644 index 0000000..c1ad08f --- /dev/null +++ b/tests/unit/update_inhibitors/test_custom.py @@ -0,0 +1,111 @@ +import pytest +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.update_inhibitors.custom import ( + run_custom_check_script, + run_custom_check_scripts, + check_custom_inhibitors, +) + + +@patch("ublue_update.update_inhibitors.custom.log") +@patch("ublue_update.update_inhibitors.custom.subprocess.run") +def test_run_custom_check_script_pass(mock_run, mock_log): + script = { + "run": "echo Hello World", + "shell": "/bin/bash", + } + mock_result = MagicMock(stdout=b"Script ran successfully.") + mock_run.return_value = mock_result + mock_result.returncode = 0 + result = run_custom_check_script(script) + assert result["passed"] + mock_run.assert_called_once_with( + ["/bin/bash", "-c", "echo Hello World"], + capture_output=True, + text=True, + check=False, + ) + + +@patch("ublue_update.update_inhibitors.custom.log") +@patch("ublue_update.update_inhibitors.custom.subprocess.run") +def test_run_custom_check_script_fail(mock_run, mock_log): + script = { + "run": "echo Hello World", + "shell": "/bin/bash", + } + mock_result = MagicMock(stdout=b"Script ran unsuccessfully.") + mock_run.return_value = mock_result + mock_result.returncode = 1 + result = run_custom_check_script(script) + assert not result["passed"] + mock_run.assert_called_once_with( + ["/bin/bash", "-c", "echo Hello World"], + capture_output=True, + text=True, + check=False, + ) + + +def test_run_custom_check_script_run_no_shell_exc(): + with pytest.raises( + Exception, + match="checks.scripts.*: 'shell' must be specified when 'run' is used", + ): + run_custom_check_script({"run": "some_command"}) + + +def test_run_custom_check_script_run_and_file_exc(): + with pytest.raises( + Exception, + match="checks.scripts.*: Only one of 'run' and 'file' must be set for a given script", + ): + run_custom_check_script( + {"run": "some_command", "file": "some_file", "shell": "some_shell"} + ) + + +@patch("ublue_update.update_inhibitors.custom.cfg") +@patch("ublue_update.update_inhibitors.custom.run_custom_check_script") +def test_run_custom_check_scripts(mock_run_custom_check_script, mock_cfg): + script = "test_script.sh" + result = {"passed": True, "message": "message"} + mock_cfg.custom_check_scripts = [script] + mock_run_custom_check_script.return_value = result + + assert run_custom_check_scripts() == [result] + mock_run_custom_check_script.assert_called_once_with(script) + + +@patch("ublue_update.update_inhibitors.custom.run_custom_check_scripts") +@patch("ublue_update.update_inhibitors.custom.log") +def test_check_custom_inhibitors_passed(mock_log, mock_run_custom_check_scripts): + mock_run_custom_check_scripts.return_value = [ + {"passed": True, "message": "message1"}, + {"passed": True, "message": "message2"}, + ] + result = check_custom_inhibitors() + assert not result[0] + assert result[1] == [] + mock_log.info.assert_called_once_with("System passed custom checks") + + +@patch("ublue_update.update_inhibitors.custom.run_custom_check_scripts") +@patch("ublue_update.update_inhibitors.custom.log") +def test_check_custom_inhibitors_failed(mock_log, mock_run_custom_check_scripts): + mock_run_custom_check_scripts.return_value = [ + {"passed": False, "message": "message1"}, + {"passed": True, "message": "message2"}, + ] + result = check_custom_inhibitors() + assert result[0] + assert result[1] == ["message1"] + mock_log.info.assert_not_called() diff --git a/tests/unit/update_inhibitors/test_hardware.py b/tests/unit/update_inhibitors/test_hardware.py new file mode 100644 index 0000000..f440a23 --- /dev/null +++ b/tests/unit/update_inhibitors/test_hardware.py @@ -0,0 +1,207 @@ +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.update_inhibitors.hardware import ( + check_network_status, + check_network_not_metered, + check_battery_status, + check_cpu_load, + check_mem_percentage, + check_hardware_inhibitors, +) + + +class BatteryStatus: + def __init__(self, percent, secsleft=None, power_plugged=False): + self.percent = percent + self.secsleft = secsleft + self.power_plugged = power_plugged + + +@patch("ublue_update.update_inhibitors.hardware.log") +@patch("ublue_update.update_inhibitors.hardware.psutil") +def test_check_network_status_up(mock_psutil, mock_log): + mock_psutil.net_if_stats.return_value = {"eth0": "snicstats"} + assert check_network_status()["passed"] + mock_psutil.net_if_stats.assert_called_once_with() + + +@patch("ublue_update.update_inhibitors.hardware.log") +@patch("ublue_update.update_inhibitors.hardware.psutil") +def test_check_network_status_down(mock_psutil, mock_log): + mock_psutil.net_if_stats.return_value = {"lo": "snicstats"} + assert not check_network_status()["passed"] + mock_psutil.net_if_stats.assert_called_once_with() + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +@patch("ublue_update.update_inhibitors.hardware.subprocess.run") +def test_check_network_not_metered_metered(mock_run, mock_cfg): + mock_cfg.network_not_metered = False + assert check_network_not_metered()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +@patch("ublue_update.update_inhibitors.hardware.subprocess.run") +def test_check_network_not_metered_unmetered(mock_run, mock_cfg): + mock_run.return_value = MagicMock(stdout="u 1") + mock_cfg.network_not_metered = True + assert not check_network_not_metered()["passed"] + mock_run.assert_called_once_with( + [ + "busctl", + "get-property", + "org.freedesktop.NetworkManager", + "/org/freedesktop/NetworkManager", + "org.freedesktop.NetworkManager", + "Metered", + ], + capture_output=True, + check=True, + text=True, + ) + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +@patch("ublue_update.update_inhibitors.hardware.psutil") +def test_check_battery_status_min_percent_above(mock_psutil, mock_cfg): + mock_cfg.min_battery_percent = 75 + mock_psutil.sensors_battery.return_value = BatteryStatus(20) + assert not check_battery_status()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +@patch("ublue_update.update_inhibitors.hardware.psutil") +def test_check_battery_status_min_percent_below(mock_psutil, mock_cfg): + mock_cfg.min_battery_percent = 75 + mock_psutil.sensors_battery.return_value = BatteryStatus(99) + assert check_battery_status()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +@patch("ublue_update.update_inhibitors.hardware.psutil") +def test_check_battery_status_min_percent_no_battery(mock_psutil, mock_cfg): + mock_cfg.min_battery_percent = 75 + mock_psutil.sensors_battery.return_value = None + assert check_battery_status()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +@patch("ublue_update.update_inhibitors.hardware.psutil") +def test_check_battery_status_min_percent_above_plugged(mock_psutil, mock_cfg): + mock_cfg.min_battery_percent = 75 + mock_psutil.sensors_battery.return_value = BatteryStatus(20, power_plugged=True) + assert check_battery_status()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +@patch("ublue_update.update_inhibitors.hardware.psutil") +def test_check_battery_status_no_min_percent(mock_psutil, mock_cfg): + mock_cfg.min_battery_percent = False + assert check_battery_status()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.psutil") +@patch("ublue_update.update_inhibitors.hardware.cfg") +def test_check_cpu_load_max_percent_above(mock_cfg, mock_psutil): + mock_cfg.max_cpu_load_percent = 50 + mock_psutil.getloadavg.return_value = [0.50, 0.49] + mock_psutil.cpu_count.return_value = 1 + assert check_cpu_load() + + +@patch("ublue_update.update_inhibitors.hardware.psutil") +@patch("ublue_update.update_inhibitors.hardware.cfg") +def test_check_cpu_load_max_percent_below(mock_cfg, mock_psutil): + mock_cfg.max_cpu_load_percent = 50 + mock_psutil.getloadavg.return_value = [0.50, 0.51] + mock_psutil.cpu_count.return_value = 1 + assert not check_cpu_load()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.psutil") +@patch("ublue_update.update_inhibitors.hardware.cfg") +def test_check_cpu_load_no_max_percent(mock_cfg, mock_psutil): + mock_cfg.max_cpu_load_percent = False + assert check_cpu_load()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.psutil") +@patch("ublue_update.update_inhibitors.hardware.cfg") +def test_check_mem_percentage_max_percent_above(mock_cfg, mock_psutil): + mock_cfg.max_mem_percent = 50 + memory = MagicMock() + memory.percent = 49 + mock_psutil.virtual_memory.return_value = memory + assert check_mem_percentage()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.psutil") +@patch("ublue_update.update_inhibitors.hardware.cfg") +def test_check_mem_percentage_max_percent_below(mock_cfg, mock_psutil): + mock_cfg.max_mem_percent = 50 + memory = MagicMock() + memory.percent = 51 + mock_psutil.virtual_memory.return_value = memory + assert not check_mem_percentage()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.cfg") +def test_check_mem_percentage_no_max_percent(mock_cfg): + mock_cfg.max_mem_percent = None + assert check_mem_percentage()["passed"] + + +@patch("ublue_update.update_inhibitors.hardware.check_network_status") +@patch("ublue_update.update_inhibitors.hardware.check_network_not_metered") +@patch("ublue_update.update_inhibitors.hardware.check_battery_status") +@patch("ublue_update.update_inhibitors.hardware.check_cpu_load") +@patch("ublue_update.update_inhibitors.hardware.check_mem_percentage") +def test_check_hardware_inhibitors_pass( + mock_check_network_status, + mock_check_network_not_metered, + mock_check_battery_status, + mock_check_cpu_load, + mock_check_mem_percentage, +): + mock_check_network_status.return_value = {"passed": True} + mock_check_network_not_metered.return_value = {"passed": True} + mock_check_battery_status.return_value = {"passed": True} + mock_check_cpu_load.return_value = {"passed": True} + mock_check_mem_percentage.return_value = {"passed": True} + + assert not check_hardware_inhibitors()[0] + + +@patch("ublue_update.update_inhibitors.hardware.check_network_status") +@patch("ublue_update.update_inhibitors.hardware.check_network_not_metered") +@patch("ublue_update.update_inhibitors.hardware.check_battery_status") +@patch("ublue_update.update_inhibitors.hardware.check_cpu_load") +@patch("ublue_update.update_inhibitors.hardware.check_mem_percentage") +def test_check_hardware_inhibitors_fail( + mock_check_network_status, + mock_check_network_not_metered, + mock_check_battery_status, + mock_check_cpu_load, + mock_check_mem_percentage, +): + failure_message = "Test failure." + mock_check_network_status.return_value = { + "passed": False, + "message": failure_message, + } + mock_check_network_not_metered.return_value = {"passed": True} + mock_check_battery_status.return_value = {"passed": True} + mock_check_cpu_load.return_value = {"passed": True} + mock_check_mem_percentage.return_value = {"passed": True} + + result = check_hardware_inhibitors() + assert result[0] + assert result[1][0] == failure_message + assert len(result[1]) == 1