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()
return f.readline().strip()
def copy_unit_to_networkd_unit_path(self, *units):
+ print()
for unit in units:
shutil.copy(os.path.join(networkd_ci_path, unit), network_unit_file_path)
if (os.path.exists(os.path.join(networkd_ci_path, unit + '.d'))):
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)
else:
subprocess.check_call('systemctl restart systemd-networkd', shell=True)
time.sleep(5)
- print()
class NetworkdNetDevTests(unittest.TestCase, Utilities):
'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',
output = subprocess.check_output(['networkctl', 'status', 'dropin-*']).rstrip().decode('utf-8')
self.assertNotRegex(output, '1: lo ')
self.assertRegex(output, 'dropin-test')
- self.assertRegex(output, 'Driver: dummy')
+
+ ret = subprocess.run(['ethtool', '--driver', 'dropin-test'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ print(ret.stdout.rstrip().decode('utf-8'))
+ if ret.returncode == 0 and re.search('driver: dummy', ret.stdout.rstrip().decode('utf-8')) != None:
+ self.assertRegex(output, 'Driver: dummy')
+ else:
+ print('ethtool does not support driver field at least for dummy interfaces, skipping test for Driver field of networkctl.')
def test_bridge(self):
self.copy_unit_to_networkd_unit_path('25-bridge.netdev')
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(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-route-section.network',
'25-route-tcp-window-settings.network',
'25-route-type.network',
+ '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')
self.assertEqual(self.read_ipv4_sysctl_attr('dummy98', 'forwarding'),'1')
self.assertEqual(self.read_ipv4_sysctl_attr('dummy98', 'proxy_arp'), '1')
+ def test_sysctl_disable_ipv6(self):
+ self.copy_unit_to_networkd_unit_path('25-sysctl-disable-ipv6.network', '12-dummy.netdev')
+
+ print('## Disable ipv6')
+ self.assertEqual(subprocess.call(['sysctl', 'net.ipv6.conf.all.disable_ipv6=1']), 0)
+ self.assertEqual(subprocess.call(['sysctl', 'net.ipv6.conf.default.disable_ipv6=1']), 0)
+
+ self.start_networkd()
+
+ self.assertTrue(self.link_exits('dummy98'))
+
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope global dummy98')
+ output = subprocess.check_output(['ip', '-6', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertEqual(output, '')
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: routable \(configured\)')
+
+ self.assertEqual(subprocess.call(['ip', 'link', 'del', 'dummy98']), 0)
+
+ print('## Enable ipv6')
+ self.assertEqual(subprocess.call(['sysctl', 'net.ipv6.conf.all.disable_ipv6=0']), 0)
+ self.assertEqual(subprocess.call(['sysctl', 'net.ipv6.conf.default.disable_ipv6=0']), 0)
+
+ self.start_networkd()
+
+ self.assertTrue(self.link_exits('dummy98'))
+
+ output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope global dummy98')
+ output = subprocess.check_output(['ip', '-6', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
+ print(output)
+ self.assertRegex(output, 'inet6 .* scope link')
+ output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
+ self.assertRegex(output, 'State: routable \(configured\)')
+
def test_bind_carrier(self):
self.copy_unit_to_networkd_unit_path('25-bind-carrier.network', '11-dummy.netdev')
self.start_networkd()
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))