1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
community.general/plugins/modules/packaging/os/xbps.py
Felix Fontein e4eead189b
Fix linting errors; fix some real bugs (#5111) (#5115)
* Fix linting errors.

* Fix bugs.

* Another linter error ignored.

* More fixes.

* Ignore sanity errors with older versions.

ci_complete

* Forgot to commit more changes.

(cherry picked from commit a54af8909c)
2022-08-12 14:37:34 +02:00

333 lines
11 KiB
Python

#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright 2016 Dino Occhialini <dino.occhialini@gmail.com>
#
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = '''
---
module: xbps
short_description: Manage packages with XBPS
description:
- Manage packages with the XBPS package manager.
author:
- "Dino Occhialini (@dinoocch)"
- "Michael Aldridge (@the-maldridge)"
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: no
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.
- Alias C(update-cache) has been deprecated and will be removed in community.general 5.0.0.
aliases: ['update-cache']
type: bool
default: yes
upgrade:
description:
- Whether or not to upgrade whole system
type: bool
default: no
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 C(no),
upgrades and installations will fail when xbps is not up to date.
type: bool
default: yes
version_added: '0.2.0'
'''
EXAMPLES = '''
- 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=yes
- 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=yes
- name: Update package cache
community.general.xbps: update_cache=yes
- name: Upgrade packages
community.general.xbps: upgrade=yes
- name: Install a package, failing if the xbps package is out of date
community.general.xbps:
name: foo
state: present
upgrade_xbps: no
'''
RETURN = '''
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 query_package(module, xbps_path, name, state="present"):
"""Returns Package info"""
if state == "present":
lcmd = "%s %s" % (xbps_path['query'], name)
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'])
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'])
rc, stdout, stderr = module.run_command(cmd, check_rc=False)
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'])
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'])
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)
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))
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, aliases=['update-cache'], type='bool',
deprecated_aliases=[dict(name='update-cache', version='5.0.0', collection_name='community.general')]),
upgrade_xbps=dict(default=True, 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()