subprocess.call(['ip', 'link', 'del', 'dev', link])
time.sleep(1)
+ def l2tp_tunnel_remove(self, tunnel_ids):
+ output = subprocess.check_output(['ip', 'l2tp', 'show', 'tunnel']).rstrip().decode('utf-8')
+ for tid in tunnel_ids:
+ words='Tunnel ' + tid + ', encap'
+ if words in output:
+ subprocess.call(['ip', 'l2tp', 'del', 'tunnel', 'tid', tid])
+ time.sleep(1)
+
def read_ipv6_sysctl_attr(self, link, attribute):
with open(os.path.join(os.path.join(network_sysctl_ipv6_path, link), attribute)) as f:
return f.readline().strip()
if os.path.exists(dnsmasq_log_file):
os.remove(dnsmasq_log_file)
- def start_networkd(self):
- if (os.path.exists(os.path.join(networkd_runtime_directory, 'state'))):
+ def start_networkd(self, remove_state_files=True):
+ if (remove_state_files and
+ os.path.exists(os.path.join(networkd_runtime_directory, 'state'))):
subprocess.check_call('systemctl stop systemd-networkd', shell=True)
os.remove(os.path.join(networkd_runtime_directory, 'state'))
subprocess.check_call('systemctl start systemd-networkd', shell=True)
'12-dummy.netdev',
'21-macvlan.netdev',
'21-macvtap.netdev',
+ '21-vlan-test1.network',
'21-vlan.netdev',
'21-vlan.network',
'25-6rd-tunnel.netdev',
'25-vxlan.netdev',
'25-wireguard-23-peers.netdev',
'25-wireguard-23-peers.network',
+ '25-wireguard-private-key.txt',
'25-wireguard.netdev',
'6rd.network',
'gre.network',
self.assertEqual('1', self.read_link_attr('bond99', 'bonding', 'tlb_dynamic_lb'))
def test_vlan(self):
- self.copy_unit_to_networkd_unit_path('21-vlan.netdev', '11-dummy.netdev', '21-vlan.network')
+ self.copy_unit_to_networkd_unit_path('21-vlan.netdev', '11-dummy.netdev',
+ '21-vlan.network', '21-vlan-test1.network')
self.start_networkd()
self.assertTrue(self.link_exits('test1'))
self.assertTrue(output, 'MVRP')
self.assertTrue(output, ' id 99 ')
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'test1']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet 192.168.24.5/24 brd 192.168.24.255 scope global test1')
+ self.assertRegex(output, 'inet 192.168.25.5/24 brd 192.168.25.255 scope global test1')
+
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'vlan99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet 192.168.23.5/24 brd 192.168.23.255 scope global vlan99')
+
def test_macvtap(self):
self.copy_unit_to_networkd_unit_path('21-macvtap.netdev', '11-dummy.netdev', 'macvtap.network')
self.start_networkd()
self.assertTrue(output, 'RDf+LSpeEre7YEIKaxg+wbpsNV7du+ktR99uBEtIiCA=\t20')
output = subprocess.check_output(['wg', 'show', 'wg99', 'endpoints']).rstrip().decode('utf-8')
self.assertTrue(output, 'RDf+LSpeEre7YEIKaxg+wbpsNV7du+ktR99uBEtIiCA=\t192.168.27.3:51820')
+ output = subprocess.check_output(['wg', 'show', 'wg99', 'private-key']).rstrip().decode('utf-8')
+ self.assertTrue(output, 'EEGlnEPYJV//kbvvIqxKkQwOiS+UENyPncC4bF46ong=')
self.assertTrue(self.link_exits('wg99'))
@expectedFailureIfModuleIsNotAvailable('wireguard')
def test_wireguard_23_peers(self):
- self.copy_unit_to_networkd_unit_path('25-wireguard-23-peers.netdev', '25-wireguard-23-peers.network')
+ self.copy_unit_to_networkd_unit_path('25-wireguard-23-peers.netdev', '25-wireguard-23-peers.network',
+ '25-wireguard-private-key.txt')
self.start_networkd()
if shutil.which('wg'):
subprocess.call('wg')
+ output = subprocess.check_output(['wg', 'show', 'wg98', 'private-key']).rstrip().decode('utf-8')
+ self.assertTrue(output, 'CJQUtcS9emY2fLYqDlpSZiE/QJyHkPWr+WHtZLZ90FU=')
self.assertTrue(self.link_exits('wg98'))
self.assertTrue(self.link_exits('ipiptun99'))
def test_vxlan(self):
- self.copy_unit_to_networkd_unit_path('25-vxlan.netdev', 'vxlan.network','11-dummy.netdev')
+ self.copy_unit_to_networkd_unit_path('25-vxlan.netdev', 'vxlan.network', '11-dummy.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('vxlan99'))
self.assertRegex(output, 'remcsumrx')
self.assertRegex(output, 'gbp')
+class NetworkdL2TPTests(unittest.TestCase, Utilities):
+
+ links =[
+ 'l2tp-ses1',
+ 'l2tp-ses2',
+ 'l2tp-ses3',
+ 'l2tp-ses4',
+ 'test1']
+
+ units = [
+ '11-dummy.netdev',
+ '25-l2tp-dummy.network',
+ '25-l2tp-ip.netdev',
+ '25-l2tp-udp.netdev']
+
+ l2tp_tunnel_ids = [ '10' ]
+
+ def setUp(self):
+ self.l2tp_tunnel_remove(self.l2tp_tunnel_ids)
+ self.link_remove(self.links)
+
+ def tearDown(self):
+ self.l2tp_tunnel_remove(self.l2tp_tunnel_ids)
+ self.link_remove(self.links)
+ self.remove_unit_from_networkd_path(self.units)
+
+ @expectedFailureIfModuleIsNotAvailable('l2tp_eth')
+ def test_l2tp_udp(self):
+ self.copy_unit_to_networkd_unit_path('11-dummy.netdev', '25-l2tp-dummy.network', '25-l2tp-udp.netdev')
+ self.start_networkd()
+
+ self.assertTrue(self.link_exits('test1'))
+ self.assertTrue(self.link_exits('l2tp-ses1'))
+ self.assertTrue(self.link_exits('l2tp-ses2'))
+
+ output = subprocess.check_output(['ip', 'l2tp', 'show', 'tunnel', 'tunnel_id', '10']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "Tunnel 10, encap UDP")
+ self.assertRegex(output, "From 192.168.30.100 to 192.168.30.101")
+ self.assertRegex(output, "Peer tunnel 11")
+ self.assertRegex(output, "UDP source / dest ports: 3000/4000")
+ self.assertRegex(output, "UDP checksum: enabled")
+
+ output = subprocess.check_output(['ip', 'l2tp', 'show', 'session', 'tid', '10', 'session_id', '15']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "Session 15 in tunnel 10")
+ self.assertRegex(output, "Peer session 16, tunnel 11")
+ self.assertRegex(output, "interface name: l2tp-ses1")
+
+ output = subprocess.check_output(['ip', 'l2tp', 'show', 'session', 'tid', '10', 'session_id', '17']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "Session 17 in tunnel 10")
+ self.assertRegex(output, "Peer session 18, tunnel 11")
+ self.assertRegex(output, "interface name: l2tp-ses2")
+
+ @expectedFailureIfModuleIsNotAvailable('l2tp_ip')
+ def test_l2tp_ip(self):
+ self.copy_unit_to_networkd_unit_path('11-dummy.netdev', '25-l2tp-dummy.network', '25-l2tp-ip.netdev')
+ self.start_networkd()
+
+ self.assertTrue(self.link_exits('test1'))
+ self.assertTrue(self.link_exits('l2tp-ses3'))
+ self.assertTrue(self.link_exits('l2tp-ses4'))
+
+ output = subprocess.check_output(['ip', 'l2tp', 'show', 'tunnel', 'tunnel_id', '10']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "Tunnel 10, encap IP")
+ self.assertRegex(output, "From 192.168.30.100 to 192.168.30.101")
+ self.assertRegex(output, "Peer tunnel 12")
+
+ output = subprocess.check_output(['ip', 'l2tp', 'show', 'session', 'tid', '10', 'session_id', '25']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "Session 25 in tunnel 10")
+ self.assertRegex(output, "Peer session 26, tunnel 12")
+ self.assertRegex(output, "interface name: l2tp-ses3")
+
+ output = subprocess.check_output(['ip', 'l2tp', 'show', 'session', 'tid', '10', 'session_id', '27']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "Session 27 in tunnel 10")
+ self.assertRegex(output, "Peer session 28, tunnel 12")
+ self.assertRegex(output, "interface name: l2tp-ses4")
+
class NetworkdNetWorkTests(unittest.TestCase, Utilities):
links = [
'bond199',
'25-sysctl-disable-ipv6.network',
'25-sysctl.network',
'configure-without-carrier.network',
- 'routing-policy-rule.network',
+ 'routing-policy-rule-dummy98.network',
+ 'routing-policy-rule-test1.network',
'test-static.network']
def setUp(self):
self.assertRegex(output, 'primary test1')
def test_routing_policy_rule(self):
- self.copy_unit_to_networkd_unit_path('routing-policy-rule.network', '11-dummy.netdev')
+ 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.assertTrue(self.link_exits('test1'))
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(trial == 0)
+
+ self.assertTrue(self.link_exits('test1'))
+ self.assertTrue(self.link_exits('dummy98'))
+
+ output = subprocess.check_output(['ip', 'rule', 'list', 'table', '7']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, '111: from 192.168.100.18 tos (?:0x08|throughput) iif test1 oif test1 lookup 7')
+
+ output = subprocess.check_output(['ip', 'rule', 'list', 'table', '8']).rstrip().decode('utf-8')
+ 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.assertTrue(self.link_exits('test1'))
@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.assertTrue(self.link_exits('test1'))
self.assertTrue(self.link_exits('dummy98'))
- output = subprocess.check_output(['ip', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
+ # This also tests address pool
+
+ output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'dummy98', 'label', '32']).rstrip().decode('utf-8')
print(output)
self.assertRegex(output, 'inet 10.2.3.4 peer 10.2.3.5/16 scope global 32')
+
+ output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'dummy98', 'label', '33']).rstrip().decode('utf-8')
+ print(output)
self.assertRegex(output, 'inet 10.6.7.8/16 brd 10.6.255.255 scope global 33')
+
+ output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'dummy98', 'label', '34']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet 192.168.[0-9]*.1/24 brd 192.168.[0-9]*.255 scope global 34')
+
+ output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'dummy98', 'label', '35']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet 172.[0-9]*.0.1/16 brd 172.[0-9]*.255.255 scope global 35')
+
+ output = subprocess.check_output(['ip', '-6', 'address', 'show', 'dev', 'dummy98']).rstrip().decode('utf-8')
+ print(output)
self.assertRegex(output, 'inet6 2001:db8::20 peer 2001:db8::10/128 scope global')
+ self.assertRegex(output, 'inet6 fd[0-9a-f:]*1/64 scope global')
output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
print(output)
print(output)
self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope link deprecated dummy98')
self.assertRegex(output, 'inet6 2001:db8:0:f101::1/64 scope global')
+ # also tests invalid [Address] section
+ self.assertNotRegex(output, '10.10.0.1/16')
+ self.assertNotRegex(output, '10.10.0.2/16')
def test_ip_route(self):
self.copy_unit_to_networkd_unit_path('25-route-section.network', '12-dummy.netdev')
output = subprocess.check_output(['networkctl', 'status', 'bond99']).rstrip().decode('utf-8')
print(output)
- self.assertRegex(output, 'State: degraded \(configured\)')
+ self.assertRegex(output, 'State: degraded-carrier \(configured\)')
self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
time.sleep(2)
self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'down']), 0)
self.assertEqual(subprocess.call(['ip', 'link', 'set', 'test1', 'down']), 0)
- time.sleep(2)
+ time.sleep(5)
output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
print(output)
output = subprocess.check_output(['networkctl', 'status', 'bond99']).rstrip().decode('utf-8')
print(output)
- self.assertRegex(output, 'State: degraded \(configured\)')
+ self.assertRegex(output, 'State: no-carrier \(configured\)')
class NetworkdNetWorkBridgeTests(unittest.TestCase, Utilities):
links = [
self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'hairpin_mode'), '1')
self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'path_cost'), '400')
self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'unicast_flood'), '1')
+ self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'multicast_flood'), '0')
self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'multicast_fast_leave'), '1')
+ if (os.path.exists('/sys/devices/virtual/net/bridge99/lower_dummy98/brport/neigh_suppress')):
+ self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'neigh_suppress'), '1')
+ self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'learning'), '0')
# CONFIG_BRIDGE_IGMP_SNOOPING=y
if (os.path.exists('/sys/devices/virtual/net/bridge00/lower_dummy98/brport/multicast_to_unicast')):
time.sleep(3)
output = subprocess.check_output(['networkctl', 'status', 'bridge99']).rstrip().decode('utf-8')
- self.assertRegex(output, 'State: degraded \(configured\)')
+ self.assertRegex(output, 'State: degraded-carrier \(configured\)')
self.assertEqual(subprocess.call(['ip', 'link', 'del', 'dummy98']), 0)
time.sleep(3)
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',
'bridge99-ignore-carrier-loss.network')
+
+ subprocess.call(['ip', 'rule', 'del', 'table', '100'])
+
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
def test_bridge_ignore_carrier_loss_frequent_loss_and_gain(self):
self.copy_unit_to_networkd_unit_path('26-bridge.netdev', '26-bridge-slave-interface-1.network',
'bridge99-ignore-carrier-loss.network')
+
+ subprocess.call(['ip', 'rule', 'del', 'table', '100'])
+
self.start_networkd()
self.assertTrue(self.link_exits('bridge99'))
'25-vrf.network',
'dhcp-client-anonymize.network',
'dhcp-client-critical-connection.network',
+ 'dhcp-client-gateway-onlink-implicit.network',
'dhcp-client-ipv4-dhcp-settings.network',
'dhcp-client-ipv4-only-ipv6-disabled.network',
'dhcp-client-ipv4-only.network',
print(output)
self.assertRegex(output, 'State: routable \(configured\)')
+ 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.assertTrue(self.link_exits('veth99'))
+
+ self.start_dnsmasq()
+
+ output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, '192.168.5')
+
+ output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'veth99', '10.0.0.0/8']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'onlink')
+ output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'veth99', '192.168.100.0/24']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'onlink')
+
if __name__ == '__main__':
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
verbosity=3))