ansible/test/units/modules/system/interfaces_file/test_interfaces_file.py
Olivier Bourdon cf3483e752 Add tests filters (#43221)
This will allow tests to be carried out condtionally if necessary
using regexp include and/or exclude filters
Reorganize imports into alphabetical order for easier insertion
2018-11-13 11:16:49 +00:00

319 lines
13 KiB
Python

# (c) 2017, Roman Belyakovsky <ihryamzik () gmail.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
from units.compat import unittest
from ansible.modules.system import interfaces_file
from shutil import copyfile, move
import difflib
import inspect
import io
import json
import os
import re
import shutil
import tempfile
class AnsibleFailJson(Exception):
pass
class ModuleMocked():
def atomic_move(self, src, dst):
move(src, dst)
def backup_local(self, path):
backupp = os.path.join("/tmp", os.path.basename(path) + ".bak")
copyfile(path, backupp)
return backupp
def fail_json(self, msg):
raise AnsibleFailJson(msg)
module = ModuleMocked()
fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures', 'input')
golden_output_path = os.path.join(os.path.dirname(__file__), 'fixtures', 'golden_output')
class TestInterfacesFileModule(unittest.TestCase):
def getTestFiles(self, include_filter=None, exclude_filter=None):
flist = next(os.walk(fixture_path))[2]
if include_filter:
flist = filter(lambda x: re.match(include_filter, x), flist)
if exclude_filter:
flist = filter(lambda x: not re.match(exclude_filter, x), flist)
return flist
def compareFileToBackup(self, path, backup):
with open(path) as f1:
with open(backup) as f2:
diffs = difflib.context_diff(f1.readlines(),
f2.readlines(),
fromfile=os.path.basename(path),
tofile=os.path.basename(backup))
# Restore backup
move(backup, path)
deltas = [d for d in diffs]
self.assertTrue(len(deltas) == 0)
def compareInterfacesLinesToFile(self, interfaces_lines, path, testname=None):
if not testname:
testname = "%s.%s" % (path, inspect.stack()[1][3])
self.compareStringWithFile("".join([d['line'] for d in interfaces_lines if 'line' in d]), testname)
def compareInterfacesToFile(self, ifaces, path, testname=None):
if not testname:
testname = "%s.%s.json" % (path, inspect.stack()[1][3])
self.compareStringWithFile(json.dumps(ifaces, sort_keys=True, indent=4, separators=(',', ': ')), testname)
def compareStringWithFile(self, string, path):
# self.assertEqual("","_",msg=path)
testfilepath = os.path.join(golden_output_path, path)
goldenstring = string
if not os.path.isfile(testfilepath):
f = io.open(testfilepath, 'wb')
f.write(string)
f.close()
else:
with open(testfilepath, 'r') as goldenfile:
goldenstring = goldenfile.read()
goldenfile.close()
self.assertEqual(string, goldenstring)
def test_no_changes(self):
for testfile in self.getTestFiles():
path = os.path.join(fixture_path, testfile)
lines, ifaces = interfaces_file.read_interfaces_file(module, path)
self.compareInterfacesLinesToFile(lines, testfile)
self.compareInterfacesToFile(ifaces, testfile)
def test_add_up_aoption_to_aggi(self):
testcases = {
"add_aggi_up": [
{
'iface': 'aggi',
'option': 'up',
'value': 'route add -net 224.0.0.0 netmask 240.0.0.0 dev aggi',
'state': 'present',
}
],
"add_and_delete_aggi_up": [
{
'iface': 'aggi',
'option': 'up',
'value': 'route add -net 224.0.0.0 netmask 240.0.0.0 dev aggi',
'state': 'present',
},
{
'iface': 'aggi',
'option': 'up',
'value': None,
'state': 'absent',
},
],
"set_aggi_slaves": [
{
'iface': 'aggi',
'option': 'slaves',
'value': 'int1 int3',
'state': 'present',
},
],
"set_aggi_and_eth0_mtu": [
{
'iface': 'aggi',
'option': 'mtu',
'value': '1350',
'state': 'present',
},
{
'iface': 'eth0',
'option': 'mtu',
'value': '1350',
'state': 'present',
},
],
}
for testname, options_list in testcases.items():
for testfile in self.getTestFiles():
path = os.path.join(fixture_path, testfile)
lines, ifaces = interfaces_file.read_interfaces_file(module, path)
fail_json_iterations = []
for i, options in enumerate(options_list):
try:
_, lines = interfaces_file.setInterfaceOption(module, lines, options['iface'], options['option'], options['value'], options['state'])
except AnsibleFailJson as e:
fail_json_iterations.append("[%d] fail_json message: %s\noptions:\n%s" %
(i, str(e), json.dumps(options, sort_keys=True, indent=4, separators=(',', ': '))))
self.compareStringWithFile("\n=====\n".join(fail_json_iterations), "%s_%s.exceptions.txt" % (testfile, testname))
self.compareInterfacesLinesToFile(lines, testfile, "%s_%s" % (testfile, testname))
self.compareInterfacesToFile(ifaces, testfile, "%s_%s.json" % (testfile, testname))
def test_revert(self):
testcases = {
"revert": [
{
'iface': 'eth0',
'option': 'mtu',
'value': '1350',
}
],
}
for testname, options_list in testcases.items():
for testfile in self.getTestFiles():
with tempfile.NamedTemporaryFile() as temp_file:
src_path = os.path.join(fixture_path, testfile)
path = temp_file.name
shutil.copy(src_path, path)
lines, ifaces = interfaces_file.read_interfaces_file(module, path)
backupp = module.backup_local(path)
options = options_list[0]
for state in ['present', 'absent']:
fail_json_iterations = []
options['state'] = state
try:
_, lines = interfaces_file.setInterfaceOption(module, lines,
options['iface'], options['option'], options['value'], options['state'])
except AnsibleFailJson as e:
fail_json_iterations.append("fail_json message: %s\noptions:\n%s" %
(str(e), json.dumps(options, sort_keys=True, indent=4, separators=(',', ': '))))
interfaces_file.write_changes(module, [d['line'] for d in lines if 'line' in d], path)
self.compareStringWithFile("\n=====\n".join(fail_json_iterations), "%s_%s.exceptions.txt" % (testfile, testname))
self.compareInterfacesLinesToFile(lines, testfile, "%s_%s" % (testfile, testname))
self.compareInterfacesToFile(ifaces, testfile, "%s_%s.json" % (testfile, testname))
self.compareFileToBackup(path, backupp)
def test_change_method(self):
testcases = {
"change_method": [
{
'iface': 'eth0',
'option': 'method',
'value': 'manual',
'state': 'present',
}
],
}
for testname, options_list in testcases.items():
for testfile in self.getTestFiles():
with tempfile.NamedTemporaryFile() as temp_file:
src_path = os.path.join(fixture_path, testfile)
path = temp_file.name
shutil.copy(src_path, path)
lines, ifaces = interfaces_file.read_interfaces_file(module, path)
backupp = module.backup_local(path)
options = options_list[0]
fail_json_iterations = []
try:
_, lines = interfaces_file.setInterfaceOption(module, lines, options['iface'], options['option'], options['value'], options['state'])
except AnsibleFailJson as e:
fail_json_iterations.append("fail_json message: %s\noptions:\n%s" %
(str(e), json.dumps(options, sort_keys=True, indent=4, separators=(',', ': '))))
interfaces_file.write_changes(module, [d['line'] for d in lines if 'line' in d], path)
self.compareStringWithFile("\n=====\n".join(fail_json_iterations), "%s_%s.exceptions.txt" % (testfile, testname))
self.compareInterfacesLinesToFile(lines, testfile, "%s_%s" % (testfile, testname))
self.compareInterfacesToFile(ifaces, testfile, "%s_%s.json" % (testfile, testname))
# Restore backup
move(backupp, path)
def test_inet_inet6(self):
testcases = {
"change_ipv4": [
{
'iface': 'eth0',
'address_family': 'inet',
'option': 'address',
'value': '192.168.0.42',
'state': 'present',
}
],
"change_ipv6": [
{
'iface': 'eth0',
'address_family': 'inet6',
'option': 'address',
'value': 'fc00::42',
'state': 'present',
}
],
"change_ipv4_pre_up": [
{
'iface': 'eth0',
'address_family': 'inet',
'option': 'pre-up',
'value': 'XXXX_ipv4',
'state': 'present',
}
],
"change_ipv6_pre_up": [
{
'iface': 'eth0',
'address_family': 'inet6',
'option': 'pre-up',
'value': 'XXXX_ipv6',
'state': 'present',
}
],
"change_ipv4_post_up": [
{
'iface': 'eth0',
'address_family': 'inet',
'option': 'post-up',
'value': 'XXXX_ipv4',
'state': 'present',
}
],
"change_ipv6_post_up": [
{
'iface': 'eth0',
'address_family': 'inet6',
'option': 'post-up',
'value': 'XXXX_ipv6',
'state': 'present',
}
],
}
for testname, options_list in testcases.items():
for testfile in self.getTestFiles():
with tempfile.NamedTemporaryFile() as temp_file:
src_path = os.path.join(fixture_path, testfile)
path = temp_file.name
shutil.copy(src_path, path)
lines, ifaces = interfaces_file.read_interfaces_file(module, path)
backupp = module.backup_local(path)
options = options_list[0]
fail_json_iterations = []
try:
_, lines = interfaces_file.setInterfaceOption(module, lines, options['iface'], options['option'],
options['value'], options['state'], options['address_family'])
except AnsibleFailJson as e:
fail_json_iterations.append("fail_json message: %s\noptions:\n%s" %
(str(e), json.dumps(options, sort_keys=True, indent=4, separators=(',', ': '))))
interfaces_file.write_changes(module, [d['line'] for d in lines if 'line' in d], path)
self.compareStringWithFile("\n=====\n".join(fail_json_iterations), "%s_%s.exceptions.txt" % (testfile, testname))
self.compareInterfacesLinesToFile(lines, testfile, "%s_%s" % (testfile, testname))
self.compareInterfacesToFile(ifaces, testfile, "%s_%s.json" % (testfile, testname))
# Restore backup
move(backupp, path)