ansible-collections/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2025-04-23 19:01:26 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

Interfaces_file - improvements (#3328) (#3396)

Browse source 
* pythonific!! no camel cases, bitte

* simplified iface attributes parsing

* some improvements, passing tests

* simplified set_interface_option()

* further simplifications

* remove unreachable stmt

* pythonified a file open

* added changelog fragment

* adjustment per PR

* PR: fixed the auto- case

* PR: added testcase and chglog frag for the misleading change report

* extra line removed

* integration is not destructive

(cherry picked from commit 7aae8d5386)

Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>
This commit is contained in:
patchback[bot] 2021-09-19 14:09:07 +02:00 • committed by GitHub
parent bc82fe36be
commit a86f31ac0f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 123 additions and 105 deletions
Download patch file Download diff file Expand all files Collapse all files

161
plugins/modules/system/interfaces_file.py
View file

@ -148,57 +148,48 @@ from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_bytes
def lineDict(line):
def line_dict(line):
return {'line': line, 'line_type': 'unknown'}
def optionDict(line, iface, option, value, address_family):
def make_option_dict(line, iface, option, value, address_family):
return {'line': line, 'iface': iface, 'option': option, 'value': value, 'line_type': 'option', 'address_family': address_family}
def getValueFromLine(s):
spaceRe = re.compile(r'\s+')
for m in spaceRe.finditer(s):
pass
valueEnd = m.start()
option = s.split()[0]
optionStart = s.find(option)
optionLen = len(option)
valueStart = re.search(r'\s', s[optionLen + optionStart:]).end() + optionLen + optionStart
return s[valueStart:valueEnd]
def get_option_value(line):
patt = re.compile(r'^\s+(?P<option>\S+)\s+(?P<value>\S?.*\S)\s*$')
match = patt.match(line)
if not match:
return None, None
return match.group("option"), match.group("value")
def read_interfaces_file(module, filename):
f = open(filename, 'r')
return read_interfaces_lines(module, f)
with open(filename, 'r') as f:
return read_interfaces_lines(module, f)
def _is_line_processing_none(first_word):
return first_word in ("source", "source-dir", "source-directory", "auto", "no-auto-down", "no-scripts") or first_word.startswith("auto-")
def read_interfaces_lines(module, line_strings):
lines = []
ifaces = {}
iface_name = None
address_family = None
currif = {}
currently_processing = None
i = 0
for line in line_strings:
i += 1
for i, line in enumerate(line_strings):
words = line.split()
if len(words) < 1:
lines.append(lineDict(line))
continue
if words[0][0] == "#":
lines.append(lineDict(line))
if not words or words[0].startswith("#"):
lines.append(line_dict(line))
continue
if words[0] == "mapping":
# currmap = calloc(1, sizeof *currmap);
lines.append(lineDict(line))
lines.append(line_dict(line))
currently_processing = "MAPPING"
elif words[0] == "source":
lines.append(lineDict(line))
currently_processing = "NONE"
elif words[0] == "source-dir":
lines.append(lineDict(line))
currently_processing = "NONE"
elif words[0] == "source-directory":
lines.append(lineDict(line))
elif _is_line_processing_none(words[0]):
lines.append(line_dict(line))
currently_processing = "NONE"
elif words[0] == "iface":
currif = {
@ -221,39 +212,26 @@ def read_interfaces_lines(module, line_strings):
ifaces[iface_name] = currif
lines.append({'line': line, 'iface': iface_name, 'line_type': 'iface', 'params': currif, 'address_family': address_family})
currently_processing = "IFACE"
elif words[0] == "auto":
lines.append(lineDict(line))
currently_processing = "NONE"
elif words[0].startswith("allow-"):
lines.append(lineDict(line))
currently_processing = "NONE"
elif words[0] == "no-auto-down":
lines.append(lineDict(line))
currently_processing = "NONE"
elif words[0] == "no-scripts":
lines.append(lineDict(line))
currently_processing = "NONE"
else:
if currently_processing == "IFACE":
option_name = words[0]
option_name, value = get_option_value(line)
# TODO: if option_name in currif.options
value = getValueFromLine(line)
lines.append(optionDict(line, iface_name, option_name, value, address_family))
lines.append(make_option_dict(line, iface_name, option_name, value, address_family))
if option_name in ["pre-up", "up", "down", "post-up"]:
currif[option_name].append(value)
else:
currif[option_name] = value
elif currently_processing == "MAPPING":
lines.append(lineDict(line))
lines.append(line_dict(line))
elif currently_processing == "NONE":
lines.append(lineDict(line))
lines.append(line_dict(line))
else:
module.fail_json(msg="misplaced option %s in line %d" % (line, i))
return None, None
module.fail_json(msg="misplaced option %s in line %d" % (line, i + 1))
return lines, ifaces
def setInterfaceOption(module, lines, iface, option, raw_value, state, address_family=None):
def set_interface_option(module, lines, iface, option, raw_value, state, address_family=None):
value = str(raw_value)
changed = False
@ -262,57 +240,54 @@ def setInterfaceOption(module, lines, iface, option, raw_value, state, address_f
iface_lines = [item for item in iface_lines
if "address_family" in item and item["address_family"] == address_family]
if len(iface_lines) < 1:
if not iface_lines:
# interface not found
module.fail_json(msg="Error: interface %s not found" % iface)
return changed, None
iface_options = list(filter(lambda i: i['line_type'] == 'option', iface_lines))
target_options = list(filter(lambda i: i['option'] == option, iface_options))
iface_options = [il for il in iface_lines if il['line_type'] == 'option']
target_options = [io for io in iface_options if io['option'] == option]
if state == "present":
if len(target_options) < 1:
changed = True
if not target_options:
# add new option
last_line_dict = iface_lines[-1]
changed, lines = addOptionAfterLine(option, value, iface, lines, last_line_dict, iface_options, address_family)
else:
if option in ["pre-up", "up", "down", "post-up"]:
if len(list(filter(lambda i: i['value'] == value, target_options))) < 1:
changed, lines = addOptionAfterLine(option, value, iface, lines, target_options[-1], iface_options, address_family)
else:
# if more than one option found edit the last one
if target_options[-1]['value'] != value:
changed = True
target_option = target_options[-1]
old_line = target_option['line']
old_value = target_option['value']
address_family = target_option['address_family']
prefix_start = old_line.find(option)
optionLen = len(option)
old_value_position = re.search(r"\s+".join(map(re.escape, old_value.split())), old_line[prefix_start + optionLen:])
start = old_value_position.start() + prefix_start + optionLen
end = old_value_position.end() + prefix_start + optionLen
line = old_line[:start] + value + old_line[end:]
index = len(lines) - lines[::-1].index(target_option) - 1
lines[index] = optionDict(line, iface, option, value, address_family)
elif state == "absent":
if len(target_options) >= 1:
return add_option_after_line(option, value, iface, lines, last_line_dict, iface_options, address_family)
if option in ["pre-up", "up", "down", "post-up"] and all(ito for ito in target_options if ito['value'] != value):
return add_option_after_line(option, value, iface, lines, target_options[-1], iface_options, address_family)
# if more than one option found edit the last one
if target_options[-1]['value'] != value:
changed = True
target_option = target_options[-1]
old_line = target_option['line']
old_value = target_option['value']
address_family = target_option['address_family']
prefix_start = old_line.find(option)
option_len = len(option)
old_value_position = re.search(r"\s+".join(map(re.escape, old_value.split())), old_line[prefix_start + option_len:])
start = old_value_position.start() + prefix_start + option_len
end = old_value_position.end() + prefix_start + option_len
line = old_line[:start] + value + old_line[end:]
index = len(lines) - lines[::-1].index(target_option) - 1
lines[index] = make_option_dict(line, iface, option, value, address_family)
return changed, lines
if state == "absent":
if target_options:
if option in ["pre-up", "up", "down", "post-up"] and value is not None and value != "None":
for target_option in filter(lambda i: i['value'] == value, target_options):
for target_option in [ito for ito in target_options if ito['value'] == value]:
changed = True
lines = list(filter(lambda ln: ln != target_option, lines))
lines = [ln for ln in lines if ln != target_option]
else:
changed = True
for target_option in target_options:
lines = list(filter(lambda ln: ln != target_option, lines))
else:
module.fail_json(msg="Error: unsupported state %s, has to be either present or absent" % state)
lines = [ln for ln in lines if ln != target_option]
return changed, lines
def addOptionAfterLine(option, value, iface, lines, last_line_dict, iface_options, address_family):
def add_option_after_line(option, value, iface, lines, last_line_dict, iface_options, address_family):
# Changing method of interface is not an addition
if option == 'method':
changed = False
@ -328,23 +303,21 @@ def addOptionAfterLine(option, value, iface, lines, last_line_dict, iface_option
suffix_start = last_line.rfind(last_line.split()[-1]) + len(last_line.split()[-1])
prefix = last_line[:prefix_start]
if len(iface_options) < 1:
if not iface_options:
# interface has no options, ident
prefix += " "
line = prefix + "%s %s" % (option, value) + last_line[suffix_start:]
option_dict = optionDict(line, iface, option, value, address_family)
option_dict = make_option_dict(line, iface, option, value, address_family)
index = len(lines) - lines[::-1].index(last_line_dict)
lines.insert(index, option_dict)
return True, lines
def write_changes(module, lines, dest):
tmpfd, tmpfile = tempfile.mkstemp()
f = os.fdopen(tmpfd, 'wb')
f.write(to_bytes(''.join(lines), errors='surrogate_or_strict'))
f.close()
with os.fdopen(tmpfd, 'wb') as f:
f.write(to_bytes(''.join(lines), errors='surrogate_or_strict'))
module.atomic_move(tmpfile, os.path.realpath(dest))
@ -382,7 +355,7 @@ def main():
changed = False
if option is not None:
changed, lines = setInterfaceOption(module, lines, iface, option, value, state, address_family)
changed, lines = set_interface_option(module, lines, iface, option, value, state, address_family)
if changed:
dummy, ifaces = read_interfaces_lines(module, [d['line'] for d in lines if 'line' in d])