# systemd-networkd tests
import argparse
+import itertools
import os
import re
import shutil
def check_link_exists(self, link):
self.assertTrue(link_exists(link))
+ def check_link_attr(self, *args):
+ self.assertEqual(read_link_attr(*args[:-1]), args[-1]);
+
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.
'vtitun99',
'vxcan99',
'vxlan99',
+ 'wg97',
'wg98',
'wg99',
]
'25-vxlan.netdev',
'25-wireguard-23-peers.netdev',
'25-wireguard-23-peers.network',
+ '25-wireguard-no-peer.netdev',
+ '25-wireguard-no-peer.network',
'25-wireguard-preshared-key.txt',
'25-wireguard-private-key.txt',
'25-wireguard.netdev',
def test_wireguard(self):
copy_unit_to_networkd_unit_path('25-wireguard.netdev', '25-wireguard.network',
'25-wireguard-23-peers.netdev', '25-wireguard-23-peers.network',
- '25-wireguard-preshared-key.txt', '25-wireguard-private-key.txt')
+ '25-wireguard-preshared-key.txt', '25-wireguard-private-key.txt',
+ '25-wireguard-no-peer.netdev', '25-wireguard-no-peer.network')
start_networkd()
- self.wait_online(['wg99:carrier', 'wg98:routable'])
+ self.wait_online(['wg99:carrier', 'wg98:routable', 'wg97:carrier'])
if shutil.which('wg'):
call('wg')
output = check_output('wg show wg98 private-key')
self.assertRegex(output, r'CJQUtcS9emY2fLYqDlpSZiE/QJyHkPWr\+WHtZLZ90FU=')
+ output = check_output('wg show wg97 listen-port')
+ self.assertRegex(output, '51821')
+ output = check_output('wg show wg97 fwmark')
+ self.assertRegex(output, '0x4d3')
+
def test_geneve(self):
copy_unit_to_networkd_unit_path('25-geneve.netdev', 'netdev-link-local-addressing-yes.network')
start_networkd()
'25-qdisc-hhf.network',
'25-qdisc-ingress-netem-compat.network',
'25-qdisc-pie.network',
+ '25-qdisc-qfq.network',
'25-route-ipv6-src.network',
'25-route-static.network',
'25-route-vrf.network',
'25-gateway-next-static.network',
'25-sysctl-disable-ipv6.network',
'25-sysctl.network',
+ '25-test1.network',
'25-veth-peer.network',
'25-veth.netdev',
'25-vrf.netdev',
'26-link-local-addressing-ipv6.network',
- 'configure-without-carrier.network',
'routing-policy-rule-dummy98.network',
'routing-policy-rule-test1.network']
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')
+ copy_unit_to_networkd_unit_path('11-dummy.netdev')
start_networkd()
- self.wait_online(['test1:routable'])
+ self.wait_operstate('test1', 'off', '')
+ check_output('ip link set dev test1 up carrier off')
- 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')
- self.assertRegex(output, 'routable')
+ copy_unit_to_networkd_unit_path('25-test1.network.d/configure-without-carrier.conf', dropins=False)
+ restart_networkd()
+ self.wait_online(['test1:no-carrier'])
+
+ carrier_map = {'on': '1', 'off': '0'}
+ routable_map = {'on': 'routable', 'off': 'no-carrier'}
+ for carrier in ['off', 'on', 'off']:
+ with self.subTest(carrier=carrier):
+ if carrier_map[carrier] != read_link_attr('test1', 'carrier'):
+ check_output(f'ip link set dev test1 carrier {carrier}')
+ self.wait_online([f'test1:{routable_map[carrier]}'])
+
+ 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')
+ self.assertRegex(output, routable_map[carrier])
+
+ def test_configure_without_carrier_yes_ignore_carrier_loss_no(self):
+ copy_unit_to_networkd_unit_path('11-dummy.netdev')
+ start_networkd()
+ self.wait_operstate('test1', 'off', '')
+ check_output('ip link set dev test1 up carrier off')
+
+ copy_unit_to_networkd_unit_path('25-test1.network')
+ restart_networkd()
+ self.wait_online(['test1:no-carrier'])
+
+ carrier_map = {'on': '1', 'off': '0'}
+ routable_map = {'on': 'routable', 'off': 'no-carrier'}
+ for (carrier, have_config) in [('off', True), ('on', True), ('off', False)]:
+ with self.subTest(carrier=carrier, have_config=have_config):
+ if carrier_map[carrier] != read_link_attr('test1', 'carrier'):
+ check_output(f'ip link set dev test1 carrier {carrier}')
+ self.wait_online([f'test1:{routable_map[carrier]}'])
+
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'test1', env=env)
+ print(output)
+ if have_config:
+ self.assertRegex(output, '192.168.0.15')
+ self.assertRegex(output, '192.168.0.1')
+ else:
+ self.assertNotRegex(output, '192.168.0.15')
+ self.assertNotRegex(output, '192.168.0.1')
+ self.assertRegex(output, routable_map[carrier])
def test_routing_policy_rule(self):
copy_unit_to_networkd_unit_path('routing-policy-rule-test1.network', '11-dummy.netdev')
self.assertEqual(read_ipv6_sysctl_attr('dummy98', 'proxy_ndp'), '1')
self.assertEqual(read_ipv4_sysctl_attr('dummy98', 'forwarding'),'1')
self.assertEqual(read_ipv4_sysctl_attr('dummy98', 'proxy_arp'), '1')
+ self.assertEqual(read_ipv4_sysctl_attr('dummy98', 'accept_local'), '1')
def test_sysctl_disable_ipv6(self):
copy_unit_to_networkd_unit_path('25-sysctl-disable-ipv6.network', '12-dummy.netdev')
output = check_output('tc qdisc show dev test1')
print(output)
self.assertRegex(output, 'qdisc netem')
- self.assertRegex(output, 'limit 100 delay 50.0ms 10.0ms loss 20%')
+ self.assertRegex(output, 'limit 100 delay 50(.0)?ms 10(.0)?ms loss 20%')
self.assertRegex(output, 'qdisc ingress')
output = check_output('tc qdisc show dev dummy98')
self.assertRegex(output, r'default (0x30|30)')
self.assertRegex(output, 'qdisc netem 30: parent 2:30')
- self.assertRegex(output, 'limit 100 delay 50.0ms 10.0ms loss 20%')
+ self.assertRegex(output, 'limit 100 delay 50(.0)?ms 10(.0)?ms loss 20%')
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')
+ self.assertRegex(output, 'limit 20480p flows 2048 quantum 1400 target 10(.0)?ms ce_threshold 100(.0)?ms interval 200(.0)?ms memory_limit 64Mb ecn')
self.assertRegex(output, 'qdisc teql1 31: parent 2:31')
self.assertRegex(output, 'maxrate 1Mbit')
self.assertRegex(output, 'qdisc codel 33: parent 2:33')
- self.assertRegex(output, 'limit 2000p target 10.0ms ce_threshold 100.0ms interval 50.0ms ecn')
+ self.assertRegex(output, 'limit 2000p target 10(.0)?ms ce_threshold 100(.0)?ms interval 50(.0)?ms ecn')
self.assertRegex(output, 'qdisc fq_codel 34: parent 2:34')
- self.assertRegex(output, 'limit 20480p flows 2048 quantum 1400 target 10.0ms ce_threshold 100.0ms interval 200.0ms memory_limit 64Mb ecn')
+ self.assertRegex(output, 'limit 20480p flows 2048 quantum 1400 target 10(.0)?ms ce_threshold 100(.0)?ms interval 200(.0)?ms memory_limit 64Mb ecn')
self.assertRegex(output, 'qdisc tbf 35: parent 2:35')
- self.assertRegex(output, 'rate 1Gbit burst 5000b peakrate 100Gbit minburst 987500b lat 70.0ms')
+ self.assertRegex(output, 'rate 1Gbit burst 5000b peakrate 100Gbit minburst 987500b lat 70(.0)?ms')
self.assertRegex(output, 'qdisc sfq 36: parent 2:36')
self.assertRegex(output, 'perturb 5sec')
self.assertRegex(output, 'prio 1 rate 1Mbit ceil 500Kbit')
def test_qdisc2(self):
- copy_unit_to_networkd_unit_path('25-qdisc-drr.network', '12-dummy.netdev')
+ copy_unit_to_networkd_unit_path('25-qdisc-drr.network', '12-dummy.netdev',
+ '25-qdisc-qfq.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)
print(output)
self.assertRegex(output, 'class drr 2:30 root quantum 2000b')
+ output = check_output('tc qdisc show dev test1')
+ print(output)
+ self.assertRegex(output, 'qdisc qfq 2: root')
+ output = check_output('tc class show dev test1')
+ print(output)
+ self.assertRegex(output, 'class qfq 2:30 root weight 2 maxpkt 16000')
+ self.assertRegex(output, 'class qfq 2:31 root weight 10 maxpkt 8000')
+
@expectedFailureIfCAKEIsNotAvailable()
def test_qdisc_cake(self):
copy_unit_to_networkd_unit_path('25-qdisc-cake.network', '12-dummy.netdev')
'11-dummy.netdev',
'12-dummy.netdev',
'26-bridge.netdev',
+ '26-bridge-configure-without-carrier.network',
'26-bridge-slave-interface-1.network',
'26-bridge-slave-interface-2.network',
'26-bridge-vlan-master.network',
print(output)
self.assertRegex(output, 'ff00::/8 table local metric 256 (linkdown )?pref medium')
+ def test_bridge_configure_without_carrier(self):
+ copy_unit_to_networkd_unit_path('26-bridge.netdev', '26-bridge-configure-without-carrier.network',
+ '11-dummy.netdev')
+ start_networkd()
+
+ # With ConfigureWithoutCarrier=yes, the bridge should remain configured for all these situations
+ for test in ['no-slave', 'add-slave', 'slave-up', 'slave-no-carrier', 'slave-carrier', 'slave-down']:
+ with self.subTest(test=test):
+ if test == 'no-slave':
+ # bridge has no slaves; it's up but *might* not have carrier
+ # It may take very long time that the interface become configured state.
+ self.wait_online(['bridge99:no-carrier'], timeout='2m', setup_state=None)
+ # due to a bug in the kernel, newly-created bridges are brought up
+ # *with* carrier, unless they have had any setting changed; e.g.
+ # their mac set, priority set, etc. Then, they will lose carrier
+ # as soon as a (down) slave interface is added, and regain carrier
+ # again once the slave interface is brought up.
+ #self.check_link_attr('bridge99', 'carrier', '0')
+ elif test == 'add-slave':
+ # add slave to bridge, but leave it down; bridge is definitely no-carrier
+ self.check_link_attr('test1', 'operstate', 'down')
+ check_output('ip link set dev test1 master bridge99')
+ self.wait_online(['bridge99:no-carrier:no-carrier'], setup_state=None)
+ self.check_link_attr('bridge99', 'carrier', '0')
+ elif test == 'slave-up':
+ # bring up slave, which will have carrier; bridge gains carrier
+ check_output('ip link set dev test1 up')
+ self.wait_online(['bridge99:routable'])
+ self.check_link_attr('bridge99', 'carrier', '1')
+ elif test == 'slave-no-carrier':
+ # drop slave carrier; bridge loses carrier
+ check_output('ip link set dev test1 carrier off')
+ self.wait_online(['bridge99:no-carrier:no-carrier'])
+ self.check_link_attr('bridge99', 'carrier', '0')
+ elif test == 'slave-carrier':
+ # restore slave carrier; bridge gains carrier
+ check_output('ip link set dev test1 carrier on')
+ self.wait_online(['bridge99:routable'])
+ self.check_link_attr('bridge99', 'carrier', '1')
+ elif test == 'slave-down':
+ # bring down slave; bridge loses carrier
+ check_output('ip link set dev test1 down')
+ self.wait_online(['bridge99:no-carrier:no-carrier'])
+ self.check_link_attr('bridge99', 'carrier', '0')
+
+ output = check_output(*networkctl_cmd, '-n', '0', 'status', 'bridge99', env=env)
+ print(output)
+ self.assertRegex(output, '10.1.2.3')
+ self.assertRegex(output, '10.1.2.1')
+
def test_bridge_ignore_carrier_loss(self):
copy_unit_to_networkd_unit_path('11-dummy.netdev', '12-dummy.netdev', '26-bridge.netdev',
'26-bridge-slave-interface-1.network', '26-bridge-slave-interface-2.network',
'dhcp-client-ipv4-dhcp-settings.network',
'dhcp-client-ipv4-only-ipv6-disabled.network',
'dhcp-client-ipv4-only.network',
- 'dhcp-client-ipv4-use-gateway-no.network',
- 'dhcp-client-ipv4-use-routes-no.network',
+ 'dhcp-client-ipv4-use-routes-use-gateway.network',
'dhcp-client-ipv6-only.network',
'dhcp-client-ipv6-rapid-commit.network',
'dhcp-client-keep-configuration-dhcp-on-stop.network',
'dhcp-client-use-dns-no.network',
'dhcp-client-use-dns-yes.network',
'dhcp-client-use-domains.network',
- 'dhcp-client-use-routes-no.network',
'dhcp-client-vrf.network',
'dhcp-client-with-ipv4ll-fallback-with-dhcp-server.network',
'dhcp-client-with-ipv4ll-fallback-without-dhcp-server.network',
'dhcp-server-decline.network',
'dhcp-server-veth-peer.network',
'dhcp-v4-server-veth-peer.network',
- 'dhcp-client-use-domains.network',
'static.network']
def setUp(self):
self.assertRegex(output, r'192.168.5.7 proto dhcp scope link src 192.168.5.181 metric 1024')
self.assertRegex(output, r'192.168.5.8 proto dhcp scope link src 192.168.5.181 metric 1024')
- def test_dhcp_client_ipv4_use_routes_no(self):
- copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-use-routes-no.network')
+ def test_dhcp_client_ipv4_use_routes_gateway(self):
+ for (routes, gateway, dnsroutes) in itertools.product([True, False, None], repeat=3):
+ self.setUp()
+ with self.subTest(routes=routes, gateway=gateway, dnsroutes=dnsroutes):
+ self._test_dhcp_client_ipv4_use_routes_gateway(routes, gateway, dnsroutes)
+ self.tearDown()
+
+ def _test_dhcp_client_ipv4_use_routes_gateway(self, routes, gateway, dnsroutes):
+ testunit = 'dhcp-client-ipv4-use-routes-use-gateway.network'
+ testunits = ['25-veth.netdev', 'dhcp-server-veth-peer.network', testunit]
+ if routes != None:
+ testunits.append(f'{testunit}.d/use-routes-{routes}.conf');
+ if gateway != None:
+ testunits.append(f'{testunit}.d/use-gateway-{gateway}.conf');
+ if dnsroutes != None:
+ testunits.append(f'{testunit}.d/use-dns-routes-{dnsroutes}.conf');
+ copy_unit_to_networkd_unit_path(*testunits, dropins=False)
start_networkd()
self.wait_online(['veth-peer:carrier'])
output = check_output('ip route show dev veth99')
print(output)
- self.assertNotRegex(output, r'192.168.5.5')
- self.assertRegex(output, r'default via 192.168.5.1 proto dhcp src 192.168.5.181 metric 1024')
- self.assertRegex(output, r'192.168.5.1 proto dhcp scope link src 192.168.5.181 metric 1024')
- def test_dhcp_client_ipv4_use_gateway_no(self):
- copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-use-gateway-no.network')
+ # UseRoutes= defaults to true
+ useroutes = routes in [True, None]
+ # UseGateway= defaults to useroutes
+ usegateway = useroutes if gateway == None else gateway
- start_networkd()
- self.wait_online(['veth-peer:carrier'])
- 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'])
+ # Check UseRoutes=
+ if useroutes:
+ self.assertRegex(output, r'192.168.5.0/24 via 192.168.5.5 proto dhcp src 192.168.5.181 metric 1024')
+ else:
+ self.assertNotRegex(output, r'192.168.5.5')
- output = check_output('ip route show dev veth99')
- print(output)
- self.assertRegex(output, r'192.168.5.0/24 via 192.168.5.5 proto dhcp src 192.168.5.181 metric 1024')
- self.assertNotRegex(output, r'default via 192.168.5.1')
+ # Check UseGateway=
+ if usegateway:
+ self.assertRegex(output, r'default via 192.168.5.1 proto dhcp src 192.168.5.181 metric 1024')
+ else:
+ self.assertNotRegex(output, r'default via 192.168.5.1')
+
+ # Check RoutesToDNS=, which defaults to false
+ if dnsroutes:
+ self.assertRegex(output, r'192.168.5.6 proto dhcp scope link src 192.168.5.181 metric 1024')
+ self.assertRegex(output, r'192.168.5.7 proto dhcp scope link src 192.168.5.181 metric 1024')
+ else:
+ self.assertNotRegex(output, r'192.168.5.6')
+ self.assertNotRegex(output, r'192.168.5.7')
def test_dhcp_client_ipv4_ipv6(self):
copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network',
projects / thirdparty / systemd.git / blobdiff