networkd_bin=shutil.which('systemd-networkd', path=which_paths)
resolved_bin=shutil.which('systemd-resolved', path=which_paths)
+udevd_bin=shutil.which('systemd-udevd', 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)
return f
+def expectedFailureIfRoutingPolicyUIDRangeIsNotAvailable():
+ def f(func):
+ support = False
+ rc = call('ip rule add from 192.168.100.19 table 7 uidrange 200-300', stderr=subprocess.DEVNULL)
+ if rc == 0:
+ ret = run('ip rule list from 192.168.100.19 table 7', stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ if ret.returncode == 0 and 'uidrange 200-300' in ret.stdout.rstrip():
+ support = True
+ call('ip rule del from 192.168.100.19 table 7 uidrange 200-300')
+
+ if support:
+ return func
+ else:
+ return unittest.expectedFailure(func)
+
+ return f
+
def expectedFailureIfLinkFileFieldIsNotSet():
def f(func):
support = False
shutil.rmtree(networkd_ci_path)
copytree(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'conf'), networkd_ci_path)
- for u in ['systemd-networkd.socket', 'systemd-networkd.service', 'systemd-resolved.service', 'firewalld.service']:
+ for u in ['systemd-networkd.socket', 'systemd-networkd.service', 'systemd-resolved.service', 'systemd-udevd.service', 'firewalld.service']:
if call(f'systemctl is-active --quiet {u}') == 0:
check_output(f'systemctl stop {u}')
running_units.append(u)
with open('/run/systemd/system/systemd-resolved.service.d/00-override.conf', mode='w') as f:
f.write('\n'.join(drop_in))
+ drop_in = [
+ '[Service]',
+ 'ExecStart=',
+ 'ExecStart=!!' + udevd_bin,
+ ]
+
+ os.makedirs('/run/systemd/system/systemd-udevd.service.d', exist_ok=True)
+ with open('/run/systemd/system/systemd-udevd.service.d/00-override.conf', mode='w') as f:
+ f.write('\n'.join(drop_in))
+
check_output('systemctl daemon-reload')
print(check_output('systemctl cat systemd-networkd.service'))
print(check_output('systemctl cat systemd-resolved.service'))
+ print(check_output('systemctl cat systemd-udevd.service'))
check_output('systemctl restart systemd-resolved')
+ check_output('systemctl restart systemd-udevd')
def tearDownModule():
global running_units
shutil.rmtree(networkd_ci_path)
- for u in ['systemd-networkd.service', 'systemd-resolved.service']:
+ for u in ['systemd-networkd.service', 'systemd-resolved.service', 'systemd-udevd.service']:
check_output(f'systemctl stop {u}')
shutil.rmtree('/run/systemd/system/systemd-networkd.service.d')
shutil.rmtree('/run/systemd/system/systemd-resolved.service.d')
+ shutil.rmtree('/run/systemd/system/systemd-udevd.service.d')
check_output('systemctl daemon-reload')
for u in running_units:
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, '-n', '0', '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.
+
+ 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']
print(output)
raise
if not bool_any and setup_state:
- # check at least once now, then once per sec for setup_timeout secs
- for secs in range(setup_timeout + 1):
- for link in links_with_operstate:
- output = check_output(*networkctl_cmd, '-n', '0', 'status', link.split(':')[0])
- print(output)
- if not re.search(rf'(?m)^\s*State:.*({setup_state}).*$', output):
- # this link isn't in the right state; break into the sleep below
- break
- else:
- # all the links were in the right state; break to exit the timer loop
- break
- # don't bother sleeping if time is up
- if secs < setup_timeout:
- time.sleep(1)
- else:
- self.fail(f'link {link} state does not match {setup_state}')
+ for link in links_with_operstate:
+ 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):
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.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')
'25-bond-active-backup-slave.netdev',
'25-fibrule-invert.network',
'25-fibrule-port-range.network',
+ '25-fibrule-uidrange.network',
'25-gre-tunnel-remote-any.netdev',
'25-ip6gre-tunnel-remote-any.netdev',
'25-ipv6-address-label-section.network',
self.assertRegex(output, 'tcp')
self.assertRegex(output, 'lookup 7')
+ @expectedFailureIfRoutingPolicyUIDRangeIsNotAvailable()
+ def test_routing_policy_rule_uidrange(self):
+ copy_unit_to_networkd_unit_path('25-fibrule-uidrange.network', '11-dummy.netdev')
+ start_networkd()
+ self.wait_online(['test1:degraded'])
+
+ output = check_output('ip rule')
+ print(output)
+ self.assertRegex(output, '111')
+ self.assertRegex(output, 'from 192.168.100.18')
+ self.assertRegex(output, 'lookup 7')
+ self.assertRegex(output, 'uidrange 100-200')
+
def test_route_static(self):
copy_unit_to_networkd_unit_path('25-route-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')
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 set dummy99 down')
time.sleep(2)
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 set dummy99 up')
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')
def test_domain(self):
copy_unit_to_networkd_unit_path('12-dummy.netdev', '24-search-domain.network')
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, 'limit 1000p flow_limit 200p buckets 512 orphan_mask 511')
+ self.assertRegex(output, 'quantum 1500')
+ self.assertRegex(output, 'initial_quantum 13000')
+ self.assertRegex(output, 'maxrate 1Mbit')
self.assertRegex(output, 'qdisc codel')
self.assertRegex(output, 'limit 2000p target 10.0ms ce_threshold 100.0ms interval 50.0ms ecn')
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)
start_networkd()
self.wait_online(['veth99:routable', 'veth-peer:degraded'])
+ output = check_output(*resolvectl_cmd, 'dns', 'veth99', env=env)
+ print(output)
+ self.assertRegex(output, 'fe80::')
+ self.assertRegex(output, '2002:da8:1::1')
+
output = check_output(*networkctl_cmd, '-n', '0', 'status', 'veth99', env=env)
print(output)
self.assertRegex(output, '2002:da8:1:0')
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('--resolved', help='Path to systemd-resolved', dest='resolved_bin')
+ parser.add_argument('--udevd', help='Path to systemd-udevd', dest='udevd_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('--resolvectl', help='Path to resolvectl', dest='resolvectl_bin')
ns, args = parser.parse_known_args(namespace=unittest)
if ns.build_dir:
- if ns.networkd_bin or ns.resolved_bin or ns.wait_online_bin or ns.networkctl_bin or ns.resolvectl_bin or ns.timedatectl_bin:
+ if ns.networkd_bin or ns.resolved_bin or ns.udevd_bin or ns.wait_online_bin or ns.networkctl_bin or ns.resolvectl_bin or ns.timedatectl_bin:
print('WARNING: --networkd, --resolved, --wait-online, --networkctl, --resolvectl, or --timedatectl options are ignored when --build-dir is specified.')
networkd_bin = os.path.join(ns.build_dir, 'systemd-networkd')
resolved_bin = os.path.join(ns.build_dir, 'systemd-resolved')
+ udevd_bin = os.path.join(ns.build_dir, 'systemd-udevd')
wait_online_bin = os.path.join(ns.build_dir, 'systemd-networkd-wait-online')
networkctl_bin = os.path.join(ns.build_dir, 'networkctl')
resolvectl_bin = os.path.join(ns.build_dir, 'resolvectl')
networkd_bin = ns.networkd_bin
if ns.resolved_bin:
resolved_bin = ns.resolved_bin
+ if ns.udevd_bin:
+ udevd_bin = ns.udevd_bin
if ns.wait_online_bin:
wait_online_bin = ns.wait_online_bin
if ns.networkctl_bin:
projects / thirdparty / systemd.git / blobdiff