ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

[PR #5718/682bb4b8 backport][stable-6] opkg: refactor module to use StateModuleHelper and CmdRunner (#5824)

Browse source 
opkg: refactor module to use StateModuleHelper and CmdRunner (#5718)

* opkg: refactor module to use StateModuleHelper and CmdRunner

* add changelog fragment

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* generate message outcome as before

* aggregated changes from 5688

* fix package query

* add unit tests

* fix sanity error

* Update plugins/modules/opkg.py

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>

* add test for specifying version

* refactor parameter name

Co-authored-by: joergho <48011876+joergho@users.noreply.github.com>
(cherry picked from commit 682bb4b88a)

Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>
This commit is contained in:
patchback[bot] 2023-01-12 21:06:11 +01:00 committed by GitHub
parent 3fd84d71b8
commit 82a9db9738
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 280 additions and 123 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
changelogs/fragments/5718-opkg-refactor.yaml Normal file
View file

@ -0,0 +1,2 @@
minor_changes:
- opkg - refactored module to use ``CmdRunner`` for executing ``opkg`` (https://github.com/ansible-collections/community.general/pull/5718).

208
plugins/modules/opkg.py
View file

@ -97,144 +97,106 @@ EXAMPLES = '''
force: overwrite
'''
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.six.moves import shlex_quote
from ansible_collections.community.general.plugins.module_utils.cmd_runner import CmdRunner, cmd_runner_fmt
from ansible_collections.community.general.plugins.module_utils.module_helper import StateModuleHelper
def update_package_db(module, opkg_path):
""" Updates packages list. """
rc, out, err = module.run_command([opkg_path, "update"])
if rc != 0:
module.fail_json(msg="could not update package db")
def query_package(module, opkg_path, name, version=None, state="present"):
""" Returns whether a package is installed or not. """
if state == "present":
rc, out, err = module.run_command([opkg_path, "list-installed", name])
if rc != 0:
return False
# variable out is one line if the package is installed:
# "NAME - VERSION - DESCRIPTION"
if version is not None:
if not out.startswith("%s - %s " % (name, version)):
return False
else:
if not out.startswith(name + " "):
return False
return True
else:
raise NotImplementedError()
def split_name_and_version(module, package):
""" Split the name and the version when using the NAME=VERSION syntax """
splitted = package.split('=', 1)
if len(splitted) == 1:
return splitted[0], None
else:
return splitted[0], splitted[1]
def remove_packages(module, opkg_path, packages):
""" Uninstalls one or more packages if installed. """
p = module.params
force = p["force"]
if force:
force = "--force-%s" % force
remove_c = 0
# Using a for loop in case of error, we can report the package that failed
for package in packages:
package, version = split_name_and_version(module, package)
# Query the package first, to see if we even need to remove
if not query_package(module, opkg_path, package):
continue
if force:
rc, out, err = module.run_command([opkg_path, "remove", force, package])
else:
rc, out, err = module.run_command([opkg_path, "remove", package])
if query_package(module, opkg_path, package):
module.fail_json(msg="failed to remove %s: %s" % (package, out))
remove_c += 1
if remove_c > 0:
module.exit_json(changed=True, msg="removed %s package(s)" % remove_c)
module.exit_json(changed=False, msg="package(s) already absent")
def install_packages(module, opkg_path, packages):
""" Installs one or more packages if not already installed. """
p = module.params
force = p["force"]
if force:
force = "--force-%s" % force
install_c = 0
for package in packages:
package, version = split_name_and_version(module, package)
if query_package(module, opkg_path, package, version) and (force != '--force-reinstall'):
continue
if version is not None:
version_str = "=%s" % version
else:
version_str = ""
if force:
rc, out, err = module.run_command([opkg_path, "install", force, package + version_str])
else:
rc, out, err = module.run_command([opkg_path, "install", package + version_str])
if not query_package(module, opkg_path, package, version):
module.fail_json(msg="failed to install %s%s: %s" % (package, version_str, out))
install_c += 1
if install_c > 0:
module.exit_json(changed=True, msg="installed %s package(s)" % (install_c))
module.exit_json(changed=False, msg="package(s) already present")
def main():
module = AnsibleModule(
class Opkg(StateModuleHelper):
module = dict(
argument_spec=dict(
name=dict(aliases=["pkg"], required=True, type="list", elements="str"),
state=dict(default="present", choices=["present", "installed", "absent", "removed"]),
force=dict(default="", choices=["", "depends", "maintainer", "reinstall", "overwrite", "downgrade", "space", "postinstall", "remove",
"checksum", "removal-of-dependent-packages"]),
update_cache=dict(default=False, type='bool'),
)
),
)
opkg_path = module.get_bin_path('opkg', True, ['/bin'])
def __init_module__(self):
self.vars.set("install_c", 0, output=False, change=True)
self.vars.set("remove_c", 0, output=False, change=True)
p = module.params
state_map = dict(
query="list-installed",
present="install",
installed="install",
absent="remove",
removed="remove",
)
if p["update_cache"]:
update_package_db(module, opkg_path)
def _force(value):
if value == "":
value = None
return cmd_runner_fmt.as_optval("--force-")(value, ctx_ignore_none=True)
pkgs = p["name"]
self.runner = CmdRunner(
self.module,
command="opkg",
arg_formats=dict(
package=cmd_runner_fmt.as_list(),
state=cmd_runner_fmt.as_map(state_map),
force=cmd_runner_fmt.as_func(_force),
update_cache=cmd_runner_fmt.as_bool("update")
),
)
if p["state"] in ["present", "installed"]:
install_packages(module, opkg_path, pkgs)
@staticmethod
def split_name_and_version(package):
""" Split the name and the version when using the NAME=VERSION syntax """
splitted = package.split('=', 1)
if len(splitted) == 1:
return splitted[0], None
else:
return splitted[0], splitted[1]
elif p["state"] in ["absent", "removed"]:
remove_packages(module, opkg_path, pkgs)
def _package_in_desired_state(self, name, want_installed, version=None):
dummy, out, dummy = self.runner("state package").run(state="query", package=name)
has_package = out.startswith(name + " - %s" % ("" if not version else (version + " ")))
return want_installed == has_package
def state_present(self):
if self.vars.update_cache:
dummy, rc, dummy = self.runner("update_cache").run()
if rc != 0:
self.do_raise("could not update package db")
with self.runner("state force package") as ctx:
for package in self.vars.name:
pkg_name, pkg_version = self.split_name_and_version(package)
if not self._package_in_desired_state(pkg_name, want_installed=True, version=pkg_version) or self.vars.force == "reinstall":
ctx.run(package=package)
if not self._package_in_desired_state(pkg_name, want_installed=True, version=pkg_version):
self.do_raise("failed to install %s" % package)
self.vars.install_c += 1
if self.vars.install_c > 0:
self.vars.msg = "installed %s package(s)" % (self.vars.install_c)
else:
self.vars.msg = "package(s) already present"
def state_absent(self):
if self.vars.update_cache:
dummy, rc, dummy = self.runner("update_cache").run()
if rc != 0:
self.do_raise("could not update package db")
with self.runner("state force package") as ctx:
for package in self.vars.name:
package, dummy = self.split_name_and_version(package)
if not self._package_in_desired_state(package, want_installed=False):
ctx.run(package=package)
if not self._package_in_desired_state(package, want_installed=False):
self.do_raise("failed to remove %s" % package)
self.vars.remove_c += 1
if self.vars.remove_c > 0:
self.vars.msg = "removed %s package(s)" % (self.vars.remove_c)
else:
self.vars.msg = "package(s) already absent"
state_installed = state_present
state_removed = state_absent
def main():
Opkg.execute()
if __name__ == '__main__':

193
tests/unit/plugins/modules/test_opkg.py Normal file
View file

@ -0,0 +1,193 @@
# -*- coding: utf-8 -*-
# Copyright (c) Alexei Znamensky (russoz@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
import json
from collections import namedtuple
from ansible_collections.community.general.plugins.modules import opkg
import pytest
TESTED_MODULE = opkg.__name__
ModuleTestCase = namedtuple("ModuleTestCase", ["id", "input", "output", "run_command_calls"])
RunCmdCall = namedtuple("RunCmdCall", ["command", "environ", "rc", "out", "err"])
@pytest.fixture
def patch_opkg(mocker):
mocker.patch('ansible.module_utils.basic.AnsibleModule.get_bin_path', return_value='/testbin/opkg')
TEST_CASES = [
ModuleTestCase(
id="install_zlibdev",
input={"name": "zlib-dev", "state": "present"},
output={
"msg": "installed 1 package(s)"
},
run_command_calls=[
RunCmdCall(
command=["/testbin/opkg", "list-installed", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out="",
err="",
),
RunCmdCall(
command=["/testbin/opkg", "install", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out=(
"Installing zlib-dev (1.2.11-6) to root..."
"Downloading https://downloads.openwrt.org/releases/22.03.0/packages/mips_24kc/base/zlib-dev_1.2.11-6_mips_24kc.ipk"
"Installing zlib (1.2.11-6) to root..."
"Downloading https://downloads.openwrt.org/releases/22.03.0/packages/mips_24kc/base/zlib_1.2.11-6_mips_24kc.ipk"
"Configuring zlib."
"Configuring zlib-dev."
),
err="",
),
RunCmdCall(
command=["/testbin/opkg", "list-installed", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out="zlib-dev - 1.2.11-6\n",
err="",
),
],
),
ModuleTestCase(
id="install_zlibdev_present",
input={"name": "zlib-dev", "state": "present"},
output={
"msg": "package(s) already present"
},
run_command_calls=[
RunCmdCall(
command=["/testbin/opkg", "list-installed", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out="zlib-dev - 1.2.11-6\n",
err="",
),
],
),
ModuleTestCase(
id="install_zlibdev_force_reinstall",
input={"name": "zlib-dev", "state": "present", "force": "reinstall"},
output={
"msg": "installed 1 package(s)"
},
run_command_calls=[
RunCmdCall(
command=["/testbin/opkg", "list-installed", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out="zlib-dev - 1.2.11-6\n",
err="",
),
RunCmdCall(
command=["/testbin/opkg", "install", "--force-reinstall", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out=(
"Installing zlib-dev (1.2.11-6) to root...\n"
"Downloading https://downloads.openwrt.org/releases/22.03.0/packages/mips_24kc/base/zlib-dev_1.2.11-6_mips_24kc.ipk\n"
"Configuring zlib-dev.\n"
),
err="",
),
RunCmdCall(
command=["/testbin/opkg", "list-installed", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out="zlib-dev - 1.2.11-6\n",
err="",
),
],
),
ModuleTestCase(
id="install_zlibdev_with_version",
input={"name": "zlib-dev=1.2.11-6", "state": "present"},
output={
"msg": "installed 1 package(s)"
},
run_command_calls=[
RunCmdCall(
command=["/testbin/opkg", "list-installed", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out="",
err="",
),
RunCmdCall(
command=["/testbin/opkg", "install", "zlib-dev=1.2.11-6"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out=(
"Installing zlib-dev (1.2.11-6) to root..."
"Downloading https://downloads.openwrt.org/releases/22.03.0/packages/mips_24kc/base/zlib-dev_1.2.11-6_mips_24kc.ipk"
"Installing zlib (1.2.11-6) to root..."
"Downloading https://downloads.openwrt.org/releases/22.03.0/packages/mips_24kc/base/zlib_1.2.11-6_mips_24kc.ipk"
"Configuring zlib."
"Configuring zlib-dev."
),
err="",
),
RunCmdCall(
command=["/testbin/opkg", "list-installed", "zlib-dev"],
environ={'environ_update': {'LANGUAGE': 'C', 'LC_ALL': 'C'}, 'check_rc': False},
rc=0,
out="zlib-dev - 1.2.11-6 \n", # This output has the extra space at the end, to satisfy the behaviour of Yocto/OpenEmbedded's opkg
err="",
),
],
),
]
TEST_CASES_IDS = [item.id for item in TEST_CASES]
@pytest.mark.parametrize('patch_ansible_module, testcase',
[[x.input, x] for x in TEST_CASES],
ids=TEST_CASES_IDS,
indirect=['patch_ansible_module'])
@pytest.mark.usefixtures('patch_ansible_module')
def test_opkg(mocker, capfd, patch_opkg, testcase):
"""
Run unit tests for test cases listen in TEST_CASES
"""
run_cmd_calls = testcase.run_command_calls
# Mock function used for running commands first
call_results = [(x.rc, x.out, x.err) for x in run_cmd_calls]
mock_run_command = mocker.patch('ansible.module_utils.basic.AnsibleModule.run_command', side_effect=call_results)
# Try to run test case
with pytest.raises(SystemExit):
opkg.main()
out, err = capfd.readouterr()
results = json.loads(out)
print("testcase =\n%s" % str(testcase))
print("results =\n%s" % results)
for test_result in testcase.output:
assert results[test_result] == testcase.output[test_result], \
"'{0}': '{1}' != '{2}'".format(test_result, results[test_result], testcase.output[test_result])
call_args_list = [(item[0][0], item[1]) for item in mock_run_command.call_args_list]
expected_call_args_list = [(item.command, item.environ) for item in run_cmd_calls]
print("call args list =\n%s" % call_args_list)
print("expected args list =\n%s" % expected_call_args_list)
assert mock_run_command.call_count == len(run_cmd_calls)
if mock_run_command.call_count:
assert call_args_list == expected_call_args_list