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'
-resolved_bin='/usr/lib/systemd/systemd-resolved'
-wait_online_bin='/usr/lib/systemd/systemd-networkd-wait-online'
-networkctl_bin='/usr/bin/networkctl'
-resolvectl_bin='/usr/bin/resolvectl'
-timedatectl_bin='/usr/bin/timedatectl'
+systemd_lib_paths=['/usr/lib/systemd', '/lib/systemd']
+which_paths=':'.join(systemd_lib_paths + os.getenv('PATH', os.defpath).lstrip(':').split(':'))
+
+networkd_bin=shutil.which('systemd-networkd', path=which_paths)
+resolved_bin=shutil.which('systemd-resolved', path=which_paths)
+wait_online_bin=shutil.which('systemd-networkd-wait-online', path=which_paths)
+networkctl_bin=shutil.which('networkctl', path=which_paths)
+resolvectl_bin=shutil.which('resolvectl', path=which_paths)
+timedatectl_bin=shutil.which('timedatectl', path=which_paths)
+
use_valgrind=False
enable_debug=True
env = {}
def is_module_available(module_name):
lsmod_output = check_output('lsmod')
module_re = re.compile(rf'^{re.escape(module_name)}\b', re.MULTILINE)
- return module_re.search(lsmod_output) or not call('modprobe', module_name)
+ return module_re.search(lsmod_output) or not call('modprobe', module_name, stderr=subprocess.DEVNULL)
def expectedFailureIfModuleIsNotAvailable(module_name):
def f(func):
def expectedFailureIfERSPANModuleIsNotAvailable():
def f(func):
- rc = call('ip link add dev erspan99 type erspan seq key 30 local 192.168.1.4 remote 192.168.1.1 erspan_ver 1 erspan 123')
+ rc = call('ip link add dev erspan99 type erspan seq key 30 local 192.168.1.4 remote 192.168.1.1 erspan_ver 1 erspan 123', stderr=subprocess.DEVNULL)
if rc == 0:
call('ip link del erspan99')
return func
def expectedFailureIfRoutingPolicyPortRangeIsNotAvailable():
def f(func):
- rc = call('ip rule add from 192.168.100.19 sport 1123-1150 dport 3224-3290 table 7')
+ rc = call('ip rule add from 192.168.100.19 sport 1123-1150 dport 3224-3290 table 7', stderr=subprocess.DEVNULL)
if rc == 0:
call('ip rule del from 192.168.100.19 sport 1123-1150 dport 3224-3290 table 7')
return func
def expectedFailureIfRoutingPolicyIPProtoIsNotAvailable():
def f(func):
- rc = call('ip rule add not from 192.168.100.19 ipproto tcp table 7')
+ rc = call('ip rule add not from 192.168.100.19 ipproto tcp table 7', stderr=subprocess.DEVNULL)
if rc == 0:
call('ip rule del not from 192.168.100.19 ipproto tcp table 7')
return func
def expectedFailureIfLinkFileFieldIsNotSet():
def f(func):
support = False
- rc = call('ip link add name dummy99 type dummy')
+ rc = call('ip link add name dummy99 type dummy', stderr=subprocess.DEVNULL)
if rc == 0:
ret = run('udevadm info -w10s /sys/class/net/dummy99', stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
if ret.returncode == 0 and 'E: ID_NET_LINK_FILE=' in ret.stdout.rstrip():
def expectedFailureIfNexthopIsNotAvailable():
def f(func):
- rc = call('ip nexthop list')
+ rc = call('ip nexthop list', stderr=subprocess.DEVNULL)
+ if rc == 0:
+ return func
+ else:
+ return unittest.expectedFailure(func)
+
+ return f
+
+def expectedFailureIfAlternativeNameIsNotAvailable():
+ def f(func):
+ call('ip link add dummy98 type dummy', stderr=subprocess.DEVNULL)
+ rc = call('ip link prop add dev dummy98 altname hogehogehogehogehoge', stderr=subprocess.DEVNULL)
if rc == 0:
return func
else:
running_units.append(u)
drop_in = [
+ '[Unit]',
+ 'StartLimitIntervalSec=0',
'[Service]',
'Restart=no',
'ExecStart=',
for u in running_units:
check_output(f'systemctl start {u}')
-def read_link_attr(link, dev, attribute):
- with open(os.path.join(os.path.join(os.path.join('/sys/class/net/', link), dev), attribute)) as f:
+def read_link_attr(*args):
+ with open(os.path.join('/sys/class/net/', *args)) as f:
return f.readline().strip()
def read_bridge_port_attr(bridge, link, attribute):
with open(os.path.join(os.path.join(network_sysctl_ipv4_path, link), attribute)) as f:
return f.readline().strip()
-def copy_unit_to_networkd_unit_path(*units):
+def copy_unit_to_networkd_unit_path(*units, dropins=True):
+ """Copy networkd unit files into the testbed.
+
+ Any networkd unit file type can be specified, as well as drop-in files.
+
+ By default, all drop-ins for a specified unit file are copied in;
+ to avoid that specify dropins=False.
+
+ When a drop-in file is specified, its unit file is also copied in automatically.
+ """
print()
for unit in units:
- shutil.copy(os.path.join(networkd_ci_path, unit), network_unit_file_path)
- if (os.path.exists(os.path.join(networkd_ci_path, unit + '.d'))):
+ if dropins and os.path.exists(os.path.join(networkd_ci_path, unit + '.d')):
copytree(os.path.join(networkd_ci_path, unit + '.d'), os.path.join(network_unit_file_path, unit + '.d'))
+ if unit.endswith('.conf'):
+ dropin = unit
+ dropindir = os.path.join(network_unit_file_path, os.path.dirname(dropin))
+ os.makedirs(dropindir, exist_ok=True)
+ shutil.copy(os.path.join(networkd_ci_path, dropin), dropindir)
+ unit = os.path.dirname(dropin).rstrip('.d')
+ shutil.copy(os.path.join(networkd_ci_path, unit), network_unit_file_path)
def remove_unit_from_networkd_path(units):
+ """Remove previously copied unit files from the testbed.
+
+ Drop-ins will be removed automatically.
+ """
for unit in units:
if (os.path.exists(os.path.join(network_unit_file_path, unit))):
os.remove(os.path.join(network_unit_file_path, unit))
stop_networkd(show_logs, remove_state_files)
start_networkd(sleep_sec)
-def get_operstate(link, show_status=True, setup_state='configured'):
- output = check_output(*networkctl_cmd, 'status', link, env=env)
- if show_status:
- print(output)
- for line in output.splitlines():
- if 'State:' in line and (not setup_state or setup_state in line):
- return line.split()[1]
- return None
class Utilities():
def check_link_exists(self, link):
self.assertTrue(link_exists(link))
- def check_operstate(self, link, expected, show_status=True, setup_state='configured'):
- self.assertRegex(get_operstate(link, show_status, setup_state), expected)
+ def wait_operstate(self, link, operstate='degraded', setup_state='configured', setup_timeout=5, fail_assert=True):
+ """Wait for the link to reach the specified operstate and/or setup state.
+
+ Specify None or '' for either operstate or setup_state to ignore that state.
+ This will recheck until the state conditions are met or the timeout expires.
+
+ If the link successfully matches the requested state, this returns True.
+ If this times out waiting for the link to match, the behavior depends on the
+ 'fail_assert' parameter; if True, this causes a test assertion failure,
+ otherwise this returns False. The default is to cause assertion failure.
+
+ Note that this function matches on *exactly* the given operstate and setup_state.
+ To wait for a link to reach *or exceed* a given operstate, use wait_online().
+ """
+ if not operstate:
+ operstate = r'\S+'
+ if not setup_state:
+ setup_state = r'\S+'
+
+ for secs in range(setup_timeout + 1):
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', link, env=env)
+ print(output)
+ if re.search(rf'(?m)^\s*State:\s+{operstate}\s+\({setup_state}\)\s*$', output):
+ return True
+ # don't bother sleeping if time is up
+ if secs < setup_timeout:
+ time.sleep(1)
+ if fail_assert:
+ self.fail(f'Timed out waiting for {link} to reach state {operstate}/{setup_state}')
+ return False
+
+ def wait_online(self, links_with_operstate, timeout='20s', bool_any=False, setup_state='configured', setup_timeout=5):
+ """Wait for the link(s) to reach the specified operstate and/or setup state.
- def wait_online(self, links_with_operstate, timeout='20s', bool_any=False, setup_state='configured'):
+ This is similar to wait_operstate() but can be used for multiple links,
+ and it also calls systemd-networkd-wait-online to wait for the given operstate.
+ The operstate should be specified in the link name, like 'eth0:degraded'.
+ If just a link name is provided, wait-online's default operstate to wait for is degraded.
+
+ The 'timeout' parameter controls the systemd-networkd-wait-online timeout, and the
+ 'setup_timeout' controls the per-link timeout waiting for the setup_state.
+
+ Set 'bool_any' to True to wait for any (instead of all) of the given links.
+ If this is set, no setup_state checks are done.
+
+ Note that this function waits for the link(s) to reach *or exceed* the given operstate.
+ However, the setup_state, if specified, must be matched *exactly*.
+
+ This returns if the link(s) reached the requested operstate/setup_state; otherwise it
+ raises CalledProcessError or fails test assertion.
+ """
args = wait_online_cmd + [f'--timeout={timeout}'] + [f'--interface={link}' for link in links_with_operstate]
if bool_any:
args += ['--any']
check_output(*args, env=env)
except subprocess.CalledProcessError:
for link in links_with_operstate:
- output = check_output(*networkctl_cmd, 'status', link.split(':')[0], env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', link.split(':')[0], env=env)
print(output)
raise
- if not bool_any:
+ if not bool_any and setup_state:
for link in links_with_operstate:
- output = check_output(*networkctl_cmd, 'status', link.split(':')[0])
- print(output)
- for line in output.splitlines():
- if 'State:' in line:
- self.assertRegex(line, setup_state)
+ self.wait_operstate(link.split(':')[0], None, setup_state, setup_timeout)
def wait_address(self, link, address_regex, scope='global', ipv='', timeout_sec=100):
for i in range(timeout_sec):
'11-dummy-mtu.netdev',
'11-dummy.network',
'12-dummy.netdev',
+ '12-dummy.link',
'25-address-static.network',
'25-veth.netdev',
'netdev-link-local-addressing-yes.network',
remove_unit_from_networkd_path(self.units)
stop_networkd(show_logs=True)
+ @expectedFailureIfAlternativeNameIsNotAvailable()
+ def test_altname(self):
+ copy_unit_to_networkd_unit_path('netdev-link-local-addressing-yes.network', '12-dummy.netdev', '12-dummy.link')
+ check_output('udevadm control --reload')
+ start_networkd()
+ self.wait_online(['dummy98:degraded'])
+
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'dummy98', env=env)
+ self.assertRegex(output, 'hogehogehogehogehogehoge')
+
def test_reconfigure(self):
copy_unit_to_networkd_unit_path('25-address-static.network', '12-dummy.netdev')
start_networkd()
copy_unit_to_networkd_unit_path('11-dummy.netdev')
check_output(*networkctl_cmd, 'reload', env=env)
- time.sleep(3)
- self.check_link_exists('test1')
- self.check_operstate('test1', 'off', setup_state='unmanaged')
+ self.wait_operstate('test1', 'off', setup_state='unmanaged')
copy_unit_to_networkd_unit_path('11-dummy.network')
check_output(*networkctl_cmd, 'reload', env=env)
remove_unit_from_networkd_path(['11-dummy.network'])
check_output(*networkctl_cmd, 'reload', env=env)
- time.sleep(1)
- self.check_operstate('test1', 'degraded', setup_state='unmanaged')
+ self.wait_operstate('test1', 'degraded', setup_state='unmanaged')
remove_unit_from_networkd_path(['11-dummy.netdev'])
check_output(*networkctl_cmd, 'reload', env=env)
- self.check_operstate('test1', 'degraded', setup_state='unmanaged')
+ self.wait_operstate('test1', 'degraded', setup_state='unmanaged')
copy_unit_to_networkd_unit_path('11-dummy.netdev', '11-dummy.network')
check_output(*networkctl_cmd, 'reload', env=env)
- self.check_operstate('test1', 'degraded')
+ self.wait_operstate('test1', 'degraded')
def test_glob(self):
copy_unit_to_networkd_unit_path('11-dummy.netdev', '11-dummy.network')
self.assertNotRegex(output, '1 lo ')
self.assertRegex(output, 'test1')
- output = check_output(*networkctl_cmd, 'status', 'te*', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'te*', env=env)
self.assertNotRegex(output, '1: lo ')
self.assertRegex(output, 'test1')
- output = check_output(*networkctl_cmd, 'status', 'tes[a-z][0-9]', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'tes[a-z][0-9]', env=env)
self.assertNotRegex(output, '1: lo ')
self.assertRegex(output, 'test1')
self.wait_online(['test1:degraded'])
- output = check_output(*networkctl_cmd, 'status', 'test1', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'test1', env=env)
self.assertRegex(output, 'MTU: 1600')
def test_type(self):
start_networkd()
self.wait_online(['test1:degraded'])
- output = check_output(*networkctl_cmd, 'status', 'test1')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'test1', env=env)
print(output)
self.assertRegex(output, 'Type: ether')
- output = check_output(*networkctl_cmd, 'status', 'lo')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'lo', env=env)
print(output)
self.assertRegex(output, 'Type: loopback')
start_networkd()
self.wait_online(['test1:degraded'])
- output = check_output(*networkctl_cmd, 'status', 'test1')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'test1', env=env)
print(output)
- self.assertRegex(output, r'Link File: (?:/usr)/lib/systemd/network/99-default.link')
+ self.assertRegex(output, r'Link File: (/usr)?/lib/systemd/network/99-default.link')
self.assertRegex(output, r'Network File: /run/systemd/network/11-dummy.network')
- output = check_output(*networkctl_cmd, 'status', 'lo')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'lo', env=env)
print(output)
- self.assertRegex(output, r'Link File: (?:/usr)/lib/systemd/network/99-default.link')
+ self.assertRegex(output, r'Link File: (/usr)?/lib/systemd/network/99-default.link')
self.assertRegex(output, r'Network File: n/a')
def test_delete_links(self):
self.wait_online(['test1:degraded', 'veth99:degraded', 'veth-peer:degraded'])
- check_output(*networkctl_cmd, 'delete', 'test1', 'veth99')
+ check_output(*networkctl_cmd, 'delete', 'test1', 'veth99', env=env)
self.assertFalse(link_exists('test1'))
self.assertFalse(link_exists('veth99'))
self.assertFalse(link_exists('veth-peer'))
'gretun97',
'gretun98',
'gretun99',
+ 'ifb99',
'ip6gretap98',
'ip6gretap99',
'ip6gretun96',
'25-gre-tunnel-local-any.netdev',
'25-gre-tunnel-remote-any.netdev',
'25-gre-tunnel.netdev',
+ '25-ifb.netdev',
'25-ip6gretap-tunnel-local-any.netdev',
'25-ip6gretap-tunnel.netdev',
'25-ip6gre-tunnel-any-any.netdev',
start_networkd()
self.wait_online(['dummy98:routable'])
- output = check_output('networkctl status dummy98')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'dummy98', env=env)
print(output)
self.assertRegex(output, 'Network File: /run/systemd/network/14-match-udev-property')
self.wait_online(['bridge99', 'test1:degraded'], bool_any=True)
- self.check_operstate('bridge99', '(?:off|no-carrier)', setup_state='configuring')
- self.check_operstate('test1', 'degraded')
+ self.wait_operstate('bridge99', '(off|no-carrier)', setup_state='configuring')
+ self.wait_operstate('test1', 'degraded')
def test_bridge(self):
copy_unit_to_networkd_unit_path('25-bridge.netdev', '25-bridge-configure-without-carrier.network')
self.assertEqual(1, int(read_link_attr('bridge99', 'bridge', 'stp_state')))
self.assertEqual(3, int(read_link_attr('bridge99', 'bridge', 'multicast_igmp_version')))
- output = check_output(*networkctl_cmd, 'status', 'bridge99')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'bridge99', env=env)
print(output)
self.assertRegex(output, 'Priority: 9')
self.assertRegex(output, 'STP: yes')
output = check_output('ip -d link show tun99')
print(output)
# Old ip command does not support IFF_ flags
- self.assertRegex(output, 'tun (?:type tun pi on vnet_hdr on multi_queue|addrgenmode) ')
+ self.assertRegex(output, 'tun (type tun pi on vnet_hdr on multi_queue|addrgenmode) ')
def test_tap(self):
copy_unit_to_networkd_unit_path('25-tap.netdev')
output = check_output('ip -d link show tap99')
print(output)
# Old ip command does not support IFF_ flags
- self.assertRegex(output, 'tun (?:type tap pi on vnet_hdr on multi_queue|addrgenmode) ')
+ self.assertRegex(output, 'tun (type tap pi on vnet_hdr on multi_queue|addrgenmode) ')
@expectedFailureIfModuleIsNotAvailable('vrf')
def test_vrf(self):
output = check_output('ip -d link show ipiptun99')
print(output)
- self.assertRegex(output, 'ipip (?:ipip |)remote 192.169.224.239 local 192.168.223.238 dev dummy98')
+ self.assertRegex(output, 'ipip (ipip )?remote 192.169.224.239 local 192.168.223.238 dev dummy98')
output = check_output('ip -d link show ipiptun98')
print(output)
- self.assertRegex(output, 'ipip (?:ipip |)remote 192.169.224.239 local any dev dummy98')
+ self.assertRegex(output, 'ipip (ipip )?remote 192.169.224.239 local any dev dummy98')
output = check_output('ip -d link show ipiptun97')
print(output)
- self.assertRegex(output, 'ipip (?:ipip |)remote any local 192.168.223.238 dev dummy98')
+ self.assertRegex(output, 'ipip (ipip )?remote any local 192.168.223.238 dev dummy98')
output = check_output('ip -d link show ipiptun96')
print(output)
- self.assertRegex(output, 'ipip (?:ipip |)remote any local any dev dummy98')
+ self.assertRegex(output, 'ipip (ipip )?remote any local any dev dummy98')
def test_gre_tunnel(self):
copy_unit_to_networkd_unit_path('12-dummy.netdev', 'gretun.network',
self.assertRegex(output, 'vti6 remote 2001:473:fece:cafe::5179 local 2a00:ffde:4567:edde::4987 dev dummy98')
output = check_output('ip -d link show vti6tun98')
print(output)
- self.assertRegex(output, 'vti6 remote 2001:473:fece:cafe::5179 local (?:any|::) dev dummy98')
+ self.assertRegex(output, 'vti6 remote 2001:473:fece:cafe::5179 local (any|::) dev dummy98')
output = check_output('ip -d link show vti6tun97')
print(output)
- self.assertRegex(output, 'vti6 remote (?:any|::) local 2a00:ffde:4567:edde::4987 dev dummy98')
+ self.assertRegex(output, 'vti6 remote (any|::) local 2a00:ffde:4567:edde::4987 dev dummy98')
def test_ip6tnl_tunnel(self):
copy_unit_to_networkd_unit_path('12-dummy.netdev', 'ip6tnl.network',
self.assertRegex(output, 'ip6tnl ip6ip6 remote 2001:473:fece:cafe::5179 local 2a00:ffde:4567:edde::4987 dev dummy98')
output = check_output('ip -d link show ip6tnl98')
print(output)
- self.assertRegex(output, 'ip6tnl ip6ip6 remote 2001:473:fece:cafe::5179 local (?:any|::) dev dummy98')
+ self.assertRegex(output, 'ip6tnl ip6ip6 remote 2001:473:fece:cafe::5179 local (any|::) dev dummy98')
output = check_output('ip -d link show ip6tnl97')
print(output)
- self.assertRegex(output, 'ip6tnl ip6ip6 remote (?:any|::) local 2a00:ffde:4567:edde::4987 dev dummy98')
+ self.assertRegex(output, 'ip6tnl ip6ip6 remote (any|::) local 2a00:ffde:4567:edde::4987 dev dummy98')
def test_sit_tunnel(self):
copy_unit_to_networkd_unit_path('12-dummy.netdev', 'sit.network',
output = check_output('ip -d link show sittun99')
print(output)
- self.assertRegex(output, "sit (?:ip6ip |)remote 10.65.223.239 local 10.65.223.238 dev dummy98")
+ self.assertRegex(output, "sit (ip6ip )?remote 10.65.223.239 local 10.65.223.238 dev dummy98")
output = check_output('ip -d link show sittun98')
print(output)
- self.assertRegex(output, "sit (?:ip6ip |)remote 10.65.223.239 local any dev dummy98")
+ self.assertRegex(output, "sit (ip6ip )?remote 10.65.223.239 local any dev dummy98")
output = check_output('ip -d link show sittun97')
print(output)
- self.assertRegex(output, "sit (?:ip6ip |)remote any local 10.65.223.238 dev dummy98")
+ self.assertRegex(output, "sit (ip6ip )?remote any local 10.65.223.238 dev dummy98")
output = check_output('ip -d link show sittun96')
print(output)
- self.assertRegex(output, "sit (?:ip6ip |)remote any local any dev dummy98")
+ self.assertRegex(output, "sit (ip6ip )?remote any local any dev dummy98")
def test_isatap_tunnel(self):
copy_unit_to_networkd_unit_path('12-dummy.netdev', 'isatap.network',
self.assertRegex(output, '00:11:22:33:44:66 dst 10.0.0.6 self permanent')
self.assertRegex(output, '00:11:22:33:44:77 dst 10.0.0.7 self permanent')
- output = check_output(*networkctl_cmd, 'status', 'vxlan99')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'vxlan99', env=env)
print(output)
self.assertRegex(output, 'VNI: 999')
self.assertRegex(output, 'Destination Port: 5555')
self.wait_online(['nlmon99:carrier'])
+ @expectedFailureIfModuleIsNotAvailable('ifb')
+ def test_ifb(self):
+ copy_unit_to_networkd_unit_path('25-ifb.netdev', 'netdev-link-local-addressing-yes.network')
+ start_networkd()
+
+ self.wait_online(['ifb99:degraded'])
+
class NetworkdL2TPTests(unittest.TestCase, Utilities):
links =[
'23-active-slave.network',
'24-keep-configuration-static.network',
'24-search-domain.network',
+ '25-address-dad-veth-peer.network',
+ '25-address-dad-veth99.network',
'25-address-link-section.network',
'25-address-preferred-lifetime-zero.network',
'25-address-static.network',
'25-neighbor-ip-dummy.network',
'25-neighbor-ip.network',
'25-nexthop.network',
- '25-qdisc.network',
+ '25-qdisc-fq-codel.network',
+ '25-qdisc-netem-and-fqcodel.network',
+ '25-qdisc-tbf-and-sfq.network',
'25-route-ipv6-src.network',
'25-route-static.network',
'25-gateway-static.network',
print(output)
self.assertRegex(output, 'default via 20.20.20.1 proto static')
+ def test_address_dad(self):
+ copy_unit_to_networkd_unit_path('25-address-dad-veth99.network', '25-address-dad-veth-peer.network',
+ '25-veth.netdev')
+ start_networkd()
+ self.wait_online(['veth99:routable', 'veth-peer:degraded'])
+
+ output = check_output('ip -4 address show dev veth99')
+ print(output)
+ self.assertRegex(output, '192.168.100.10/24')
+
+ output = check_output('ip -4 address show dev veth-peer')
+ print(output)
+ self.assertNotRegex(output, '192.168.100.10/24')
+
def test_configure_without_carrier(self):
copy_unit_to_networkd_unit_path('configure-without-carrier.network', '11-dummy.netdev')
start_networkd()
self.wait_online(['test1:routable'])
- output = check_output(*networkctl_cmd, 'status', 'test1')
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'test1', env=env)
print(output)
self.assertRegex(output, '192.168.0.15')
self.assertRegex(output, '192.168.0.1')
print(output)
self.assertRegex(output, '111:')
self.assertRegex(output, 'from 192.168.100.18')
- self.assertRegex(output, r'tos (?:0x08|throughput)\s')
+ self.assertRegex(output, r'tos (0x08|throughput)\s')
self.assertRegex(output, 'iif test1')
self.assertRegex(output, 'oif test1')
self.assertRegex(output, 'lookup 7')
output = check_output('ip rule list table 7')
print(output)
- self.assertRegex(output, '111: from 192.168.100.18 tos (?:0x08|throughput) iif test1 oif test1 lookup 7')
+ self.assertRegex(output, '111: from 192.168.100.18 tos (0x08|throughput) iif test1 oif test1 lookup 7')
output = check_output('ip rule list table 8')
print(output)
- self.assertRegex(output, '112: from 192.168.101.18 tos (?:0x08|throughput) iif dummy98 oif dummy98 lookup 8')
+ self.assertRegex(output, '112: from 192.168.101.18 tos (0x08|throughput) iif dummy98 oif dummy98 lookup 8')
stop_networkd(remove_state_files=False)
start_networkd()
self.wait_online(['dummy98:routable'])
- output = check_output(*networkctl_cmd, 'status', 'dummy98', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'dummy98', env=env)
print(output)
print('### ip -6 route show dev dummy98')
print(output)
self.assertRegex(output, 'prohibit 202.54.1.4 proto static')
+ print('### ip route show 192.168.10.1')
+ output = check_output('ip route show 192.168.10.1')
+ print(output)
+ self.assertRegex(output, '192.168.10.1 proto static')
+ self.assertRegex(output, 'nexthop via 149.10.124.59 dev dummy98 weight 10')
+ self.assertRegex(output, 'nexthop via 149.10.124.60 dev dummy98 weight 5')
+
+ print('### ip route show 192.168.10.2')
+ output = check_output('ip route show 192.168.10.2')
+ print(output)
+ # old ip command does not show IPv6 gateways...
+ self.assertRegex(output, '192.168.10.2 proto static')
+ self.assertRegex(output, 'nexthop')
+ self.assertRegex(output, 'dev dummy98 weight 10')
+ self.assertRegex(output, 'dev dummy98 weight 5')
+
+ print('### ip -6 route show 2001:1234:5:7fff:ff:ff:ff:ff')
+ output = check_output('ip -6 route show 2001:1234:5:7fff:ff:ff:ff:ff')
+ print(output)
+ # old ip command does not show 'nexthop' keyword and weight...
+ self.assertRegex(output, '2001:1234:5:7fff:ff:ff:ff:ff')
+ self.assertRegex(output, 'via 2001:1234:5:8fff:ff:ff:ff:ff dev dummy98')
+ self.assertRegex(output, 'via 2001:1234:5:9fff:ff:ff:ff:ff dev dummy98')
+
def test_gateway_reconfigure(self):
copy_unit_to_networkd_unit_path('25-gateway-static.network', '12-dummy.netdev')
start_networkd()
self.check_link_exists('dummy98')
- self.check_operstate('dummy98', 'off', setup_state='unmanaged')
+ self.wait_operstate('dummy98', 'off', setup_state='unmanaged')
def test_ipv6_address_label(self):
copy_unit_to_networkd_unit_path('25-ipv6-address-label-section.network', '12-dummy.netdev')
self.assertRegex(output, 'default via 2607:5300:203:39ff:ff:ff:ff:ff proto static')
def test_bind_carrier(self):
+ check_output('ip link add dummy98 type dummy')
+ check_output('ip link set dummy98 up')
+ time.sleep(2)
+
copy_unit_to_networkd_unit_path('25-bind-carrier.network', '11-dummy.netdev')
start_networkd()
self.wait_online(['test1:routable'])
- check_output('ip link add dummy98 type dummy')
- check_output('ip link set dummy98 up')
- time.sleep(2)
output = check_output('ip address show test1')
print(output)
self.assertRegex(output, 'UP,LOWER_UP')
self.assertRegex(output, 'inet 192.168.10.30/24 brd 192.168.10.255 scope global test1')
- self.check_operstate('test1', 'routable')
+ self.wait_operstate('test1', 'routable')
check_output('ip link add dummy99 type dummy')
check_output('ip link set dummy99 up')
print(output)
self.assertRegex(output, 'UP,LOWER_UP')
self.assertRegex(output, 'inet 192.168.10.30/24 brd 192.168.10.255 scope global test1')
- self.check_operstate('test1', 'routable')
+ self.wait_operstate('test1', 'routable')
check_output('ip link del dummy98')
time.sleep(2)
print(output)
self.assertRegex(output, 'UP,LOWER_UP')
self.assertRegex(output, 'inet 192.168.10.30/24 brd 192.168.10.255 scope global test1')
- self.check_operstate('test1', 'routable')
+ self.wait_operstate('test1', 'routable')
- check_output('ip link del dummy99')
+ check_output('ip link set dummy99 down')
time.sleep(2)
output = check_output('ip address show test1')
print(output)
self.assertNotRegex(output, 'UP,LOWER_UP')
self.assertRegex(output, 'DOWN')
self.assertNotRegex(output, '192.168.10')
- self.check_operstate('test1', 'off')
+ self.wait_operstate('test1', 'off')
- check_output('ip link add dummy98 type dummy')
- check_output('ip link set dummy98 up')
+ check_output('ip link set dummy99 up')
time.sleep(2)
output = check_output('ip address show test1')
print(output)
self.assertRegex(output, 'UP,LOWER_UP')
self.assertRegex(output, 'inet 192.168.10.30/24 brd 192.168.10.255 scope global test1')
- self.check_operstate('test1', 'routable')
+ self.wait_operstate('test1', 'routable')
def test_domain(self):
copy_unit_to_networkd_unit_path('12-dummy.netdev', '24-search-domain.network')
start_networkd()
self.wait_online(['dummy98:routable'])
- output = check_output(*networkctl_cmd, 'status', 'dummy98', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'dummy98', env=env)
print(output)
self.assertRegex(output, 'Address: 192.168.42.100')
self.assertRegex(output, 'DNS: 192.168.42.1')
self.assertRegex(output, '192.168.5.1')
def test_qdisc(self):
- copy_unit_to_networkd_unit_path('25-qdisc.network', '12-dummy.netdev')
+ copy_unit_to_networkd_unit_path('25-qdisc-netem-and-fqcodel.network', '12-dummy.netdev',
+ '25-qdisc-tbf-and-sfq.network', '11-dummy.netdev')
start_networkd()
- self.wait_online(['dummy98:routable'])
+ self.wait_online(['dummy98:routable', 'test1:routable'])
output = check_output('tc qdisc show dev dummy98')
print(output)
+ self.assertRegex(output, 'qdisc netem')
self.assertRegex(output, 'limit 100 delay 50.0ms 10.0ms loss 20%')
- self.assertRegex(output, 'limit 200 delay 100.0ms 13.0ms loss 20.5%')
+ self.assertRegex(output, 'qdisc fq_codel')
+ self.assertRegex(output, 'limit 20480p flows 2048 quantum 1400 target 10.0ms ce_threshold 100.0ms interval 200.0ms memory_limit 64Mb ecn')
+ output = check_output('tc qdisc show dev test1')
+ print(output)
+ self.assertRegex(output, 'qdisc tbf')
+ self.assertRegex(output, 'rate 1Gbit burst 5000b peakrate 100Gbit minburst 987500b lat 70.0ms')
+ self.assertRegex(output, 'qdisc sfq')
+ self.assertRegex(output, 'perturb 5sec')
+
+ def test_qdisc2(self):
+ copy_unit_to_networkd_unit_path('25-qdisc-fq-codel.network', '12-dummy.netdev')
+ start_networkd()
+
+ self.wait_online(['dummy98:routable'])
+
+ output = check_output('tc qdisc show dev dummy98')
+ print(output)
+ self.assertRegex(output, 'qdisc fq')
+ self.assertRegex(output, 'limit 1000p flow_limit 200p buckets 512 orphan_mask 511 quantum 1500 initial_quantum 13000 maxrate 1Mbit')
+ self.assertRegex(output, 'qdisc codel')
+ self.assertRegex(output, 'limit 2000p target 10.0ms ce_threshold 100.0ms interval 50.0ms ecn')
class NetworkdStateFileTests(unittest.TestCase, Utilities):
links = [
self.assertRegex(data, r'LLMNR=no')
self.assertRegex(data, r'MDNS=yes')
self.assertRegex(data, r'DNSSEC=no')
- self.assertRegex(data, r'ADDRESSES=192.168.(?:10.10|12.12)/24 192.168.(?:12.12|10.10)/24')
+ self.assertRegex(data, r'ADDRESSES=192.168.(10.10|12.12)/24 192.168.(12.12|10.10)/24')
check_output(*resolvectl_cmd, 'dns', 'dummy98', '10.10.10.12', '10.10.10.13', env=env)
check_output(*resolvectl_cmd, 'domain', 'dummy98', 'hogehogehoge', '~foofoofoo', env=env)
print(output)
self.assertRegex(output, 'MASTER,UP,LOWER_UP')
- self.check_operstate('dummy98', 'enslaved')
- self.check_operstate('test1', 'enslaved')
- self.check_operstate('bond99', 'routable')
+ self.wait_operstate('dummy98', 'enslaved')
+ self.wait_operstate('test1', 'enslaved')
+ self.wait_operstate('bond99', 'routable')
check_output('ip link set dummy98 down')
- time.sleep(2)
- self.check_operstate('dummy98', 'off')
- self.check_operstate('test1', 'enslaved')
- self.check_operstate('bond99', 'degraded-carrier')
+ self.wait_operstate('dummy98', 'off')
+ self.wait_operstate('test1', 'enslaved')
+ self.wait_operstate('bond99', 'degraded-carrier')
check_output('ip link set dummy98 up')
- time.sleep(2)
- self.check_operstate('dummy98', 'enslaved')
- self.check_operstate('test1', 'enslaved')
- self.check_operstate('bond99', 'routable')
+ self.wait_operstate('dummy98', 'enslaved')
+ self.wait_operstate('test1', 'enslaved')
+ self.wait_operstate('bond99', 'routable')
check_output('ip link set dummy98 down')
check_output('ip link set test1 down')
- time.sleep(2)
- self.check_operstate('dummy98', 'off')
- self.check_operstate('test1', 'off')
+ self.wait_operstate('dummy98', 'off')
+ self.wait_operstate('test1', 'off')
- for trial in range(30):
- if trial > 0:
- time.sleep(1)
- output = check_output('ip address show bond99')
- print(output)
- if get_operstate('bond99') == 'no-carrier':
- break
- else:
+ if not self.wait_operstate('bond99', 'no-carrier', setup_timeout=30, fail_assert=False):
# Huh? Kernel does not recognize that all slave interfaces are down?
# Let's confirm that networkd's operstate is consistent with ip's result.
self.assertNotRegex(output, 'NO-CARRIER')
self.assertRegex(output, 'ff00::/8 table local metric 256 pref medium')
self.assertEqual(call('ip link del test1'), 0)
- time.sleep(3)
- self.check_operstate('bridge99', 'degraded-carrier')
+ self.wait_operstate('bridge99', 'degraded-carrier')
check_output('ip link del dummy98')
- time.sleep(3)
- self.check_operstate('bridge99', 'no-carrier')
+ self.wait_operstate('bridge99', 'no-carrier')
output = check_output('ip address show bridge99')
print(output)
print('### ip -6 route list table all dev bridge99')
output = check_output('ip -6 route list table all dev bridge99')
print(output)
- self.assertRegex(output, 'ff00::/8 table local metric 256 (?:linkdown |)pref medium')
+ self.assertRegex(output, 'ff00::/8 table local metric 256 (linkdown )?pref medium')
def test_bridge_ignore_carrier_loss(self):
copy_unit_to_networkd_unit_path('11-dummy.netdev', '12-dummy.netdev', '26-bridge.netdev',
start_networkd()
self.wait_online(['veth99:routable', 'veth-peer:degraded'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, '2002:da8:1:0')
start_networkd()
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, '192.168.5.*')
self.assertRegex(output, 'Gateway: 192.168.5.1')
start_networkd()
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, 'Gateway: 192.168.5.*')
self.assertRegex(output, '192.168.5.*')
'25-vrf.netdev',
'25-vrf.network',
'dhcp-client-anonymize.network',
+ 'dhcp-client-decline.network',
+ 'dhcp-client-gateway-ipv4.network',
+ 'dhcp-client-gateway-ipv6.network',
'dhcp-client-gateway-onlink-implicit.network',
'dhcp-client-ipv4-dhcp-settings.network',
'dhcp-client-ipv4-only-ipv6-disabled.network',
'dhcp-client-with-ipv4ll-fallback-without-dhcp-server.network',
'dhcp-client-with-static-address.network',
'dhcp-client.network',
+ 'dhcp-server-decline.network',
'dhcp-server-veth-peer.network',
'dhcp-v4-server-veth-peer.network',
'dhcp-client-use-domains.network',
start_dnsmasq()
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, '2600::')
self.assertNotRegex(output, '192.168.5')
start_dnsmasq(additional_options='--dhcp-option=option:dns-server,192.168.5.6,192.168.5.7', lease_time='2m')
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertNotRegex(output, '2600::')
self.assertRegex(output, '192.168.5')
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertNotRegex(output, '2600::')
self.assertRegex(output, '192.168.5')
# 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 scope global dynamic', ipv='-4')
- self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, '2600::')
self.assertRegex(output, '192.168.5')
output = check_output('ip address show dev veth99 scope global')
print(output)
- self.assertRegex(output, r'inet6 2600::[0-9a-f]*/128 scope global (?:noprefixroute dynamic|dynamic noprefixroute)')
+ self.assertRegex(output, r'inet6 2600::[0-9a-f]*/128 scope global (noprefixroute dynamic|dynamic noprefixroute)')
output = check_output('ip -6 route show dev veth99')
print(output)
print(output)
self.assertRegex(output, r'192.168.5.*')
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, r'192.168.5.*')
print(output)
self.assertRegex(output, r'192.168.5.*')
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, r'192.168.5.*')
print(output)
self.assertRegex(output, r'192.168.5.*')
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, r'192.168.5.*')
print(output)
self.assertRegex(output, r'192.168.5.*')
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, r'192.168.5.*')
# 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 scope global dynamic', ipv='-4')
- self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
output = check_output('ip address show dev veth99 scope global')
print(output)
# 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 scope global dynamic', ipv='-4')
- self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
print('## ip -d link show dev vrf99')
output = check_output('ip -d link show dev vrf99')
print(output)
self.assertRegex(output, 'inet 169.254.[0-9]*.[0-9]*/16 brd 169.254.255.255 scope link veth99')
self.assertRegex(output, 'inet 192.168.5.[0-9]*/24 brd 192.168.5.255 scope global dynamic veth99')
- self.assertRegex(output, 'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)')
+ self.assertRegex(output, 'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)')
self.assertRegex(output, 'inet6 .* scope link')
print('## ip address show dev veth99')
print(output)
self.assertRegex(output, 'inet 169.254.[0-9]*.[0-9]*/16 brd 169.254.255.255 scope link veth99')
self.assertRegex(output, 'inet 192.168.5.[0-9]*/24 brd 192.168.5.255 scope global dynamic veth99')
- self.assertRegex(output, 'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)')
+ self.assertRegex(output, 'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)')
self.assertRegex(output, 'inet6 .* scope link')
print('## ip route show vrf vrf99')
print(output)
self.assertEqual(output, '')
+ def test_dhcp_client_gateway_ipv4(self):
+ copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network',
+ 'dhcp-client-gateway-ipv4.network')
+ start_networkd()
+ self.wait_online(['veth-peer:carrier'])
+ start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ output = check_output('ip route list dev veth99 10.0.0.0/8')
+ print(output)
+ self.assertRegex(output, '10.0.0.0/8 via 192.168.5.1 proto static')
+
+ def test_dhcp_client_gateway_ipv6(self):
+ copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network',
+ 'dhcp-client-gateway-ipv6.network')
+ start_networkd()
+ self.wait_online(['veth-peer:carrier'])
+ start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ output = check_output('ip -6 route list dev veth99 2001:1234:5:9fff:ff:ff:ff:ff')
+ print(output)
+ self.assertRegex(output, 'via fe80::1034:56ff:fe78:9abd')
+
def test_dhcp_client_gateway_onlink_implicit(self):
copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network',
'dhcp-client-gateway-onlink-implicit.network')
start_dnsmasq()
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, '192.168.5')
# 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 scope global dynamic', ipv='-4')
- self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
time.sleep(3)
output = check_output(*resolvectl_cmd, 'dns', 'veth99', env=env)
# 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 scope global dynamic', ipv='-4')
- self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
time.sleep(3)
output = check_output(*resolvectl_cmd, 'dns', 'veth99', env=env)
# 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 scope global dynamic', ipv='-4')
- self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
time.sleep(3)
output = check_output(*resolvectl_cmd, 'dns', 'veth99', env=env)
# 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 scope global dynamic', ipv='-4')
- self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (?:dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
time.sleep(3)
output = check_output(*resolvectl_cmd, 'dns', 'veth99', env=env)
start_dnsmasq('--dhcp-option=option:domain-search,example.com')
self.wait_online(['veth99:routable', 'veth-peer:routable'])
- output = check_output(*networkctl_cmd, 'status', 'veth99', env=env)
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, 'Search Domains: example.com')
print(output)
self.assertRegex(output, 'example.com')
+ def test_dhcp_client_decline(self):
+ copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-decline.network', 'dhcp-client-decline.network')
+
+ start_networkd()
+ self.wait_online(['veth-peer:carrier'])
+ rc = call(*wait_online_cmd, '--timeout=10s', '--interface=veth99:routable', env=env)
+ self.assertTrue(rc == 1)
+
class NetworkdIPv6PrefixTests(unittest.TestCase, Utilities):
links = ['veth99']
print(output)
self.assertRegex(output, '2001:db8:0:1::/64 proto ra')
+class NetworkdMTUTests(unittest.TestCase, Utilities):
+ links = ['dummy98']
+
+ units = [
+ '12-dummy.netdev',
+ '12-dummy-mtu.netdev',
+ '12-dummy-mtu.link',
+ '12-dummy.network',
+ ]
+
+ def setUp(self):
+ remove_links(self.links)
+ stop_networkd(show_logs=False)
+
+ def tearDown(self):
+ remove_log_file()
+ remove_links(self.links)
+ remove_unit_from_networkd_path(self.units)
+ stop_networkd(show_logs=True)
+
+ def check_mtu(self, mtu, ipv6_mtu=None, reset=True):
+ if not ipv6_mtu:
+ ipv6_mtu = mtu
+
+ # test normal start
+ start_networkd()
+ self.wait_online(['dummy98:routable'])
+ self.assertEqual(read_ipv6_sysctl_attr('dummy98', 'mtu'), ipv6_mtu)
+ self.assertEqual(read_link_attr('dummy98', 'mtu'), mtu)
+
+ # test normal restart
+ restart_networkd()
+ self.wait_online(['dummy98:routable'])
+ self.assertEqual(read_ipv6_sysctl_attr('dummy98', 'mtu'), ipv6_mtu)
+ self.assertEqual(read_link_attr('dummy98', 'mtu'), mtu)
+
+ if reset:
+ self.reset_check_mtu(mtu, ipv6_mtu)
+
+ def reset_check_mtu(self, mtu, ipv6_mtu=None):
+ ''' test setting mtu/ipv6_mtu with interface already up '''
+ stop_networkd()
+
+ # note - changing the device mtu resets the ipv6 mtu
+ run('ip link set up mtu 1501 dev dummy98')
+ run('ip link set up mtu 1500 dev dummy98')
+ self.assertEqual(read_link_attr('dummy98', 'mtu'), '1500')
+ self.assertEqual(read_ipv6_sysctl_attr('dummy98', 'mtu'), '1500')
+
+ self.check_mtu(mtu, ipv6_mtu, reset=False)
+
+ def test_mtu_network(self):
+ copy_unit_to_networkd_unit_path('12-dummy.netdev', '12-dummy.network.d/mtu.conf')
+ self.check_mtu('1600')
+
+ def test_mtu_netdev(self):
+ copy_unit_to_networkd_unit_path('12-dummy-mtu.netdev', '12-dummy.network', dropins=False)
+ # note - MTU set by .netdev happens ONLY at device creation!
+ self.check_mtu('1600', reset=False)
+
+ def test_mtu_link(self):
+ copy_unit_to_networkd_unit_path('12-dummy.netdev', '12-dummy-mtu.link', '12-dummy.network', dropins=False)
+ # must reload udev because it only picks up new files after 3 second delay
+ call('udevadm control --reload')
+ # note - MTU set by .link happens ONLY at udev processing of device 'add' uevent!
+ self.check_mtu('1600', reset=False)
+
+ def test_ipv6_mtu(self):
+ ''' set ipv6 mtu without setting device mtu '''
+ copy_unit_to_networkd_unit_path('12-dummy.netdev', '12-dummy.network.d/ipv6-mtu-1400.conf')
+ self.check_mtu('1500', '1400')
+
+ def test_ipv6_mtu_toolarge(self):
+ ''' try set ipv6 mtu over device mtu (it shouldn't work) '''
+ copy_unit_to_networkd_unit_path('12-dummy.netdev', '12-dummy.network.d/ipv6-mtu-1550.conf')
+ self.check_mtu('1500', '1500')
+
+ def test_mtu_network_ipv6_mtu(self):
+ ''' set ipv6 mtu and set device mtu via network file '''
+ copy_unit_to_networkd_unit_path('12-dummy.netdev', '12-dummy.network.d/mtu.conf', '12-dummy.network.d/ipv6-mtu-1550.conf')
+ self.check_mtu('1600', '1550')
+
+ def test_mtu_netdev_ipv6_mtu(self):
+ ''' set ipv6 mtu and set device mtu via netdev file '''
+ copy_unit_to_networkd_unit_path('12-dummy-mtu.netdev', '12-dummy.network.d/ipv6-mtu-1550.conf')
+ self.check_mtu('1600', '1550', reset=False)
+
+ def test_mtu_link_ipv6_mtu(self):
+ ''' set ipv6 mtu and set device mtu via link file '''
+ copy_unit_to_networkd_unit_path('12-dummy.netdev', '12-dummy-mtu.link', '12-dummy.network.d/ipv6-mtu-1550.conf')
+ # must reload udev because it only picks up new files after 3 second delay
+ call('udevadm control --reload')
+ self.check_mtu('1600', '1550', reset=False)
+
+
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--build-dir', help='Path to build dir', dest='build_dir')
projects / thirdparty / systemd.git / blobdiff