test/test-network/systemd-networkd-tests.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: LGPL-2.1+
   3 # systemd-networkd tests
   4
   5 import os
   6 import sys
   7 import unittest
   8 import subprocess
   9 import time
  10 import re
  11 import shutil
  12 import signal
  13 import socket
  14 import threading
  15 from shutil import copytree
  16
  17 network_unit_file_path='/var/run/systemd/network'
  18 networkd_ci_path='/var/run/networkd-ci'
  19 network_sysctl_ipv6_path='/proc/sys/net/ipv6/conf'
  20 network_sysctl_ipv4_path='/proc/sys/net/ipv4/conf'
  21
  22 dnsmasq_config_file='/var/run/networkd-ci/test-dnsmasq.conf'
  23 dnsmasq_pid_file='/var/run/networkd-ci/test-test-dnsmasq.pid'
  24 dnsmasq_log_file='/var/run/networkd-ci/test-dnsmasq-log-file'
  25
  26 def is_module_available(module_name):
  27     lsmod_output = subprocess.check_output('lsmod', universal_newlines=True)
  28     module_re = re.compile(r'^{0}\b'.format(re.escape(module_name)), re.MULTILINE)
  29     return module_re.search(lsmod_output) or not subprocess.call(["modprobe", module_name])
  30
  31 def expectedFailureIfModuleIsNotAvailable(module_name):
  32     def f(func):
  33         if not is_module_available(module_name):
  34             return unittest.expectedFailure(func)
  35         return func
  36
  37     return f
  38
  39 def setUpModule():
  40
  41     os.makedirs(network_unit_file_path, exist_ok=True)
  42     os.makedirs(networkd_ci_path, exist_ok=True)
  43
  44     shutil.rmtree(networkd_ci_path)
  45     copytree(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'conf'), networkd_ci_path)
  46
  47 def tearDownModule():
  48     shutil.rmtree(networkd_ci_path)
  49
  50 class Utilities():
  51     dhcp_server_data = []
  52
  53     def read_link_attr(self, link, dev, attribute):
  54         with open(os.path.join(os.path.join(os.path.join('/sys/class/net/', link), dev), attribute)) as f:
  55             return f.readline().strip()
  56
  57     def link_exits(self, link):
  58         return os.path.exists(os.path.join('/sys/class/net', link))
  59
  60     def link_remove(self, links):
  61         for link in links:
  62             if os.path.exists(os.path.join('/sys/class/net', link)):
  63                 subprocess.call(['ip', 'link', 'del', 'dev', link])
  64         time.sleep(1)
  65
  66     def read_ipv6_sysctl_attr(self, link, attribute):
  67         with open(os.path.join(os.path.join(network_sysctl_ipv6_path, link), attribute)) as f:
  68             return f.readline().strip()
  69
  70     def read_ipv4_sysctl_attr(self, link, attribute):
  71         with open(os.path.join(os.path.join(network_sysctl_ipv4_path, link), attribute)) as f:
  72             return f.readline().strip()
  73
  74     def copy_unit_to_networkd_unit_path(self, *units):
  75         for unit in units:
  76             shutil.copy(os.path.join(networkd_ci_path, unit), network_unit_file_path)
  77
  78     def remove_unit_from_networkd_path(self, units):
  79         for unit in units:
  80             if (os.path.exists(os.path.join(network_unit_file_path, unit))):
  81                 os.remove(os.path.join(network_unit_file_path, unit))
  82
  83     def start_dnsmasq(self):
  84         subprocess.check_call('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 --dhcp-range=192.168.5.10,192.168.5.200 -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', shell=True)
  85
  86         time.sleep(10)
  87
  88     def stop_dnsmasq(self, pid_file):
  89         if os.path.exists(pid_file):
  90             with open(pid_file, 'r') as f:
  91                 pid = f.read().rstrip(' \t\r\n\0')
  92                 os.kill(int(pid), signal.SIGTERM)
  93
  94             os.remove(pid_file)
  95
  96     def search_words_in_file(self, word):
  97         if os.path.exists(dnsmasq_log_file):
  98             with open (dnsmasq_log_file) as in_file:
  99                 contents = in_file.read()
 100                 print(contents)
 101                 for part in contents.split():
 102                     if word in part:
 103                         in_file.close()
 104                         print("%s, %s" % (word, part))
 105                         return True
 106         return False
 107
 108     def remove_lease_file(self):
 109         if os.path.exists(os.path.join(networkd_ci_path, 'lease')):
 110             os.remove(os.path.join(networkd_ci_path, 'lease'))
 111
 112     def remove_log_file(self):
 113         if os.path.exists(dnsmasq_log_file):
 114             os.remove(dnsmasq_log_file)
 115
 116     def start_networkd(self):
 117         subprocess.check_call('systemctl restart systemd-networkd', shell=True)
 118         time.sleep(5)
 119
 120 global ip
 121 global port
 122
 123 class DHCPServer(threading.Thread):
 124     def __init__(self, name):
 125         threading.Thread.__init__(self)
 126         self.name = name
 127
 128     def run(self):
 129         self.start_dhcp_server()
 130
 131     def start_dhcp_server(self):
 132         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 133
 134         server_address = ('0.0.0.0', 67)
 135         sock.bind(server_address)
 136
 137         print('Starting DHCP Server ...\n')
 138         data, addr = sock.recvfrom(1024) # buffer size is 1024 bytes
 139
 140         global ip
 141         ip = addr[0]
 142
 143         global port
 144         port = addr[1]
 145         sock.close()
 146
 147 class NetworkdNetDevTests(unittest.TestCase, Utilities):
 148
 149     links =['bridge99', 'bond99', 'bond99', 'vlan99', 'test1', 'macvtap99',
 150             'macvlan99', 'ipvlan99', 'vxlan99', 'veth99', 'vrf99', 'tun99',
 151             'tap99', 'vcan99', 'geneve99', 'dummy98', 'ipiptun99', 'sittun99', '6rdtun99',
 152             'gretap99', 'vtitun99', 'vti6tun99','ip6tnl99', 'gretun99', 'ip6gretap99', 'wg99']
 153
 154     units = ['25-bridge.netdev', '25-bond.netdev', '21-vlan.netdev', '11-dummy.netdev', '21-vlan.network',
 155              '21-macvtap.netdev', 'macvtap.network', '21-macvlan.netdev', 'macvlan.network', 'vxlan.network',
 156              '25-vxlan.netdev', '25-ipvlan.netdev', 'ipvlan.network', '25-veth.netdev', '25-vrf.netdev',
 157              '25-tun.netdev', '25-tun.netdev', '25-vcan.netdev', '25-geneve.netdev', '25-ipip-tunnel.netdev',
 158              '25-ip6tnl-tunnel.netdev', '25-ip6gre-tunnel.netdev','25-sit-tunnel.netdev', '25-6rd-tunnel.netdev',
 159              '25-gre-tunnel.netdev', '25-gretap-tunnel.netdev', '25-vti-tunnel.netdev', '25-vti6-tunnel.netdev',
 160              '12-dummy.netdev', 'gre.network', 'ipip.network', 'ip6gretap.network', 'gretun.network',
 161              'ip6tnl.network', '25-tap.netdev', 'vti6.network', 'vti.network', 'gretap.network', 'sit.network',
 162              '25-ipip-tunnel-independent.netdev', '25-wireguard.netdev', '6rd.network']
 163
 164     def setUp(self):
 165         self.link_remove(self.links)
 166
 167     def tearDown(self):
 168         self.link_remove(self.links)
 169         self.remove_unit_from_networkd_path(self.units)
 170
 171     def test_bridge(self):
 172         self.copy_unit_to_networkd_unit_path('25-bridge.netdev')
 173         self.start_networkd()
 174
 175         self.assertTrue(self.link_exits('bridge99'))
 176
 177         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge', 'hello_time'))
 178         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge', 'max_age'))
 179         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge','forward_delay'))
 180         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge','ageing_time'))
 181         self.assertEqual('9',   self.read_link_attr('bridge99', 'bridge','priority'))
 182         self.assertEqual('1',   self.read_link_attr('bridge99', 'bridge','multicast_querier'))
 183         self.assertEqual('1',   self.read_link_attr('bridge99', 'bridge','multicast_snooping'))
 184         self.assertEqual('1',   self.read_link_attr('bridge99', 'bridge','stp_state'))
 185
 186     def test_bond(self):
 187         self.copy_unit_to_networkd_unit_path('25-bond.netdev')
 188         self.start_networkd()
 189
 190         self.assertTrue(self.link_exits('bond99'))
 191
 192         self.assertEqual('802.3ad 4',         self.read_link_attr('bond99', 'bonding', 'mode'))
 193         self.assertEqual('layer3+4 1',        self.read_link_attr('bond99', 'bonding', 'xmit_hash_policy'))
 194         self.assertEqual('1000',              self.read_link_attr('bond99', 'bonding', 'miimon'))
 195         self.assertEqual('fast 1',            self.read_link_attr('bond99', 'bonding', 'lacp_rate'))
 196         self.assertEqual('2000',              self.read_link_attr('bond99', 'bonding', 'updelay'))
 197         self.assertEqual('2000',              self.read_link_attr('bond99', 'bonding', 'downdelay'))
 198         self.assertEqual('4',                 self.read_link_attr('bond99', 'bonding', 'resend_igmp'))
 199         self.assertEqual('1',                 self.read_link_attr('bond99', 'bonding', 'min_links'))
 200         self.assertEqual('1218',              self.read_link_attr('bond99', 'bonding', 'ad_actor_sys_prio'))
 201         self.assertEqual('811',               self.read_link_attr('bond99', 'bonding', 'ad_user_port_key'))
 202         self.assertEqual('00:11:22:33:44:55', self.read_link_attr('bond99', 'bonding', 'ad_actor_system'))
 203
 204     def test_vlan(self):
 205         self.copy_unit_to_networkd_unit_path('21-vlan.netdev', '11-dummy.netdev', '21-vlan.network')
 206
 207         self.start_networkd()
 208
 209         self.assertTrue(self.link_exits('vlan99'))
 210
 211         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vlan99']).rstrip().decode('utf-8')
 212         self.assertTrue(output, 'REORDER_HDR')
 213         self.assertTrue(output, 'LOOSE_BINDING')
 214         self.assertTrue(output, 'GVRP')
 215         self.assertTrue(output, 'MVRP')
 216         self.assertTrue(output, '99')
 217
 218     def test_macvtap(self):
 219         self.copy_unit_to_networkd_unit_path('21-macvtap.netdev', '11-dummy.netdev', 'macvtap.network')
 220
 221         self.start_networkd()
 222
 223         self.assertTrue(self.link_exits('macvtap99'))
 224
 225     def test_macvlan(self):
 226         self.copy_unit_to_networkd_unit_path('21-macvlan.netdev', '11-dummy.netdev', 'macvlan.network')
 227
 228         self.start_networkd()
 229
 230         self.assertTrue(self.link_exits('macvlan99'))
 231
 232     @expectedFailureIfModuleIsNotAvailable('ipvlan')
 233     def test_ipvlan(self):
 234         self.copy_unit_to_networkd_unit_path('25-ipvlan.netdev', '11-dummy.netdev', 'ipvlan.network')
 235
 236         self.start_networkd()
 237
 238         self.assertTrue(self.link_exits('ipvlan99'))
 239
 240     def test_veth(self):
 241         self.copy_unit_to_networkd_unit_path('25-veth.netdev')
 242
 243         self.start_networkd()
 244
 245         self.assertTrue(self.link_exits('veth99'))
 246
 247     def test_dummy(self):
 248         self.copy_unit_to_networkd_unit_path('11-dummy.netdev')
 249
 250         self.start_networkd()
 251
 252         self.assertTrue(self.link_exits('test1'))
 253
 254     def test_tun(self):
 255         self.copy_unit_to_networkd_unit_path('25-tun.netdev')
 256
 257         self.start_networkd()
 258
 259         self.assertTrue(self.link_exits('tun99'))
 260
 261     def test_tap(self):
 262         self.copy_unit_to_networkd_unit_path('25-tap.netdev')
 263
 264         self.start_networkd()
 265
 266         self.assertTrue(self.link_exits('tap99'))
 267
 268     @expectedFailureIfModuleIsNotAvailable('vrf')
 269     def test_vrf(self):
 270         self.copy_unit_to_networkd_unit_path('25-vrf.netdev')
 271
 272         self.start_networkd()
 273
 274         self.assertTrue(self.link_exits('vrf99'))
 275
 276     @expectedFailureIfModuleIsNotAvailable('vcan')
 277     def test_vcan(self):
 278         self.copy_unit_to_networkd_unit_path('25-vcan.netdev')
 279
 280         self.start_networkd()
 281
 282         self.assertTrue(self.link_exits('vcan99'))
 283
 284     @expectedFailureIfModuleIsNotAvailable('wireguard')
 285     def test_wireguard(self):
 286         self.copy_unit_to_networkd_unit_path('25-wireguard.netdev')
 287
 288         self.start_networkd()
 289
 290         if shutil.which('wg'):
 291             subprocess.call('wg')
 292
 293         self.assertTrue(self.link_exits('wg99'))
 294
 295     def test_geneve(self):
 296         self.copy_unit_to_networkd_unit_path('25-geneve.netdev')
 297
 298         self.start_networkd()
 299
 300         self.assertTrue(self.link_exits('geneve99'))
 301
 302         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'geneve99']).rstrip().decode('utf-8')
 303         self.assertTrue(output, '192.168.22.1')
 304         self.assertTrue(output, '6082')
 305         self.assertTrue(output, 'udpcsum')
 306         self.assertTrue(output, 'udp6zerocsumrx')
 307
 308     def test_ipip_tunnel(self):
 309         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ipip-tunnel.netdev', 'ipip.network')
 310         self.start_networkd()
 311
 312         self.assertTrue(self.link_exits('dummy98'))
 313         self.assertTrue(self.link_exits('ipiptun99'))
 314
 315     def test_gre_tunnel(self):
 316         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gre-tunnel.netdev', 'gretun.network')
 317         self.start_networkd()
 318
 319         self.assertTrue(self.link_exits('dummy98'))
 320         self.assertTrue(self.link_exits('gretun99'))
 321
 322     def test_gretap_tunnel(self):
 323         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gretap-tunnel.netdev', 'gretap.network')
 324         self.start_networkd()
 325
 326         self.assertTrue(self.link_exits('dummy98'))
 327         self.assertTrue(self.link_exits('gretap99'))
 328
 329     def test_ip6gretap_tunnel(self):
 330         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6gre-tunnel.netdev', 'ip6gretap.network')
 331         self.start_networkd()
 332
 333         self.assertTrue(self.link_exits('dummy98'))
 334         self.assertTrue(self.link_exits('ip6gretap99'))
 335
 336     def test_vti_tunnel(self):
 337         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti-tunnel.netdev', 'vti.network')
 338         self.start_networkd()
 339
 340         self.assertTrue(self.link_exits('dummy98'))
 341         self.assertTrue(self.link_exits('vtitun99'))
 342
 343     def test_vti6_tunnel(self):
 344         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti6-tunnel.netdev', 'vti6.network')
 345         self.start_networkd()
 346
 347         self.assertTrue(self.link_exits('dummy98'))
 348         self.assertTrue(self.link_exits('vti6tun99'))
 349
 350     def test_ip6tnl_tunnel(self):
 351         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6tnl-tunnel.netdev', 'ip6tnl.network')
 352         self.start_networkd()
 353
 354         self.assertTrue(self.link_exits('dummy98'))
 355         self.assertTrue(self.link_exits('ip6tnl99'))
 356
 357     def test_sit_tunnel(self):
 358         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-sit-tunnel.netdev', 'sit.network')
 359         self.start_networkd()
 360
 361         self.assertTrue(self.link_exits('dummy98'))
 362         self.assertTrue(self.link_exits('sittun99'))
 363
 364     def test_6rd_tunnel(self):
 365         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-6rd-tunnel.netdev', '6rd.network')
 366         self.start_networkd()
 367
 368         self.assertTrue(self.link_exits('dummy98'))
 369         self.assertTrue(self.link_exits('sittun99'))
 370
 371     def test_tunnel_independent(self):
 372         self.copy_unit_to_networkd_unit_path('25-ipip-tunnel-independent.netdev')
 373
 374         self.start_networkd()
 375         self.assertTrue(self.link_exits('ipiptun99'))
 376
 377     def test_vxlan(self):
 378         self.copy_unit_to_networkd_unit_path('25-vxlan.netdev', 'vxlan.network','11-dummy.netdev')
 379
 380         self.start_networkd()
 381
 382         self.assertTrue(self.link_exits('vxlan99'))
 383
 384         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vxlan99']).rstrip().decode('utf-8')
 385         self.assertRegex(output, "999")
 386         self.assertRegex(output, '5555')
 387         self.assertRegex(output, 'l2miss')
 388         self.assertRegex(output, 'l3miss')
 389         self.assertRegex(output, 'udpcsum')
 390         self.assertRegex(output, 'udp6zerocsumtx')
 391         self.assertRegex(output, 'udp6zerocsumrx')
 392         self.assertRegex(output, 'remcsumtx')
 393         self.assertRegex(output, 'remcsumrx')
 394         self.assertRegex(output, 'gbp')
 395
 396 class NetworkdNetWorkTests(unittest.TestCase, Utilities):
 397     links = ['dummy98', 'test1', 'bond199']
 398
 399     units = ['12-dummy.netdev', 'test-static.network', 'configure-without-carrier.network', '11-dummy.netdev',
 400              '23-primary-slave.network', '23-test1-bond199.network', '11-dummy.netdev', '23-bond199.network',
 401              '25-bond-active-backup-slave.netdev', '12-dummy.netdev', '23-active-slave.network',
 402              'routing-policy-rule.network', '25-address-section.network', '25-address-section-miscellaneous.network',
 403              '25-route-section.network', '25-route-type.network', '25-route-tcp-window-settings.network',
 404              '25-address-link-section.network', '25-ipv6-address-label-section.network', '25-link-section-unmanaged.network',
 405              '25-sysctl.network']
 406
 407     def setUp(self):
 408         self.link_remove(self.links)
 409
 410     def tearDown(self):
 411         self.link_remove(self.links)
 412         self.remove_unit_from_networkd_path(self.units)
 413
 414     def test_static_address(self):
 415         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', 'test-static.network')
 416         self.start_networkd()
 417
 418         self.assertTrue(self.link_exits('dummy98'))
 419         output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
 420         print(output)
 421         self.assertRegex(output, '192.168.0.15')
 422         self.assertRegex(output, '192.168.0.1')
 423         self.assertRegex(output, 'routable')
 424
 425     def test_configure_without_carrier(self):
 426         self.copy_unit_to_networkd_unit_path('configure-without-carrier.network', '11-dummy.netdev')
 427         self.start_networkd()
 428
 429         self.assertTrue(self.link_exits('test1'))
 430         output = subprocess.check_output(['networkctl', 'status', 'test1']).rstrip().decode('utf-8')
 431         print(output)
 432         self.assertRegex(output, '192.168.0.15')
 433         self.assertRegex(output, '192.168.0.1')
 434         self.assertRegex(output, 'routable')
 435
 436     def test_bond_active_slave(self):
 437         self.copy_unit_to_networkd_unit_path('23-active-slave.network', '23-bond199.network', '25-bond-active-backup-slave.netdev', '12-dummy.netdev')
 438         self.start_networkd()
 439
 440         self.assertTrue(self.link_exits('dummy98'))
 441         self.assertTrue(self.link_exits('bond199'))
 442         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'bond199']).rstrip().decode('utf-8')
 443         print(output)
 444         self.assertRegex(output, 'active_slave dummy98')
 445
 446     def test_bond_primary_slave(self):
 447         self.copy_unit_to_networkd_unit_path('23-primary-slave.network', '23-test1-bond199.network', '25-bond-active-backup-slave.netdev', '11-dummy.netdev')
 448         self.start_networkd()
 449
 450         self.assertTrue(self.link_exits('test1'))
 451         self.assertTrue(self.link_exits('bond199'))
 452         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'bond199']).rstrip().decode('utf-8')
 453         print(output)
 454         self.assertRegex(output, 'primary test1')
 455
 456     def test_routing_policy_rule(self):
 457         self.copy_unit_to_networkd_unit_path('routing-policy-rule.network', '11-dummy.netdev')
 458         self.start_networkd()
 459
 460         self.assertTrue(self.link_exits('test1'))
 461         output = subprocess.check_output(['ip', 'rule']).rstrip().decode('utf-8')
 462         print(output)
 463         self.assertRegex(output, '111')
 464         self.assertRegex(output, 'from 192.168.100.18')
 465         self.assertRegex(output, r'tos (?:0x08|throughput)\s')
 466         self.assertRegex(output, 'iif test1')
 467         self.assertRegex(output, 'oif test1')
 468         self.assertRegex(output, 'lookup 7')
 469
 470     def test_address_preferred_lifetime_zero_ipv6(self):
 471         self.copy_unit_to_networkd_unit_path('25-address-section-miscellaneous.network', '12-dummy.netdev')
 472         self.start_networkd()
 473
 474         self.assertTrue(self.link_exits('dummy98'))
 475
 476         output = subprocess.check_output(['ip', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
 477         print(output)
 478         self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope link deprecated dummy98')
 479         self.assertRegex(output, 'inet6 2001:db8:0:f101::1/64 scope global')
 480
 481     def test_ip_route(self):
 482         self.copy_unit_to_networkd_unit_path('25-route-section.network', '12-dummy.netdev')
 483         self.start_networkd()
 484
 485         self.assertTrue(self.link_exits('dummy98'))
 486
 487         output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'dummy98']).rstrip().decode('utf-8')
 488         print(output)
 489         self.assertRegex(output, '192.168.0.1')
 490         self.assertRegex(output, 'static')
 491         self.assertRegex(output, '192.168.0.0/24')
 492
 493     def test_ip_route_blackhole_unreachable_prohibit(self):
 494         self.copy_unit_to_networkd_unit_path('25-route-type.network', '12-dummy.netdev')
 495         self.start_networkd()
 496
 497         self.assertTrue(self.link_exits('dummy98'))
 498
 499         output = subprocess.check_output(['ip', 'route', 'list']).rstrip().decode('utf-8')
 500         print(output)
 501         self.assertRegex(output, 'blackhole')
 502         self.assertRegex(output, 'unreachable')
 503         self.assertRegex(output, 'prohibit')
 504
 505         subprocess.call(['ip', 'route', 'del', 'blackhole', '202.54.1.2'])
 506         subprocess.call(['ip', 'route', 'del', 'unreachable', '202.54.1.3'])
 507         subprocess.call(['ip', 'route', 'del', 'prohibit', '202.54.1.4'])
 508
 509     def test_ip_route_tcp_window(self):
 510         self.copy_unit_to_networkd_unit_path('25-route-tcp-window-settings.network', '11-dummy.netdev')
 511         self.start_networkd()
 512
 513         self.assertTrue(self.link_exits('test1'))
 514
 515         output = subprocess.check_output(['ip', 'route', 'list']).rstrip().decode('utf-8')
 516         print(output)
 517         self.assertRegex(output, 'initcwnd 20')
 518         self.assertRegex(output, 'initrwnd 30')
 519
 520     def test_ip_link_mac_address(self):
 521         self.copy_unit_to_networkd_unit_path('25-address-link-section.network', '12-dummy.netdev')
 522         self.start_networkd()
 523
 524         self.assertTrue(self.link_exits('dummy98'))
 525
 526         output = subprocess.check_output(['ip', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
 527         print(output)
 528         self.assertRegex(output, '00:01:02:aa:bb:cc')
 529
 530     def test_ip_link_unmanaged(self):
 531         self.copy_unit_to_networkd_unit_path('25-link-section-unmanaged.network', '12-dummy.netdev')
 532         self.start_networkd()
 533
 534         self.assertTrue(self.link_exits('dummy98'))
 535
 536         output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
 537         print(output)
 538         self.assertRegex(output, 'unmanaged')
 539
 540     def test_ipv6_address_label(self):
 541         self.copy_unit_to_networkd_unit_path('25-ipv6-address-label-section.network', '12-dummy.netdev')
 542         self.start_networkd()
 543
 544         self.assertTrue(self.link_exits('dummy98'))
 545
 546         output = subprocess.check_output(['ip', 'addrlabel', 'list']).rstrip().decode('utf-8')
 547         print(output)
 548         self.assertRegex(output, '2004:da8:1::/64')
 549
 550     def test_sysctl(self):
 551         self.copy_unit_to_networkd_unit_path('25-sysctl.network', '12-dummy.netdev')
 552         self.start_networkd()
 553
 554         self.assertTrue(self.link_exits('dummy98'))
 555
 556         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'forwarding'), '1')
 557         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'use_tempaddr'), '2')
 558         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'dad_transmits'), '3')
 559         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'hop_limit'), '5')
 560         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'proxy_ndp'), '1')
 561         self.assertEqual(self.read_ipv4_sysctl_attr('dummy98', 'forwarding'),'1')
 562         self.assertEqual(self.read_ipv4_sysctl_attr('dummy98', 'proxy_arp'), '1')
 563
 564 class NetworkdNetWorkBrideTests(unittest.TestCase, Utilities):
 565     links = ['dummy98', 'test1', 'bridge99']
 566
 567     units = ['11-dummy.netdev', '12-dummy.netdev', '26-bridge.netdev', '26-bridge-slave-interface-1.network',
 568              '26-bridge-slave-interface-2.network', 'bridge99.network']
 569
 570     def setUp(self):
 571         self.link_remove(self.links)
 572
 573     def tearDown(self):
 574         self.link_remove(self.links)
 575         self.remove_unit_from_networkd_path(self.units)
 576
 577     def test_bridge_property(self):
 578         self.copy_unit_to_networkd_unit_path('11-dummy.netdev', '12-dummy.netdev', '26-bridge.netdev',
 579                                              '26-bridge-slave-interface-1.network', '26-bridge-slave-interface-2.network',
 580                                              'bridge99.network')
 581         self.start_networkd()
 582
 583         self.assertTrue(self.link_exits('dummy98'))
 584         self.assertTrue(self.link_exits('test1'))
 585         self.assertTrue(self.link_exits('bridge99'))
 586
 587         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1']).rstrip().decode('utf-8')
 588         print(output)
 589         self.assertRegex(output, 'master')
 590         self.assertRegex(output, 'bridge')
 591
 592         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
 593         print(output)
 594         self.assertRegex(output, 'master')
 595         self.assertRegex(output, 'bridge')
 596
 597         output = subprocess.check_output(['ip', 'addr', 'show', 'bridge99']).rstrip().decode('utf-8')
 598         print(output)
 599         self.assertRegex(output, '192.168.0.15')
 600         self.assertRegex(output, '192.168.0.1')
 601
 602         output = subprocess.check_output(['bridge', '-d', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
 603         print(output)
 604         self.assertRegex(output, 'cost 400')
 605         self.assertRegex(output, 'hairpin on')
 606         self.assertRegex(output, 'flood on')
 607         self.assertRegex(output, 'fastleave on')
 608
 609 class NetworkdNetWorkLLDPTests(unittest.TestCase, Utilities):
 610     links = ['veth99']
 611
 612     units = ['23-emit-lldp.network', '24-lldp.network', '25-veth.netdev']
 613
 614     def setUp(self):
 615         self.link_remove(self.links)
 616
 617     def tearDown(self):
 618         self.link_remove(self.links)
 619         self.remove_unit_from_networkd_path(self.units)
 620
 621     def test_lldp(self):
 622         self.copy_unit_to_networkd_unit_path('23-emit-lldp.network', '24-lldp.network', '25-veth.netdev')
 623         self.start_networkd()
 624
 625         self.assertTrue(self.link_exits('veth99'))
 626
 627         output = subprocess.check_output(['networkctl', 'lldp']).rstrip().decode('utf-8')
 628         print(output)
 629         self.assertRegex(output, 'veth-peer')
 630         self.assertRegex(output, 'veth99')
 631
 632 class NetworkdNetworkRATests(unittest.TestCase, Utilities):
 633     links = ['veth99']
 634
 635     units = ['25-veth.netdev', 'ipv6-prefix.network', 'ipv6-prefix-veth.network']
 636
 637     def setUp(self):
 638         self.link_remove(self.links)
 639
 640     def tearDown(self):
 641         self.link_remove(self.links)
 642         self.remove_unit_from_networkd_path(self.units)
 643
 644     def test_ipv6_prefix_delegation(self):
 645         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'ipv6-prefix.network', 'ipv6-prefix-veth.network')
 646         self.start_networkd()
 647
 648         self.assertTrue(self.link_exits('veth99'))
 649
 650         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 651         print(output)
 652         self.assertRegex(output, '2002:da8:1:0')
 653
 654 class NetworkdNetworkDHCPServerTests(unittest.TestCase, Utilities):
 655     links = ['veth99', 'dummy98']
 656
 657     units = ['25-veth.netdev', 'dhcp-client.network', 'dhcp-server.network', '12-dummy.netdev', '24-search-domain.network',
 658              'dhcp-client-timezone-router.network', 'dhcp-server-timezone-router.network']
 659
 660     def setUp(self):
 661         self.link_remove(self.links)
 662
 663     def tearDown(self):
 664         self.link_remove(self.links)
 665         self.remove_unit_from_networkd_path(self.units)
 666
 667     def test_dhcp_server(self):
 668         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client.network', 'dhcp-server.network')
 669         self.start_networkd()
 670
 671         self.assertTrue(self.link_exits('veth99'))
 672
 673         time.sleep(5)
 674
 675         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 676         print(output)
 677         self.assertRegex(output, '192.168.5.*')
 678         self.assertRegex(output, 'Gateway: 192.168.5.1')
 679         self.assertRegex(output, 'DNS: 192.168.5.1')
 680         self.assertRegex(output, 'NTP: 192.168.5.1')
 681
 682     def test_domain(self):
 683         self.copy_unit_to_networkd_unit_path( '12-dummy.netdev', '24-search-domain.network')
 684         self.start_networkd()
 685
 686         self.assertTrue(self.link_exits('dummy98'))
 687
 688         output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
 689         print(output)
 690         self.assertRegex(output, 'Address: 192.168.42.100')
 691         self.assertRegex(output, 'DNS: 192.168.42.1')
 692         self.assertRegex(output, 'Search Domains: one')
 693
 694     def test_emit_router_timezone(self):
 695         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client-timezone-router.network', 'dhcp-server-timezone-router.network')
 696         self.start_networkd()
 697
 698         self.assertTrue(self.link_exits('veth99'))
 699
 700         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 701         print(output)
 702         self.assertRegex(output, 'Gateway: 192.168.5.*')
 703         self.assertRegex(output, '192.168.5.*')
 704         self.assertRegex(output, 'Europe/Berlin')
 705
 706 class NetworkdNetworkDHCPClientTests(unittest.TestCase, Utilities):
 707     links = ['veth99', 'dummy98']
 708
 709     units = ['25-veth.netdev', 'dhcp-server-veth-peer.network','dhcp-client-ipv6-only.network',
 710              'dhcp-client-ipv4-only-ipv6-disabled.network', 'dhcp-client-ipv4-only.network',
 711              'dhcp-client-ipv4-dhcp-settings.network', 'dhcp-client-anonymize.network',
 712              'dhcp-client-ipv6-rapid-commit.network', 'dhcp-client-route-table.network',
 713              'dhcp-v4-server-veth-peer.network', 'dhcp-client-listen-port.network',
 714              'dhcp-client-route-metric.network', 'dhcp-client-critical-connection.network']
 715
 716     def setUp(self):
 717         self.link_remove(self.links)
 718         self.stop_dnsmasq(dnsmasq_pid_file)
 719
 720     def tearDown(self):
 721         self.link_remove(self.links)
 722         self.remove_unit_from_networkd_path(self.units)
 723         self.stop_dnsmasq(dnsmasq_pid_file)
 724         self.remove_lease_file()
 725         self.remove_log_file()
 726
 727     def test_dhcp_client_ipv6_only(self):
 728         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network','dhcp-client-ipv6-only.network')
 729         self.start_networkd()
 730
 731         self.assertTrue(self.link_exits('veth99'))
 732
 733         self.start_dnsmasq()
 734
 735         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 736         print(output)
 737         self.assertRegex(output, '2600::')
 738         self.assertNotRegex(output, '192.168.5')
 739
 740     def test_dhcp_client_ipv4_only(self):
 741         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network','dhcp-client-ipv4-only-ipv6-disabled.network')
 742         self.start_networkd()
 743
 744         self.assertTrue(self.link_exits('veth99'))
 745
 746         self.start_dnsmasq()
 747
 748         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 749         print(output)
 750         self.assertNotRegex(output, '2600::')
 751         self.assertRegex(output, '192.168.5')
 752
 753     def test_dhcp_client_ipv4_ipv6(self):
 754         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network',
 755                                              'dhcp-client-ipv4-only.network')
 756         self.start_networkd()
 757
 758         self.assertTrue(self.link_exits('veth99'))
 759
 760         self.start_dnsmasq()
 761
 762         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 763         print(output)
 764         self.assertRegex(output, '2600::')
 765         self.assertRegex(output, '192.168.5')
 766
 767     def test_dhcp_client_settings(self):
 768         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-dhcp-settings.network')
 769         self.start_networkd()
 770
 771         self.assertTrue(self.link_exits('veth99'))
 772
 773         self.start_dnsmasq()
 774
 775         output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
 776         print(output)
 777         self.assertRegex(output, '12:34:56:78:9a:bc')
 778         self.assertRegex(output, '192.168.5')
 779         self.assertRegex(output, '1492')
 780
 781         output = subprocess.check_output(['ip', 'route']).rstrip().decode('utf-8')
 782         print(output)
 783         self.assertRegex(output, 'default.*dev veth99 proto dhcp')
 784
 785         self.search_words_in_file('vendor class: SusantVendorTest')
 786         self.search_words_in_file('client MAC address: 12:34:56:78:9a:bc')
 787         self.search_words_in_file('client provides name: test-hostname')
 788         self.search_words_in_file('26:mtu')
 789
 790     def test_dhcp6_client_settings_rapidcommit_true(self):
 791         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network')
 792         self.start_networkd()
 793
 794         self.assertTrue(self.link_exits('veth99'))
 795
 796         self.start_dnsmasq()
 797
 798         output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
 799         print(output)
 800         self.assertRegex(output, '12:34:56:78:9a:bc')
 801
 802         self.assertTrue(self.search_words_in_file('14:rapid-commit'))
 803
 804     def test_dhcp6_client_settings_rapidcommit_false(self):
 805         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-rapid-commit.network')
 806         self.start_networkd()
 807
 808         self.assertTrue(self.link_exits('veth99'))
 809
 810         self.start_dnsmasq()
 811
 812         output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
 813         print(output)
 814         self.assertRegex(output, '12:34:56:78:9a:bc')
 815
 816         self.assertFalse(self.search_words_in_file('14:rapid-commit'))
 817
 818     def test_dhcp_client_settings_anonymize(self):
 819         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-anonymize.network')
 820         self.start_networkd()
 821
 822         self.assertTrue(self.link_exits('veth99'))
 823
 824         self.start_dnsmasq()
 825         self.assertFalse(self.search_words_in_file('VendorClassIdentifier=SusantVendorTest'))
 826         self.assertFalse(self.search_words_in_file('test-hostname'))
 827         self.assertFalse(self.search_words_in_file('26:mtu'))
 828
 829     def test_dhcp_client_listen_port(self):
 830         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-listen-port.network')
 831         dh_server = DHCPServer("dhcp_server")
 832         dh_server.start()
 833
 834         self.start_networkd()
 835
 836         self.assertTrue(self.link_exits('veth99'))
 837
 838         global port
 839         global ip
 840
 841         self.assertRegex(str(port), '5555')
 842         self.assertRegex(str(ip), '0.0.0.0')
 843
 844         dh_server.join()
 845
 846     def test_dhcp_route_table_id(self):
 847         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-table.network')
 848         self.start_networkd()
 849         self.start_dnsmasq()
 850
 851         self.assertTrue(self.link_exits('veth99'))
 852
 853         output = subprocess.check_output(['ip', 'route', 'show', 'table', '12']).rstrip().decode('utf-8')
 854         print(output)
 855
 856         self.assertRegex(output, 'veth99 proto dhcp')
 857         self.assertRegex(output, '192.168.5.1')
 858
 859     def test_dhcp_route_metric(self):
 860         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-metric.network')
 861         self.start_networkd()
 862         self.start_dnsmasq()
 863
 864         self.assertTrue(self.link_exits('veth99'))
 865
 866         output = subprocess.check_output(['ip', 'route', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
 867         print(output)
 868
 869         self.assertRegex(output, 'metric 24')
 870
 871     def test_dhcp_route_criticalconnection_true(self):
 872         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-critical-connection.network')
 873         self.start_networkd()
 874         self.start_dnsmasq()
 875
 876         self.assertTrue(self.link_exits('veth99'))
 877
 878         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 879         print(output)
 880
 881         self.assertRegex(output, '192.168.5.*')
 882         # Stoping dnsmasq as networkd won't be allowed to renew the DHCP lease.
 883         self.stop_dnsmasq(dnsmasq_pid_file)
 884
 885         # Sleep for 120 sec as the dnsmasq minimum lease time can only be set to 120
 886         time.sleep(125)
 887
 888         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 889         print(output)
 890         self.assertRegex(output, '192.168.5.*')
 891
 892 if __name__ == '__main__':
 893     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
 894                                                      verbosity=3))