return f
-def expectedFailureIf_ip6gre_do_not_support_ipv6ll():
- def f(func):
- success = False
- rc = subprocess.call(['ip', 'link', 'add', 'name', 'test1', 'type', 'dummy'])
- if rc == 0:
- time.sleep(1)
- rc = subprocess.call(['ip', 'tunnel', 'add', 'tun99', 'local', '2a00:ffde:4567:edde::4986', 'remote', '2001:473:fece:cafe::5178', 'mode', 'ip6gre', 'dev', 'test1'])
- if rc == 0:
- time.sleep(1)
- # Not sure why, but '0' or '2' do not work.
- subprocess.call(['sysctl', '-w', 'net.ipv6.conf.tun99.addr_gen_mode=3'])
-
- output = subprocess.check_output(['ip', '-6', 'address', 'show', 'dev', 'tun99', 'scope', 'link'], universal_newlines=True).rstrip()
- print(output)
- success = 'inet6' in output
-
- subprocess.run(['ip', 'tunnel', 'del', 'tun99'])
-
- subprocess.run(['ip', 'link', 'del', 'test1'])
-
- if success:
- return func
- else:
- return unittest.expectedFailure(func)
-
- return f
-
def setUpModule():
os.makedirs(network_unit_file_path, exist_ok=True)
os.makedirs(networkd_ci_path, exist_ok=True)
if (os.path.exists(os.path.join(network_unit_file_path, unit + '.d'))):
shutil.rmtree(os.path.join(network_unit_file_path, unit + '.d'))
- def start_dnsmasq(self, additional_options='', lease_time='1h'):
- dnsmasq_command = f'dnsmasq -8 /var/run/networkd-ci/test-dnsmasq-log-file --log-queries=extra --log-dhcp --pid-file=/var/run/networkd-ci/test-test-dnsmasq.pid --conf-file=/dev/null --interface=veth-peer --enable-ra --dhcp-range=2600::10,2600::20,{lease_time} --dhcp-range=192.168.5.10,192.168.5.200,{lease_time} -R --dhcp-leasefile=/var/run/networkd-ci/lease --dhcp-option=26,1492 --dhcp-option=option:router,192.168.5.1 --dhcp-option=33,192.168.5.4,192.168.5.5 --port=0 ' + additional_options
+ def warn_about_firewalld(self):
+ rc = subprocess.call(['systemctl', '-q', 'is-active', 'firewalld.service'])
+ if rc == 0:
+ print('\nWARNING: firewalld.service is active. The test may fail.')
+
+ def start_dnsmasq(self, additional_options='', ipv4_range='192.168.5.10,192.168.5.200', ipv6_range='2600::10,2600::20', lease_time='1h'):
+ self.warn_about_firewalld()
+ dnsmasq_command = f'dnsmasq -8 /var/run/networkd-ci/test-dnsmasq-log-file --log-queries=extra --log-dhcp --pid-file=/var/run/networkd-ci/test-test-dnsmasq.pid --conf-file=/dev/null --interface=veth-peer --enable-ra --dhcp-range={ipv6_range},{lease_time} --dhcp-range={ipv4_range},{lease_time} -R --dhcp-leasefile=/var/run/networkd-ci/lease --dhcp-option=26,1492 --dhcp-option=option:router,192.168.5.1 --dhcp-option=33,192.168.5.4,192.168.5.5 --port=0 ' + additional_options
subprocess.check_call(dnsmasq_command, shell=True)
time.sleep(10)
'vtitun97',
'vtitun98',
'vtitun99',
+ 'vxcan99',
'vxlan99',
'wg98',
'wg99']
'25-vti-tunnel-local-any.netdev',
'25-vti-tunnel-remote-any.netdev',
'25-vti-tunnel.netdev',
+ '25-vxcan.netdev',
'25-vxlan.netdev',
'25-wireguard-23-peers.netdev',
'25-wireguard-23-peers.network',
'macsec.network',
'macvlan.network',
'macvtap.network',
+ 'netdev-link-local-addressing-yes.network',
'sit.network',
'vti6.network',
'vti.network',
self.start_networkd(0)
self.wait_online(['bridge99', 'test1:degraded'], bool_any=True)
- self.check_link_exists('bridge99')
- self.check_link_exists('test1')
self.check_operstate('bridge99', '(?:off|no-carrier)', setup_state='configuring')
self.check_operstate('test1', 'degraded')
'21-vlan.network', '21-vlan-test1.network')
self.start_networkd(0)
- self.wait_online(['test1', 'vlan99'])
+ self.wait_online(['test1:degraded', 'vlan99:routable'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1'], universal_newlines=True).rstrip()
print(output)
with self.subTest(mode=mode):
if mode != 'private':
self.tearDown()
- self.copy_unit_to_networkd_unit_path('21-macvtap.netdev', '11-dummy.netdev', 'macvtap.network')
+ self.copy_unit_to_networkd_unit_path('21-macvtap.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'macvtap.network')
with open(os.path.join(network_unit_file_path, '21-macvtap.netdev'), mode='a') as f:
f.write('[MACVTAP]\nMode=' + mode)
self.start_networkd(0)
- self.wait_online(['macvtap99:off', 'test1:degraded'])
+ self.wait_online(['macvtap99:degraded', 'test1:degraded'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'macvtap99'], universal_newlines=True).rstrip()
print(output)
with self.subTest(mode=mode):
if mode != 'private':
self.tearDown()
- self.copy_unit_to_networkd_unit_path('21-macvlan.netdev', '11-dummy.netdev', 'macvlan.network')
+ self.copy_unit_to_networkd_unit_path('21-macvlan.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'macvlan.network')
with open(os.path.join(network_unit_file_path, '21-macvlan.netdev'), mode='a') as f:
f.write('[MACVLAN]\nMode=' + mode)
self.start_networkd(0)
- self.wait_online(['macvlan99:off', 'test1:degraded'])
+ self.wait_online(['macvlan99:degraded', 'test1:degraded'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1'], universal_newlines=True).rstrip()
print(output)
with self.subTest(mode=mode, flag=flag):
if mode != 'L2':
self.tearDown()
- self.copy_unit_to_networkd_unit_path('25-ipvlan.netdev', '11-dummy.netdev', 'ipvlan.network')
+ self.copy_unit_to_networkd_unit_path('25-ipvlan.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'ipvlan.network')
with open(os.path.join(network_unit_file_path, '25-ipvlan.netdev'), mode='a') as f:
f.write('[IPVLAN]\nMode=' + mode + '\nFlags=' + flag)
self.start_networkd(0)
- self.wait_online(['ipvlan99:off', 'test1:degraded'])
+ self.wait_online(['ipvlan99:degraded', 'test1:degraded'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ipvlan99'], universal_newlines=True).rstrip()
print(output)
with self.subTest(mode=mode, flag=flag):
if mode != 'L2':
self.tearDown()
- self.copy_unit_to_networkd_unit_path('25-ipvtap.netdev', '11-dummy.netdev', 'ipvtap.network')
+ self.copy_unit_to_networkd_unit_path('25-ipvtap.netdev', 'netdev-link-local-addressing-yes.network',
+ '11-dummy.netdev', 'ipvtap.network')
with open(os.path.join(network_unit_file_path, '25-ipvtap.netdev'), mode='a') as f:
f.write('[IPVTAP]\nMode=' + mode + '\nFlags=' + flag)
self.start_networkd(0)
- self.wait_online(['ipvtap99:off', 'test1:degraded'])
+ self.wait_online(['ipvtap99:degraded', 'test1:degraded'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ipvtap99'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'ipvtap *mode ' + mode.lower() + ' ' + flag)
def test_veth(self):
- self.copy_unit_to_networkd_unit_path('25-veth.netdev')
+ self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['veth99:off', 'veth-peer:off'])
-
- def test_dummy(self):
- self.copy_unit_to_networkd_unit_path('11-dummy.netdev')
- self.start_networkd(0)
+ self.wait_online(['veth99:degraded', 'veth-peer:degraded'])
- self.wait_online(['test1:off'])
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'veth99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'link/ether 12:34:56:78:9a:bc')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'veth-peer'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'link/ether 12:34:56:78:9a:bd')
def test_tun(self):
self.copy_unit_to_networkd_unit_path('25-tun.netdev')
self.wait_online(['tun99:off'])
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'tun99'], universal_newlines=True).rstrip()
+ print(output)
+ # Old ip command does not support IFF_ flags
+ self.assertRegex(output, 'tun (?:type tun pi on vnet_hdr on multi_queue|addrgenmode) ')
+
def test_tap(self):
self.copy_unit_to_networkd_unit_path('25-tap.netdev')
self.start_networkd(0)
self.wait_online(['tap99:off'])
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'tap99'], universal_newlines=True).rstrip()
+ print(output)
+ # Old ip command does not support IFF_ flags
+ self.assertRegex(output, 'tun (?:type tap pi on vnet_hdr on multi_queue|addrgenmode) ')
+
@expectedFailureIfModuleIsNotAvailable('vrf')
def test_vrf(self):
- self.copy_unit_to_networkd_unit_path('25-vrf.netdev')
+ self.copy_unit_to_networkd_unit_path('25-vrf.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['vrf99:off'])
+ self.wait_online(['vrf99:carrier'])
@expectedFailureIfModuleIsNotAvailable('vcan')
def test_vcan(self):
- self.copy_unit_to_networkd_unit_path('25-vcan.netdev')
+ self.copy_unit_to_networkd_unit_path('25-vcan.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['vcan99:off'])
+ self.wait_online(['vcan99:carrier'])
+
+ @expectedFailureIfModuleIsNotAvailable('vxcan')
+ def test_vxcan(self):
+ self.copy_unit_to_networkd_unit_path('25-vxcan.netdev', 'netdev-link-local-addressing-yes.network')
+ self.start_networkd(0)
+
+ self.wait_online(['vxcan99:carrier', 'vxcan-peer:carrier'])
@expectedFailureIfModuleIsNotAvailable('wireguard')
def test_wireguard(self):
self.assertRegex(output, 'CJQUtcS9emY2fLYqDlpSZiE/QJyHkPWr\+WHtZLZ90FU=')
def test_geneve(self):
- self.copy_unit_to_networkd_unit_path('25-geneve.netdev')
+ self.copy_unit_to_networkd_unit_path('25-geneve.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['geneve99:off'])
+ self.wait_online(['geneve99:degraded'])
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'geneve99'], universal_newlines=True).rstrip()
print(output)
self.assertNotRegex(output, 'iseq')
self.assertNotRegex(output, 'oseq')
- @expectedFailureIf_ip6gre_do_not_support_ipv6ll()
def test_ip6gre_tunnel(self):
self.copy_unit_to_networkd_unit_path('12-dummy.netdev', 'ip6gretun.network',
'25-ip6gre-tunnel.netdev', '25-tunnel.network',
'25-ip6gre-tunnel-remote-any.netdev', '25-tunnel-remote-any.network')
self.start_networkd()
+ # Old kernels seem not to support IPv6LL address on ip6gre tunnel, So please do not use wait_online() here.
+
self.check_link_exists('dummy98')
self.check_link_exists('ip6gretun99')
self.check_link_exists('ip6gretun98')
print(output)
self.assertRegex(output, 'ip6gre remote any local 2a00:ffde:4567:edde::4987 dev dummy98')
- # Old kernels may not support IPv6LL address on ip6gre tunnel, and the following test may fails.
- self.wait_online(['ip6gretun99:routable', 'ip6gretun98:routable', 'ip6gretun97:routable', 'dummy98:degraded'])
-
def test_gretap_tunnel(self):
self.copy_unit_to_networkd_unit_path('12-dummy.netdev', 'gretap.network',
'25-gretap-tunnel.netdev', '25-tunnel.network',
self.start_networkd(0)
self.wait_online(['isataptun99:routable', 'dummy98:degraded'])
- self.check_link_exists('dummy98')
- self.check_link_exists('isataptun99')
-
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'isataptun99'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, "isatap ")
self.assertRegex(output, 'oseq')
def test_tunnel_independent(self):
- self.copy_unit_to_networkd_unit_path('25-ipip-tunnel-independent.netdev')
+ self.copy_unit_to_networkd_unit_path('25-ipip-tunnel-independent.netdev', 'netdev-link-local-addressing-yes.network')
self.start_networkd(0)
- self.wait_online(['ipiptun99:off'])
+ self.wait_online(['ipiptun99:carrier'])
@expectedFailureIfModuleIsNotAvailable('fou')
def test_fou(self):
self.start_networkd(0)
self.wait_online(['dummy98:degraded'])
- self.check_link_exists('dummy98')
-
self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'forwarding'), '1')
self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'use_tempaddr'), '2')
self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'dad_transmits'), '3')
self.start_networkd(0)
self.wait_online(['dummy98:routable'])
- self.check_link_exists('dummy98')
-
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dummy98'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope global dummy98')
self.start_networkd(0)
self.wait_online(['dummy98:routable'])
- self.check_link_exists('dummy98')
-
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dummy98'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope global dummy98')
self.check_operstate('dummy98', 'off')
self.check_operstate('test1', 'off')
- bond_has_no_carrier=False
for trial in range(30):
if trial > 0:
time.sleep(1)
self.assertEqual(subprocess.call(['ip', 'link', 'add', 'dummy98', 'type', 'dummy']), 0)
self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
- time.sleep(3)
+ for trial in range(30):
+ if trial > 0:
+ time.sleep(1)
+ if self.get_operstate('bridge99') == 'routable' and self.get_operstate('dummy98') == 'enslaved':
+ break
+ else:
+ self.assertTrue(False)
output = subprocess.check_output(['ip', 'address', 'show', 'bridge99'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, 'inet 192.168.0.15/24 brd 192.168.0.255 scope global bridge99')
- self.check_operstate('bridge99', 'routable')
- self.check_operstate('dummy98', 'enslaved')
-
output = subprocess.check_output(['ip', 'rule', 'list', 'table', '100'], universal_newlines=True).rstrip()
print(output)
self.assertEqual(output, '0: from all to 8.8.8.8 lookup 100')
self.remove_unit_from_networkd_path(self.units)
def test_ipv6_prefix_delegation(self):
+ self.warn_about_firewalld()
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'ipv6-prefix.network', 'ipv6-prefix-veth.network')
self.start_networkd()
self.remove_unit_from_networkd_path(self.units)
def test_dhcp_server(self):
+ self.warn_about_firewalld()
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client.network', 'dhcp-server.network')
self.start_networkd()
self.assertRegex(output, 'Search Domains: one')
def test_emit_router_timezone(self):
+ self.warn_about_firewalld()
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client-timezone-router.network', 'dhcp-server-timezone-router.network')
self.start_networkd()
'static.network']
def setUp(self):
- self.link_remove(self.links)
self.stop_dnsmasq(dnsmasq_pid_file)
+ self.link_remove(self.links)
def tearDown(self):
- self.link_remove(self.links)
- self.remove_unit_from_networkd_path(self.units)
self.stop_dnsmasq(dnsmasq_pid_file)
self.remove_lease_file()
self.remove_log_file()
+ self.link_remove(self.links)
+ self.remove_unit_from_networkd_path(self.units)
def test_dhcp_client_ipv6_only(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network')
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'link'], universal_newlines=True).rstrip()
self.assertRegex(output, 'inet .* scope link')
+ def test_dhcp_client_route_remove_on_renew(self):
+ self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network',
+ 'dhcp-client-ipv4-only-ipv6-disabled.network')
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
+ self.start_dnsmasq(ipv4_range='192.168.5.100,192.168.5.199', lease_time='2m')
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ # test for issue #12490
+
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'global', 'dynamic'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'inet 192.168.5.1[0-9]*/24 brd 192.168.5.255 scope global dynamic veth99')
+ address1=None
+ for line in output.splitlines():
+ if 'brd 192.168.5.255 scope global dynamic veth99' in line:
+ address1 = line.split()[1].split('/')[0]
+ break
+
+ output = subprocess.check_output(['ip', '-4', 'route', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, f'default via 192.168.5.1 proto dhcp src {address1} metric 1024')
+ self.assertRegex(output, f'192.168.5.1 proto dhcp scope link src {address1} metric 1024')
+
+ self.stop_dnsmasq(dnsmasq_pid_file)
+ self.start_dnsmasq(ipv4_range='192.168.5.200,192.168.5.250', lease_time='2m')
+
+ print('Wait for the dynamic address to be expired')
+ time.sleep(130)
+
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'global', 'dynamic'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertRegex(output, 'inet 192.168.5.2[0-9]*/24 brd 192.168.5.255 scope global dynamic veth99')
+ address2=None
+ for line in output.splitlines():
+ if 'brd 192.168.5.255 scope global dynamic veth99' in line:
+ address2 = line.split()[1].split('/')[0]
+ break
+
+ self.assertNotEqual(address1, address2)
+
+ output = subprocess.check_output(['ip', '-4', 'route', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
+ print(output)
+ self.assertNotRegex(output, f'default via 192.168.5.1 proto dhcp src {address1} metric 1024')
+ self.assertNotRegex(output, f'192.168.5.1 proto dhcp scope link src {address1} metric 1024')
+ self.assertRegex(output, f'default via 192.168.5.1 proto dhcp src {address2} metric 1024')
+ self.assertRegex(output, f'192.168.5.1 proto dhcp scope link src {address2} metric 1024')
+
if __name__ == '__main__':
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
verbosity=3))
projects / thirdparty / systemd.git / blobdiff