mirror of
https://github.com/ansible-collections/community.general.git
synced 2024-09-14 20:13:21 +02:00
* Initial Commit
* Adding changelog fragment
* Ensured params are present during verbose output and enhanced check_mode
* Making specific to builtins
* Removing unneccessary external call
* Acutal bugfix
(cherry picked from commit d180390dbc)
Co-authored-by: Ajpantuso <ajpantuso@gmail.com>
This commit is contained in:
patchback[bot]
GitHub
parent
48b1bc7d47
commit
0760f60ca5
3 changed files with 263 additions and 53 deletions
|
|
@ -0,0 +1,3 @@
|
|||
---
|
||||
bugfixes:
|
||||
- modprobe - added additional checks to ensure module load/unload is effective (https://github.com/ansible-collections/community.general/issues/1608).
|
||||
|
|
@ -50,11 +50,90 @@ EXAMPLES = '''
|
|||
'''
|
||||
|
||||
import os.path
|
||||
import platform
|
||||
import shlex
|
||||
import traceback
|
||||
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible.module_utils._text import to_native
|
||||
from ansible.module_utils.common.text.converters import to_native
|
||||
|
||||
RELEASE_VER = platform.release()
|
||||
|
||||
|
||||
class Modprobe(object):
|
||||
def __init__(self, module):
|
||||
self.module = module
|
||||
self.modprobe_bin = module.get_bin_path('modprobe', True)
|
||||
|
||||
self.check_mode = module.check_mode
|
||||
self.desired_state = module.params['state']
|
||||
self.name = module.params['name']
|
||||
self.params = module.params['params']
|
||||
|
||||
self.changed = False
|
||||
|
||||
def load_module(self):
|
||||
command = [self.modprobe_bin]
|
||||
if self.check_mode:
|
||||
command.append('-n')
|
||||
command.extend([self.name] + shlex.split(self.params))
|
||||
|
||||
rc, out, err = self.module.run_command(command)
|
||||
|
||||
if rc != 0:
|
||||
return self.module.fail_json(msg=err, rc=rc, stdout=out, stderr=err, **self.result)
|
||||
|
||||
if self.check_mode or self.module_loaded():
|
||||
self.changed = True
|
||||
else:
|
||||
rc, stdout, stderr = self.module.run_command(
|
||||
[self.modprobe_bin, '-n', '--first-time', self.name] + shlex.split(self.params)
|
||||
)
|
||||
if rc != 0:
|
||||
self.module.warn(stderr)
|
||||
|
||||
def module_loaded(self):
|
||||
is_loaded = False
|
||||
try:
|
||||
with open('/proc/modules') as modules:
|
||||
module_name = self.name.replace('-', '_') + ' '
|
||||
for line in modules:
|
||||
if line.startswith(module_name):
|
||||
is_loaded = True
|
||||
break
|
||||
|
||||
if not is_loaded:
|
||||
module_file = '/' + self.name + '.ko'
|
||||
builtin_path = os.path.join('/lib/modules/', RELEASE_VER, 'modules.builtin')
|
||||
with open(builtin_path) as builtins:
|
||||
for line in builtins:
|
||||
if line.rstrip().endswith(module_file):
|
||||
is_loaded = True
|
||||
break
|
||||
except (IOError, OSError) as e:
|
||||
self.module.fail_json(msg=to_native(e), exception=traceback.format_exc(), **self.result)
|
||||
|
||||
return is_loaded
|
||||
|
||||
def unload_module(self):
|
||||
command = [self.modprobe_bin, '-r', self.name]
|
||||
if self.check_mode:
|
||||
command.append('-n')
|
||||
|
||||
rc, out, err = self.module.run_command(command)
|
||||
if rc != 0:
|
||||
return self.module.fail_json(msg=err, rc=rc, stdout=out, stderr=err, **self.result)
|
||||
|
||||
self.changed = True
|
||||
|
||||
@property
|
||||
def result(self):
|
||||
return {
|
||||
'changed': self.changed,
|
||||
'name': self.name,
|
||||
'params': self.params,
|
||||
'state': self.desired_state,
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
|
|
@ -67,60 +146,14 @@ def main():
|
|||
supports_check_mode=True,
|
||||
)
|
||||
|
||||
name = module.params['name']
|
||||
params = module.params['params']
|
||||
state = module.params['state']
|
||||
modprobe = Modprobe(module)
|
||||
|
||||
# FIXME: Adding all parameters as result values is useless
|
||||
result = dict(
|
||||
changed=False,
|
||||
name=name,
|
||||
params=params,
|
||||
state=state,
|
||||
)
|
||||
if modprobe.desired_state == 'present' and not modprobe.module_loaded():
|
||||
modprobe.load_module()
|
||||
elif modprobe.desired_state == 'absent' and modprobe.module_loaded():
|
||||
modprobe.unload_module()
|
||||
|
||||
# Check if module is present
|
||||
try:
|
||||
present = False
|
||||
with open('/proc/modules') as modules:
|
||||
module_name = name.replace('-', '_') + ' '
|
||||
for line in modules:
|
||||
if line.startswith(module_name):
|
||||
present = True
|
||||
break
|
||||
if not present:
|
||||
command = [module.get_bin_path('uname', True), '-r']
|
||||
rc, uname_kernel_release, err = module.run_command(command)
|
||||
module_file = '/' + name + '.ko'
|
||||
builtin_path = os.path.join('/lib/modules/', uname_kernel_release.strip(),
|
||||
'modules.builtin')
|
||||
with open(builtin_path) as builtins:
|
||||
for line in builtins:
|
||||
if line.endswith(module_file):
|
||||
present = True
|
||||
break
|
||||
except IOError as e:
|
||||
module.fail_json(msg=to_native(e), exception=traceback.format_exc(), **result)
|
||||
|
||||
# Add/remove module as needed
|
||||
if state == 'present':
|
||||
if not present:
|
||||
if not module.check_mode:
|
||||
command = [module.get_bin_path('modprobe', True), name]
|
||||
command.extend(shlex.split(params))
|
||||
rc, out, err = module.run_command(command)
|
||||
if rc != 0:
|
||||
module.fail_json(msg=err, rc=rc, stdout=out, stderr=err, **result)
|
||||
result['changed'] = True
|
||||
elif state == 'absent':
|
||||
if present:
|
||||
if not module.check_mode:
|
||||
rc, out, err = module.run_command([module.get_bin_path('modprobe', True), '-r', name])
|
||||
if rc != 0:
|
||||
module.fail_json(msg=err, rc=rc, stdout=out, stderr=err, **result)
|
||||
result['changed'] = True
|
||||
|
||||
module.exit_json(**result)
|
||||
module.exit_json(**modprobe.result)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
174
tests/unit/plugins/modules/system/test_modprobe.py
Normal file
174
tests/unit/plugins/modules/system/test_modprobe.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# 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
|
||||
|
||||
from ansible_collections.community.general.tests.unit.plugins.modules.utils import ModuleTestCase, set_module_args
|
||||
from ansible_collections.community.general.tests.unit.compat.mock import patch
|
||||
from ansible_collections.community.general.tests.unit.compat.mock import Mock
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible_collections.community.general.plugins.modules.system.modprobe import Modprobe
|
||||
|
||||
|
||||
class TestLoadModule(ModuleTestCase):
|
||||
def setUp(self):
|
||||
super(TestLoadModule, self).setUp()
|
||||
|
||||
self.mock_module_loaded = patch(
|
||||
'ansible_collections.community.general.plugins.modules.system.modprobe.Modprobe.module_loaded'
|
||||
)
|
||||
self.mock_run_command = patch('ansible.module_utils.basic.AnsibleModule.run_command')
|
||||
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
|
||||
|
||||
self.module_loaded = self.mock_module_loaded.start()
|
||||
self.run_command = self.mock_run_command.start()
|
||||
self.get_bin_path = self.mock_get_bin_path.start()
|
||||
|
||||
def tearDown(self):
|
||||
"""Teardown."""
|
||||
super(TestLoadModule, self).tearDown()
|
||||
self.mock_module_loaded.stop()
|
||||
self.mock_run_command.stop()
|
||||
self.mock_get_bin_path.stop()
|
||||
|
||||
def test_load_module_success(self):
|
||||
set_module_args(dict(
|
||||
name='test',
|
||||
state='present',
|
||||
))
|
||||
|
||||
module = AnsibleModule(
|
||||
argument_spec=dict(
|
||||
name=dict(type='str', required=True),
|
||||
state=dict(type='str', default='present', choices=['absent', 'present']),
|
||||
params=dict(type='str', default=''),
|
||||
),
|
||||
supports_check_mode=True,
|
||||
)
|
||||
|
||||
self.get_bin_path.side_effect = ['modprobe']
|
||||
self.module_loaded.side_effect = [True]
|
||||
self.run_command.side_effect = [(0, '', '')]
|
||||
|
||||
modprobe = Modprobe(module)
|
||||
modprobe.load_module()
|
||||
|
||||
assert modprobe.result == {
|
||||
'changed': True,
|
||||
'name': 'test',
|
||||
'params': '',
|
||||
'state': 'present',
|
||||
}
|
||||
|
||||
def test_load_module_unchanged(self):
|
||||
set_module_args(dict(
|
||||
name='test',
|
||||
state='present',
|
||||
))
|
||||
|
||||
module = AnsibleModule(
|
||||
argument_spec=dict(
|
||||
name=dict(type='str', required=True),
|
||||
state=dict(type='str', default='present', choices=['absent', 'present']),
|
||||
params=dict(type='str', default=''),
|
||||
),
|
||||
supports_check_mode=True,
|
||||
)
|
||||
|
||||
module.warn = Mock()
|
||||
|
||||
self.get_bin_path.side_effect = ['modprobe']
|
||||
self.module_loaded.side_effect = [False]
|
||||
self.run_command.side_effect = [(0, '', ''), (1, '', '')]
|
||||
|
||||
modprobe = Modprobe(module)
|
||||
modprobe.load_module()
|
||||
|
||||
module.warn.assert_called_once_with('')
|
||||
|
||||
|
||||
class TestUnloadModule(ModuleTestCase):
|
||||
def setUp(self):
|
||||
super(TestUnloadModule, self).setUp()
|
||||
|
||||
self.mock_module_loaded = patch(
|
||||
'ansible_collections.community.general.plugins.modules.system.modprobe.Modprobe.module_loaded'
|
||||
)
|
||||
self.mock_run_command = patch('ansible.module_utils.basic.AnsibleModule.run_command')
|
||||
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
|
||||
|
||||
self.module_loaded = self.mock_module_loaded.start()
|
||||
self.run_command = self.mock_run_command.start()
|
||||
self.get_bin_path = self.mock_get_bin_path.start()
|
||||
|
||||
def tearDown(self):
|
||||
"""Teardown."""
|
||||
super(TestUnloadModule, self).tearDown()
|
||||
self.mock_module_loaded.stop()
|
||||
self.mock_run_command.stop()
|
||||
self.mock_get_bin_path.stop()
|
||||
|
||||
def test_unload_module_success(self):
|
||||
set_module_args(dict(
|
||||
name='test',
|
||||
state='absent',
|
||||
))
|
||||
|
||||
module = AnsibleModule(
|
||||
argument_spec=dict(
|
||||
name=dict(type='str', required=True),
|
||||
state=dict(type='str', default='present', choices=['absent', 'present']),
|
||||
params=dict(type='str', default=''),
|
||||
),
|
||||
supports_check_mode=True,
|
||||
)
|
||||
|
||||
self.get_bin_path.side_effect = ['modprobe']
|
||||
self.module_loaded.side_effect = [False]
|
||||
self.run_command.side_effect = [(0, '', '')]
|
||||
|
||||
modprobe = Modprobe(module)
|
||||
modprobe.unload_module()
|
||||
|
||||
assert modprobe.result == {
|
||||
'changed': True,
|
||||
'name': 'test',
|
||||
'params': '',
|
||||
'state': 'absent',
|
||||
}
|
||||
|
||||
def test_unload_module_failure(self):
|
||||
set_module_args(dict(
|
||||
name='test',
|
||||
state='absent',
|
||||
))
|
||||
|
||||
module = AnsibleModule(
|
||||
argument_spec=dict(
|
||||
name=dict(type='str', required=True),
|
||||
state=dict(type='str', default='present', choices=['absent', 'present']),
|
||||
params=dict(type='str', default=''),
|
||||
),
|
||||
supports_check_mode=True,
|
||||
)
|
||||
|
||||
module.fail_json = Mock()
|
||||
|
||||
self.get_bin_path.side_effect = ['modprobe']
|
||||
self.module_loaded.side_effect = [True]
|
||||
self.run_command.side_effect = [(1, '', '')]
|
||||
|
||||
modprobe = Modprobe(module)
|
||||
modprobe.unload_module()
|
||||
|
||||
dummy_result = {
|
||||
'changed': False,
|
||||
'name': 'test',
|
||||
'state': 'absent',
|
||||
'params': '',
|
||||
}
|
||||
|
||||
module.fail_json.assert_called_once_with(
|
||||
msg='', rc=1, stdout='', stderr='', **dummy_result
|
||||
)
|
||||
Loading…
Reference in a new issue