115
142
route_n = fp.read()
116
143
logger.debug("route -n:\n{}".format(route_n))
145
with open(os.path.join(self.td.collect, "route_6_n")) as fp:
146
route_6_n = fp.read()
147
logger.debug("route -6 -n:\n{}".format(route_6_n))
118
153
interfaces = network_state.get('interfaces')
119
154
for iface in interfaces.values():
120
subnets = iface.get('subnets', {})
122
for index, subnet in zip(range(0, len(subnets)), subnets):
123
iface['index'] = index
125
ifname = "{name}".format(**iface)
127
ifname = "{name}:{index}".format(**iface)
129
self.check_interface(iface,
130
ifconfig_dict.get(ifname),
134
self.check_interface(iface,
135
ifconfig_dict.get(iface['name']),
138
def check_interface(self, iface, ifconfig, route_n):
140
'testing iface:\n{}\n\nifconfig:\n{}'.format(iface, ifconfig))
141
subnets = iface.get('subnets', {})
142
if subnets and iface['index'] != 0:
143
ifname = "{name}:{index}".format(**iface)
145
ifname = "{name}".format(**iface)
155
print("\nnetwork_state iface: %s" % (
156
yaml.dump(iface, default_flow_style=False, indent=4)))
157
self.check_interface(iface['name'],
159
ip_dict.get(iface['name']),
162
def check_interface(self, ifname, iface, ipcfg, routes):
163
print('check_interface: testing '
164
'ifname:{}\niface:\n{}\n\nipcfg:\n{}'.format(ifname, iface,
166
# FIXME: remove check?
147
167
# initial check, do we have the correct iface ?
148
logger.debug('ifname={}'.format(ifname))
149
logger.debug("ifconfig['interface']={}".format(ifconfig['interface']))
150
self.assertEqual(ifname, ifconfig['interface'])
152
# check physical interface attributes
153
for key in ['mac_address', 'mtu']:
168
print('ifname={}'.format(ifname))
169
self.assertEqual(ifname, ipcfg['interface'])
170
print("ipcfg['interface']={}".format(ipcfg['interface']))
172
# check physical interface attributes (skip bond members, macs change)
173
if iface['type'] in ['physical'] and 'bond-master' not in iface:
174
for key in ['mac_address']:
175
print("checking mac on iface: %s" % iface['name'])
176
if key in iface and iface[key]:
177
self.assertEqual(iface[key].lower(),
180
# we can check mtu on all interfaces
154
182
if key in iface and iface[key]:
155
self.assertEqual(iface[key],
158
def __get_subnet(subnets, subidx):
159
for index, subnet in zip(range(0, len(subnets)), subnets):
164
# check subnet related attributes, and specifically only
165
# the subnet specified by iface['index']
166
subnets = iface.get('subnets', {})
168
subnet = __get_subnet(subnets, iface['index'])
183
print("checking mtu on iface: %s" % iface['name'])
184
self.assertEqual(int(iface[key]),
187
# check subnet related attributes
188
subnets = iface.get('subnets')
191
for subnet in subnets:
192
config_inet_iface = None
193
found_inet_iface = None
194
print('validating subnet:\n%s' % subnet)
169
195
if 'address' in subnet and subnet['address']:
196
# we will create to ipaddress.IPvXInterface objects
197
# one based on config, and other from collected data
199
config_ipstr = subnet['address']
200
if 'netmask' in subnet:
201
config_ipstr += "/%s" % subnet['netmask']
203
# One more bit is how to construct the
204
# right Version interface, detecting on ":" in address
170
206
if ':' in subnet['address']:
171
inet_iface = ipaddress.IPv6Interface(
174
inet_iface = ipaddress.IPv4Interface(
178
self.assertEqual(str(inet_iface.ip),
181
self.assertEqual(str(inet_iface.netmask),
185
str(inet_iface.network.broadcast_address),
186
ifconfig['broadcast'])
188
# handle gateway by looking at routing table
189
if 'gateway' in subnet and subnet['gateway']:
190
gw_ip = subnet['gateway']
191
gateways = [line for line in route_n.split('\n')
192
if 'UG' in line and gw_ip in line]
193
logger.debug('matching gateways:\n{}'.format(gateways))
194
self.assertEqual(len(gateways), 1)
195
[gateways] = gateways
196
(dest, gw, genmask, flags, metric, ref, use, iface) = \
198
logger.debug('expected gw:{} found gw:{}'.format(gw_ip, gw))
199
self.assertEqual(gw_ip, gw)
202
class TestNetworkStaticAbs(TestNetworkAbs):
203
conf_file = "examples/tests/basic_network_static.yaml"
206
class TestNetworkVlanAbs(TestNetworkAbs):
207
conf_file = "examples/tests/vlan_network.yaml"
208
collect_scripts = TestNetworkAbs.collect_scripts + [textwrap.dedent("""
210
dpkg-query -W -f '${Status}' vlan > vlan_installed
211
ip -d link show interface1.2667 > ip_link_show_interface1.2667
212
ip -d link show interface1.2668 > ip_link_show_interface1.2668
213
ip -d link show interface1.2669 > ip_link_show_interface1.2669
214
ip -d link show interface1.2670 > ip_link_show_interface1.2670
218
network_state = self.get_network_state()
219
logger.debug('get_vlans ns:\n{}'.format(
220
yaml.dump(network_state, default_flow_style=False, indent=4)))
221
interfaces = network_state.get('interfaces')
222
return [iface for iface in interfaces.values()
223
if iface['type'] == 'vlan']
225
def test_output_files_exist_vlan(self):
226
link_files = ["ip_link_show_{}".format(vlan['name'])
227
for vlan in self.get_vlans()]
228
self.output_files_exist(["vlan_installed"] + link_files)
230
def test_vlan_installed(self):
231
with open(os.path.join(self.td.collect, "vlan_installed")) as fp:
232
status = fp.read().strip()
233
logger.debug('vlan installed?: {}'.format(status))
234
self.assertEqual('install ok installed', status)
236
def test_vlan_enabled(self):
238
# we must have at least one
239
self.assertGreaterEqual(len(self.get_vlans()), 1)
241
# did they get configured?
242
for vlan in self.get_vlans():
243
link_file = "ip_link_show_" + vlan['name']
244
vlan_msg = "vlan protocol 802.1Q id " + str(vlan['vlan_id'])
245
self.check_file_regex(link_file, vlan_msg)
248
class TestNetworkENISource(TestNetworkAbs):
249
""" Curtin now emits a source /etc/network/interfaces.d/*.cfg
250
line. This test exercises this feature by emitting additional
251
network configuration in /etc/network/interfaces.d/eth2.cfg
253
This relies on the network_config.yaml of the TestClass to
254
define a spare nic with no configuration. This ensures that
255
a udev rule for eth2 is emitted so we can reference the interface
256
in our injected configuration.
258
Note, ifupdown allows multiple stanzas with the same iface name
259
and combines the options together during ifup. We rely on this
260
feature allowing etc/network/interfaces to have an unconfigured
261
iface eth2 inet manual line, and then defer the configuration
262
to /etc/network/interfaces.d/eth2.cfg
264
This testcase then uses curtin.net.deb_parse_config method to
265
extract information about what curtin wrote and compare that
266
with what was actually configured (which we capture via ifconfig)
208
config_inet_iface = ipaddress.IPv6Interface(config_ipstr)
209
ip_func = ipaddress.IPv6Interface
210
addresses = ipcfg.get('inet6', [])
213
config_inet_iface = ipaddress.IPv4Interface(config_ipstr)
214
ip_func = ipaddress.IPv4Interface
215
addresses = ipcfg.get('inet4', [])
218
print('found addresses: %s' % addresses)
220
print('cur ip=%s\nsubnet=%s' % (ip, subnet))
221
# drop /CIDR if present for matching
222
if (ip['address'].split("/")[0] ==
223
subnet['address'].split("/")[0]):
224
print('found a match!')
225
found_ipstr = ip['address']
226
if ('netmask' in subnet or '/' in subnet['address']):
227
found_ipstr += "/%s" % ip.get('prefixlen')
228
found_inet_iface = ip_func(found_ipstr)
229
print('returning inet iface')
232
# check ipaddress interface matches (config vs. found)
233
self.assertIsNotNone(config_inet_iface)
234
self.assertIsNotNone(found_inet_iface)
235
self.assertEqual(config_inet_iface, found_inet_iface)
237
def __find_gw_config(subnet):
239
if 'gateway' in subnet:
240
gateways.append(subnet.get('gateway'))
241
for route in subnet.get('routes', []):
242
gateways += __find_gw_config(route)
245
# handle gateways by looking at routing table
246
configured_gws = __find_gw_config(subnet)
247
print('iface:%s configured_gws: %s' % (ifname, configured_gws))
248
for gw_ip in configured_gws:
249
logger.debug('found a gateway in subnet config: %s', gw_ip)
251
route_d = routes['6']
253
route_d = routes['4']
255
found_gws = [line for line in route_d.split('\n')
256
if 'UG' in line and gw_ip in line]
257
logger.debug('found gateways in guest output:\n%s', found_gws)
259
print('found_gws: %s\nexpected: %s' % (found_gws,
261
self.assertEqual(len(found_gws), len(configured_gws))
262
for fgw in found_gws:
264
(dest, gw, flags, metric, ref, use, iface) = \
267
(dest, gw, genmask, flags, metric, ref, use, iface) = \
269
logger.debug('configured gw:%s found gw:%s', gw_ip, gw)
270
self.assertEqual(gw_ip, gw)
273
class TestNetworkBasicAbs(TestNetworkBaseTestsAbs):
274
""" Basic network testing with ipv4
269
conf_file = "examples/tests/network_source.yaml"
270
collect_scripts = [textwrap.dedent("""
272
ifconfig -a > ifconfig_a
273
cp -av /etc/network/interfaces .
274
cp -a /etc/network/interfaces.d .
275
find /etc/network/interfaces.d > find_interfacesd
276
cp /etc/resolv.conf .
277
cp -av /etc/udev/rules.d/70-persistent-net.rules .
278
ip -o route show > ip_route_show
282
def test_source_cfg_exists(self):
283
"""Test that our curthooks wrote our injected config."""
284
self.output_files_exist(["interfaces.d/interface2.cfg"])
286
def test_etc_network_interfaces_source_cfg(self):
287
""" Compare injected configuration as parsed by curtin matches
288
how ifup configured the interface."""
289
# interfaces uses absolute paths, fix for test-case
290
interfaces = os.path.join(self.td.collect, "interfaces")
291
cmd = ['sed', '-i.orig', '-e', 's,/etc/network/,,g',
292
'{}'.format(interfaces)]
293
subprocess.check_call(cmd, stderr=subprocess.STDOUT)
295
curtin_ifaces = self.parse_deb_config(interfaces)
296
logger.debug('parsed eni dict:\n{}'.format(
297
yaml.dump(curtin_ifaces, default_flow_style=False, indent=4)))
298
print('parsed eni dict:\n{}'.format(
299
yaml.dump(curtin_ifaces, default_flow_style=False, indent=4)))
301
with open(os.path.join(self.td.collect, "ifconfig_a")) as fp:
302
ifconfig_a = fp.read()
303
logger.debug('ifconfig -a:\n{}'.format(ifconfig_a))
305
ifconfig_dict = helpers.ifconfig_to_dict(ifconfig_a)
306
logger.debug('parsed ifconfig dict:\n{}'.format(
307
yaml.dump(ifconfig_dict, default_flow_style=False, indent=4)))
308
print('parsed ifconfig dict:\n{}'.format(
309
yaml.dump(ifconfig_dict, default_flow_style=False, indent=4)))
312
self.assertTrue(iface in curtin_ifaces)
314
expected_address = curtin_ifaces[iface].get('address', None)
315
self.assertIsNotNone(expected_address)
317
# handle CIDR notation
319
return addr.split("/")[0]
320
actual_address = ifconfig_dict[iface].get('address', "")
321
self.assertEqual(_nocidr(expected_address), _nocidr(actual_address))
324
class PreciseHWETTestNetwork(relbase.precise_hwe_t, TestNetworkAbs):
325
# FIXME: off due to hang at test: Starting execute cloud user/final scripts
329
class PreciseHWETTestNetworkStatic(relbase.precise_hwe_t,
330
TestNetworkStaticAbs):
331
# FIXME: off due to hang at test: Starting execute cloud user/final scripts
335
class TrustyTestNetwork(relbase.trusty, TestNetworkAbs):
339
class TrustyTestNetworkStatic(relbase.trusty, TestNetworkStaticAbs):
343
class TrustyHWEUTestNetwork(relbase.trusty_hwe_u, TrustyTestNetwork):
344
# Working, off by default to safe test suite runtime, covered by bonding
348
class TrustyHWEUTestNetworkStatic(relbase.trusty_hwe_u,
349
TestNetworkStaticAbs):
350
# Working, off by default to safe test suite runtime, covered by bonding
354
class TrustyHWEVTestNetwork(relbase.trusty_hwe_v, TrustyTestNetwork):
355
# Working, off by default to safe test suite runtime, covered by bonding
359
class TrustyHWEVTestNetworkStatic(relbase.trusty_hwe_v,
360
TestNetworkStaticAbs):
361
# Working, off by default to safe test suite runtime, covered by bonding
365
class TrustyHWEWTestNetwork(relbase.trusty_hwe_w, TrustyTestNetwork):
366
# Working, off by default to safe test suite runtime, covered by bonding
370
class TrustyHWEWTestNetworkStatic(relbase.trusty_hwe_w,
371
TestNetworkStaticAbs):
372
# Working, off by default to safe test suite runtime, covered by bonding
376
class WilyTestNetwork(relbase.wily, TestNetworkAbs):
380
class WilyTestNetworkStatic(relbase.wily, TestNetworkStaticAbs):
384
class XenialTestNetwork(relbase.xenial, TestNetworkAbs):
388
class XenialTestNetworkStatic(relbase.xenial, TestNetworkStaticAbs):
392
class YakketyTestNetwork(relbase.yakkety, TestNetworkAbs):
396
class YakketyTestNetworkStatic(relbase.yakkety, TestNetworkStaticAbs):
400
class PreciseTestNetworkVlan(relbase.precise, TestNetworkVlanAbs):
403
# precise ip -d link show output is different (of course)
404
def test_vlan_enabled(self):
406
# we must have at least one
407
self.assertGreaterEqual(len(self.get_vlans()), 1)
409
# did they get configured?
410
for vlan in self.get_vlans():
411
link_file = "ip_link_show_" + vlan['name']
412
vlan_msg = "vlan id " + str(vlan['vlan_id'])
413
self.check_file_regex(link_file, vlan_msg)
416
class TrustyTestNetworkVlan(relbase.trusty, TestNetworkVlanAbs):
420
class WilyTestNetworkVlan(relbase.wily, TestNetworkVlanAbs):
424
class XenialTestNetworkVlan(relbase.xenial, TestNetworkVlanAbs):
428
class YakketyTestNetworkVlan(relbase.yakkety, TestNetworkVlanAbs):
432
class PreciseTestNetworkENISource(relbase.precise, TestNetworkENISource):
434
# not working, still debugging though; possible older ifupdown doesn't
435
# like the multiple iface method.
438
class TrustyTestNetworkENISource(relbase.trusty, TestNetworkENISource):
442
class WilyTestNetworkENISource(relbase.wily, TestNetworkENISource):
446
class XenialTestNetworkENISource(relbase.xenial, TestNetworkENISource):
450
class YakketyTestNetworkENISource(relbase.yakkety, TestNetworkENISource):
276
conf_file = "examples/tests/basic_network.yaml"
279
class PreciseHWETTestNetworkBasic(relbase.precise_hwe_t, TestNetworkBasicAbs):
280
# FIXME: off due to hang at test: Starting execute cloud user/final scripts
284
class TrustyTestNetworkBasic(relbase.trusty, TestNetworkBasicAbs):
288
class TrustyHWEUTestNetworkBasic(relbase.trusty_hwe_u, TrustyTestNetworkBasic):
289
# Working, off by default to safe test suite runtime, covered by bonding
293
class TrustyHWEVTestNetworkBasic(relbase.trusty_hwe_v, TrustyTestNetworkBasic):
294
# Working, off by default to safe test suite runtime, covered by bonding
298
class TrustyHWEWTestNetworkBasic(relbase.trusty_hwe_w, TrustyTestNetworkBasic):
299
# Working, off by default to safe test suite runtime, covered by bonding
303
class XenialTestNetworkBasic(relbase.xenial, TestNetworkBasicAbs):
307
class YakketyTestNetworkBasic(relbase.yakkety, TestNetworkBasicAbs):
added
removed
tests/vmtests/test_network.py
