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)
def check_link_exists(self, link):
self.assertTrue(self.link_exists(link))
- def link_remove(self, links):
+ def remove_links(self, links):
for link in links:
if self.link_exists(link):
subprocess.call(['ip', 'link', 'del', 'dev', link])
time.sleep(1)
+ def remove_fou_ports(self, ports):
+ for port in ports:
+ subprocess.call(['ip', 'fou', 'del', 'port', port])
+
+ def remove_routing_policy_rule_tables(self, tables):
+ for table in tables:
+ subprocess.call(['ip', 'rule', 'del', 'table', table])
+
+ def remove_routes(self, routes):
+ for route_type, addr in routes:
+ subprocess.call(['ip', 'route', 'del', route_type, addr])
+
def l2tp_tunnel_remove(self, tunnel_ids):
output = subprocess.check_output(['ip', 'l2tp', 'show', 'tunnel'], universal_newlines=True).rstrip()
for tid in tunnel_ids:
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
- subprocess.check_call(dnsmasq_command, shell=True)
+ 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.')
- time.sleep(10)
+ 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)
def stop_dnsmasq(self, pid_file):
if os.path.exists(pid_file):
def check_operstate(self, link, expected, show_status=True, setup_state='configured'):
self.assertRegex(self.get_operstate(link, show_status, setup_state), expected)
+ def wait_address(self, link, address_regex, scope='global', ipv='', timeout_sec=10):
+ for i in range(timeout_sec):
+ if i > 0:
+ time.sleep(1)
+ output = subprocess.check_output(['ip', ipv, 'address', 'show', 'dev', link, 'scope', scope], universal_newlines=True).rstrip()
+ if re.search(address_regex, output):
+ break
+ else:
+ self.assertRegex(output, address_regex)
+
class NetworkdNetDevTests(unittest.TestCase, Utilities):
'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',
'vxlan-test1.network',
'vxlan.network']
+ fou_ports = [
+ '55555',
+ '55556']
+
def setUp(self):
- self.link_remove(self.links)
+ self.remove_fou_ports(self.fou_ports)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_fou_ports(self.fou_ports)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_dropin_and_networkctl_glob(self):
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)
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'])
+ self.wait_online(['veth99:degraded', 'veth-peer:degraded'])
- def test_dummy(self):
- self.copy_unit_to_networkd_unit_path('11-dummy.netdev')
- self.start_networkd(0)
-
- 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')
@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: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(['vcan99:off'])
+ 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):
print(output)
self.assertRegex(output, 'encap fou encap-sport auto encap-dport 55556')
- subprocess.call(['ip', 'fou', 'del', 'port', '55555'])
- subprocess.call(['ip', 'fou', 'del', 'port', '55556'])
-
def test_vxlan(self):
self.copy_unit_to_networkd_unit_path('25-vxlan.netdev', 'vxlan.network',
'11-dummy.netdev', 'vxlan-test1.network')
def setUp(self):
self.l2tp_tunnel_remove(self.l2tp_tunnel_ids)
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
self.l2tp_tunnel_remove(self.l2tp_tunnel_ids)
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
@expectedFailureIfModuleIsNotAvailable('l2tp_eth')
self.assertRegex(output, "Peer session 28, tunnel 12")
self.assertRegex(output, "interface name: l2tp-ses4")
-class NetworkdNetWorkTests(unittest.TestCase, Utilities):
+class NetworkdNetworkTests(unittest.TestCase, Utilities):
links = [
'bond199',
'dummy98',
'routing-policy-rule-dummy98.network',
'routing-policy-rule-test1.network']
+ routing_policy_rule_tables = ['7', '8']
+ routes = [['blackhole', '202.54.1.2'], ['unreachable', '202.54.1.3'], ['prohibit', '202.54.1.4']]
+
def setUp(self):
- self.link_remove(self.links)
+ self.remove_routing_policy_rule_tables(self.routing_policy_rule_tables)
+ self.remove_routes(self.routes)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_routing_policy_rule_tables(self.routing_policy_rule_tables)
+ self.remove_routes(self.routes)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_address_static(self):
def test_routing_policy_rule(self):
self.copy_unit_to_networkd_unit_path('routing-policy-rule-test1.network', '11-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
self.start_networkd()
self.check_link_exists('test1')
self.assertRegex(output, 'oif test1')
self.assertRegex(output, 'lookup 7')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
def test_routing_policy_rule_issue_11280(self):
self.copy_unit_to_networkd_unit_path('routing-policy-rule-test1.network', '11-dummy.netdev',
'routing-policy-rule-dummy98.network', '12-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
- subprocess.call(['ip', 'rule', 'del', 'table', '8'])
-
for trial in range(3):
# Remove state files only first time
self.start_networkd(remove_state_files=(trial == 0))
print(output)
self.assertRegex(output, '112: from 192.168.101.18 tos (?:0x08|throughput) iif dummy98 oif dummy98 lookup 8')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
- subprocess.call(['ip', 'rule', 'del', 'table', '8'])
-
@expectedFailureIfRoutingPolicyPortRangeIsNotAvailable()
def test_routing_policy_rule_port_range(self):
self.copy_unit_to_networkd_unit_path('25-fibrule-port-range.network', '11-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
self.start_networkd()
self.check_link_exists('test1')
self.assertRegex(output, 'tcp')
self.assertRegex(output, 'lookup 7')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
@expectedFailureIfRoutingPolicyIPProtoIsNotAvailable()
def test_routing_policy_rule_invert(self):
self.copy_unit_to_networkd_unit_path('25-fibrule-invert.network', '11-dummy.netdev')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
self.start_networkd()
self.check_link_exists('test1')
self.assertRegex(output, 'tcp')
self.assertRegex(output, 'lookup 7')
- subprocess.call(['ip', 'rule', 'del', 'table', '7'])
-
def test_route_static(self):
self.copy_unit_to_networkd_unit_path('25-route-static.network', '12-dummy.netdev')
self.start_networkd(0)
print(output)
self.assertRegex(output, 'prohibit 202.54.1.4 proto static')
- subprocess.call(['ip', 'route', 'del', 'blackhole', '202.54.1.2'])
- subprocess.call(['ip', 'route', 'del', 'unreachable', '202.54.1.3'])
- subprocess.call(['ip', 'route', 'del', 'prohibit', '202.54.1.4'])
-
def test_ip_route_ipv6_src_route(self):
# a dummy device does not make the addresses go through tentative state, so we
# reuse a bond from an earlier test, which does make the addresses go through
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.assertRegex(output, 'inet 192.168.10.30/24 brd 192.168.10.255 scope global test1')
self.check_operstate('test1', 'routable')
-class NetworkdNetWorkBondTests(unittest.TestCase, Utilities):
+class NetworkdBondTests(unittest.TestCase, Utilities):
links = [
'bond199',
'bond99',
'bond-slave.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_bond_active_slave(self):
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)
# Let's confirm that networkd's operstate is consistent with ip's result.
self.assertNotRegex(output, 'NO-CARRIER')
-class NetworkdNetWorkBridgeTests(unittest.TestCase, Utilities):
+class NetworkdBridgeTests(unittest.TestCase, Utilities):
links = [
'bridge99',
'dummy98',
'bridge99.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_bridge_property(self):
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')
subprocess.call(['ip', 'rule', 'del', 'table', '100'])
-class NetworkdNetWorkLLDPTests(unittest.TestCase, Utilities):
+class NetworkdLLDPTests(unittest.TestCase, Utilities):
links = ['veth99']
units = [
'25-veth.netdev']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
self.remove_unit_from_networkd_path(self.units)
def test_lldp(self):
self.assertRegex(output, 'veth-peer')
self.assertRegex(output, 'veth99')
-class NetworkdNetworkRATests(unittest.TestCase, Utilities):
+class NetworkdRATests(unittest.TestCase, Utilities):
links = ['veth99']
units = [
'ipv6-prefix-veth.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
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()
print(output)
self.assertRegex(output, '2002:da8:1:0')
-class NetworkdNetworkDHCPServerTests(unittest.TestCase, Utilities):
+class NetworkdDHCPServerTests(unittest.TestCase, Utilities):
links = [
'dummy98',
'veth99']
'dhcp-server-timezone-router.network']
def setUp(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
def tearDown(self):
- self.link_remove(self.links)
+ self.remove_links(self.links)
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()
self.assertRegex(output, '192.168.5.*')
self.assertRegex(output, 'Europe/Berlin')
-class NetworkdNetworkDHCPClientTests(unittest.TestCase, Utilities):
+class NetworkdDHCPClientTests(unittest.TestCase, Utilities):
links = [
- 'dummy98',
'veth99',
'vrf99']
'static.network']
def setUp(self):
- self.link_remove(self.links)
self.stop_dnsmasq(dnsmasq_pid_file)
+ self.remove_links(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.remove_links(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')
def test_dhcp_client_ipv4_only(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()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_client_ipv4_ipv6(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network',
'dhcp-client-ipv4-only.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ # 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', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_client_settings(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-dhcp-settings.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
print('## ip address show dev veth99')
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
def test_dhcp6_client_settings_rapidcommit_true(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp6_client_settings_rapidcommit_false(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-rapid-commit.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_client_settings_anonymize(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-anonymize.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
self.assertFalse(self.search_words_in_dnsmasq_log('VendorClassIdentifier=SusantVendorTest', True))
self.assertFalse(self.search_words_in_dnsmasq_log('test-hostname'))
def test_dhcp_client_listen_port(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-listen-port.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq('--dhcp-alternate-port=67,5555')
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ # 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', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_route_table_id(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-table.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'route', 'show', 'table', '12'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_route_metric(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-metric.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', 'route', 'show', 'dev', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_route_criticalconnection_true(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-critical-connection.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_client_reuse_address_as_static(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+ # 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', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99', 'scope', 'global'], universal_newlines=True).rstrip()
print(output)
self.assertRegex(output, '192.168.5')
self.assertRegex(output, '2600::')
- ipv4_address = re.search('192\.168\.5\.[0-9]*/24', output)
- ipv6_address = re.search('2600::[0-9a-f:]*/128', output)
+ ipv4_address = re.search(r'192.168.5.[0-9]*/24', output)
+ ipv6_address = re.search(r'2600::[0-9a-f:]*/128', output)
static_network = '\n'.join(['[Match]', 'Name=veth99', '[Network]', 'IPv6AcceptRA=no', 'Address=' + ipv4_address.group(), 'Address=' + ipv6_address.group()])
print(static_network)
with open(os.path.join(network_unit_file_path, 'static.network'), mode='w') as f:
f.write(static_network)
- self.start_networkd()
-
- self.check_link_exists('veth99')
+ # When networkd started, the links are already configured, so let's wait for 5 seconds
+ # the links to be re-configured.
+ self.start_networkd(5)
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'global'], universal_newlines=True).rstrip()
print(output)
def test_dhcp_client_vrf(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-vrf.network',
'25-vrf.netdev', '25-vrf.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
- self.check_link_exists('vrf99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable', 'vrf99:carrier'])
+
+ # 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', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128', ipv='-6')
print('## ip -d link show dev vrf99')
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'dev', 'vrf99'], universal_newlines=True).rstrip()
def test_dhcp_client_gateway_onlink_implicit(self):
self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network',
'dhcp-client-gateway-onlink-implicit.network')
- self.start_networkd()
-
- self.check_link_exists('veth99')
-
+ self.start_networkd(0)
+ self.wait_online(['veth-peer:carrier'])
self.start_dnsmasq()
+ self.wait_online(['veth99:routable', 'veth-peer:routable'])
output = subprocess.check_output(['networkctl', 'status', 'veth99'], universal_newlines=True).rstrip()
print(output)
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