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

add persistent option for modprobe (#5424)

* add persistent option for modprobe

* add suggested changes + fix broken test

* change modprobe module path in tests due to rebase

* change persistent option type from bool to str with choices

* fix unused import

* add example with persistent option

* fix some minor issues after review

- move regexps compiling to __init__
- move AnsibleModule to build_module function and use this function in tests instead of AnsibleModule
- fix terminlogy issue in documentation

* fix unused-import
This commit is contained in:
Alex Groshev 2023-02-26 14:58:58 +01:00 committed by GitHub
parent 617be6e124
commit 29f5033737
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 500 additions and 36 deletions

View file

@ -0,0 +1,3 @@
---
minor_changes:
- modprobe - add ``persistent`` option (https://github.com/ansible-collections/community.general/issues/4028, https://github.com/ansible-collections/community.general/pull/542).

View file

@ -42,6 +42,21 @@ options:
description:
- Modules parameters.
default: ''
persistent:
type: str
choices: [ disabled, absent, present ]
default: disabled
description:
- Persistency between reboots for configured module.
- This option creates files in C(/etc/modules-load.d/) and C(/etc/modprobe.d/) that make your module configuration persistent during reboots.
- If C(present), adds module name to C(/etc/modules-load.d/) and params to C(/etc/modprobe.d/) so the module will be loaded on next reboot.
- If C(absent), will comment out module name from C(/etc/modules-load.d/) and comment out params from C(/etc/modprobe.d/) so the module will not be
loaded on next reboot.
- If C(disabled), will not toch anything and leave C(/etc/modules-load.d/) and C(/etc/modprobe.d/) as it is.
- Note that it is usually a better idea to rely on the automatic module loading by PCI IDs, USB IDs, DMI IDs or similar triggers encoded in the
kernel modules themselves instead of configuration like this.
- In fact, most modern kernel modules are prepared for automatic loading already.
- "B(Note:) This option works only with distributions that use C(systemd) when set to values other than C(disabled)."
'''
EXAMPLES = '''
@ -55,20 +70,31 @@ EXAMPLES = '''
name: dummy
state: present
params: 'numdummies=2'
- name: Add the dummy module and make sure it is loaded after reboots
community.general.modprobe:
name: dummy
state: present
params: 'numdummies=2'
persistent: present
'''
import os.path
import platform
import shlex
import traceback
import re
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_native
RELEASE_VER = platform.release()
MODULES_LOAD_LOCATION = '/etc/modules-load.d'
PARAMETERS_FILES_LOCATION = '/etc/modprobe.d'
class Modprobe(object):
def __init__(self, module):
self.module = module
self.modprobe_bin = module.get_bin_path('modprobe', True)
@ -77,9 +103,14 @@ class Modprobe(object):
self.desired_state = module.params['state']
self.name = module.params['name']
self.params = module.params['params']
self.persistent = module.params['persistent']
self.changed = False
self.re_find_module = re.compile(r'^ *{0} *(?:[#;].*)?\n?\Z'.format(self.name))
self.re_find_params = re.compile(r'^options {0} \w+=\S+ *(?:[#;].*)?\n?\Z'.format(self.name))
self.re_get_params_and_values = re.compile(r'^options {0} (\w+=\S+) *(?:[#;].*)?\n?\Z'.format(self.name))
def load_module(self):
command = [self.modprobe_bin]
if self.check_mode:
@ -100,6 +131,117 @@ class Modprobe(object):
if rc != 0:
self.module.warn(stderr)
@property
def module_is_loaded_persistently(self):
for module_file in self.modules_files:
with open(module_file) as file:
for line in file:
if self.re_find_module.match(line):
return True
return False
@property
def params_is_set(self):
desired_params = set(self.params.split())
return desired_params == self.permanent_params
@property
def permanent_params(self):
params = set()
for modprobe_file in self.modprobe_files:
with open(modprobe_file) as file:
for line in file:
match = self.re_get_params_and_values.match(line)
if match:
params.add(match.group(1))
return params
def create_module_file(self):
file_path = os.path.join(MODULES_LOAD_LOCATION,
self.name + '.conf')
with open(file_path, 'w') as file:
file.write(self.name + '\n')
@property
def module_options_file_content(self):
file_content = ['options {0} {1}'.format(self.name, param)
for param in self.params.split()]
return '\n'.join(file_content) + '\n'
def create_module_options_file(self):
new_file_path = os.path.join(PARAMETERS_FILES_LOCATION,
self.name + '.conf')
with open(new_file_path, 'w') as file:
file.write(self.module_options_file_content)
def disable_old_params(self):
for modprobe_file in self.modprobe_files:
with open(modprobe_file) as file:
file_content = file.readlines()
content_changed = False
for index, line in enumerate(file_content):
if self.re_find_params.match(line):
file_content[index] = '#' + line
content_changed = True
if content_changed:
with open(modprobe_file, 'w') as file:
file.write('\n'.join(file_content))
def disable_module_permanent(self):
for module_file in self.modules_files:
with open(module_file) as file:
file_content = file.readlines()
content_changed = False
for index, line in enumerate(file_content):
if self.re_find_module.match(line):
file_content[index] = '#' + line
content_changed = True
if content_changed:
with open(module_file, 'w') as file:
file.write('\n'.join(file_content))
def load_module_permanent(self):
if not self.module_is_loaded_persistently:
self.create_module_file()
self.changed = True
if not self.params_is_set:
self.disable_old_params()
self.create_module_options_file()
self.changed = True
def unload_module_permanent(self):
if self.module_is_loaded_persistently:
self.disable_module_permanent()
self.changed = True
if self.permanent_params:
self.disable_old_params()
self.changed = True
@property
def modules_files(self):
modules_paths = [os.path.join(MODULES_LOAD_LOCATION, path)
for path in os.listdir(MODULES_LOAD_LOCATION)]
return [path for path in modules_paths if os.path.isfile(path)]
@property
def modprobe_files(self):
modules_paths = [os.path.join(PARAMETERS_FILES_LOCATION, path)
for path in os.listdir(PARAMETERS_FILES_LOCATION)]
return [path for path in modules_paths if os.path.isfile(path)]
def module_loaded(self):
is_loaded = False
try:
@ -144,16 +286,21 @@ class Modprobe(object):
}
def main():
module = AnsibleModule(
def build_module():
return AnsibleModule(
argument_spec=dict(
name=dict(type='str', required=True),
state=dict(type='str', default='present', choices=['absent', 'present']),
params=dict(type='str', default=''),
persistent=dict(type='str', default='disabled', choices=['disabled', 'present', 'absent']),
),
supports_check_mode=True,
)
def main():
module = build_module()
modprobe = Modprobe(module)
if modprobe.desired_state == 'present' and not modprobe.module_loaded():
@ -161,6 +308,11 @@ def main():
elif modprobe.desired_state == 'absent' and modprobe.module_loaded():
modprobe.unload_module()
if modprobe.persistent == 'present' and not (modprobe.module_is_loaded_persistently and modprobe.params_is_set):
modprobe.load_module_permanent()
elif modprobe.persistent == 'absent' and (modprobe.module_is_loaded_persistently or modprobe.permanent_params):
modprobe.unload_module_permanent()
module.exit_json(**modprobe.result)

View file

@ -6,11 +6,12 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import sys
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.modprobe import Modprobe
from ansible_collections.community.general.tests.unit.compat.mock import mock_open
from ansible_collections.community.general.plugins.modules.modprobe import Modprobe, build_module
class TestLoadModule(ModuleTestCase):
@ -40,14 +41,7 @@ class TestLoadModule(ModuleTestCase):
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 = build_module()
self.get_bin_path.side_effect = ['modprobe']
self.module_loaded.side_effect = [True]
@ -69,14 +63,7 @@ class TestLoadModule(ModuleTestCase):
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 = build_module()
module.warn = Mock()
@ -117,14 +104,7 @@ class TestUnloadModule(ModuleTestCase):
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 = build_module()
self.get_bin_path.side_effect = ['modprobe']
self.module_loaded.side_effect = [False]
@ -146,14 +126,7 @@ class TestUnloadModule(ModuleTestCase):
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 = build_module()
module.fail_json = Mock()
@ -174,3 +147,339 @@ class TestUnloadModule(ModuleTestCase):
module.fail_json.assert_called_once_with(
msg='', rc=1, stdout='', stderr='', **dummy_result
)
class TestModuleIsLoadedPersistently(ModuleTestCase):
def setUp(self):
if (sys.version_info[0] == 3 and sys.version_info[1] < 7) or (sys.version_info[0] == 2 and sys.version_info[1] < 7):
self.skipTest('open_mock doesnt support readline in earlier python versions')
super(TestModuleIsLoadedPersistently, self).setUp()
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
self.get_bin_path = self.mock_get_bin_path.start()
def tearDown(self):
"""Teardown."""
super(TestModuleIsLoadedPersistently, self).tearDown()
self.mock_get_bin_path.stop()
def test_module_is_loaded(self):
set_module_args(dict(
name='dummy',
state='present',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open(read_data='dummy')) as mocked_file:
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modules_files'):
modprobe.modules_files = ['/etc/modules-load.d/dummy.conf']
assert modprobe.module_is_loaded_persistently
mocked_file.assert_called_once_with('/etc/modules-load.d/dummy.conf')
def test_module_is_not_loaded_empty_file(self):
set_module_args(dict(
name='dummy',
state='present',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open(read_data='')) as mocked_file:
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modules_files'):
modprobe.modules_files = ['/etc/modules-load.d/dummy.conf']
assert not modprobe.module_is_loaded_persistently
mocked_file.assert_called_once_with('/etc/modules-load.d/dummy.conf')
def test_module_is_not_loaded_no_files(self):
set_module_args(dict(
name='dummy',
state='present',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modules_files'):
modprobe.modules_files = []
assert not modprobe.module_is_loaded_persistently
class TestPermanentParams(ModuleTestCase):
def setUp(self):
if (sys.version_info[0] == 3 and sys.version_info[1] < 7) or (sys.version_info[0] == 2 and sys.version_info[1] < 7):
self.skipTest('open_mock doesnt support readline in earlier python versions')
super(TestPermanentParams, self).setUp()
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
self.get_bin_path = self.mock_get_bin_path.start()
def tearDown(self):
"""Teardown."""
super(TestPermanentParams, self).tearDown()
self.mock_get_bin_path.stop()
def test_module_permanent_params_exist(self):
files_content = [
'options dummy numdummies=4\noptions dummy dummy_parameter1=6',
'options dummy dummy_parameter2=5 #Comment\noptions notdummy notdummy_param=5'
]
mock_files_content = [mock_open(read_data=content).return_value for content in files_content]
set_module_args(dict(
name='dummy',
state='present',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open()) as mocked_file:
mocked_file.side_effect = mock_files_content
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modprobe_files'):
modprobe.modprobe_files = ['/etc/modprobe.d/dummy1.conf', '/etc/modprobe.d/dummy2.conf']
assert modprobe.permanent_params == set(['numdummies=4', 'dummy_parameter1=6', 'dummy_parameter2=5'])
def test_module_permanent_params_empty(self):
files_content = [
'',
''
]
mock_files_content = [mock_open(read_data=content).return_value for content in files_content]
set_module_args(dict(
name='dummy',
state='present',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open(read_data='')) as mocked_file:
mocked_file.side_effect = mock_files_content
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modprobe_files'):
modprobe.modprobe_files = ['/etc/modprobe.d/dummy1.conf', '/etc/modprobe.d/dummy2.conf']
assert modprobe.permanent_params == set()
class TestCreateModuleFIle(ModuleTestCase):
def setUp(self):
super(TestCreateModuleFIle, self).setUp()
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
self.get_bin_path = self.mock_get_bin_path.start()
def tearDown(self):
"""Teardown."""
super(TestCreateModuleFIle, self).tearDown()
self.mock_get_bin_path.stop()
def test_create_file(self):
set_module_args(dict(
name='dummy',
state='present',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open()) as mocked_file:
modprobe.create_module_file()
mocked_file.assert_called_once_with('/etc/modules-load.d/dummy.conf', 'w')
mocked_file().write.assert_called_once_with('dummy\n')
class TestCreateModuleOptionsFIle(ModuleTestCase):
def setUp(self):
super(TestCreateModuleOptionsFIle, self).setUp()
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
self.get_bin_path = self.mock_get_bin_path.start()
def tearDown(self):
"""Teardown."""
super(TestCreateModuleOptionsFIle, self).tearDown()
self.mock_get_bin_path.stop()
def test_create_file(self):
set_module_args(dict(
name='dummy',
state='present',
params='numdummies=4',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open()) as mocked_file:
modprobe.create_module_options_file()
mocked_file.assert_called_once_with('/etc/modprobe.d/dummy.conf', 'w')
mocked_file().write.assert_called_once_with('options dummy numdummies=4\n')
class TestDisableOldParams(ModuleTestCase):
def setUp(self):
super(TestDisableOldParams, self).setUp()
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
self.get_bin_path = self.mock_get_bin_path.start()
def tearDown(self):
"""Teardown."""
super(TestDisableOldParams, self).tearDown()
self.mock_get_bin_path.stop()
def test_disable_old_params_file_changed(self):
mock_data = 'options dummy numdummies=4'
set_module_args(dict(
name='dummy',
state='present',
params='numdummies=4',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open(read_data=mock_data)) as mocked_file:
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modprobe_files'):
modprobe.modprobe_files = ['/etc/modprobe.d/dummy1.conf']
modprobe.disable_old_params()
mocked_file.assert_called_with('/etc/modprobe.d/dummy1.conf', 'w')
mocked_file().write.assert_called_once_with('#options dummy numdummies=4')
def test_disable_old_params_file_unchanged(self):
mock_data = 'options notdummy numdummies=4'
set_module_args(dict(
name='dummy',
state='present',
params='numdummies=4',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open(read_data=mock_data)) as mocked_file:
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modprobe_files'):
modprobe.modprobe_files = ['/etc/modprobe.d/dummy1.conf']
modprobe.disable_old_params()
mocked_file.assert_called_once_with('/etc/modprobe.d/dummy1.conf')
class TestDisableModulePermanent(ModuleTestCase):
def setUp(self):
super(TestDisableModulePermanent, self).setUp()
self.mock_get_bin_path = patch('ansible.module_utils.basic.AnsibleModule.get_bin_path')
self.get_bin_path = self.mock_get_bin_path.start()
def tearDown(self):
"""Teardown."""
super(TestDisableModulePermanent, self).tearDown()
self.mock_get_bin_path.stop()
def test_disable_module_permanent_file_changed(self):
set_module_args(dict(
name='dummy',
state='present',
params='numdummies=4',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open(read_data='dummy')) as mocked_file:
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modules_files'):
modprobe.modules_files = ['/etc/modules-load.d/dummy.conf']
modprobe.disable_module_permanent()
mocked_file.assert_called_with('/etc/modules-load.d/dummy.conf', 'w')
mocked_file().write.assert_called_once_with('#dummy')
def test_disable_module_permanent_file_unchanged(self):
set_module_args(dict(
name='dummy',
state='present',
params='numdummies=4',
persistent='present'
))
module = build_module()
self.get_bin_path.side_effect = ['modprobe']
modprobe = Modprobe(module)
with patch('ansible_collections.community.general.plugins.modules.modprobe.open', mock_open(read_data='notdummy')) as mocked_file:
with patch('ansible_collections.community.general.plugins.modules.modprobe.Modprobe.modules_files'):
modprobe.modules_files = ['/etc/modules-load.d/dummy.conf']
modprobe.disable_module_permanent()
mocked_file.assert_called_once_with('/etc/modules-load.d/dummy.conf')