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)
'bridge99',
'dropin-test',
'dummy98',
- 'erspan-test',
+ 'erspan98',
+ 'erspan99',
'geneve99',
+ 'gretap98',
'gretap99',
+ 'gretun97',
+ 'gretun98',
'gretun99',
+ 'ip6gretap98',
'ip6gretap99',
+ 'ip6gretun97',
+ 'ip6gretun98',
+ 'ip6gretun99',
+ 'ip6tnl97',
+ 'ip6tnl98',
'ip6tnl99',
+ 'ipiptun97',
+ 'ipiptun98',
'ipiptun99',
'ipvlan99',
'isataptun99',
'macvlan99',
'macvtap99',
+ 'sittun97',
+ 'sittun98',
'sittun99',
'tap99',
'test1',
'veth99',
'vlan99',
'vrf99',
+ 'vti6tun97',
+ 'vti6tun98',
'vti6tun99',
+ 'vtitun97',
+ 'vtitun98',
'vtitun99',
'vxlan99',
'wg98',
'12-dummy.netdev',
'21-macvlan.netdev',
'21-macvtap.netdev',
+ '21-vlan-test1.network',
'21-vlan.netdev',
'21-vlan.network',
'25-6rd-tunnel.netdev',
'25-bond.netdev',
'25-bond-balanced-tlb.netdev',
'25-bridge.netdev',
+ '25-erspan-tunnel-local-any.netdev',
'25-erspan-tunnel.netdev',
'25-geneve.netdev',
+ '25-gretap-tunnel-local-any.netdev',
'25-gretap-tunnel.netdev',
+ '25-gre-tunnel-local-any.netdev',
+ '25-gre-tunnel-remote-any.netdev',
'25-gre-tunnel.netdev',
+ '25-ip6gretap-tunnel-local-any.netdev',
+ '25-ip6gretap-tunnel.netdev',
+ '25-ip6gre-tunnel-local-any.netdev',
+ '25-ip6gre-tunnel-remote-any.netdev',
'25-ip6gre-tunnel.netdev',
+ '25-ip6tnl-tunnel-remote-any.netdev',
+ '25-ip6tnl-tunnel-local-any.netdev',
'25-ip6tnl-tunnel.netdev',
'25-ipip-tunnel-independent.netdev',
+ '25-ipip-tunnel-local-any.netdev',
+ '25-ipip-tunnel-remote-any.netdev',
'25-ipip-tunnel.netdev',
'25-ipvlan.netdev',
'25-isatap-tunnel.netdev',
+ '25-sit-tunnel-local-any.netdev',
+ '25-sit-tunnel-remote-any.netdev',
'25-sit-tunnel.netdev',
'25-tap.netdev',
'25-tun.netdev',
'25-vcan.netdev',
'25-veth.netdev',
'25-vrf.netdev',
+ '25-vti6-tunnel-local-any.netdev',
+ '25-vti6-tunnel-remote-any.netdev',
'25-vti6-tunnel.netdev',
+ '25-vti-tunnel-local-any.netdev',
+ '25-vti-tunnel-remote-any.netdev',
'25-vti-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',
+ 'erspan.network',
'gre.network',
'gretap.network',
'gretun.network',
'ip6gretap.network',
+ 'ip6gretun.network',
'ip6tnl.network',
'ipip.network',
'ipvlan.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(self.link_exits('vlan99'))
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1']).rstrip().decode('utf-8')
+ print(output)
+ self.assertTrue(output, ' mtu 2004 ')
+
output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vlan99']).rstrip().decode('utf-8')
print(output)
+ self.assertTrue(output, ' mtu 2000 ')
self.assertTrue(output, 'REORDER_HDR')
self.assertTrue(output, 'LOOSE_BINDING')
self.assertTrue(output, 'GVRP')
self.assertTrue(output, 'MVRP')
- self.assertTrue(output, '99')
+ 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.copy_unit_to_networkd_unit_path('21-macvlan.netdev', '11-dummy.netdev', 'macvlan.network')
self.start_networkd()
+ self.assertTrue(self.link_exits('test1'))
self.assertTrue(self.link_exits('macvlan99'))
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1']).rstrip().decode('utf-8')
+ print(output)
+ self.assertTrue(output, ' mtu 2000 ')
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'macvlan99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertTrue(output, ' mtu 2000 ')
+
@expectedFailureIfModuleIsNotAvailable('ipvlan')
def test_ipvlan(self):
self.copy_unit_to_networkd_unit_path('25-ipvlan.netdev', '11-dummy.netdev', 'ipvlan.network')
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(output, 'udp6zerocsumrx')
def test_ipip_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ipip-tunnel.netdev', 'ipip.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ipip-tunnel.netdev', 'ipip.network',
+ '25-ipip-tunnel-local-any.netdev', '25-ipip-tunnel-remote-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('ipiptun99'))
+ self.assertTrue(self.link_exits('ipiptun98'))
+ self.assertTrue(self.link_exits('ipiptun97'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ipiptun99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ipip (?:ipip |)remote 192.169.224.239 local 192.168.223.238 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ipiptun98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ipip (?:ipip |)remote 192.169.224.239 local any dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ipiptun97']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ipip (?:ipip |)remote any local 192.168.223.238 dev dummy98')
def test_gre_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gre-tunnel.netdev', 'gretun.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gre-tunnel.netdev', 'gretun.network',
+ '25-gre-tunnel-local-any.netdev', '25-gre-tunnel-remote-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('gretun99'))
+ self.assertTrue(self.link_exits('gretun98'))
+ self.assertTrue(self.link_exits('gretun97'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'gretun99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'gre remote 10.65.223.239 local 10.65.223.238 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'gretun98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'gre remote 10.65.223.239 local any dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'gretun97']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'gre remote any local 10.65.223.238 dev dummy98')
+
+ def test_ip6gre_tunnel(self):
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6gre-tunnel.netdev', 'ip6gretun.network',
+ '25-ip6gre-tunnel-local-any.netdev', '25-ip6gre-tunnel-remote-any.netdev')
+ self.start_networkd()
+
+ self.assertTrue(self.link_exits('dummy98'))
+ self.assertTrue(self.link_exits('ip6gretun99'))
+ self.assertTrue(self.link_exits('ip6gretun98'))
+ self.assertTrue(self.link_exits('ip6gretun97'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6gretun99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6gre remote 2001:473:fece:cafe::5179 local 2a00:ffde:4567:edde::4987 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6gretun98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6gre remote 2001:473:fece:cafe::5179 local any dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6gretun97']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6gre remote any local 2a00:ffde:4567:edde::4987 dev dummy98')
def test_gretap_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gretap-tunnel.netdev', 'gretap.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gretap-tunnel.netdev', 'gretap.network',
+ '25-gretap-tunnel-local-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('gretap99'))
+ self.assertTrue(self.link_exits('gretap98'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'gretap99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'gretap remote 10.65.223.239 local 10.65.223.238 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'gretap98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'gretap remote 10.65.223.239 local any dev dummy98')
def test_ip6gretap_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6gre-tunnel.netdev', 'ip6gretap.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6gretap-tunnel.netdev', 'ip6gretap.network',
+ '25-ip6gretap-tunnel-local-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('ip6gretap99'))
+ self.assertTrue(self.link_exits('ip6gretap98'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6gretap99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6gretap remote 2001:473:fece:cafe::5179 local 2a00:ffde:4567:edde::4987 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6gretap98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6gretap remote 2001:473:fece:cafe::5179 local any dev dummy98')
def test_vti_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti-tunnel.netdev', 'vti.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti-tunnel.netdev', 'vti.network',
+ '25-vti-tunnel-local-any.netdev', '25-vti-tunnel-remote-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('vtitun99'))
+ self.assertTrue(self.link_exits('vtitun98'))
+ self.assertTrue(self.link_exits('vtitun97'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vtitun99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'vti remote 10.65.223.239 local 10.65.223.238 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vtitun98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'vti remote 10.65.223.239 local any dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vtitun97']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'vti remote any local 10.65.223.238 dev dummy98')
def test_vti6_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti6-tunnel.netdev', 'vti6.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti6-tunnel.netdev', 'vti6.network',
+ '25-vti6-tunnel-local-any.netdev', '25-vti6-tunnel-remote-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('vti6tun99'))
+ self.assertTrue(self.link_exits('vti6tun98'))
+ self.assertTrue(self.link_exits('vti6tun97'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vti6tun99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'vti6 remote 2001:473:fece:cafe::5179 local 2a00:ffde:4567:edde::4987 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vti6tun98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'vti6 remote 2001:473:fece:cafe::5179 local (?:any|::) dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vti6tun97']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'vti6 remote (?:any|::) local 2a00:ffde:4567:edde::4987 dev dummy98')
def test_ip6tnl_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6tnl-tunnel.netdev', 'ip6tnl.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6tnl-tunnel.netdev', 'ip6tnl.network',
+ '25-ip6tnl-tunnel-local-any.netdev', '25-ip6tnl-tunnel-remote-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('ip6tnl99'))
+ self.assertTrue(self.link_exits('ip6tnl98'))
+ self.assertTrue(self.link_exits('ip6tnl97'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6tnl99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6tnl ip6ip6 remote 2001:473:fece:cafe::5179 local 2a00:ffde:4567:edde::4987 dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6tnl98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6tnl ip6ip6 remote 2001:473:fece:cafe::5179 local (?:any|::) dev dummy98')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'ip6tnl97']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'ip6tnl ip6ip6 remote (?:any|::) local 2a00:ffde:4567:edde::4987 dev dummy98')
def test_sit_tunnel(self):
- self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-sit-tunnel.netdev', 'sit.network')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-sit-tunnel.netdev', 'sit.network',
+ '25-sit-tunnel-local-any.netdev',
+ '25-sit-tunnel-remote-any.netdev')
self.start_networkd()
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('sittun99'))
+ self.assertTrue(self.link_exits('sittun98'))
+ self.assertTrue(self.link_exits('sittun97'))
+
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'sittun99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "sit (?:ip6ip |)remote 10.65.223.239 local 10.65.223.238 dev dummy98")
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'sittun98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "sit (?:ip6ip |)remote 10.65.223.239 local any dev dummy98")
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'sittun97']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, "sit (?:ip6ip |)remote any local 10.65.223.238 dev dummy98")
def test_isatap_tunnel(self):
self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-isatap-tunnel.netdev', 'isatap.network')
self.assertTrue(self.link_exits('dummy98'))
self.assertTrue(self.link_exits('sittun99'))
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'sittun99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, '6rd-prefix 2602::/24')
+
@expectedFailureIfERSPANModuleIsNotAvailable()
def test_erspan_tunnel(self):
- self.copy_unit_to_networkd_unit_path('25-erspan-tunnel.netdev')
+ self.copy_unit_to_networkd_unit_path('12-dummy.netdev', 'erspan.network',
+ '25-erspan-tunnel.netdev', '25-erspan-tunnel-local-any.netdev')
self.start_networkd()
- self.assertTrue(self.link_exits('erspan-test'))
+ self.assertTrue(self.link_exits('dummy98'))
+ self.assertTrue(self.link_exits('erspan99'))
+ self.assertTrue(self.link_exits('erspan98'))
- output = subprocess.check_output(['ip', '-d', 'link', 'show', 'erspan-test']).rstrip().decode('utf-8')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'erspan99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'erspan remote 172.16.1.100 local 172.16.1.200')
+ self.assertRegex(output, '101')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'erspan98']).rstrip().decode('utf-8')
print(output)
- self.assertTrue(output, '172.16.1.200')
- self.assertTrue(output, '172.16.1.100')
- self.assertTrue(output, '101')
+ self.assertRegex(output, 'erspan remote 172.16.1.100 local any')
+ self.assertRegex(output, '102')
def test_tunnel_independent(self):
self.copy_unit_to_networkd_unit_path('25-ipip-tunnel-independent.netdev')
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')
class NetworkdNetWorkBondTests(unittest.TestCase, Utilities):
links = [
'bond99',
- 'veth99']
+ 'dummy98',
+ 'test1']
units = [
+ '11-dummy.netdev',
+ '12-dummy.netdev',
'25-bond.netdev',
- '25-veth.netdev',
'bond99.network',
- 'dhcp-server.network',
- 'veth-bond.network']
+ 'bond-slave.network']
def setUp(self):
self.link_remove(self.links)
self.link_remove(self.links)
self.remove_unit_from_networkd_path(self.units)
- def test_bridge_property(self):
- self.copy_unit_to_networkd_unit_path('25-bond.netdev', '25-veth.netdev', 'bond99.network',
- 'dhcp-server.network', 'veth-bond.network')
+ def test_bond_operstate(self):
+ self.copy_unit_to_networkd_unit_path('25-bond.netdev', '11-dummy.netdev', '12-dummy.netdev',
+ 'bond99.network','bond-slave.network')
self.start_networkd()
self.assertTrue(self.link_exits('bond99'))
- self.assertTrue(self.link_exits('veth99'))
- self.assertTrue(self.link_exits('veth-peer'))
+ self.assertTrue(self.link_exits('dummy98'))
+ self.assertTrue(self.link_exits('test1'))
- output = subprocess.check_output(['ip', '-d', 'link', 'show', 'veth-peer']).rstrip().decode('utf-8')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
print(output)
- self.assertRegex(output, 'UP,LOWER_UP')
+ self.assertRegex(output, 'SLAVE,UP,LOWER_UP')
- output = subprocess.check_output(['ip', '-d', 'link', 'show', 'veth99']).rstrip().decode('utf-8')
+ output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1']).rstrip().decode('utf-8')
print(output)
self.assertRegex(output, 'SLAVE,UP,LOWER_UP')
print(output)
self.assertRegex(output, 'MASTER,UP,LOWER_UP')
- output = subprocess.check_output(['networkctl', 'status', 'veth-peer']).rstrip().decode('utf-8')
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
print(output)
- self.assertRegex(output, 'State: routable \(configured\)')
+ self.assertRegex(output, 'State: enslaved \(configured\)')
- output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
+ output = subprocess.check_output(['networkctl', 'status', 'test1']).rstrip().decode('utf-8')
print(output)
self.assertRegex(output, 'State: enslaved \(configured\)')
print(output)
self.assertRegex(output, 'State: routable \(configured\)')
- self.assertEqual(subprocess.call(['ip', 'link', 'set', 'veth99', 'down']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'down']), 0)
time.sleep(2)
- output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
print(output)
self.assertRegex(output, 'State: off \(configured\)')
+ output = subprocess.check_output(['networkctl', 'status', 'test1']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'State: enslaved \(configured\)')
+
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', 'veth99', 'up']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
time.sleep(2)
- output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'State: enslaved \(configured\)')
+
+ output = subprocess.check_output(['networkctl', 'status', 'test1']).rstrip().decode('utf-8')
print(output)
self.assertRegex(output, 'State: enslaved \(configured\)')
print(output)
self.assertRegex(output, 'State: routable \(configured\)')
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'down']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'test1', 'down']), 0)
+ time.sleep(5)
+
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'State: off \(configured\)')
+
+ output = subprocess.check_output(['networkctl', 'status', 'test1']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'State: off \(configured\)')
+
+ output = subprocess.check_output(['networkctl', 'status', 'bond99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'State: no-carrier \(configured\)')
+
class NetworkdNetWorkBridgeTests(unittest.TestCase, Utilities):
links = [
'bridge99',
output = subprocess.check_output(['ip', 'addr', 'show', 'bridge99']).rstrip().decode('utf-8')
print(output)
- self.assertRegex(output, '192.168.0.15')
- self.assertRegex(output, '192.168.0.1')
+ self.assertRegex(output, '192.168.0.15/24')
output = subprocess.check_output(['bridge', '-d', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
print(output)
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')):
self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'multicast_to_unicast'), '1')
+ output = subprocess.check_output(['networkctl', 'status', 'test1']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: enslaved \(configured\)')
+
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: enslaved \(configured\)')
+
+ output = subprocess.check_output(['networkctl', 'status', 'bridge99']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: routable \(configured\)')
+
self.assertEqual(subprocess.call(['ip', 'address', 'add', '192.168.0.16/24', 'dev', 'bridge99']), 0)
time.sleep(1)
+ output = subprocess.check_output(['ip', 'addr', 'show', 'bridge99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, '192.168.0.16/24')
+
+ output = subprocess.check_output(['networkctl', 'status', 'bridge99']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: routable \(configured\)')
+
self.assertEqual(subprocess.call(['ip', 'link', 'del', 'test1']), 0)
+ time.sleep(3)
+
+ output = subprocess.check_output(['networkctl', 'status', 'bridge99']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: degraded-carrier \(configured\)')
+
self.assertEqual(subprocess.call(['ip', 'link', 'del', 'dummy98']), 0)
time.sleep(3)
+ output = subprocess.check_output(['networkctl', 'status', 'bridge99']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: no-carrier \(configured\)')
+
output = subprocess.check_output(['ip', 'address', 'show', 'bridge99']).rstrip().decode('utf-8')
print(output)
self.assertRegex(output, 'NO-CARRIER')
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'))
self.assertRegex(output, 'inet 192.168.0.15/24 brd 192.168.0.255 scope global bridge99')
self.assertRegex(output, 'inet 192.168.0.16/24 scope global secondary bridge99')
+ subprocess.call(['ip', 'rule', 'del', 'table', '100'])
+
+ 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'))
+
+ self.assertEqual(subprocess.call(['ip', 'link', 'add', 'dummy98', 'type', 'dummy']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'del', 'dummy98']), 0)
+
+ self.assertEqual(subprocess.call(['ip', 'link', 'add', 'dummy98', 'type', 'dummy']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'del', 'dummy98']), 0)
+
+ self.assertEqual(subprocess.call(['ip', 'link', 'add', 'dummy98', 'type', 'dummy']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'del', 'dummy98']), 0)
+
+ self.assertEqual(subprocess.call(['ip', 'link', 'add', 'dummy98', 'type', 'dummy']), 0)
+ self.assertEqual(subprocess.call(['ip', 'link', 'set', 'dummy98', 'up']), 0)
+
+ time.sleep(3)
+
+ output = subprocess.check_output(['ip', 'address', 'show', 'bridge99']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet 192.168.0.15/24 brd 192.168.0.255 scope global bridge99')
+
+ output = subprocess.check_output(['networkctl', 'status', 'bridge99']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: routable \(configured\)')
+
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: enslaved \(configured\)')
+
+ output = subprocess.check_output(['ip', 'rule', 'list', 'table', '100']).rstrip().decode('utf-8')
+ 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):
links = ['veth99']
'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))
projects / thirdparty / systemd.git / blobdiff