diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 52b76da..33081f2 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -20,6 +20,7 @@ lint: - ci/codecov-wrapper -F unittests before_script: # install dependencies + - sudo qubes-dom0-update -y - sudo qubes-dom0-update -y ansible python3-pytest python3-coverage perl-Digest-SHA # install from artifacts - find $CI_PROJECT_DIR/artifacts/repository/host-fc* -name '*.noarch.rpm' -exec sudo dnf install -y {} \+ diff --git a/ansible_collections/qubesos/core/plugins/modules/qubes_dom0_update.py b/ansible_collections/qubesos/core/plugins/modules/qubes_dom0_update.py new file mode 100644 index 0000000..e987f71 --- /dev/null +++ b/ansible_collections/qubesos/core/plugins/modules/qubes_dom0_update.py @@ -0,0 +1,461 @@ +#!/usr/bin/python3 +# Copyright (C) 2026 Guillaume Chinal (guiiix) +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +DOCUMENTATION = r""" +--- +module: qubes_dom0_update + +short_description: Manage dom0 packages in Qubes OS + +description: + - Installs, removes, or upgrades packages in Qubes OS dom0 using C(qubes-dom0-update). + - Can also manage audio daemon. + +version_added: "1.0.0" + +author: + - Guillaume Chinal + +options: + name: + description: + - List of package names to manage + - Use C(*) alone with O(state=latest) to upgrade all installed packages. + - Globs are not supported for update only. + type: list + elements: str + aliases: + - pkg + default: [] + + state: + description: + - Desired state of the packages. + - Use V(present) or V(installed) to install missing packages. + - Use V(latest) to install and upgrade packages to their latest available version. + - Use V(absent) or V(removed) to remove packages. + type: str + default: present + choices: + - absent + - installed + - present + - removed + - latest + + force_xen_upgrade: + description: + - Pass C(--force-xen-upgrade) to C(qubes-dom0-update). + - Force major Xen upgrade even if some qubes are running. + type: bool + default: false + + skip_boot_check: + description: + - Pass C(--skip-boot-check) to C(qubes-dom0-update). + - Does not check if /boot & /boot/efi should be mounted. + type: bool + default: false + + switch_audio_server: + description: + - Switch the dom0 audio daemon to the specified backend. + - Mutually exclusive with O(name). + type: str + choices: + - pipewire + - pulseaudio + +notes: + - This module must be run as root. +""" + +EXAMPLES = r""" +- name: Install a package + qubesos.core.qubes_dom0_update: + name: nano + state: present + +- name: Install multiple packages + qubesos.core.qubes_dom0_update: + name: + - nano + - htop + state: present + +- name: Remove a package + qubesos.core.qubes_dom0_update: + name: nano + state: absent + +- name: Upgrade specific packages + qubesos.core.qubes_dom0_update: + name: + - qubes-core-dom0 + - qubes-gui-daemon + state: latest + +- name: Upgrade all dom0 packages + qubesos.core.qubes_dom0_update: + name: "*" + state: latest + +- name: Upgrade all dom0 packages, forcing Xen upgrade + qubesos.core.qubes_dom0_update: + name: "*" + state: latest + force_xen_upgrade: true + +- name: Switch audio server to PipeWire + qubesos.core.qubes_dom0_update: + switch_audio_server: pipewire +""" + +import libdnf5 +import libdnf5.transaction +import os + +from ansible.module_utils.basic import AnsibleModule +from ansible.module_utils.common.locale import get_best_parsable_locale + + +ARGUMENT_SPEC = dict( + force_xen_upgrade=dict(type="bool", default=False), + name=dict(type="list", elements="str", aliases=["pkg"], default=[]), + skip_boot_check=dict(type="bool", default=False), + state=dict( + type="str", + default="present", + choices=["absent", "installed", "present", "removed", "latest"], + ), + switch_audio_server=dict( + type="str", + choices=[ + "pipewire", + "pulseaudio", + ], + ), +) + + +class QubesDom0UpdateModule: + def __init__(self, module): + self.module = module + + self.force_xen_upgrade = module.params["force_xen_upgrade"] + self.names = [p.strip() for p in self.module.params["name"]] + self.skip_boot_check = module.params["skip_boot_check"] + self.state = self.module.params["state"] + self.switch_audio_server = self.module.params["switch_audio_server"] + + locale = get_best_parsable_locale(self.module) + os.environ["LC_ALL"] = os.environ["LC_MESSAGES"] = locale + os.environ["LANGUAGE"] = os.environ["LANG"] = locale + + self.dnf_base = None + self.dnf_conf = None + + def _call_qubes_dom0_update(self, options=None, args=None): + opts = self._get_dnf_opts() + if options: + opts += options + if args: + opts += args + return self.module.run_command(["/usr/bin/qubes-dom0-update", *opts]) + + def _get_dnf_opts(self): + args = ["-y"] + if self.force_xen_upgrade: + args.append("--force-xen-upgrade") + if self.skip_boot_check: + args.append("--skip-boot-check") + return args + + def _init_dnf(self): + self.dnf_base = libdnf5.base.Base() + self.conf = self.dnf_base.get_config() + try: + self.dnf_base.load_config() + except RuntimeError as e: + self.module.fail_json( + msg=str(e), + conf_file=self.dnf_conf.config_file_path, + failures=[], + rc=1, + ) + self.dnf_base.setup() + log_router = self.dnf_base.get_logger() + global_logger = libdnf5.logger.GlobalLogger() + global_logger.set(log_router.get(), libdnf5.logger.Logger.Level_DEBUG) + # FIXME hardcoding the filename does not seem right, should libdnf5 expose the default file name? + logger = libdnf5.logger.create_file_logger(self.dnf_base, "dnf5.log") + log_router.add_logger(logger) + sack = self.dnf_base.get_repo_sack() + sack.create_repos_from_system_configuration() + + # Disable all repo as we'll work with local RPM database only + repo_query = libdnf5.repo.RepoQuery(self.dnf_base) + repo_query.filter_id("*", libdnf5.common.QueryCmp_IGLOB) + for repo in repo_query: + repo.disable() + sack.load_repos() + + def _process_install(self): + """Install packages if not present + + This will check in local RPM database if wanted packages are + installed If not, qubes-dom0-update is called and RPM database is read + to return the installed version. + + We need to check in local RPM database because qubes-dom0-update will + always update our package even when using --action=install + """ + results = [] + packages_to_install = set() + for pkg_name in self.names: + if self.get_package_info(pkg_name) is None: + packages_to_install.add(pkg_name) + + if packages_to_install: + rc, stdout, stderr = self._call_qubes_dom0_update( + ["--action=install"], + packages_to_install, + ) + + if rc != 0: + self.module.exit_json( + msg="Failed to installed the specified package", + failures=stderr, + rc=1, + ) + + self._init_dnf() + + for pkg in packages_to_install: + pkg_info = self.get_package_info(pkg) + if pkg_info: + results.append(f"Installed: {pkg_info.get_nevra()}") + else: + results.append(f"Installed: {pkg}") + + if results: + self.module.exit_json( + results=results, + changed=True, + ) + + else: + self.module.exit_json(msg="Nothing to do") + + def _process_remove(self): + """Rely on local dnf to remove packages""" + results = [] + goal = libdnf5.base.Goal(self.dnf_base) + settings = libdnf5.base.GoalJobSettings() + settings.set_group_with_name(True) + + for pkg_name in self.names: + try: + goal.add_remove(pkg_name, settings) + except RuntimeError as e: + self.module.fail_json(msg=str(e), failures=[], rc=1) + + try: + transaction = goal.resolve() + except RuntimeError as e: + self.module.fail_json(msg=str(e), failures=[], rc=1) + + if transaction.get_problems(): + failures = [ + log_event.to_string() + for log_event in transaction.get_resolve_logs() + ] + + if ( + transaction.get_problems() + & libdnf5.base.GoalProblem_SOLVER_ERROR + != 0 + ): + msg = "Depsolve Error occurred" + else: + msg = "Failed to install some of the specified packages" + + self.module.fail_json( + msg=msg, + failures=failures, + rc=1, + ) + + for pkg in transaction.get_transaction_packages(): + results.append(f"Removed: {pkg.get_package().get_nevra()}") + + transaction.set_description( + "ansible qubesos.core.qubes_dom0_update module" + ) + result = transaction.run() + if result != libdnf5.base.Transaction.TransactionRunResult_SUCCESS: + self.module.fail_json( + msg="Transaction failure", + failures=[ + "{}: {}".format( + transaction.transaction_result_to_string(result), + log, + ) + for log in transaction.get_transaction_problems() + ], + rc=1, + ) + + if not results: + self.module.exit_json(msg="Nothing to do") + + self.module.exit_json(changed=True, results=results) + + def _process_update(self): + """Call qubes-dom0-update and read last transaction""" + if self.names == ["*"]: + rc, stdout, stderr = self._call_qubes_dom0_update( + ["--action=update", "--clean"] + ) + elif "*" in self.names: + self.module.fail_json(msg="'*' cannot be used with other packages") + else: + rc, stdout, stderr = self._call_qubes_dom0_update( + ["--clean"], self.names + ) + + # If system is up to date, an error may occur because nothing has + # been downloaded. Check stderr to confirm + # Fixed in recent versions of qubes-core-dom0-linux + # https://github.com/QubesOS/qubes-core-admin-linux/pull/210 + if "Nothing to do." in (stdout + stderr): + self.module.exit_json(msg="Nothing to do") + + if rc != 0: + self.module.fail_json( + msg="Failed to update the specified packages", + rc=1, + failures=stderr, + ) + + # Return last transaction details + history = libdnf5.transaction.TransactionHistory(self.dnf_base) + last_tx = history.list_all_transactions()[-1] + changes = [] + for pkg in last_tx.get_packages(): + pkg_action = libdnf5.transaction.transaction_item_action_to_string( + pkg.get_action() + ) + pkg_nevra = pkg.to_string() + changes.append(f"{pkg_action}: {pkg_nevra}") + self.module.exit_json( + changed=True, + results=changes, + ) + + def _switch_audio_server(self): + required_packages = { + "pipewire": ["pipewire", "pipewire-pulseaudio"], + "pulseaudio": ["pulseaudio"], + } + self._init_dnf() + + if all( + self.get_package_info(pkg) is not None + for pkg in required_packages[self.switch_audio_server] + ): + self.module.exit_json(msg="Nothing to do") + + rc, stdout, stderr = self._call_qubes_dom0_update( + [f"--switch-audio-server-to={self.switch_audio_server}"] + ) + + if rc != 0: + self.module.exit_json( + msg=f"Failed to switch audio daemon to {self.switch_audio_server}", + failures=stderr, + rc=1, + ) + + self.module.exit_json( + changed=True, + msg=f"Audio daemon switched to {self.switch_audio_server}", + ) + + def get_package_info(self, pkg_name): + """Extracted from is_installed() function of dnf5 module""" + # settings = libdnf5.base.ResolveSpecSettings() + installed_query = libdnf5.rpm.PackageQuery(self.dnf_base) + installed_query.filter_installed() + installed_query.filter_name(pkg_name) + + pkgs = list(installed_query) + + if len(pkgs) == 0: + return None + + return pkgs[0] + + def run(self): + if os.geteuid() != 0: + self.module.fail_json( + msg="This command has to be run under the root user.", + failures=[], + rc=1, + ) + + if self.names: + if self.state in {"present", "installed"}: + pkg_with_glob = [pkg for pkg in self.names if "*" in pkg] + if pkg_with_glob: + self.module.fail_json( + msg="Globs are not supported with state present and installed", + failures=pkg_with_glob, + rc=1, + ) + self._init_dnf() + + if self.names: + if self.state in {"installed", "present"}: + self._process_install() + elif self.state in {"removed", "absent"}: + self._process_remove() + elif self.state == "latest": + self._process_update() + + if self.switch_audio_server: + self._switch_audio_server() + + +def main(): + module = AnsibleModule( + argument_spec=ARGUMENT_SPEC, + required_one_of=[["name", "switch_audio_server"]], + mutually_exclusive=[["name", "switch_audio_server"]], + ) + + QubesDom0UpdateModule(module).run() + + +if __name__ == "__main__": + main() diff --git a/debian/install b/debian/install index 159ad76..aa536b2 100644 --- a/debian/install +++ b/debian/install @@ -8,4 +8,5 @@ /usr/share/ansible/collections/ansible_collections/qubesos/core/plugins/module_utils/qubes_module_qube.py /usr/share/ansible/collections/ansible_collections/qubesos/core/plugins/modules/command.py /usr/share/ansible/collections/ansible_collections/qubesos/core/plugins/modules/host_devices_facts.py -/usr/share/ansible/collections/ansible_collections/qubesos/core/plugins/modules/qube.py \ No newline at end of file +/usr/share/ansible/collections/ansible_collections/qubesos/core/plugins/modules/qube.py +/usr/share/ansible/collections/ansible_collections/qubesos/core/plugins/modules/qubes_dom0_update.py \ No newline at end of file diff --git a/tests/qubes/conftest.py b/tests/qubes/conftest.py index 26a15f6..d3b896a 100644 --- a/tests/qubes/conftest.py +++ b/tests/qubes/conftest.py @@ -1,3 +1,4 @@ +import subprocess import sys import uuid @@ -11,8 +12,12 @@ AnsibleFailJson, run_module_qubesos_core_qube_main as run_module, ) +from pathlib import Path +from typing import List + DEBIAN_TEMPLATE = "debian-12-minimal" +PLUGIN_PATH = Path(__file__).parent / "plugins" / "modules" @pytest.fixture(scope="function") @@ -169,3 +174,41 @@ def block_device(): # Assume the block device under test is always present # See fepitre/qubes-g2g-continuous-integration return "block:dom0:vdb" + + +@pytest.fixture +def run_playbook(tmp_path, ansible_config): + """ + Helper to write a playbook and execute it with ansible-playbook. + """ + + ansible_config_path = Path(__file__).parent.parent / f"{ansible_config}.cfg" + assert ansible_config_path.is_file() + + def _run(playbook_content: List[dict], vms: List[str] = []): + # Create playbook file + pb_file = tmp_path / "playbook.yml" + import yaml + + pb_file.write_text(yaml.dump(playbook_content)) + # Run ansible-playbook + cmd = [ + "ansible-playbook", + "-i", + f"localhost,dom0,{','.join(vms)}", + "-c", + "local", + "-M", + str(PLUGIN_PATH), + str(pb_file), + ] + result = subprocess.run( + cmd, + cwd=tmp_path, + capture_output=True, + text=True, + env={"ANSIBLE_CONFIG": str(ansible_config_path)}, + ) + return result + + return _run diff --git a/tests/qubes/test_cli.py b/tests/qubes/test_cli.py index b7f64fc..6bf9e03 100644 --- a/tests/qubes/test_cli.py +++ b/tests/qubes/test_cli.py @@ -1,50 +1,9 @@ import subprocess import uuid import json -from typing import List - import pytest -from pathlib import Path - -PLUGIN_PATH = Path(__file__).parent / "plugins" / "modules" - - -@pytest.fixture -def run_playbook(tmp_path, ansible_config): - """ - Helper to write a playbook and execute it with ansible-playbook. - """ - - ansible_config_path = Path(__file__).parent.parent / f"{ansible_config}.cfg" - assert ansible_config_path.is_file() - - def _run(playbook_content: List[dict], vms: List[str] = []): - # Create playbook file - pb_file = tmp_path / "playbook.yml" - import yaml - - pb_file.write_text(yaml.dump(playbook_content)) - # Run ansible-playbook - cmd = [ - "ansible-playbook", - "-i", - f"localhost,dom0,{','.join(vms)}", - "-c", - "local", - "-M", - str(PLUGIN_PATH), - str(pb_file), - ] - result = subprocess.run( - cmd, - cwd=tmp_path, - capture_output=True, - text=True, - env={"ANSIBLE_CONFIG": str(ansible_config_path)}, - ) - return result - return _run +from conftest import PLUGIN_PATH @pytest.mark.parametrize( diff --git a/tests/qubes/test_legacy_cli.py b/tests/qubes/test_legacy_cli.py index aaeb397..3cfa521 100644 --- a/tests/qubes/test_legacy_cli.py +++ b/tests/qubes/test_legacy_cli.py @@ -1,50 +1,9 @@ import subprocess import uuid import json -from typing import List - import pytest -from pathlib import Path - -PLUGIN_PATH = Path(__file__).parent / "plugins" / "modules" - - -@pytest.fixture -def run_playbook(tmp_path, ansible_config): - """ - Helper to write a playbook and execute it with ansible-playbook. - """ - - ansible_config_path = Path(__file__).parent.parent / f"{ansible_config}.cfg" - assert ansible_config_path.is_file() - - def _run(playbook_content: List[dict], vms: List[str] = []): - # Create playbook file - pb_file = tmp_path / "playbook.yml" - import yaml - - pb_file.write_text(yaml.dump(playbook_content)) - # Run ansible-playbook - cmd = [ - "ansible-playbook", - "-i", - f"localhost,dom0,{','.join(vms)}", - "-c", - "local", - "-M", - str(PLUGIN_PATH), - str(pb_file), - ] - result = subprocess.run( - cmd, - cwd=tmp_path, - capture_output=True, - text=True, - env={"ANSIBLE_CONFIG": str(ansible_config_path)}, - ) - return result - return _run +from conftest import PLUGIN_PATH @pytest.mark.parametrize( diff --git a/tests/qubes/test_module_dom0_update_integ.py b/tests/qubes/test_module_dom0_update_integ.py new file mode 100644 index 0000000..ebd09f9 --- /dev/null +++ b/tests/qubes/test_module_dom0_update_integ.py @@ -0,0 +1,305 @@ +#!/usr/bin/python3 +# Copyright (C) 2026 Guillaume Chinal (guiiix) +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: GPL-3.0-or-later + +import json +import os +import pytest +import subprocess +import rpm + + +@pytest.fixture(autouse=True) +def skip_if_not_dom0(): + if not os.path.exists("/usr/bin/qubes-dom0-update"): + pytest.skip("Can be tested on dom0 only") + + +@pytest.fixture +def nano_installed(): + if is_package_installed("nano"): + return + + result = subprocess.run( + ["/usr/bin/qubes-dom0-update", "-y", "nano"], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + + +@pytest.fixture +def nano_uninstalled(): + result = subprocess.run( + ["dnf", "remove", "-y", "nano"], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + + +@pytest.fixture +def notif_daemon_downgraded(): + result = subprocess.run( + [ + "/usr/bin/qubes-dom0-update", + "--action=downgrade", + "-y", + "qubes-notification-daemon", + ], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + + +def get_installed_packages(package_name) -> rpm.mi: + ts = rpm.TransactionSet() + return ts.dbMatch("name", package_name) + + +def is_package_installed(package_name): + return len(get_installed_packages(package_name)) > 0 + + +@pytest.mark.parametrize( + "ansible_config", + ["ansible_linear_strategy"], +) +def test_playbook_pkg_nano_installed(run_playbook, nano_uninstalled): + assert not is_package_installed("nano") + + playbook = [ + { + "hosts": "dom0", + "gather_facts": False, + "tasks": [ + { + "name": "Install nano", + "qubesos.core.qubes_dom0_update": { + "name": "nano", + "state": "present", + }, + }, + ], + } + ] + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + task_results = returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"][ + "results" + ][0] + + mi = get_installed_packages("nano") + assert len(mi) == 1 + pkg = list(mi)[0] + assert task_results == f"Installed: {pkg.nevra}" + + +@pytest.mark.parametrize( + "ansible_config", + ["ansible_linear_strategy"], +) +def test_playbook_pkg_nano_installed_using_package( + run_playbook, nano_uninstalled +): + assert not is_package_installed("nano") + + playbook = [ + { + "hosts": "dom0", + "gather_facts": False, + "tasks": [ + { + "name": "Install nano", + "package": { + "name": "nano", + "state": "present", + "use": "qubesos.core.qubes_dom0_update", + }, + }, + ], + } + ] + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + task_results = returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"][ + "results" + ][0] + + mi = get_installed_packages("nano") + assert len(mi) == 1 + pkg = list(mi)[0] + assert task_results == f"Installed: {pkg.nevra}" + + +@pytest.mark.parametrize( + "ansible_config", + ["ansible_linear_strategy"], +) +def test_playbook_pkg_nano_removed(run_playbook, nano_installed): + mi = get_installed_packages("nano") + assert len(mi) == 1 + pkg = list(mi)[0] + + playbook = [ + { + "hosts": "dom0", + "gather_facts": False, + "tasks": [ + { + "name": "Install nano", + "qubesos.core.qubes_dom0_update": { + "name": "nano", + "state": "absent", + }, + }, + ], + } + ] + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + task_results = returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"][ + "results" + ][0] + + assert task_results == f"Removed: {pkg.nevra}" + + +@pytest.mark.parametrize( + "ansible_config", + ["ansible_linear_strategy"], +) +def test_installed_package_should_be_upgraded_only_when_state_latest( + run_playbook, notif_daemon_downgraded +): + mi = get_installed_packages("qubes-notification-daemon") + assert len(mi) == 1 + pkg_before = list(mi)[0] + + playbook = [ + { + "hosts": "dom0", + "gather_facts": False, + "tasks": [ + { + "name": "Install qubes-notification-daemon", + "qubesos.core.qubes_dom0_update": { + "name": "qubes-notification-daemon", + "state": "present", + }, + }, + ], + } + ] + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert not returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] + mi = get_installed_packages("qubes-notification-daemon") + assert len(mi) == 1 + pkg_after = list(mi)[0] + assert ( + rpm.labelCompare( + (pkg_before.epoch, pkg_before.version, pkg_before.release), + (pkg_after.epoch, pkg_after.version, pkg_after.release), + ) + == 0 + ) + + playbook[0]["tasks"][0]["qubesos.core.qubes_dom0_update"][ + "state" + ] = "latest" + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] + mi = get_installed_packages("qubes-notification-daemon") + assert len(mi) == 1 + pkg_after = list(mi)[0] + assert ( + rpm.labelCompare( + (pkg_before.epoch, pkg_before.version, pkg_before.release), + (pkg_after.epoch, pkg_after.version, pkg_after.release), + ) + == -1 + ) + + +@pytest.mark.parametrize( + "ansible_config", + ["ansible_linear_strategy"], +) +def test_idempotence(run_playbook, nano_uninstalled, notif_daemon_downgraded): + assert not is_package_installed("nano") + + playbook = [ + { + "hosts": "dom0", + "gather_facts": False, + "tasks": [ + { + "name": "Install nano", + "qubesos.core.qubes_dom0_update": { + "name": "nano", + "state": "present", + }, + }, + ], + } + ] + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] + + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert not returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] + + playbook[0]["tasks"][0]["qubesos.core.qubes_dom0_update"][ + "state" + ] = "absent" + + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] + + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert not returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] + + playbook[0]["tasks"][0]["qubesos.core.qubes_dom0_update"][ + "state" + ] = "latest" + playbook[0]["tasks"][0]["qubesos.core.qubes_dom0_update"]["name"] = "*" + + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] + + result = run_playbook(playbook) + assert result.returncode == 0, result.stderr + returned_data = json.loads(result.stdout) + assert not returned_data["plays"][0]["tasks"][0]["hosts"]["dom0"]["changed"] diff --git a/tests/qubes/test_module_dom0_update_units.py b/tests/qubes/test_module_dom0_update_units.py new file mode 100644 index 0000000..882d187 --- /dev/null +++ b/tests/qubes/test_module_dom0_update_units.py @@ -0,0 +1,560 @@ +#!/usr/bin/python3 +# Copyright (C) 2026 Guillaume Chinal (guiiix) +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: GPL-3.0-or-later + +import json +import sys +import pytest + + +from ansible.module_utils import basic +from ansible.module_utils.common.text.converters import to_bytes +from unittest.mock import MagicMock, patch + + +_libdnf5 = MagicMock() +sys.modules["libdnf5"] = _libdnf5 +sys.modules["libdnf5.base"] = _libdnf5 +sys.modules["libdnf5.rpm"] = _libdnf5 +sys.modules["libdnf5.repo"] = _libdnf5 +sys.modules["libdnf5.common"] = _libdnf5 +sys.modules["libdnf5.logger"] = _libdnf5 +sys.modules["libdnf5.transaction"] = _libdnf5 + +from ansible_collections.qubesos.core.plugins.modules.qubes_dom0_update import ( + QubesDom0UpdateModule, + main, + ARGUMENT_SPEC, +) + +# --------------------------------------------------------------------------- +# Ansible test helpers +# --------------------------------------------------------------------------- + + +def set_module_args(args) -> None: + """prepare arguments so that they will be picked up during module creation + (https://docs.ansible.com/projects/ansible/latest/dev_guide/testing_units_modules.html) + """ + basic._ANSIBLE_ARGS = to_bytes(json.dumps({"ANSIBLE_MODULE_ARGS": args})) + + +class AnsibleExitJson(Exception): + """Exception class to be raised by module.exit_json and caught by the test case""" + + +class AnsibleFailJson(Exception): + """Exception class to be raised by module.fail_json and caught by the test case""" + + +def exit_json(*args, **kwargs): + """function to patch over exit_json; package return data into an exception""" + kwargs.setdefault("changed", False) + raise AnsibleExitJson(kwargs) + + +def fail_json(*args, **kwargs): + kwargs["failed"] = True + raise AnsibleFailJson(kwargs) + + +def fake_pkg(nevra: str) -> MagicMock: + pkg = MagicMock() + pkg.get_nevra.return_value = nevra + return pkg + + +def init_module(params=None) -> QubesDom0UpdateModule: + if params is None: + params = {} + set_module_args(params) + + return QubesDom0UpdateModule( + basic.AnsibleModule(argument_spec=ARGUMENT_SPEC) + ) + + +@pytest.fixture(autouse=True) +def setup(): + _libdnf5.reset_mock() + with patch.multiple( + basic.AnsibleModule, exit_json=exit_json, fail_json=fail_json + ), patch("os.geteuid", return_value=0), patch( + "ansible_collections.qubesos.core.plugins.modules.qubes_dom0_update.get_best_parsable_locale", + return_value="C", + ), patch.dict( + "os.environ", {}, clear=False + ): + yield + + +# --------------------------------------------------------------------------- +# Call qubes_dom0_update +# --------------------------------------------------------------------------- + + +def test_call_qubes_dom0_update_default_options(): + m = init_module() + with patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "", "") + ) as mock_run: + m._call_qubes_dom0_update() + assert mock_run.call_args[0][0] == ["/usr/bin/qubes-dom0-update", "-y"] + + +def test_call_qubes_dom0_opts_force_xen_upgrade(): + m = init_module({"force_xen_upgrade": True}) + with patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "", "") + ) as mock_run: + m._call_qubes_dom0_update() + assert mock_run.call_args[0][0] == [ + "/usr/bin/qubes-dom0-update", + "-y", + "--force-xen-upgrade", + ] + + +def test_call_qubes_dom0_opts_skip_boot_check(): + m = init_module({"skip_boot_check": True}) + with patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "", "") + ) as mock_run: + m._call_qubes_dom0_update() + assert mock_run.call_args[0][0] == [ + "/usr/bin/qubes-dom0-update", + "-y", + "--skip-boot-check", + ] + + +def test_call_qubes_dom0_update_with_options_and_args(): + m = init_module({"force_xen_upgrade": True}) + with patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "", "") + ) as mock_run: + m._call_qubes_dom0_update(options=["--action=install"], args=["pkg1"]) + cmd = mock_run.call_args[0][0] + assert cmd[0] == "/usr/bin/qubes-dom0-update" + assert "-y" in cmd + assert "--force-xen-upgrade" in cmd + assert "--action=install" in cmd + assert "pkg1" in cmd + + +# --------------------------------------------------------------------------- +# Module inputs checks +# --------------------------------------------------------------------------- + + +def test_run_fails_when_not_root(): + with patch("os.geteuid", return_value=1000) as _: + module = init_module() + with pytest.raises(AnsibleFailJson) as exc: + module.run() + assert ( + exc.value.args[0]["msg"] + == "This command has to be run under the root user." + ) + + +def test_run_rejects_glob_with_state_present(): + set_module_args({"name": ["pkg*"], "state": "present"}) + with pytest.raises(AnsibleFailJson) as exc: + main() + assert "Globs are not supported" in exc.value.args[0]["msg"] + assert "pkg*" in exc.value.args[0]["failures"] + + +def test_run_rejects_glob_with_state_installed(): + set_module_args({"name": ["*-devel"], "state": "installed"}) + with pytest.raises(AnsibleFailJson) as exc: + main() + assert "*-devel" in exc.value.args[0]["failures"] + + +def test_run_rejects_glob_with_other_packages(): + set_module_args({"name": ["*", "pkg"], "state": "latest"}) + with pytest.raises(AnsibleFailJson) as exc: + main() + assert exc.value.args[0]["msg"] == "'*' cannot be used with other packages" + + +# --------------------------------------------------------------------------- +# Install +# --------------------------------------------------------------------------- + + +def test_process_install_nothing_to_do(): + set_module_args({"name": ["pkg1"], "state": "present"}) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, + "get_package_info", + return_value=fake_pkg("pkg1-1.0-1.x86_64"), + ), patch.object(basic.AnsibleModule, "run_command") as mock_run: + mock_run.return_value = (0, "", "") + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["msg"] == "Nothing to do" + mock_run.assert_not_called() + + +def test_process_install_installs_missing_package(): + set_module_args({"name": ["pkg1"], "state": "present"}) + pkg = fake_pkg("pkg1-1.0-1.x86_64") + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, "get_package_info", side_effect=[None, pkg] + ), patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "", "") + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["changed"] is True + assert "Installed: pkg1-1.0-1.x86_64" in exc.value.args[0]["results"] + + +def test_process_install_multiple_missing_packages(): + set_module_args({"name": ["pkg1", "pkg2"], "state": "present"}) + pkg1 = fake_pkg("pkg1-1.0-1.x86_64") + pkg2 = fake_pkg("pkg2-2.0-1.x86_64") + + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, + "get_package_info", + side_effect=[None, None, pkg1, pkg2], + ), patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "", "") + ) as mock_run: + with pytest.raises(AnsibleExitJson) as exc: + main() + results = exc.value.args[0]["results"] + assert any("pkg1" in r for r in results) + assert any("pkg2" in r for r in results) + mock_run.assert_called_once() + assert [ + "/usr/bin/qubes-dom0-update", + "-y", + "--action=install", + ] == mock_run.call_args_list[0].args[0][0:3] + assert ["pkg1", "pkg2"] == sorted(mock_run.call_args_list[0].args[0][3:]) + + +def test_process_install_exits_on_command_failure(): + set_module_args({"name": ["pkg1"], "state": "present"}) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, "get_package_info", return_value=None + ), patch.object( + basic.AnsibleModule, + "run_command", + return_value=(1, "", "download failed"), + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["rc"] == 1 + + +# --------------------------------------------------------------------------- +# Update +# --------------------------------------------------------------------------- + +_QUBES_DOM0_UPDATE_NOTHING_TO_DO_STDOUT = """ +Nothing to do. +""" + +_QUBES_DOM0_UPDATE_NOTHING_TO_DO_STDERR = """ +Using sys-firewall as UpdateVM for Dom0 +Downloading updates. This may take a while... +Updating and loading repositories: + Qubes Host Repository (updates) 100% | 4.0 KiB/s | 2.7 KiB | 00m01s + Fedora 41 - x86_64 - Updates 100% | 6.6 KiB/s | 3.4 KiB | 00m01s + Fedora 41 - x86_64 100% | 51.2 KiB/s | 3.6 KiB | 00m00s +Repositories loaded. +Updating and loading repositories: + Qubes OS Repository for Dom0 100% | 0.0 B/s | 1.5 KiB | 00m00s +Repositories loaded. +""" + +_QUBES_DOM0_UPDATE_ERROR_STDERR = """ +Using sys-firewall as UpdateVM for Dom0 +Downloading updates. This may take a while... +Updating and loading repositories: + Fedora 41 - x86_64 - Updates ???% | 0.0 B 0.0 B + Fedora 41 - x86_64 - Updates ???% | 0.0 BKiB_[1A + Qubes Host Repository (updates) 100% | 2.7 KiB_[1A + Fedora 41 - x86_64 - Updates 100% | 3.4 KiB + Fedora 41 - x86_64 100% | 23.9 MiB + +Broadcast message from root@sys-firewall (Wed 2026-04-29 10:01:55 CEST): + +The system will power off now! + + +Session terminated, killing shell... +""" + + +def test_process_update_nothing_to_do(): + set_module_args({"name": ["pkg1"], "state": "latest"}) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + basic.AnsibleModule, + "run_command", + return_value=( + 0, + _QUBES_DOM0_UPDATE_NOTHING_TO_DO_STDOUT, + _QUBES_DOM0_UPDATE_NOTHING_TO_DO_STDERR, + ), + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["msg"] == "Nothing to do" + + +def test_process_update_fails_on_qubes_error(): + set_module_args({"name": ["pkg1"], "state": "latest"}) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + basic.AnsibleModule, + "run_command", + return_value=(1, "", _QUBES_DOM0_UPDATE_ERROR_STDERR), + ): + with pytest.raises(AnsibleFailJson) as exc: + main() + assert "Failed to update" in exc.value.args[0]["msg"] + + +def _setup_update_transaction_history(packages_info): + """Set up libdnf5.transaction mocks for the update path. + + packages_info: list of (action_string, nevra_string) tuples. + """ + pkg_mocks = [] + action_map = {} + for action_str, nevra_str in packages_info: + pkg_mock = MagicMock() + pkg_mock.to_string.return_value = nevra_str + action_mock = MagicMock() + pkg_mock.get_action.return_value = action_mock + action_map[action_mock] = action_str + pkg_mocks.append(pkg_mock) + + last_tx_mock = MagicMock() + last_tx_mock.get_packages.return_value = pkg_mocks + + history_mock = MagicMock() + history_mock.list_all_transactions.return_value = [last_tx_mock] + + _libdnf5.transaction.TransactionHistory.return_value = history_mock + _libdnf5.transaction.transaction_item_action_to_string.side_effect = ( + lambda a: action_map[a] + ) + + +def test_process_update_returns_transaction_history(): + set_module_args({"name": ["pkg1"], "state": "latest"}) + _setup_update_transaction_history( + [ + ("Upgrade", "pkg1-2.0-1.x86_64"), + ("Replaced", "pkg1-1.0-1.x86_64"), + ] + ) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "updated", "") + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + + assert exc.value.args[0]["changed"] is True + assert exc.value.args[0]["results"] == [ + "Upgrade: pkg1-2.0-1.x86_64", + "Replaced: pkg1-1.0-1.x86_64", + ] + + +def test_process_update_returns_multiple_package_changes(): + set_module_args({"name": ["pkg1", "pkg2", "pkg3"], "state": "latest"}) + _setup_update_transaction_history( + [ + ("Upgrade", "pkg1-2.0-1.x86_64"), + ("Upgrade", "pkg2-3.0-1.x86_64"), + ("Install", "pkg3-1.0-1.x86_64"), + ] + ) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "updated", "") + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + + results = exc.value.args[0]["results"] + assert "Upgrade: pkg1-2.0-1.x86_64" in results + assert "Upgrade: pkg2-3.0-1.x86_64" in results + assert "Install: pkg3-1.0-1.x86_64" in results + + +def test_process_update_returns_empty_changes_when_no_packages(): + set_module_args({"name": ["pkg1"], "state": "latest"}) + _setup_update_transaction_history([]) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "updated", "") + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + + assert exc.value.args[0]["changed"] is True + assert exc.value.args[0]["results"] == [] + + +# --------------------------------------------------------------------------- +# Remove +# --------------------------------------------------------------------------- + + +def _setup_remove_transaction(transaction_packages, run_succeeds=True): + success_val = object() + _libdnf5.base.Transaction.TransactionRunResult_SUCCESS = success_val + + transaction_mock = MagicMock() + transaction_mock.get_problems.return_value = 0 + transaction_mock.get_transaction_packages.return_value = ( + transaction_packages + ) + transaction_mock.run.return_value = ( + success_val if run_succeeds else MagicMock() + ) + + goal_mock = MagicMock() + goal_mock.resolve.return_value = transaction_mock + _libdnf5.base.Goal.return_value = goal_mock + _libdnf5.base.GoalJobSettings.return_value = MagicMock() + return transaction_mock + + +def test_process_remove_nothing_to_do(): + set_module_args({"name": ["pkg1"], "state": "absent"}) + _setup_remove_transaction([]) + with patch.object(QubesDom0UpdateModule, "_init_dnf"): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["msg"] == "Nothing to do" + + +def test_process_remove_success(): + set_module_args({"name": ["pkg1"], "state": "absent"}) + pkg_mock = MagicMock() + pkg_mock.get_package.return_value = fake_pkg("pkg1-1.0-1.x86_64") + _setup_remove_transaction([pkg_mock]) + with patch.object(QubesDom0UpdateModule, "_init_dnf"): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["changed"] is True + assert "Removed: pkg1-1.0-1.x86_64" in exc.value.args[0]["results"] + + +def test_process_remove_fails_on_transaction_run_error(): + set_module_args({"name": ["pkg1"], "state": "absent"}) + pkg_mock = MagicMock() + pkg_mock.get_package.return_value = fake_pkg("pkg1-1.0-1.x86_64") + _setup_remove_transaction([pkg_mock], run_succeeds=False) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), pytest.raises( + AnsibleFailJson + ) as exc: + main() + assert "Transaction failure" in exc.value.args[0]["msg"] + + +def test_process_remove_fails_on_goal_resolve_error(): + set_module_args({"name": ["pkg1"], "state": "absent"}) + goal_mock = MagicMock() + goal_mock.resolve.side_effect = RuntimeError("depsolve error") + _libdnf5.base.Goal.return_value = goal_mock + _libdnf5.base.GoalJobSettings.return_value = MagicMock() + with patch.object(QubesDom0UpdateModule, "_init_dnf"), pytest.raises( + AnsibleFailJson + ) as exc: + main() + assert exc.value.args[0]["msg"] == "depsolve error" + + +def test_process_remove_fails_on_add_remove_error(): + set_module_args({"name": ["pkg1"], "state": "absent"}) + goal_mock = MagicMock() + goal_mock.add_remove.side_effect = RuntimeError("cannot remove pkg1") + _libdnf5.base.Goal.return_value = goal_mock + _libdnf5.base.GoalJobSettings.return_value = MagicMock() + with patch.object(QubesDom0UpdateModule, "_init_dnf"), pytest.raises( + AnsibleFailJson + ) as exc: + main() + assert exc.value.args[0]["msg"] == "cannot remove pkg1" + + +# --------------------------------------------------------------------------- +# _switch_audio_server -- tested via main() +# --------------------------------------------------------------------------- + + +def test_switch_audio_server_pipewire_nothing_to_do(): + set_module_args({"switch_audio_server": "pipewire"}) + pkg = fake_pkg("pipewire-1.0-1.x86_64") + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, "get_package_info", return_value=pkg + ), patch.object(basic.AnsibleModule, "run_command") as mock_run: + mock_run.return_value = (0, "", "") + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["msg"] == "Nothing to do" + mock_run.assert_not_called() + + +def test_switch_audio_server_pulseaudio_nothing_to_do(): + set_module_args({"switch_audio_server": "pulseaudio"}) + pkg = fake_pkg("pulseaudio-1.0-1.x86_64") + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, "get_package_info", return_value=pkg + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["msg"] == "Nothing to do" + + +def test_switch_audio_server_success(): + set_module_args({"switch_audio_server": "pulseaudio"}) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, "get_package_info", return_value=None + ), patch.object( + basic.AnsibleModule, "run_command", return_value=(0, "", "") + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["changed"] is True + assert exc.value.args[0]["msg"] == "Audio daemon switched to pulseaudio" + + +def test_switch_audio_server_command_fails(): + set_module_args({"switch_audio_server": "pipewire"}) + with patch.object(QubesDom0UpdateModule, "_init_dnf"), patch.object( + QubesDom0UpdateModule, "get_package_info", return_value=None + ), patch.object( + basic.AnsibleModule, + "run_command", + return_value=(1, "", "switch failed"), + ): + with pytest.raises(AnsibleExitJson) as exc: + main() + assert exc.value.args[0]["rc"] == 1 + assert ( + exc.value.args[0]["msg"] == "Failed to switch audio daemon to pipewire" + )