# SPDX-License-Identifier: LGPL-2.1+
# systemd-networkd tests
+import argparse
import os
import re
import shutil
dnsmasq_pid_file='/run/networkd-ci/test-test-dnsmasq.pid'
dnsmasq_log_file='/run/networkd-ci/test-dnsmasq-log-file'
+networkd_bin='/usr/lib/systemd/systemd-networkd'
wait_online_bin='/usr/lib/systemd/systemd-networkd-wait-online'
+networkctl_bin='/usr/bin/networkctl'
+use_valgrind=False
+enable_debug=False
+env = {}
+asan_options=None
+ubsan_options=None
def is_module_available(module_name):
lsmod_output = subprocess.check_output('lsmod', universal_newlines=True)
return f
-def expectedFailureIf_ip6gre_do_not_support_ipv6ll():
- def f(func):
- success = False
- rc = subprocess.call(['ip', 'link', 'add', 'name', 'test1', 'type', 'dummy'])
- if rc == 0:
- time.sleep(1)
- rc = subprocess.call(['ip', 'tunnel', 'add', 'tun99', 'local', '2a00:ffde:4567:edde::4986', 'remote', '2001:473:fece:cafe::5178', 'mode', 'ip6gre', 'dev', 'test1'])
- if rc == 0:
- time.sleep(1)
- # Not sure why, but '0' or '2' do not work.
- subprocess.call(['sysctl', '-w', 'net.ipv6.conf.tun99.addr_gen_mode=3'])
-
- output = subprocess.check_output(['ip', '-6', 'address', 'show', 'dev', 'tun99', 'scope', 'link'], universal_newlines=True).rstrip()
- print(output)
- success = 'inet6' in output
-
- subprocess.run(['ip', 'tunnel', 'del', 'tun99'])
-
- subprocess.run(['ip', 'link', 'del', 'test1'])
-
- if success:
- return func
- else:
- return unittest.expectedFailure(func)
-
- return f
-
def setUpModule():
os.makedirs(network_unit_file_path, exist_ok=True)
os.makedirs(networkd_ci_path, exist_ok=True)
subprocess.check_call('systemctl stop systemd-networkd.socket', shell=True)
+ drop_in = [
+ '[Service]',
+ 'Restart=no',
+ 'ExecStart=',
+ ]
+ if use_valgrind:
+ drop_in += [
+ 'ExecStart=!!valgrind --track-origins=yes --leak-check=full --show-leak-kinds=all ' + networkd_bin,
+ 'PrivateTmp=yes'
+ ]
+ else:
+ drop_in += ['ExecStart=!!' + networkd_bin]
+ if enable_debug:
+ drop_in += ['Environment=SYSTEMD_LOG_LEVEL=debug']
+ if asan_options:
+ drop_in += ['Environment=ASAN_OPTIONS="' + asan_options + '"']
+ if ubsan_options:
+ drop_in += ['Environment=UBSAN_OPTIONS="' + ubsan_options + '"']
+ if use_valgrind or asan_options or ubsan_options:
+ drop_in += ['MemoryDenyWriteExecute=no']
+
+ drop_in_str = '\n'.join(drop_in)
+ print(drop_in_str)
+
+ os.makedirs('/run/systemd/system/systemd-networkd.service.d', exist_ok=True)
+ with open('/run/systemd/system/systemd-networkd.service.d/00-override.conf', mode='w') as f:
+ f.write(drop_in_str)
+
+ subprocess.check_call('systemctl daemon-reload', shell=True)
+
def tearDownModule():
shutil.rmtree(networkd_ci_path)
+ shutil.rmtree('/run/systemd/system/systemd-networkd.service.d')
+ subprocess.check_call('systemctl daemon-reload', shell=True)
subprocess.check_call('systemctl stop systemd-networkd.service', shell=True)
subprocess.check_call('systemctl start systemd-networkd.socket', shell=True)
def check_link_exists(self, link):
self.assertTrue(self.link_exists(link))
- def link_remove(self, links):
+ def remove_links(self, links):
for link in links:
if self.link_exists(link):
subprocess.call(['ip', 'link', 'del', 'dev', link])
time.sleep(1)
+ def remove_fou_ports(self, ports):
+ for port in ports:
+ subprocess.call(['ip', 'fou', 'del', 'port', port])
+
+ def remove_routing_policy_rule_tables(self, tables):
+ for table in tables:
+ subprocess.call(['ip', 'rule', 'del', 'table', table])
+
+ def remove_routes(self, routes):
+ for route_type, addr in routes:
+ subprocess.call(['ip', 'route', 'del', route_type, addr])
+
def l2tp_tunnel_remove(self, tunnel_ids):
output = subprocess.check_output(['ip', 'l2tp', 'show', 'tunnel'], universal_newlines=True).rstrip()
for tid in tunnel_ids:
if (os.path.exists(os.path.join(network_unit_file_path, unit + '.d'))):
shutil.rmtree(os.path.join(network_unit_file_path, unit + '.d'))
- def start_dnsmasq(self, additional_options='', lease_time='1h'):
- dnsmasq_command = f'dnsmasq -8 /var/run/networkd-ci/test-dnsmasq-log-file --log-queries=extra --log-dhcp --pid-file=/var/run/networkd-ci/test-test-dnsmasq.pid --conf-file=/dev/null --interface=veth-peer --enable-ra --dhcp-range=2600::10,2600::20,{lease_time} --dhcp-range=192.168.5.10,192.168.5.200,{lease_time} -R --dhcp-leasefile=/var/run/networkd-ci/lease --dhcp-option=26,1492 --dhcp-option=option:router,192.168.5.1 --dhcp-option=33,192.168.5.4,192.168.5.5 --port=0 ' + additional_options
- subprocess.check_call(dnsmasq_command, shell=True)
+ def warn_about_firewalld(self):
+ rc = subprocess.call(['systemctl', '-q', 'is-active', 'firewalld.service'])
+ if rc == 0:
+ print('\nWARNING: firewalld.service is active. The test may fail.')
- time.sleep(10)
+ def start_dnsmasq(self, additional_options='', ipv4_range='192.168.5.10,192.168.5.200', ipv6_range='2600::10,2600::20', lease_time='1h'):
+ self.warn_about_firewalld()
+ dnsmasq_command = f'dnsmasq -8 /var/run/networkd-ci/test-dnsmasq-log-file --log-queries=extra --log-dhcp --pid-file=/var/run/networkd-ci/test-test-dnsmasq.pid --conf-file=/dev/null --interface=veth-peer --enable-ra --dhcp-range={ipv6_range},{lease_time} --dhcp-range={ipv4_range},{lease_time} -R --dhcp-leasefile=/var/run/networkd-ci/lease --dhcp-option=26,1492 --dhcp-option=option:router,192.168.5.1 --dhcp-option=33,192.168.5.4,192.168.5.5 --port=0 ' + additional_options
+ subprocess.check_call(dnsmasq_command, shell=True)
def stop_dnsmasq(self, pid_file):
if os.path.exists(pid_file):
time.sleep(sleep_sec)
def wait_online(self, links_with_operstate, timeout='20s', bool_any=False):
- args = [wait_online_bin, f'--timeout={timeout}'] + [f'--interface={link}' for link in links_with_operstate]
+ args = wait_online_cmd + [f'--timeout={timeout}'] + [f'--interface={link}' for link in links_with_operstate]
if bool_any:
args += ['--any']
try:
- subprocess.check_call(args)
+ subprocess.check_call(args, env=env)
except subprocess.CalledProcessError:
for link in links_with_operstate:
- output = subprocess.check_output(['networkctl', 'status', link.split(':')[0]], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', link.split(':')[0]], universal_newlines=True, env=env).rstrip()
print(output)
raise
def get_operstate(self, link, show_status=True, setup_state='configured'):
- output = subprocess.check_output(['networkctl', 'status', link], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', link], universal_newlines=True, env=env).rstrip()
if show_status:
print(output)
for line in output.splitlines():
def check_operstate(self, link, expected, show_status=True, setup_state='configured'):
self.assertRegex(self.get_operstate(link, show_status, setup_state), expected)
+ def wait_address(self, link, address_regex, scope='global', ipv='', timeout_sec=100):
+ for i in range(timeout_sec):
+ if i > 0:
+ time.sleep(1)
+ output = subprocess.check_output(['ip', ipv, 'address', 'show', 'dev', link, 'scope', scope], universal_newlines=True).rstrip()
+ if re.search(address_regex, output):
+ break
+ else:
+ self.assertRegex(output, address_regex)
+
class NetworkdNetDevTests(unittest.TestCase, Utilities):
'isataptun99',
'macvlan99',
'macvtap99',
+ 'nlmon99',
'sittun96',
'sittun97',
'sittun98',
'vtitun97',
'vtitun98',
'vtitun99',
+ 'vxcan99',
'vxlan99',
'wg98',
'wg99']
'25-macsec.key',
'25-macsec.netdev',
'25-macsec.network',
+ '25-nlmon.netdev',
'25-sit-tunnel-local-any.netdev',
'25-sit-tunnel-remote-any.netdev',
'25-sit-tunnel.netdev',
'25-vti-tunnel-local-any.netdev',
'25-vti-tunnel-remote-any.netdev',
'25-vti-tunnel.netdev',
+ '25-vxcan.netdev',
'25-vxlan.netdev',
'25-wireguard-23-peers.netdev',
'25-wireguard-23-peers.network',
'macsec.network',
'macvlan.network',
'macvtap.network',
+ 'netdev-link-local-addressing-yes.network',
'sit.network',
'vti6.network',
'vti.network',
'vxlan-test1.network',
'vxlan.network']
+ fou_ports = [
+ '55555',
+ '55556']
+
def setUp(self):
- self.link_remove(self.links)
+ self.remove_fou_ports(self.fou_ports)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_fou_ports(self.fou_ports)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_dropin_and_networkctl_glob(self):
print(output)
self.assertRegex(output, '00:50:56:c0:00:28')
- output = subprocess.check_output(['networkctl', 'list'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['list'], universal_newlines=True, env=env).rstrip()
self.assertRegex(output, '1 lo ')
self.assertRegex(output, 'dropin-test')
- output = subprocess.check_output(['networkctl', 'list', 'dropin-test'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['list', 'dropin-test'], universal_newlines=True, env=env).rstrip()
self.assertNotRegex(output, '1 lo ')
self.assertRegex(output, 'dropin-test')
- output = subprocess.check_output(['networkctl', 'list', 'dropin-*'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['list', 'dropin-*'], universal_newlines=True, env=env).rstrip()
self.assertNotRegex(output, '1 lo ')
self.assertRegex(output, 'dropin-test')
- output = subprocess.check_output(['networkctl', 'status', 'dropin-*'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'dropin-*'], universal_newlines=True, env=env).rstrip()
self.assertNotRegex(output, '1: lo ')
self.assertRegex(output, 'dropin-test')
self.start_networkd(0)
self.wait_online(['bridge99', 'test1:degraded'], bool_any=True)
- self.check_link_exists('bridge99')
- self.check_link_exists('test1')
self.check_operstate('bridge99', '(?:off|no-carrier)', setup_state='configuring')
self.check_operstate('test1', 'degraded')
self.wait_online(['bridge99:off'])
- self.assertEqual('900', self.read_link_attr('bridge99', 'bridge', 'hello_time'))
- self.assertEqual('900', self.read_link_attr('bridge99', 'bridge', 'max_age'))
- self.assertEqual('900', self.read_link_attr('bridge99', 'bridge','forward_delay'))
- self.assertEqual('900', self.read_link_attr('bridge99', 'bridge','ageing_time'))
- self.assertEqual('9', self.read_link_attr('bridge99', 'bridge','priority'))
- self.assertEqual('1', self.read_link_attr('bridge99', 'bridge','multicast_querier'))
- self.assertEqual('1', self.read_link_attr('bridge99', 'bridge','multicast_snooping'))
- self.assertEqual('1', self.read_link_attr('bridge99', 'bridge','stp_state'))
+ tick = os.sysconf('SC_CLK_TCK')
+ self.assertEqual(9, round(float(self.read_link_attr('bridge99', 'bridge', 'hello_time')) / tick))
+ self.assertEqual(9, round(float(self.read_link_attr('bridge99', 'bridge', 'max_age')) / tick))
+ self.assertEqual(9, round(float(self.read_link_attr('bridge99', 'bridge','forward_delay')) / tick))
+ self.assertEqual(9, round(float(self.read_link_attr('bridge99', 'bridge','ageing_time')) / tick))
+ self.assertEqual(9, int(self.read_link_attr('bridge99', 'bridge','priority')))
+ self.assertEqual(1, int(self.read_link_attr('bridge99', 'bridge','multicast_querier')))
+ self.assertEqual(1, int(self.read_link_attr('bridge99', 'bridge','multicast_snooping')))
+ self.assertEqual(1, int(self.read_link_attr('bridge99', 'bridge','stp_state')))
def test_bond(self):
self.copy_unit_to_networkd_unit_path('25-bond.netdev', '25-bond-balanced-tlb.netdev')
'21-vlan.network', '21-vlan-test1.network')
self.start_networkd(0)
- self.wait_online(['test1', 'vlan99'])
+ self.wait_online(['test1:degraded', 'vlan99:routable'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'inet 192.168.23.5/24 brd 192.168.23.255 scope global vlan99')
def test_macvtap(self):
- self.copy_unit_to_networkd_unit_path('21-macvtap.netdev', '11-dummy.netdev', 'macvtap.network')
- self.start_networkd(0)
-
- self.wait_online(['macvtap99:off', 'test1:degraded'])
+ for mode in ['private', 'vepa', 'bridge', 'passthru']:
+ with self.subTest(mode=mode):
+ if mode != 'private':
+ self.tearDown()
+ self.copy_unit_to_networkd_unit_path('21-macvtap.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'macvtap.network')
+ with open(os.path.join(network_unit_file_path, '21-macvtap.netdev'), mode='a') as f:
+ f.write('[MACVTAP]\nMode=' + mode)
+ self.start_networkd(0)
+
+ self.wait_online(['macvtap99:degraded', 'test1:degraded'])
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'macvtap99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'macvtap mode ' + mode + ' ')
def test_macvlan(self):
- self.copy_unit_to_networkd_unit_path('21-macvlan.netdev', '11-dummy.netdev', 'macvlan.network')
- self.start_networkd(0)
-
- self.wait_online(['macvlan99:off', 'test1:degraded'])
-
- output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1'], universal_newlines=True).rstrip()
- print(output)
- self.assertRegex(output, ' mtu 2000 ')
+ for mode in ['private', 'vepa', 'bridge', 'passthru']:
+ with self.subTest(mode=mode):
+ if mode != 'private':
+ self.tearDown()
+ self.copy_unit_to_networkd_unit_path('21-macvlan.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'macvlan.network')
+ with open(os.path.join(network_unit_file_path, '21-macvlan.netdev'), mode='a') as f:
+ f.write('[MACVLAN]\nMode=' + mode)
+ self.start_networkd(0)
+
+ self.wait_online(['macvlan99:degraded', 'test1:degraded'])
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, ' mtu 2000 ')
- output = subprocess.check_output(['ip', '-d', 'link', 'show', 'macvlan99'], universal_newlines=True).rstrip()
- print(output)
- self.assertRegex(output, ' mtu 2000 ')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'macvlan99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, ' mtu 2000 ')
+ self.assertRegex(output, 'macvlan mode ' + mode + ' ')
@expectedFailureIfModuleIsNotAvailable('ipvlan')
def test_ipvlan(self):
- self.copy_unit_to_networkd_unit_path('25-ipvlan.netdev', '11-dummy.netdev', 'ipvlan.network')
- self.start_networkd(0)
-
- self.wait_online(['ipvlan99:off', 'test1:degraded'])
+ for mode, flag in [['L2', 'private'], ['L3', 'vepa'], ['L3S', 'bridge']]:
+ with self.subTest(mode=mode, flag=flag):
+ if mode != 'L2':
+ self.tearDown()
+ self.copy_unit_to_networkd_unit_path('25-ipvlan.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'ipvlan.network')
+ with open(os.path.join(network_unit_file_path, '25-ipvlan.netdev'), mode='a') as f:
+ f.write('[IPVLAN]\nMode=' + mode + '\nFlags=' + flag)
+
+ self.start_networkd(0)
+ self.wait_online(['ipvlan99:degraded', 'test1:degraded'])
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ipvlan99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'ipvlan *mode ' + mode.lower() + ' ' + flag)
@expectedFailureIfModuleIsNotAvailable('ipvtap')
def test_ipvtap(self):
- self.copy_unit_to_networkd_unit_path('25-ipvtap.netdev', '11-dummy.netdev', 'ipvtap.network')
- self.start_networkd(0)
-
- self.wait_online(['ipvtap99:off', 'test1:degraded'])
+ for mode, flag in [['L2', 'private'], ['L3', 'vepa'], ['L3S', 'bridge']]:
+ with self.subTest(mode=mode, flag=flag):
+ if mode != 'L2':
+ self.tearDown()
+ self.copy_unit_to_networkd_unit_path('25-ipvtap.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'ipvtap.network')
+ with open(os.path.join(network_unit_file_path, '25-ipvtap.netdev'), mode='a') as f:
+ f.write('[IPVTAP]\nMode=' + mode + '\nFlags=' + flag)
+
+ self.start_networkd(0)
+ self.wait_online(['ipvtap99:degraded', 'test1:degraded'])
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ipvtap99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'ipvtap *mode ' + mode.lower() + ' ' + flag)
def test_veth(self):
- self.copy_unit_to_networkd_unit_path('25-veth.netdev')
+ self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['veth99:off', 'veth-peer:off'])
-
- def test_dummy(self):
- self.copy_unit_to_networkd_unit_path('11-dummy.netdev')
- self.start_networkd(0)
+ self.wait_online(['veth99:degraded', 'veth-peer:degraded'])
- self.wait_online(['test1:off'])
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'veth99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'link/ether 12:34:56:78:9a:bc')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'veth-peer'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'link/ether 12:34:56:78:9a:bd')
def test_tun(self):
self.copy_unit_to_networkd_unit_path('25-tun.netdev')
self.wait_online(['tun99:off'])
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'tun99'], universal_newlines=True).rstrip()
+ print(output)
+ # Old ip command does not support IFF_ flags
+ self.assertRegex(output, 'tun (?:type tun pi on vnet_hdr on multi_queue|addrgenmode) ')
+
def test_tap(self):
self.copy_unit_to_networkd_unit_path('25-tap.netdev')
self.start_networkd(0)
self.wait_online(['tap99:off'])
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'tap99'], universal_newlines=True).rstrip()
+ print(output)
+ # Old ip command does not support IFF_ flags
+ self.assertRegex(output, 'tun (?:type tap pi on vnet_hdr on multi_queue|addrgenmode) ')
+
@expectedFailureIfModuleIsNotAvailable('vrf')
def test_vrf(self):
- self.copy_unit_to_networkd_unit_path('25-vrf.netdev')
+ self.copy_unit_to_networkd_unit_path('25-vrf.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['vrf99:off'])
+ self.wait_online(['vrf99:carrier'])
@expectedFailureIfModuleIsNotAvailable('vcan')
def test_vcan(self):
- self.copy_unit_to_networkd_unit_path('25-vcan.netdev')
+ self.copy_unit_to_networkd_unit_path('25-vcan.netdev', 'netdev-link-local-addressing-yes.network')
+ self.start_networkd(0)
+
+ self.wait_online(['vcan99:carrier'])
+
+ @expectedFailureIfModuleIsNotAvailable('vxcan')
+ def test_vxcan(self):
+ self.copy_unit_to_networkd_unit_path('25-vxcan.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['vcan99:off'])
+ self.wait_online(['vxcan99:carrier', 'vxcan-peer:carrier'])
@expectedFailureIfModuleIsNotAvailable('wireguard')
def test_wireguard(self):
self.assertRegex(output, 'CJQUtcS9emY2fLYqDlpSZiE/QJyHkPWr\+WHtZLZ90FU=')
def test_geneve(self):
- self.copy_unit_to_networkd_unit_path('25-geneve.netdev')
+ self.copy_unit_to_networkd_unit_path('25-geneve.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['geneve99:off'])
+ self.wait_online(['geneve99:degraded'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'geneve99'], universal_newlines=True).rstrip()
print(output)
self.assertNotRegex(output, 'iseq')
self.assertNotRegex(output, 'oseq')
- @expectedFailureIf_ip6gre_do_not_support_ipv6ll()
def test_ip6gre_tunnel(self):
self.copy_unit_to_networkd_unit_path('12-dummy.netdev', 'ip6gretun.network',
'25-ip6gre-tunnel.netdev', '25-tunnel.network',
'25-ip6gre-tunnel-remote-any.netdev', '25-tunnel-remote-any.network')
self.start_networkd()
+ # Old kernels seem not to support IPv6LL address on ip6gre tunnel, So please do not use wait_online() here.
+
self.check_link_exists('dummy98')
self.check_link_exists('ip6gretun99')
self.check_link_exists('ip6gretun98')
print(output)
self.assertRegex(output, 'ip6gre remote any local 2a00:ffde:4567:edde::4987 dev dummy98')
- # Old kernels may not support IPv6LL address on ip6gre tunnel, and the following test may fails.
- self.wait_online(['ip6gretun99:routable', 'ip6gretun98:routable', 'ip6gretun97:routable', 'dummy98:degraded'])
-
def test_gretap_tunnel(self):
self.copy_unit_to_networkd_unit_path('12-dummy.netdev', 'gretap.network',
'25-gretap-tunnel.netdev', '25-tunnel.network',
self.start_networkd(0)
self.wait_online(['isataptun99:routable', 'dummy98:degraded'])
- self.check_link_exists('dummy98')
- self.check_link_exists('isataptun99')
-
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'isataptun99'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, "isatap ")
self.assertRegex(output, 'oseq')
def test_tunnel_independent(self):
- self.copy_unit_to_networkd_unit_path('25-ipip-tunnel-independent.netdev')
+ self.copy_unit_to_networkd_unit_path('25-ipip-tunnel-independent.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['ipiptun99:off'])
+ self.wait_online(['ipiptun99:carrier'])
@expectedFailureIfModuleIsNotAvailable('fou')
def test_fou(self):
print(output)
self.assertRegex(output, 'encap fou encap-sport auto encap-dport 55556')
- subprocess.call(['ip', 'fou', 'del', 'port', '55555'])
- subprocess.call(['ip', 'fou', 'del', 'port', '55556'])
-
def test_vxlan(self):
self.copy_unit_to_networkd_unit_path('25-vxlan.netdev', 'vxlan.network',
'11-dummy.netdev', 'vxlan-test1.network')
self.assertRegex(output, 'RXSC: 8c16456c83a90002, state on')
self.assertRegex(output, '0: PN [0-9]*, state off, key 02030400000000000000000000000000')
+ def test_nlmon(self):
+ self.copy_unit_to_networkd_unit_path('25-nlmon.netdev', 'netdev-link-local-addressing-yes.network')
+ self.start_networkd()
+
+ self.wait_online(['nlmon99:carrier'])
class NetworkdL2TPTests(unittest.TestCase, Utilities):
def setUp(self):
self.l2tp_tunnel_remove(self.l2tp_tunnel_ids)
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
self.l2tp_tunnel_remove(self.l2tp_tunnel_ids)
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
@expectedFailureIfModuleIsNotAvailable('l2tp_eth')
self.assertRegex(output, "Peer session 28, tunnel 12")
self.assertRegex(output, "interface name: l2tp-ses4")
-class NetworkdNetWorkTests(unittest.TestCase, Utilities):
+class NetworkdNetworkTests(unittest.TestCase, Utilities):
links = [
'bond199',
'dummy98',
'11-dummy.netdev',
'12-dummy.netdev',
'23-active-slave.network',
+ '24-search-domain.network',
'25-address-link-section.network',
'25-address-preferred-lifetime-zero-ipv6.network',
'25-address-static.network',
'routing-policy-rule-dummy98.network',
'routing-policy-rule-test1.network']
+ routing_policy_rule_tables = ['7', '8']
+ routes = [['blackhole', '202.54.1.2'], ['unreachable', '202.54.1.3'], ['prohibit', '202.54.1.4']]
+
def setUp(self):
- self.link_remove(self.links)
+ self.remove_routing_policy_rule_tables(self.routing_policy_rule_tables)
+ self.remove_routes(self.routes)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_routing_policy_rule_tables(self.routing_policy_rule_tables)
+ self.remove_routes(self.routes)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_address_static(self):
self.check_link_exists('test1')
- output = subprocess.check_output(['networkctl', 'status', 'test1'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'test1'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '192.168.0.15')
self.assertRegex(output, '192.168.0.1')
def test_routing_policy_rule(self):
self.copy_unit_to_networkd_unit_path('routing-policy-rule-test1.network', '11-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
self.start_networkd()
self.check_link_exists('test1')
self.assertRegex(output, 'oif test1')
self.assertRegex(output, 'lookup 7')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
def test_routing_policy_rule_issue_11280(self):
self.copy_unit_to_networkd_unit_path('routing-policy-rule-test1.network', '11-dummy.netdev',
'routing-policy-rule-dummy98.network', '12-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
- subprocess.call(['ip', 'rule', 'del', 'table', '8'])
-
for trial in range(3):
# Remove state files only first time
self.start_networkd(remove_state_files=(trial == 0))
print(output)
self.assertRegex(output, '112: from 192.168.101.18 tos (?:0x08|throughput) iif dummy98 oif dummy98 lookup 8')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
- subprocess.call(['ip', 'rule', 'del', 'table', '8'])
-
@expectedFailureIfRoutingPolicyPortRangeIsNotAvailable()
def test_routing_policy_rule_port_range(self):
self.copy_unit_to_networkd_unit_path('25-fibrule-port-range.network', '11-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
self.start_networkd()
self.check_link_exists('test1')
self.assertRegex(output, 'tcp')
self.assertRegex(output, 'lookup 7')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
@expectedFailureIfRoutingPolicyIPProtoIsNotAvailable()
def test_routing_policy_rule_invert(self):
self.copy_unit_to_networkd_unit_path('25-fibrule-invert.network', '11-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
self.start_networkd()
self.check_link_exists('test1')
self.assertRegex(output, 'tcp')
self.assertRegex(output, 'lookup 7')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
def test_route_static(self):
self.copy_unit_to_networkd_unit_path('25-route-static.network', '12-dummy.netdev')
self.start_networkd(0)
print(output)
self.assertRegex(output, 'prohibit 202.54.1.4 proto static')
- subprocess.call(['ip', 'route', 'del', 'blackhole', '202.54.1.2'])
- subprocess.call(['ip', 'route', 'del', 'unreachable', '202.54.1.3'])
- subprocess.call(['ip', 'route', 'del', 'prohibit', '202.54.1.4'])
-
def test_ip_route_ipv6_src_route(self):
# a dummy device does not make the addresses go through tentative state, so we
# reuse a bond from an earlier test, which does make the addresses go through
self.check_link_exists('dummy98')
- output = subprocess.check_output(['networkctl', 'status', 'dummy98'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'dummy98'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, 'unmanaged')
self.start_networkd(0)
self.wait_online(['dummy98:degraded'])
- self.check_link_exists('dummy98')
-
self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'forwarding'), '1')
self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'use_tempaddr'), '2')
self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'dad_transmits'), '3')
self.start_networkd(0)
self.wait_online(['dummy98:routable'])
- self.check_link_exists('dummy98')
-
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dummy98'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope global dummy98')
self.start_networkd(0)
self.wait_online(['dummy98:routable'])
- self.check_link_exists('dummy98')
-
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dummy98'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope global dummy98')
self.assertRegex(output, 'inet 192.168.10.30/24 brd 192.168.10.255 scope global test1')
self.check_operstate('test1', 'routable')
-class NetworkdNetWorkBondTests(unittest.TestCase, Utilities):
+ def test_domain(self):
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '24-search-domain.network')
+ self.start_networkd(0)
+ self.wait_online(['dummy98:routable'])
+
+ output = subprocess.check_output(networkctl_cmd + ['status', 'dummy98'], universal_newlines=True, env=env).rstrip()
+ print(output)
+ self.assertRegex(output, 'Address: 192.168.42.100')
+ self.assertRegex(output, 'DNS: 192.168.42.1')
+ self.assertRegex(output, 'Search Domains: one')
+
+class NetworkdBondTests(unittest.TestCase, Utilities):
links = [
'bond199',
'bond99',
'bond-slave.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_bond_active_slave(self):
self.check_operstate('dummy98', 'off')
self.check_operstate('test1', 'off')
- bond_has_no_carrier=False
for trial in range(30):
if trial > 0:
time.sleep(1)
# Let's confirm that networkd's operstate is consistent with ip's result.
self.assertNotRegex(output, 'NO-CARRIER')
-class NetworkdNetWorkBridgeTests(unittest.TestCase, Utilities):
+class NetworkdBridgeTests(unittest.TestCase, Utilities):
links = [
'bridge99',
'dummy98',
'bridge99.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_bridge_property(self):
self.assertEqual(subprocess.call(['ip', 'link', 'add', 'dummy98', 'type', 'dummy']), 0)
self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
- time.sleep(3)
+ for trial in range(30):
+ if trial > 0:
+ time.sleep(1)
+ if self.get_operstate('bridge99') == 'routable' and self.get_operstate('dummy98') == 'enslaved':
+ break
+ else:
+ self.assertTrue(False)
output = subprocess.check_output(['ip', 'address', 'show', 'bridge99'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'inet 192.168.0.15/24 brd 192.168.0.255 scope global bridge99')
- self.check_operstate('bridge99', 'routable')
- self.check_operstate('dummy98', 'enslaved')
-
output = subprocess.check_output(['ip', 'rule', 'list', 'table', '100'], universal_newlines=True).rstrip()
print(output)
self.assertEqual(output, '0: from all to 8.8.8.8 lookup 100')
subprocess.call(['ip', 'rule', 'del', 'table', '100'])
-class NetworkdNetWorkLLDPTests(unittest.TestCase, Utilities):
+class NetworkdLLDPTests(unittest.TestCase, Utilities):
links = ['veth99']
units = [
'25-veth.netdev']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_lldp(self):
self.copy_unit_to_networkd_unit_path('23-emit-lldp.network', '24-lldp.network', '25-veth.netdev')
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth99:degraded', 'veth-peer:degraded'])
- output = subprocess.check_output(['networkctl', 'lldp'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['lldp'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, 'veth-peer')
self.assertRegex(output, 'veth99')
-class NetworkdNetworkRATests(unittest.TestCase, Utilities):
+class NetworkdRATests(unittest.TestCase, Utilities):
links = ['veth99']
units = [
'ipv6-prefix-veth.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_ipv6_prefix_delegation(self):
+ self.warn_about_firewalld()
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'ipv6-prefix.network', 'ipv6-prefix-veth.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth99:routable', 'veth-peer:degraded'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '2002:da8:1:0')
-class NetworkdNetworkDHCPServerTests(unittest.TestCase, Utilities):
- links = [
- 'dummy98',
- 'veth99']
+class NetworkdDHCPServerTests(unittest.TestCase, Utilities):
+ links = ['veth99']
units = [
- '12-dummy.netdev',
- '24-search-domain.network',
'25-veth.netdev',
'dhcp-client.network',
'dhcp-client-timezone-router.network',
'dhcp-server-timezone-router.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_dhcp_server(self):
+ self.warn_about_firewalld()
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client.network', 'dhcp-server.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '192.168.5.*')
self.assertRegex(output, 'Gateway: 192.168.5.1')
self.assertRegex(output, 'DNS: 192.168.5.1')
self.assertRegex(output, 'NTP: 192.168.5.1')
- def test_domain(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '24-search-domain.network')
- self.start_networkd()
-
- self.check_link_exists('dummy98')
-
- output = subprocess.check_output(['networkctl', 'status', 'dummy98'], universal_newlines=True).rstrip()
- print(output)
- self.assertRegex(output, 'Address: 192.168.42.100')
- self.assertRegex(output, 'DNS: 192.168.42.1')
- self.assertRegex(output, 'Search Domains: one')
-
def test_emit_router_timezone(self):
+ self.warn_about_firewalld()
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client-timezone-router.network', 'dhcp-server-timezone-router.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, 'Gateway: 192.168.5.*')
self.assertRegex(output, '192.168.5.*')
self.assertRegex(output, 'Europe/Berlin')
-class NetworkdNetworkDHCPClientTests(unittest.TestCase, Utilities):
+class NetworkdDHCPClientTests(unittest.TestCase, Utilities):
links = [
- 'dummy98',
'veth99',
'vrf99']
'static.network']
def setUp(self):
- self.link_remove(self.links)
self.stop_dnsmasq(dnsmasq_pid_file)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
- self.remove_unit_from_networkd_path(self.units)
self.stop_dnsmasq(dnsmasq_pid_file)
self.remove_lease_file()
self.remove_log_file()
+ self.remove_links(self.links)
+ self.remove_unit_from_networkd_path(self.units)
def test_dhcp_client_ipv6_only(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network')
self.start_dnsmasq()
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '2600::')
self.assertNotRegex(output, '192.168.5')
def test_dhcp_client_ipv4_only(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-only-ipv6-disabled.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertNotRegex(output, '2600::')
self.assertRegex(output, '192.168.5')
def test_dhcp_client_ipv4_ipv6(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network',
'dhcp-client-ipv4-only.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ # link become 'routable' when at least one protocol provide an valid address.
+ self.wait_address('veth99', r'inet 192.168.5.[0-9]*/24 brd 192.168.5.255', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
+
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '2600::')
self.assertRegex(output, '192.168.5')
def test_dhcp_client_settings(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-dhcp-settings.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
print('## ip address show dev veth99')
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
def test_dhcp6_client_settings_rapidcommit_true(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp6_client_settings_rapidcommit_false(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-rapid-commit.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_client_settings_anonymize(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-anonymize.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
self.assertFalse(self.search_words_in_dnsmasq_log('VendorClassIdentifier=SusantVendorTest', True))
self.assertFalse(self.search_words_in_dnsmasq_log('test-hostname'))
def test_dhcp_client_listen_port(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-listen-port.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq('--dhcp-alternate-port=67,5555')
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ # link become 'routable' when at least one protocol provide an valid address.
+ self.wait_address('veth99', r'inet 192.168.5.[0-9]*/24 brd 192.168.5.255', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_route_table_id(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-table.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'route', 'show', 'table', '12'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_route_metric(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-metric.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'route', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_route_criticalconnection_true(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-critical-connection.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '192.168.5.*')
# Sleep for 120 sec as the dnsmasq minimum lease time can only be set to 120
time.sleep(125)
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '192.168.5.*')
def test_dhcp_client_reuse_address_as_static(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ # link become 'routable' when at least one protocol provide an valid address.
+ self.wait_address('veth99', r'inet 192.168.5.[0-9]*/24 brd 192.168.5.255', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99', 'scope', 'global'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, '192.168.5')
self.assertRegex(output, '2600::')
- ipv4_address = re.search('192\.168\.5\.[0-9]*/24', output)
- ipv6_address = re.search('2600::[0-9a-f:]*/128', output)
+ ipv4_address = re.search(r'192.168.5.[0-9]*/24', output)
+ ipv6_address = re.search(r'2600::[0-9a-f:]*/128', output)
static_network = '\n'.join(['[Match]', 'Name=veth99', '[Network]', 'IPv6AcceptRA=no', 'Address=' + ipv4_address.group(), 'Address=' + ipv6_address.group()])
print(static_network)
with open(os.path.join(network_unit_file_path, 'static.network'), mode='w') as f:
f.write(static_network)
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ # When networkd started, the links are already configured, so let's wait for 5 seconds
+ # the links to be re-configured.
+ self.start_networkd(5)
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'global'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_client_vrf(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-vrf.network',
'25-vrf.netdev', '25-vrf.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
- self.check_link_exists('vrf99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable', 'vrf99:carrier'])
+
+ # link become 'routable' when at least one protocol provide an valid address.
+ self.wait_address('veth99', r'inet 192.168.5.[0-9]*/24 brd 192.168.5.255', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
print('## ip -d link show dev vrf99')
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'dev', 'vrf99'], universal_newlines=True).rstrip()
def test_dhcp_client_gateway_onlink_implicit(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network',
'dhcp-client-gateway-onlink-implicit.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
+ output = subprocess.check_output(networkctl_cmd + ['status', 'veth99'], universal_newlines=True, env=env).rstrip()
print(output)
self.assertRegex(output, '192.168.5')
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'link'], universal_newlines=True).rstrip()
self.assertRegex(output, 'inet .* scope link')
+ def test_dhcp_client_route_remove_on_renew(self):
+ self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network',
+ 'dhcp-client-ipv4-only-ipv6-disabled.network')
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
+ self.start_dnsmasq(ipv4_range='192.168.5.100,192.168.5.199', lease_time='2m')
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ # test for issue #12490
+
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'global', 'dynamic'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'inet 192.168.5.1[0-9]*/24 brd 192.168.5.255 scope global dynamic veth99')
+ address1=None
+ for line in output.splitlines():
+ if 'brd 192.168.5.255 scope global dynamic veth99' in line:
+ address1 = line.split()[1].split('/')[0]
+ break
+
+ output = subprocess.check_output(['ip', '-4', 'route', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, f'default via 192.168.5.1 proto dhcp src {address1} metric 1024')
+ self.assertRegex(output, f'192.168.5.1 proto dhcp scope link src {address1} metric 1024')
+
+ self.stop_dnsmasq(dnsmasq_pid_file)
+ self.start_dnsmasq(ipv4_range='192.168.5.200,192.168.5.250', lease_time='2m')
+
+ print('Wait for the dynamic address to be expired')
+ time.sleep(130)
+
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'global', 'dynamic'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'inet 192.168.5.2[0-9]*/24 brd 192.168.5.255 scope global dynamic veth99')
+ address2=None
+ for line in output.splitlines():
+ if 'brd 192.168.5.255 scope global dynamic veth99' in line:
+ address2 = line.split()[1].split('/')[0]
+ break
+
+ self.assertNotEqual(address1, address2)
+
+ output = subprocess.check_output(['ip', '-4', 'route', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertNotRegex(output, f'default via 192.168.5.1 proto dhcp src {address1} metric 1024')
+ self.assertNotRegex(output, f'192.168.5.1 proto dhcp scope link src {address1} metric 1024')
+ self.assertRegex(output, f'default via 192.168.5.1 proto dhcp src {address2} metric 1024')
+ self.assertRegex(output, f'192.168.5.1 proto dhcp scope link src {address2} metric 1024')
+
if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--build-dir', help='Path to build dir', dest='build_dir')
+ parser.add_argument('--networkd', help='Path to systemd-networkd', dest='networkd_bin')
+ parser.add_argument('--wait-online', help='Path to systemd-networkd-wait-online', dest='wait_online_bin')
+ parser.add_argument('--networkctl', help='Path to networkctl', dest='networkctl_bin')
+ parser.add_argument('--valgrind', help='Enable valgrind', dest='use_valgrind', type=bool, nargs='?', const=True, default=use_valgrind)
+ parser.add_argument('--debug', help='Generate debugging logs', dest='enable_debug', type=bool, nargs='?', const=True, default=enable_debug)
+ parser.add_argument('--asan-options', help='ASAN options', dest='asan_options')
+ parser.add_argument('--ubsan-options', help='UBSAN options', dest='ubsan_options')
+ ns, args = parser.parse_known_args(namespace=unittest)
+
+ if ns.build_dir:
+ if ns.networkd_bin or ns.wait_online_bin or ns.networkctl_bin:
+ print('WARNING: --networkd, --wait-online, or --networkctl options are ignored when --build-dir is specified.')
+ networkd_bin = os.path.join(ns.build_dir, 'systemd-networkd')
+ wait_online_bin = os.path.join(ns.build_dir, 'systemd-networkd-wait-online')
+ networkctl_bin = os.path.join(ns.build_dir, 'networkctl')
+ else:
+ if ns.networkd_bin:
+ networkd_bin = ns.networkd_bin
+ if ns.wait_online_bin:
+ wait_online_bin = ns.wait_online_bin
+ if ns.networkctl_bin:
+ networkctl_bin = ns.networkctl_bin
+
+ use_valgrind = ns.use_valgrind
+ enable_debug = ns.enable_debug
+ asan_options = ns.asan_options
+ ubsan_options = ns.ubsan_options
+
+ if use_valgrind:
+ networkctl_cmd = ['valgrind', '--track-origins=yes', '--leak-check=full', '--show-leak-kinds=all', networkctl_bin]
+ wait_online_cmd = ['valgrind', '--track-origins=yes', '--leak-check=full', '--show-leak-kinds=all', wait_online_bin]
+ else:
+ networkctl_cmd = [networkctl_bin]
+ wait_online_cmd = [wait_online_bin]
+
+ if enable_debug:
+ env.update({ 'SYSTEMD_LOG_LEVEL' : 'debug' })
+ if asan_options:
+ env.update({ 'ASAN_OPTIONS' : asan_options })
+ if ubsan_options:
+ env.update({ 'UBSAN_OPTIONS' : ubsan_options })
+
+ sys.argv[1:] = args
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
verbosity=3))