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()
'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',
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)
projects / thirdparty / systemd.git / blobdiff