Mirrors/network
1
0
Fork 0
mirror of https://github.com/linux-system-roles/network.git synced 2026-07-23 19:07:45 +00:00
Code Issues Projects Releases Wiki Activity

library: merge branch 'th/static-routes'

Browse source 
https://github.com/linux-system-roles/network/pull/21
https://bugzilla.redhat.com/show_bug.cgi?id=1487747
This commit is contained in:
Thomas Haller 2017-11-22 08:55:03 +01:00
commit 5e29382bc9
2 changed files with 708 additions and 155 deletions
Download patch file Download diff file Expand all files Collapse all files

384
library/network_connections.py
View file

@ -48,6 +48,10 @@ class ValidationError(MyError):
self.error_message = message
self.name = name
@staticmethod
def from_connection(idx, message):
return ValidationError('connection[' + str(idx) + ']', message)
class Util:
PY3 = (sys.version_info[0] == 3)
@ -243,6 +247,7 @@ class Util:
if addr is None:
return (None, None)
if family is not None:
Util.addr_family_check(family)
a = socket.inet_pton(family, addr)
else:
a = None
@ -255,6 +260,11 @@ class Util:
family = socket.AF_INET6
return (socket.inet_ntop(family, a), family)
@staticmethod
def addr_family_check(family):
if family != socket.AF_INET and family != socket.AF_INET6:
raise MyError('invalid address family %s' % (family))
@staticmethod
def addr_family_to_v(family):
if family is None:
@ -265,27 +275,43 @@ class Util:
return 'v6'
raise MyError('invalid address family "%s"' % (family))
@staticmethod
def addr_family_default_prefix(family):
Util.addr_family_check(family)
if family == socket.AF_INET:
return 24
else:
return 64
@staticmethod
def addr_family_valid_prefix(family, prefix):
Util.addr_family_check(family)
if family == socket.AF_INET:
m = 32
else:
m = 128
return prefix >= 0 and prefix <= m
@staticmethod
def parse_address(address, family = None):
result = {}
try:
parts = address.split()
addr_parts = parts[0].split('/')
if len(addr_parts) != 2:
raise MyError('expect two addr-parts: ADDR/PLEN')
a, family = Util.parse_ip(addr_parts[0], family)
result['address'] = a
result['is_v4'] = (family == socket.AF_INET)
result['family'] = family
prefix = int(addr_parts[1])
if not (prefix >=0 and prefix <= (32 if family == socket.AF_INET else 128)):
if not Util.addr_family_valid_prefix(family, prefix):
raise MyError('invalid prefix %s' % (prefix))
result['prefix'] = prefix
if len(parts) > 1:
raise MyError('too many parts')
return {
'address': a,
'family': family,
'prefix': prefix,
}
except Exception as e:
raise MyError('invalid address "%s"' % (address))
return result
###############################################################################
@ -420,7 +446,6 @@ class ArgUtil:
c = ArgUtil.connection_find_by_name(name, connections, n_connections)
if not c:
raise MyError('invalid master/parent "%s"' % (name))
assert c.get('nm.uuid', None)
return c['nm.uuid']
@staticmethod
@ -517,46 +542,6 @@ class ArgValidatorBool(ArgValidator):
pass
raise ValidationError(name, 'must be an boolean but is "%s"' % (value))
class ArgValidatorIP(ArgValidatorStr):
def __init__(self, name, family = None, required = False, default_value = None, plain_address = True):
ArgValidatorStr.__init__(self, name, required, default_value, None)
self.family = family
self.plain_address = plain_address
def _validate(self, value, name):
v = ArgValidatorStr._validate(self, value, name)
try:
addr, family = Util.parse_ip(v, self.family)
except:
raise ValidationError(name, 'value "%s" is not a valid IP%s address' % (value, Util.addr_family_to_v(self.family)))
if self.plain_address:
return addr
return { 'is_v4': family == socket.AF_INET, 'family': family, 'address': addr }
class ArgValidatorMac(ArgValidatorStr):
def __init__(self, name, force_len = None, required = False, default_value = None):
ArgValidatorStr.__init__(self, name, required, default_value, None)
self.force_len = force_len
def _validate(self, value, name):
v = ArgValidatorStr._validate(self, value, name)
try:
addr = Util.mac_aton(v, self.force_len)
except MyError as e:
raise ValidationError(name, 'value "%s" is not a valid MAC address' % (value))
if not addr:
raise ValidationError(name, 'value "%s" is not a valid MAC address' % (value))
return Util.mac_ntoa(addr)
class ArgValidatorIPAddr(ArgValidatorStr):
def __init__(self, name, family = None, required = False, default_value = None):
ArgValidatorStr.__init__(self, name, required, default_value, None)
self.family = family
def _validate(self, value, name):
v = ArgValidatorStr._validate(self, value, name)
try:
return Util.parse_address(v, self.family)
except:
raise ValidationError(name, 'value "%s" is not a valid IP%s address with prefix length' % (value, Util.addr_family_to_v(self.family)))
class ArgValidatorDict(ArgValidator):
def __init__(self, name = None, required = False, nested = None, default_value = None, all_missing_during_validate = False):
ArgValidator.__init__(self, name, required, default_value)
@ -617,6 +602,106 @@ class ArgValidatorList(ArgValidator):
result.append(vv)
return result
class ArgValidatorIP(ArgValidatorStr):
def __init__(self, name, family = None, required = False, default_value = None, plain_address = True):
ArgValidatorStr.__init__(self, name, required, default_value, None)
self.family = family
self.plain_address = plain_address
def _validate(self, value, name):
v = ArgValidatorStr._validate(self, value, name)
try:
addr, family = Util.parse_ip(v, self.family)
except:
raise ValidationError(name, 'value "%s" is not a valid IP%s address' % (value, Util.addr_family_to_v(self.family)))
if self.plain_address:
return addr
return { 'family': family, 'address': addr }
class ArgValidatorMac(ArgValidatorStr):
def __init__(self, name, force_len = None, required = False, default_value = None):
ArgValidatorStr.__init__(self, name, required, default_value, None)
self.force_len = force_len
def _validate(self, value, name):
v = ArgValidatorStr._validate(self, value, name)
try:
addr = Util.mac_aton(v, self.force_len)
except MyError as e:
raise ValidationError(name, 'value "%s" is not a valid MAC address' % (value))
if not addr:
raise ValidationError(name, 'value "%s" is not a valid MAC address' % (value))
return Util.mac_ntoa(addr)
class ArgValidatorIPAddr(ArgValidatorDict):
def __init__(self, name, family = None, required = False, default_value = None):
ArgValidatorDict.__init__(self,
name,
required,
nested = [
ArgValidatorIP ('address', family = family, required = True, plain_address = False),
ArgValidatorNum('prefix', default_value = None, val_min = 0),
],
)
self.family = family
def _validate(self, value, name):
if isinstance(value, Util.STRING_TYPE):
v = str(value)
if not v:
raise ValidationError(name, 'cannot be empty')
try:
return Util.parse_address(v, self.family)
except:
raise ValidationError(name, 'value "%s" is not a valid IP%s address with prefix length' % (value, Util.addr_family_to_v(self.family)))
v = ArgValidatorDict._validate(self, value, name)
return {
'address': v['address']['address'],
'family': v['address']['family'],
'prefix': v['prefix'],
}
def _validate_post(self, value, name, result):
family = result['family']
prefix = result['prefix']
if prefix is None:
prefix = Util.addr_family_default_prefix(family)
result['prefix'] = prefix
elif not Util.addr_family_valid_prefix(family, prefix):
raise ValidationError(name, 'invalid prefix %s in "%s"' % (prefix, value))
return result
class ArgValidatorIPRoute(ArgValidatorDict):
def __init__(self, name, family = None, required = False, default_value = None):
ArgValidatorDict.__init__(self,
name,
required,
nested = [
ArgValidatorIP ('network', family = family, required = True, plain_address = False),
ArgValidatorNum('prefix', default_value = None, val_min = 0),
ArgValidatorIP ('gateway', family = family, default_value = None, plain_address = False),
ArgValidatorNum('metric', default_value = -1, val_min = -1, val_max = 0xFFFFFFFF),
],
)
self.family = family
def _validate_post(self, value, name, result):
network = result['network']
family = network['family']
result['network'] = network['address']
result['family'] = family
gateway = result['gateway']
if gateway is not None:
if family != gateway['family']:
raise ValidationError(name, 'conflicting address family between network and gateway \"%s\"' % (gateway['address']))
result['gateway'] = gateway['address']
prefix = result['prefix']
if prefix is None:
prefix = Util.addr_family_default_prefix(family)
result['prefix'] = prefix
elif not Util.addr_family_valid_prefix(family, prefix):
raise ValidationError(name, 'invalid prefix %s in "%s"' % (prefix, value))
return result
class ArgValidator_DictIP(ArgValidatorDict):
def __init__(self):
ArgValidatorDict.__init__(self,
@ -633,6 +718,12 @@ class ArgValidator_DictIP(ArgValidatorDict):
nested = ArgValidatorIPAddr('address[?]'),
default_value = list,
),
ArgValidatorList('route',
nested = ArgValidatorIPRoute('route[?]'),
default_value = list,
),
ArgValidatorBool('route_append_only'),
ArgValidatorBool('rule_append_only'),
ArgValidatorList('dns',
nested = ArgValidatorIP('dns[?]', plain_address=False),
default_value = list,
@ -651,6 +742,9 @@ class ArgValidator_DictIP(ArgValidatorDict):
'gateway6': None,
'route_metric6': None,
'address': [],
'route': [],
'route_append_only': False,
'rule_append_only': False,
'dns': [],
'dns_search': [],
},
@ -658,9 +752,9 @@ class ArgValidator_DictIP(ArgValidatorDict):
def _validate_post(self, value, name, result):
if result['dhcp4'] is None:
result['dhcp4'] = result['dhcp4_send_hostname'] is not None or not any([a for a in result['address'] if a['is_v4']])
result['dhcp4'] = result['dhcp4_send_hostname'] is not None or not any([a for a in result['address'] if a['family'] == socket.AF_INET])
if result['auto6'] is None:
result['auto6'] = not any([a for a in result['address'] if not a['is_v4']])
result['auto6'] = not any([a for a in result['address'] if a['family'] == socket.AF_INET6])
if result['dhcp4_send_hostname'] is not None:
if not result['dhcp4']:
raise ValidationError(name, '"dhcp4_send_hostname" is only valid if "dhcp4" is enabled')
@ -879,6 +973,31 @@ class ArgValidator_ListConnections(ArgValidatorList):
raise ValidationError(name + '[' + str(idx) + '].parent', 'references non-existing "parent" connection "%s"' % (connection['parent']))
return result
VALIDATE_ONE_MODE_NM = 'nm'
VALIDATE_ONE_MODE_INITSCRIPTS = 'initscripts'
def validate_connection_one(self, mode, connections, idx):
connection = connections[idx]
if 'type' not in connection:
return
if ( (connection['parent'])
and ( ( (mode == self.VALIDATE_ONE_MODE_INITSCRIPTS)
and (connection['type'] == 'vlan'))
or ( (connection['type'] == 'infiniband')
and (connection['infiniband_p_key'] not in [ None, -1 ])))):
try:
ArgUtil.connection_find_master(connection['parent'], connections, idx)
except MyError as e:
raise ValidationError.from_connection(idx, 'profile references a parent "%s" which has \'interface_name\' missing' % (connection['parent']))
if ( (connection['master'])
and (mode == self.VALIDATE_ONE_MODE_INITSCRIPTS)):
try:
ArgUtil.connection_find_master(connection['master'], connections, idx)
except MyError as e:
raise ValidationError.from_connection(idx, 'profile references a master "%s" which has \'interface_name\' missing' % (connection['master']))
###############################################################################
class IfcfgUtil:
@ -970,14 +1089,40 @@ class IfcfgUtil:
return s
@classmethod
def ifcfg_create(cls, connections, idx, warn_fcn):
def _ifcfg_route_merge(cls, route, append_only, current):
if not append_only or current is None:
if not route:
return None
return '\n'.join(route) + '\n'
if route:
# the 'route' file is processed line by line by initscripts' ifup-route. Hence,
# the order of the route matters. _ifcfg_route_merge() is not sophisticated
# enough to understand pre-existing lines. It will only append lines that
# don't exist yet, which hopefully is correct.
# It's better to always rewrite the entire file with route_append_only=False.
changed = False
c_lines = list(current.split('\n'))
for r in route:
if r not in c_lines:
changed = True
c_lines.append(r)
if changed:
return '\n'.join(c_lines) + '\n'
return current
@classmethod
def ifcfg_create(cls, connections, idx, warn_fcn = lambda msg: None, content_current = None):
connection = connections[idx]
ip = connection['ip']
ifcfg_all = {}
for file_type in cls.FILE_TYPES:
ifcfg_all[file_type] = {}
ifcfg = ifcfg_all['ifcfg']
ifcfg = {}
keys_file = None
route4_file = None
route6_file = None
rule4_file = None
rule6_file = None
if ip['dhcp4_send_hostname'] is not None:
warn_fcn('ip.dhcp4_send_hostname is not supported by initscripts provider')
@ -1042,9 +1187,13 @@ class IfcfgUtil:
ifcfg['DEVICETYPE'] = 'TeamPort'
else:
raise MyError('invalid slave_type "%s"' % (connection['slave_type']))
if ip['route_append_only'] and content_current:
route4_file = content_current['route']
route6_file = content_current['route6']
else:
addrs4 = list([a for a in ip['address'] if a['is_v4']])
addrs6 = list([a for a in ip['address'] if not a['is_v4']])
addrs4 = list([a for a in ip['address'] if a['family'] == socket.AF_INET])
addrs6 = list([a for a in ip['address'] if a['family'] == socket.AF_INET6])
if ip['dhcp4']:
ifcfg['BOOTPROTO'] = 'dhcp'
@ -1079,16 +1228,47 @@ class IfcfgUtil:
if ip['gateway6'] is not None:
ifcfg['IPV6_DEFAULTGW'] = ip['gateway6']
for file_type in cls.FILE_TYPES:
h = ifcfg_all[file_type]
for key in h.keys():
if h[key] is None:
del h[key]
continue
if type(h[key]) == type(True):
h[key] = 'yes' if h[key] else 'no'
route4 = []
route6 = []
for r in ip['route']:
line = r['network'] + '/' + str(r['prefix'])
if r['gateway']:
line += ' via ' + r['gateway']
if r['metric'] != -1:
line += ' metric ' + str(r['metric'])
return ifcfg_all
if r['family'] == socket.AF_INET:
route4.append(line)
else:
route6.append(line)
route4_file = cls._ifcfg_route_merge(route4,
ip['route_append_only'] and content_current,
content_current['route'] if content_current else None)
route6_file = cls._ifcfg_route_merge(route6,
ip['route_append_only'] and content_current,
content_current['route6'] if content_current else None)
if ip['rule_append_only'] and content_current:
rule4_file = content_current['rule']
rule6_file = content_current['rule6']
for key in list(ifcfg.keys()):
v = ifcfg[key]
if v is None:
del ifcfg[key]
continue
if type(v) == type(True):
ifcfg[key] = 'yes' if v else 'no'
return {
'ifcfg': ifcfg,
'keys': keys_file,
'route': route4_file,
'route6': route6_file,
'rule': rule4_file,
'rule6': rule6_file,
}
@classmethod
def ifcfg_parse_line(cls, line):
@ -1135,18 +1315,18 @@ class IfcfgUtil:
content = {}
for file_type in cls._file_types(file_type):
h = ifcfg_all[file_type]
if not h:
if file_type != 'ifcfg':
content[file_type] = None
continue
s = cls.ANSIBLE_MANAGED + '\n'
for key in sorted(h.keys()):
value = h[key]
if not cls.KeyValid(key):
raise MyError('invalid ifcfg key %s' % (key))
if value is not None:
s += key + '=' + cls.ValueEscape(value) + '\n'
content[file_type] = s
if file_type == 'ifcfg':
s = cls.ANSIBLE_MANAGED + '\n'
for key in sorted(h.keys()):
value = h[key]
if not cls.KeyValid(key):
raise MyError('invalid ifcfg key %s' % (key))
if value is not None:
s += key + '=' + cls.ValueEscape(value) + '\n'
content[file_type] = s
else:
content[file_type] = h
return content
@classmethod
@ -1223,6 +1403,11 @@ class NMUtil:
nmclient = Util.NM().Client.new(None)
self.nmclient = nmclient
def setting_ip_config_get_routes(self, s_ip):
if s_ip is not None:
for i in range(0, s_ip.get_num_routes()):
yield s_ip.get_route(i)
def connection_ensure_setting(self, connection, setting_type):
setting = connection.get_setting(setting_type)
if not setting:
@ -1313,7 +1498,7 @@ class NMUtil:
return True
return False
def connection_create(self, connections, idx):
def connection_create(self, connections, idx, connection_current = None):
NM = Util.NM()
connection = connections[idx]
@ -1378,8 +1563,8 @@ class NMUtil:
s_ip4.set_property(NM.SETTING_IP_CONFIG_METHOD, 'auto')
s_ip6.set_property(NM.SETTING_IP_CONFIG_METHOD, 'auto')
addrs4 = list([a for a in ip['address'] if a['is_v4']])
addrs6 = list([a for a in ip['address'] if not a['is_v4']])
addrs4 = list([a for a in ip['address'] if a['family'] == socket.AF_INET])
addrs6 = list([a for a in ip['address'] if a['family'] == socket.AF_INET6])
if ip['dhcp4']:
s_ip4.set_property(NM.SETTING_IP_CONFIG_METHOD, 'auto')
@ -1395,7 +1580,7 @@ class NMUtil:
if ip['route_metric4'] is not None and ip['route_metric4'] >= 0:
s_ip4.set_property(NM.SETTING_IP_CONFIG_ROUTE_METRIC, ip['route_metric4'])
for d in ip['dns']:
if d['is_v4']:
if d['family'] == socket.AF_INET:
s_ip4.add_dns(d['address'])
for s in ip['dns_search']:
s_ip4.add_dns_search(s)
@ -1413,9 +1598,21 @@ class NMUtil:
if ip['route_metric6'] is not None and ip['route_metric6'] >= 0:
s_ip6.set_property(NM.SETTING_IP_CONFIG_ROUTE_METRIC, ip['route_metric6'])
for d in ip['dns']:
if not d['is_v4']:
if d['family'] == socket.AF_INET6:
s_ip6.add_dns(d['address'])
if ip['route_append_only'] and connection_current:
for r in self.setting_ip_config_get_routes(connection_current.get_setting(NM.SettingIP4Config)):
s_ip4.add_route(r)
for r in self.setting_ip_config_get_routes(connection_current.get_setting(NM.SettingIP6Config)):
s_ip6.add_route(r)
for r in ip['route']:
rr = NM.IPRoute.new(r['family'], r['network'], r['prefix'], r['gateway'], r['metric'])
if r['family'] == socket.AF_INET:
s_ip4.add_route(rr)
else:
s_ip6.add_route(rr)
try:
con.normalize()
except Exception as e:
@ -1925,6 +2122,13 @@ class Cmd:
AnsibleUtil.fail_json('unsupported provider %s' % (provider))
def run(self):
for idx, connection in enumerate(AnsibleUtil.connections):
try:
AnsibleUtil.ARGS_CONNECTIONS.validate_connection_one(self.validate_one_type,
AnsibleUtil.connections,
idx)
except ValidationError as e:
AnsibleUtil.log_fatal(idx, str(e))
self.run_prepare()
while AnsibleUtil.check_mode_next() != CheckMode.DONE:
for idx, connection in enumerate(AnsibleUtil.connections):
@ -1986,6 +2190,7 @@ class Cmd_nm(Cmd):
def __init__(self):
self._nmutil = None
self.validate_one_type = ArgValidator_ListConnections.VALIDATE_ONE_MODE_NM
@property
def nmutil(self):
@ -2052,7 +2257,7 @@ class Cmd_nm(Cmd):
def run_state_present(self, idx):
connection = AnsibleUtil.connections[idx]
con_cur = Util.first(self.nmutil.connection_list(name = connection['name'], uuid = connection['nm.uuid']))
con_new = self.nmutil.connection_create(AnsibleUtil.connections, idx)
con_new = self.nmutil.connection_create(AnsibleUtil.connections, idx, con_cur)
changed = False
if con_cur is None:
AnsibleUtil.log_info(idx, 'add connection %s, %s' % (connection['name'], connection['nm.uuid']))
@ -2175,6 +2380,9 @@ class Cmd_nm(Cmd):
class Cmd_initscripts(Cmd):
def __init__(self):
self.validate_one_type = ArgValidator_ListConnections.VALIDATE_ONE_MODE_INITSCRIPTS
def check_name(self, idx, name = None):
if name is None:
name = AnsibleUtil.connections[idx]['name']
@ -2228,10 +2436,12 @@ class Cmd_initscripts(Cmd):
connection = AnsibleUtil.connections[idx]
name = connection['name']
ifcfg_all = IfcfgUtil.ifcfg_create(AnsibleUtil.connections, idx,
lambda msg: AnsibleUtil.log_warn(idx, msg))
old_content = IfcfgUtil.content_from_file(name)
ifcfg_all = IfcfgUtil.ifcfg_create(AnsibleUtil.connections, idx,
lambda msg: AnsibleUtil.log_warn(idx, msg),
old_content)
new_content = IfcfgUtil.content_from_dict(ifcfg_all)
if old_content == new_content:

479
library/test_network_connections.py
View file

@ -3,6 +3,8 @@
import sys
import os
import unittest
import socket
import itertools
sys.path.insert(1, os.path.dirname(os.path.abspath(__file__)))
@ -20,11 +22,24 @@ except AttributeError:
try:
nmutil = n.NMUtil()
assert(nmutil)
except:
# NMUtil is not supported, for example on RHEL 6 or without
# pygobject.
nmutil = None
if nmutil:
NM = n.Util.NM()
GObject = n.Util.GObject()
def pprint(msg, obj):
print('PRINT: %s\n' % (msg))
import pprint
p = pprint.PrettyPrinter(indent = 4)
p.pprint(obj)
if nmutil is not None and isinstance(obj, NM.Connection):
obj.dump()
class TestValidator(unittest.TestCase):
def assertValidationError(self, v, value):
@ -32,6 +47,86 @@ class TestValidator(unittest.TestCase):
v.validate,
value)
def assert_nm_connection_routes_expected(self, connection, route_list_expected):
parser = n.ArgValidatorIPRoute('route[?]')
route_list_exp = [parser.validate(r) for r in route_list_expected]
route_list_new = itertools.chain(nmutil.setting_ip_config_get_routes(connection.get_setting(NM.SettingIP4Config)),
nmutil.setting_ip_config_get_routes(connection.get_setting(NM.SettingIP6Config)))
route_list_new = [{
'family': r.get_family(),
'network': r.get_dest(),
'prefix': int(r.get_prefix()),
'gateway': r.get_next_hop(),
'metric': int(r.get_metric()),
} for r in route_list_new]
self.assertEqual(route_list_exp, route_list_new)
def do_connections_check_invalid(self, input_connections):
self.assertValidationError(n.AnsibleUtil.ARGS_CONNECTIONS, input_connections)
def do_connections_validate_nm(self, input_connections, **kwargs):
if not nmutil:
return
connections = n.AnsibleUtil.ARGS_CONNECTIONS.validate(input_connections)
for connection in connections:
if 'type' in connection:
connection['nm.exists'] = False
connection['nm.uuid'] = n.Util.create_uuid()
mode = n.ArgValidator_ListConnections.VALIDATE_ONE_MODE_INITSCRIPTS
for idx, connection in enumerate(connections):
try:
n.AnsibleUtil.ARGS_CONNECTIONS.validate_connection_one(mode, connections, idx)
except n.ValidationError as e:
continue
if 'type' in connection:
con_new = nmutil.connection_create(connections, idx)
self.assertTrue(con_new)
self.assertTrue(con_new.verify())
if 'nm_route_list_current' in kwargs:
parser = n.ArgValidatorIPRoute('route[?]')
s4 = con_new.get_setting(NM.SettingIP4Config)
s6 = con_new.get_setting(NM.SettingIP6Config)
s4.clear_routes()
s6.clear_routes()
for r in kwargs['nm_route_list_current'][idx]:
r = parser.validate(r)
r = NM.IPRoute.new(r['family'], r['network'], r['prefix'], r['gateway'], r['metric'])
if r.get_family() == socket.AF_INET:
s4.add_route(r)
else:
s6.add_route(r)
con_new = nmutil.connection_create(connections, idx,
connection_current = con_new)
self.assertTrue(con_new)
self.assertTrue(con_new.verify())
if 'nm_route_list_expected' in kwargs:
self.assert_nm_connection_routes_expected(con_new, kwargs['nm_route_list_expected'][idx])
def do_connections_validate_ifcfg(self, input_connections, **kwargs):
mode = n.ArgValidator_ListConnections.VALIDATE_ONE_MODE_INITSCRIPTS
connections = n.AnsibleUtil.ARGS_CONNECTIONS.validate(input_connections)
for idx, connection in enumerate(connections):
try:
n.AnsibleUtil.ARGS_CONNECTIONS.validate_connection_one(mode, connections, idx)
except n.ValidationError as e:
continue
if 'type' in connection:
content_current = kwargs.get('initscripts_content_current', None)
if content_current:
content_current = content_current[idx]
c = n.IfcfgUtil.ifcfg_create(connections, idx, content_current = content_current)
#pprint("con[%s] = \"%s\"" % (idx, connections[idx]['name']), c)
exp = kwargs.get('initscripts_dict_expected', None)
if exp is not None:
self.assertEqual(exp[idx], c)
def do_connections_validate(self, expected_connections, input_connections, **kwargs):
connections = n.AnsibleUtil.ARGS_CONNECTIONS.validate(input_connections)
self.assertEqual(expected_connections, connections)
self.do_connections_validate_nm(input_connections, **kwargs)
self.do_connections_validate_ifcfg(input_connections, **kwargs)
def test_validate_str(self):
v = n.ArgValidatorStr('state')
@ -126,12 +221,12 @@ class TestValidator(unittest.TestCase):
self.maxDiff = None
self.assertEqual(
self.do_connections_validate(
[],
[],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([]),
)
self.assertEqual(
self.do_connections_validate(
[
{
'name': '5',
@ -146,6 +241,9 @@ class TestValidator(unittest.TestCase):
'auto6': True,
'dhcp4': True,
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
'route_metric6': None,
'dhcp4_send_hostname': None,
'dns': [],
@ -170,14 +268,14 @@ class TestValidator(unittest.TestCase):
'ignore_errors': None,
}
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{ 'name': '5',
'type': 'ethernet',
},
{ 'name': '5' }
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'name': '5',
@ -192,6 +290,9 @@ class TestValidator(unittest.TestCase):
'auto6': True,
'dhcp4': True,
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
'dns': [],
'dns_search': [],
'route_metric6': None,
@ -211,18 +312,17 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{ 'name': '5',
'state': 'up',
'type': 'ethernet',
},
]),
],
)
self.assertValidationError(n.AnsibleUtil.ARGS_CONNECTIONS,
[ { 'name': 'a', 'autoconnect': True }])
self.do_connections_check_invalid([ { 'name': 'a', 'autoconnect': True }])
self.assertEqual(
self.do_connections_validate(
[
{
'name': '5',
@ -230,15 +330,15 @@ class TestValidator(unittest.TestCase):
'ignore_errors': None,
}
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{
'name': '5',
'state': 'absent',
}
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'autoconnect': True,
@ -256,12 +356,14 @@ class TestValidator(unittest.TestCase):
'dns': [],
'address': [
{
'is_v4': True,
'prefix': 24,
'family': 2,
'family': socket.AF_INET,
'address': '192.168.174.5'
}
]
],
'route_append_only': False,
'rule_append_only': False,
'route': [],
},
'state': 'up',
'mtu': 1450,
@ -279,7 +381,7 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{
'name': 'prod1',
'state': 'up',
@ -291,10 +393,10 @@ class TestValidator(unittest.TestCase):
'address': '192.168.174.5/24',
}
}
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'autoconnect': True,
@ -305,18 +407,19 @@ class TestValidator(unittest.TestCase):
'auto6': True,
'address': [
{
'is_v4': True,
'prefix': 24,
'family': 2,
'family': socket.AF_INET,
'address': '192.168.176.5'
},
{
'is_v4': True,
'prefix': 24,
'family': 2,
'family': socket.AF_INET,
'address': '192.168.177.5'
}
],
'route_append_only': False,
'rule_append_only': False,
'route': [],
'route_metric6': None,
'route_metric4': None,
'dns_search': [],
@ -352,16 +455,31 @@ class TestValidator(unittest.TestCase):
'dhcp4_send_hostname': None,
'gateway6': None,
'gateway4': None,
'auto6': True,
'auto6': False,
'dns': [],
'address': [
{
'is_v4': True,
'prefix': 24,
'family': 2,
'family': socket.AF_INET,
'address': '192.168.174.5'
}
]
},
{
'prefix': 65,
'family': socket.AF_INET6,
'address': 'a:b:c::6',
},
],
'route_append_only': False,
'rule_append_only': False,
'route': [
{
'family': socket.AF_INET,
'network': '192.168.5.0',
'prefix': 24,
'gateway': None,
'metric': -1,
},
],
},
'mac': None,
'mtu': None,
@ -379,7 +497,7 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
}
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{
'name': 'prod1',
'state': 'up',
@ -398,13 +516,26 @@ class TestValidator(unittest.TestCase):
'parent': 'prod1',
'vlan_id': '100',
'ip': {
'address': [ '192.168.174.5/24' ],
'address': [
'192.168.174.5/24',
{
'address': 'a:b:c::6',
'prefix': 65,
},
],
'route_append_only': False,
'rule_append_only': False,
'route': [
{
'network': '192.168.5.0',
},
],
}
}
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'autoconnect': True,
@ -420,7 +551,10 @@ class TestValidator(unittest.TestCase):
'gateway4': None,
'auto6': False,
'dns': [],
'address': []
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
},
'mac': None,
'mtu': None,
@ -445,6 +579,9 @@ class TestValidator(unittest.TestCase):
'dhcp4': True,
'auto6': True,
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
'route_metric6': None,
'route_metric4': None,
'dns_search': [],
@ -469,7 +606,7 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
}
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{
'name': 'prod2',
'state': 'up',
@ -487,10 +624,10 @@ class TestValidator(unittest.TestCase):
'interface_name': 'eth1',
'master': 'prod2',
}
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'autoconnect': True,
@ -506,7 +643,10 @@ class TestValidator(unittest.TestCase):
'gateway4': None,
'auto6': True,
'dns': [],
'address': []
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
},
'mac': None,
'mtu': None,
@ -528,16 +668,16 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{
'name': 'bond1',
'state': 'up',
'type': 'bond',
},
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'autoconnect': True,
@ -553,7 +693,10 @@ class TestValidator(unittest.TestCase):
'gateway4': None,
'auto6': True,
'dns': [],
'address': []
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
},
'mac': None,
'mtu': None,
@ -575,7 +718,7 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{
'name': 'bond1',
'state': 'up',
@ -584,21 +727,22 @@ class TestValidator(unittest.TestCase):
'mode': 'active-backup',
},
},
]),
],
)
self.assertValidationError(n.AnsibleUtil.ARGS_CONNECTIONS,
[ { } ])
self.assertValidationError(n.AnsibleUtil.ARGS_CONNECTIONS,
[ { 'name': 'b', 'xxx': 5 } ])
self.do_connections_check_invalid([ { } ])
self.do_connections_check_invalid([ { 'name': 'b', 'xxx': 5 } ])
self.assertEqual(
self.do_connections_validate(
[
{
'autoconnect': True,
'interface_name': None,
'ip': {
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
'auto6': True,
'dhcp4': True,
'dhcp4_send_hostname': None,
@ -624,16 +768,16 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{
'name': '5',
'type': 'ethernet',
'mac': 'AA:bb:cC:DD:ee:FF',
}
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'name': '5',
@ -648,6 +792,9 @@ class TestValidator(unittest.TestCase):
'auto6': True,
'dhcp4': True,
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
'dns': [],
'dns_search': [ ],
'route_metric6': None,
@ -667,15 +814,15 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{ 'name': '5',
'state': 'up',
'type': 'ethernet',
},
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'name': '5',
@ -690,6 +837,9 @@ class TestValidator(unittest.TestCase):
'auto6': True,
'dhcp4': True,
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [],
'dns': [],
'dns_search': [ ],
'route_metric6': None,
@ -709,20 +859,20 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
[
{ 'name': '5',
'state': 'up',
'type': 'ethernet',
'ip': {
},
},
]),
],
)
self.assertEqual(
self.do_connections_validate(
[
{
'name': '5',
'name': '555',
'state': 'up',
'type': 'ethernet',
'autoconnect': True,
@ -734,6 +884,24 @@ class TestValidator(unittest.TestCase):
'auto6': True,
'dhcp4': True,
'address': [],
'route_append_only': False,
'rule_append_only': False,
'route': [
{
'family': socket.AF_INET,
'network': '192.168.45.0',
'prefix': 24,
'gateway': None,
'metric': 545,
},
{
'family': socket.AF_INET,
'network': '192.168.46.0',
'prefix': 30,
'gateway': None,
'metric': -1,
},
],
'dns': [],
'dns_search': [ 'aa', 'bb' ],
'route_metric6': None,
@ -753,29 +921,204 @@ class TestValidator(unittest.TestCase):
'infiniband_transport_mode': None,
},
],
n.AnsibleUtil.ARGS_CONNECTIONS.validate([
{ 'name': '5',
[
{ 'name': '555',
'state': 'up',
'type': 'ethernet',
'ip': {
'dns_search': [ 'aa', 'bb' ],
'route': [
{
'network': '192.168.45.0',
'metric': 545,
},
{
'network': '192.168.46.0',
'prefix': 30,
},
],
},
},
]),
],
initscripts_dict_expected = [
{
'ifcfg': {
'BOOTPROTO': 'dhcp',
'DOMAIN': 'aa bb',
'IPV6INIT': 'yes',
'IPV6_AUTOCONF': 'yes',
'NM_CONTROLLED': 'no',
'ONBOOT': 'yes',
'TYPE': 'Ethernet',
},
'keys': None,
'route': '192.168.45.0/24 metric 545\n192.168.46.0/30\n',
'route6': None,
'rule': None,
'rule6': None,
},
],
)
self.do_connections_validate(
[
{
'name': 'e556',
'state': 'up',
'type': 'ethernet',
'autoconnect': True,
'parent': None,
'ip': {
'gateway6': None,
'gateway4': None,
'route_metric4': None,
'auto6': True,
'dhcp4': True,
'address': [],
'route_append_only': True,
'rule_append_only': False,
'route': [
{
'family': socket.AF_INET,
'network': '192.168.45.0',
'prefix': 24,
'gateway': None,
'metric': 545,
},
{
'family': socket.AF_INET,
'network': '192.168.46.0',
'prefix': 30,
'gateway': None,
'metric': -1,
},
{
'family': socket.AF_INET6,
'network': 'a:b:c:d::',
'prefix': 64,
'gateway': None,
'metric': -1,
},
],
'dns': [],
'dns_search': [ 'aa', 'bb' ],
'route_metric6': None,
'dhcp4_send_hostname': None,
},
'mac': None,
'mtu': None,
'master': None,
'vlan_id': None,
'ignore_errors': None,
'interface_name': None,
'check_iface_exists': True,
'force_state_change': None,
'slave_type': None,
'wait': None,
'infiniband_p_key': None,
'infiniband_transport_mode': None,
},
],
[
{ 'name': 'e556',
'state': 'up',
'type': 'ethernet',
'ip': {
'dns_search': [ 'aa', 'bb' ],
'route_append_only': True,
'rule_append_only': False,
'route': [
{
'network': '192.168.45.0',
'metric': 545,
},
{
'network': '192.168.46.0',
'prefix': 30,
},
{
'network': 'a:b:c:d::',
},
],
},
},
],
nm_route_list_current = [
[
{
'network': '192.168.40.0',
'prefix': 24,
'metric': 545,
},
{
'network': '192.168.46.0',
'prefix': 30,
},
{
'network': 'a:b:c:f::',
},
],
],
nm_route_list_expected = [
[
{
'network': '192.168.40.0',
'prefix': 24,
'metric': 545,
},
{
'network': '192.168.46.0',
'prefix': 30,
},
{
'network': '192.168.45.0',
'prefix': 24,
'metric': 545,
},
{
'network': 'a:b:c:f::',
},
{
'network': 'a:b:c:d::',
},
],
],
initscripts_content_current = [
{
'ifcfg': '',
'keys': None,
'route': '192.168.40.0/24 metric 545\n192.168.46.0/30',
'route6': 'a:b:c:f::/64',
'rule': None,
'rule6': None,
},
],
initscripts_dict_expected = [
{
'ifcfg': {
'BOOTPROTO': 'dhcp',
'DOMAIN': 'aa bb',
'IPV6INIT': 'yes',
'IPV6_AUTOCONF': 'yes',
'NM_CONTROLLED': 'no',
'ONBOOT': 'yes',
'TYPE': 'Ethernet',
},
'keys': None,
'route': '192.168.40.0/24 metric 545\n192.168.46.0/30\n192.168.45.0/24 metric 545\n',
'route6': 'a:b:c:f::/64\na:b:c:d::/64\n',
'rule': None,
'rule6': None,
},
],
)
self.assertValidationError(n.AnsibleUtil.ARGS_CONNECTIONS,
[ { 'name': 'b', 'type': 'ethernet', 'mac': 'aa:b' } ])
self.do_connections_check_invalid([ { 'name': 'b', 'type': 'ethernet', 'mac': 'aa:b' } ])
@my_test_skipIf(nmutil is None, 'no support for NM (libnm via pygobject)')
class TestNM(unittest.TestCase):
def test_connection_ensure_setting(self):
NM = n.Util.NM()
GObject = n.Util.GObject()
con = NM.SimpleConnection.new()
self.assertIsNotNone(con)
self.assertTrue(GObject.type_is_a(con, NM.Connection))