# 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
+lsan_options=None
+ubsan_options=None
def is_module_available(module_name):
lsmod_output = subprocess.check_output('lsmod', universal_newlines=True)
copytree(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'conf'), networkd_ci_path)
subprocess.check_call('systemctl stop systemd-networkd.socket', shell=True)
+ subprocess.check_call('systemctl stop systemd-networkd.service', 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 lsan_options:
+ drop_in += ['Environment=LSAN_OPTIONS="' + lsan_options + '"']
+ if ubsan_options:
+ drop_in += ['Environment=UBSAN_OPTIONS="' + ubsan_options + '"']
+ if use_valgrind or asan_options or lsan_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)
subprocess.check_call('systemctl stop systemd-networkd.service', shell=True)
+
+ shutil.rmtree('/run/systemd/system/systemd-networkd.service.d')
+ subprocess.check_call('systemctl daemon-reload', shell=True)
+
subprocess.check_call('systemctl start systemd-networkd.socket', shell=True)
subprocess.check_call('systemctl start systemd-networkd.service', shell=True)
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=10):
+ 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)
'isataptun99',
'macvlan99',
'macvtap99',
+ 'nlmon99',
'sittun96',
'sittun97',
'sittun98',
'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',
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.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):
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')
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:routable'])
- 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, 'Address: 192.168.42.100')
self.assertRegex(output, 'DNS: 192.168.42.1')
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')
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')
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')
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.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')
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')
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', '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.assertRegex(output, '192.168.5')
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.*')
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')
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('--lsan-options', help='LSAN options', dest='lsan_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
+ lsan_options = ns.lsan_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 lsan_options:
+ env.update({ 'LSAN_OPTIONS' : lsan_options })
+ if ubsan_options:
+ env.update({ 'UBSAN_OPTIONS' : ubsan_options })
+
+ sys.argv[1:] = args
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
verbosity=3))
projects / thirdparty / systemd.git / blobdiff