cloud-images/ansible/collections/ansible_collections/community/general/plugins/modules/xbps.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

# Copyright 2016 Dino Occhialini <dino.occhialini@gmail.com>
#
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later

from __future__ import absolute_import, division, print_function
__metaclass__ = type


DOCUMENTATION = r"""
module: xbps
short_description: Manage packages with XBPS
description:
  - Manage packages with the XBPS package manager.
author:
  - "Dino Occhialini (@dinoocch)"
  - "Michael Aldridge (@the-maldridge)"
extends_documentation_fragment:
  - community.general.attributes
attributes:
  check_mode:
    support: full
  diff_mode:
    support: none
options:
  name:
    description:
      - Name of the package to install, upgrade, or remove.
    aliases: [pkg, package]
    type: list
    elements: str
  state:
    description:
      - Desired state of the package.
    default: "present"
    choices: ["present", "absent", "latest", "installed", "removed"]
    type: str
  recurse:
    description:
      - When removing a package, also remove its dependencies, provided that they are not required by other packages and were not explicitly installed
        by a user.
    type: bool
    default: false
  update_cache:
    description:
      - Whether or not to refresh the master package lists. This can be run as part of a package installation or as a separate step.
    type: bool
    default: true
  upgrade:
    description:
      - Whether or not to upgrade whole system.
    type: bool
    default: false
  upgrade_xbps:
    description:
      - Whether or not to upgrade the xbps package when necessary. Before installing new packages, xbps requires the user to update the xbps package
        itself. Thus when this option is set to V(false), upgrades and installations will fail when xbps is not up to date.
    type: bool
    default: true
    version_added: '0.2.0'
  root:
    description:
      - The full path for the target root directory.
    type: path
    version_added: '10.2.0'
  repositories:
    description:
      - Repository URL(s) to prepend to the repository list for the package installation. The URL can be a URL to a repository for remote repositories
        or a path for local repositories.
    type: list
    elements: str
    version_added: '10.2.0'
  accept_pubkey:
    description:
      - Whether or not repository signing keys should be automatically accepted.
    type: bool
    default: false
    version_added: '10.2.0'
"""

EXAMPLES = r"""
- name: Install package foo (automatically updating the xbps package if needed)
  community.general.xbps:
    name: foo
    state: present

- name: Upgrade package foo
  community.general.xbps:
    name: foo
    state: latest
    update_cache: true

- name: Remove packages foo and bar
  community.general.xbps:
    name:
      - foo
      - bar
    state: absent

- name: Recursively remove package foo
  community.general.xbps:
    name: foo
    state: absent
    recurse: true

- name: Update package cache
  community.general.xbps:
    update_cache: true

- name: Upgrade packages
  community.general.xbps:
    upgrade: true

- name: Install a package, failing if the xbps package is out of date
  community.general.xbps:
    name: foo
    state: present
    upgrade_xbps: false

- name: Find repository keys to install into a new void system on a mounted partition
  ansible.builtin.find:
    path: /var/db/xbps/keys
    pattern: "*.plist"
  register: xbps_keys

- name: Copy repository keys to into a new void system on a mounted partition
  ansible.builtin.copy:
    remote_src: true
    src: "{{ item }}"
    dest: "/mnt/{{ item }}"
    owner: root
    group: root
    mode: "0644"
  when: xbps_keys.matched > 0
  loop: "{{ xbps_keys.files | map(attribute='path') }}"

- name: Install a new void system on a mounted partition
  community.general.xbps:
    name: base-system
    state: present
    repositories: https://repo-default.voidlinux.org/current
    root: /mnt
"""

RETURN = r"""
msg:
  description: Message about results.
  returned: success
  type: str
  sample: "System Upgraded"
packages:
  description: Packages that are affected/would be affected.
  type: list
  sample: ["ansible"]
  returned: success
"""


import os

from ansible.module_utils.basic import AnsibleModule


def is_installed(xbps_output):
    """Returns package install state"""
    return bool(len(xbps_output))


def append_flags(module, xbps_path, cmd, skip_repo=False):
    """Appends the repository/root flags when needed"""
    if module.params["root"]:
        cmd = "%s -r %s" % (cmd, module.params["root"])
    if module.params["repositories"] and not cmd.startswith(xbps_path["remove"]) and not skip_repo:
        for repo in module.params["repositories"]:
            cmd = "%s --repository=%s" % (cmd, repo)

    return cmd


def query_package(module, xbps_path, name, state="present"):
    """Returns Package info"""
    if state == "present":
        lcmd = "%s %s" % (xbps_path['query'], name)
        lcmd = append_flags(module, xbps_path, lcmd, skip_repo=True)
        lrc, lstdout, lstderr = module.run_command(lcmd, check_rc=False)
        if not is_installed(lstdout):
            # package is not installed locally
            return False, False

        rcmd = "%s -Sun" % (xbps_path['install'])
        rcmd = append_flags(module, xbps_path, rcmd)
        rrc, rstdout, rstderr = module.run_command(rcmd, check_rc=False)
        if rrc == 0 or rrc == 17:
            """Return True to indicate that the package is installed locally,
            and the result of the version number comparison to determine if the
            package is up-to-date"""
            return True, name not in rstdout

        return False, False


def update_package_db(module, xbps_path):
    """Returns True if update_package_db changed"""
    cmd = "%s -S" % (xbps_path['install'])
    cmd = append_flags(module, xbps_path, cmd)
    if module.params['accept_pubkey']:
        stdin = "y\n"
    else:
        stdin = "n\n"
    rc, stdout, stderr = module.run_command(cmd, check_rc=False, data=stdin)

    if "Failed to import pubkey" in stderr:
        module.fail_json(msg="Failed to import pubkey for repository")
    if rc != 0:
        module.fail_json(msg="Could not update package db")
    if "avg rate" in stdout:
        return True
    else:
        return False


def upgrade_xbps(module, xbps_path, exit_on_success=False):
    cmdupgradexbps = "%s -uy xbps" % (xbps_path['install'])
    cmdupgradexbps = append_flags(module, xbps_path, cmdupgradexbps)
    rc, stdout, stderr = module.run_command(cmdupgradexbps, check_rc=False)
    if rc != 0:
        module.fail_json(msg='Could not upgrade xbps itself')


def upgrade(module, xbps_path):
    """Returns true is full upgrade succeeds"""
    cmdupgrade = "%s -uy" % (xbps_path['install'])
    cmdneedupgrade = "%s -un" % (xbps_path['install'])
    cmdupgrade = append_flags(module, xbps_path, cmdupgrade)
    cmdneedupgrade = append_flags(module, xbps_path, cmdneedupgrade)

    rc, stdout, stderr = module.run_command(cmdneedupgrade, check_rc=False)
    if rc == 0:
        if len(stdout.splitlines()) == 0:
            module.exit_json(changed=False, msg='Nothing to upgrade')
        elif module.check_mode:
            module.exit_json(changed=True, msg='Would have performed upgrade')
        else:
            rc, stdout, stderr = module.run_command(cmdupgrade, check_rc=False)
            if rc == 0:
                module.exit_json(changed=True, msg='System upgraded')
            elif rc == 16 and module.params['upgrade_xbps']:
                upgrade_xbps(module, xbps_path)
                # avoid loops by not trying self-upgrade again
                module.params['upgrade_xbps'] = False
                upgrade(module, xbps_path)
            else:
                module.fail_json(msg="Could not upgrade")
    else:
        module.fail_json(msg="Could not upgrade")


def remove_packages(module, xbps_path, packages):
    """Returns true if package removal succeeds"""
    changed_packages = []
    # Using a for loop in case of error, we can report the package that failed
    for package in packages:
        # Query the package first, to see if we even need to remove
        installed, updated = query_package(module, xbps_path, package)
        if not installed:
            continue

        cmd = "%s -y %s" % (xbps_path['remove'], package)
        cmd = append_flags(module, xbps_path, cmd, skip_repo=True)
        rc, stdout, stderr = module.run_command(cmd, check_rc=False)

        if rc != 0:
            module.fail_json(msg="failed to remove %s" % (package))

        changed_packages.append(package)

    if len(changed_packages) > 0:

        module.exit_json(changed=True, msg="removed %s package(s)" %
                         len(changed_packages), packages=changed_packages)

    module.exit_json(changed=False, msg="package(s) already absent")


def install_packages(module, xbps_path, state, packages):
    """Returns true if package install succeeds."""
    toInstall = []
    for i, package in enumerate(packages):
        """If the package is installed and state == present or state == latest
        and is up-to-date then skip"""
        installed, updated = query_package(module, xbps_path, package)
        if installed and (state == 'present' or
                          (state == 'latest' and updated)):
            continue

        toInstall.append(package)

    if len(toInstall) == 0:
        module.exit_json(changed=False, msg="Nothing to Install")

    cmd = "%s -y %s" % (xbps_path['install'], " ".join(toInstall))
    cmd = append_flags(module, xbps_path, cmd)
    rc, stdout, stderr = module.run_command(cmd, check_rc=False)

    if rc == 16 and module.params['upgrade_xbps']:
        upgrade_xbps(module, xbps_path)
        # avoid loops by not trying self-update again
        module.params['upgrade_xbps'] = False
        install_packages(module, xbps_path, state, packages)
    elif rc != 0 and not (state == 'latest' and rc == 17):
        module.fail_json(msg="failed to install %s packages(s)"
                         % (len(toInstall)),
                         packages=toInstall)

    module.exit_json(changed=True, msg="installed %s package(s)"
                     % (len(toInstall)),
                     packages=toInstall)


def check_packages(module, xbps_path, packages, state):
    """Returns change status of command"""
    would_be_changed = []
    for package in packages:
        installed, updated = query_package(module, xbps_path, package)
        if ((state in ["present", "latest"] and not installed) or
                (state == "absent" and installed) or
                (state == "latest" and not updated)):
            would_be_changed.append(package)
    if would_be_changed:
        if state == "absent":
            state = "removed"
        module.exit_json(changed=True, msg="%s package(s) would be %s" % (
            len(would_be_changed), state),
            packages=would_be_changed)
    else:
        module.exit_json(changed=False, msg="package(s) already %s" % state,
                         packages=[])


def update_cache(module, xbps_path, upgrade_planned):
    """Update package cache"""
    if module.check_mode:
        if upgrade_planned:
            return
        module.exit_json(
            changed=True, msg='Would have updated the package cache'
        )
    changed = update_package_db(module, xbps_path)
    if not upgrade_planned:
        module.exit_json(changed=changed, msg=(
            'Updated the package master lists' if changed
            else 'Package list already up to date'
        ))


def main():
    """Returns, calling appropriate command"""

    module = AnsibleModule(
        argument_spec=dict(
            name=dict(default=None, aliases=['pkg', 'package'], type='list', elements='str'),
            state=dict(default='present', choices=['present', 'installed',
                                                   'latest', 'absent',
                                                   'removed']),
            recurse=dict(default=False, type='bool'),
            upgrade=dict(default=False, type='bool'),
            update_cache=dict(default=True, type='bool'),
            upgrade_xbps=dict(default=True, type='bool'),
            root=dict(type='path'),
            repositories=dict(type='list', elements='str'),
            accept_pubkey=dict(default=False, type='bool')
        ),
        required_one_of=[['name', 'update_cache', 'upgrade']],
        supports_check_mode=True)

    xbps_path = dict()
    xbps_path['install'] = module.get_bin_path('xbps-install', True)
    xbps_path['query'] = module.get_bin_path('xbps-query', True)
    xbps_path['remove'] = module.get_bin_path('xbps-remove', True)

    if not os.path.exists(xbps_path['install']):
        module.fail_json(msg="cannot find xbps, in path %s"
                         % (xbps_path['install']))

    p = module.params

    # normalize the state parameter
    if p['state'] in ['present', 'installed']:
        p['state'] = 'present'
    elif p['state'] in ['absent', 'removed']:
        p['state'] = 'absent'

    if p['update_cache']:
        update_cache(module, xbps_path, (p['name'] or p['upgrade']))

    if p['upgrade']:
        upgrade(module, xbps_path)

    if p['name']:
        pkgs = p['name']

        if module.check_mode:
            check_packages(module, xbps_path, pkgs, p['state'])

        if p['state'] in ['present', 'latest']:
            install_packages(module, xbps_path, p['state'], pkgs)
        elif p['state'] == 'absent':
            remove_packages(module, xbps_path, pkgs)


if __name__ == "__main__":
    main()