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 re
   7 import shutil
   8 import signal
   9 import socket
  10 import subprocess
  11 import sys
  12 import time
  13 import unittest
  14 from shutil import copytree
  15
  16 network_unit_file_path='/run/systemd/network'
  17 networkd_runtime_directory='/run/systemd/netif'
  18 networkd_ci_path='/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_pid_file='/run/networkd-ci/test-test-dnsmasq.pid'
  23 dnsmasq_log_file='/run/networkd-ci/test-dnsmasq-log-file'
  24
  25 def is_module_available(module_name):
  26     lsmod_output = subprocess.check_output('lsmod', universal_newlines=True)
  27     module_re = re.compile(r'^{0}\b'.format(re.escape(module_name)), re.MULTILINE)
  28     return module_re.search(lsmod_output) or not subprocess.call(["modprobe", module_name])
  29
  30 def expectedFailureIfModuleIsNotAvailable(module_name):
  31     def f(func):
  32         if not is_module_available(module_name):
  33             return unittest.expectedFailure(func)
  34         return func
  35
  36     return f
  37
  38 def expectedFailureIfERSPANModuleIsNotAvailable():
  39     def f(func):
  40         rc = subprocess.call(['ip', 'link', 'add', 'dev', 'erspan99', 'type', 'erspan', 'seq', 'key', '30', 'local', '192.168.1.4', 'remote', '192.168.1.1', 'erspan_ver', '1', 'erspan', '123'])
  41         if rc == 0:
  42             subprocess.call(['ip', 'link', 'del', 'erspan99'])
  43             return func
  44         else:
  45             return unittest.expectedFailure(func)
  46
  47     return f
  48
  49 def expectedFailureIfRoutingPolicyPortRangeIsNotAvailable():
  50     def f(func):
  51         rc = subprocess.call(['ip', 'rule', 'add', 'from', '192.168.100.19', 'sport', '1123-1150', 'dport', '3224-3290', 'table', '7'])
  52         if rc == 0:
  53             subprocess.call(['ip', 'rule', 'del', 'from', '192.168.100.19', 'sport', '1123-1150', 'dport', '3224-3290', 'table', '7'])
  54             return func
  55         else:
  56             return unittest.expectedFailure(func)
  57
  58     return f
  59
  60 def expectedFailureIfRoutingPolicyIPProtoIsNotAvailable():
  61     def f(func):
  62         rc = subprocess.call(['ip', 'rule', 'add', 'not', 'from', '192.168.100.19', 'ipproto', 'tcp', 'table', '7'])
  63         if rc == 0:
  64             subprocess.call(['ip', 'rule', 'del', 'not', 'from', '192.168.100.19', 'ipproto', 'tcp', 'table', '7'])
  65             return func
  66         else:
  67             return unittest.expectedFailure(func)
  68
  69     return f
  70
  71 def setUpModule():
  72
  73     os.makedirs(network_unit_file_path, exist_ok=True)
  74     os.makedirs(networkd_ci_path, exist_ok=True)
  75
  76     shutil.rmtree(networkd_ci_path)
  77     copytree(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'conf'), networkd_ci_path)
  78
  79     subprocess.check_call('systemctl stop systemd-networkd.socket', shell=True)
  80
  81 def tearDownModule():
  82     shutil.rmtree(networkd_ci_path)
  83
  84     subprocess.check_call('systemctl stop systemd-networkd.service', shell=True)
  85     subprocess.check_call('systemctl start systemd-networkd.socket', shell=True)
  86     subprocess.check_call('systemctl start systemd-networkd.service', shell=True)
  87
  88 class Utilities():
  89     dhcp_server_data = []
  90
  91     def read_link_attr(self, link, dev, attribute):
  92         with open(os.path.join(os.path.join(os.path.join('/sys/class/net/', link), dev), attribute)) as f:
  93             return f.readline().strip()
  94
  95     def read_bridge_port_attr(self, bridge, link, attribute):
  96
  97         path_bridge = os.path.join('/sys/devices/virtual/net', bridge)
  98         path_port = 'lower_' + link + '/brport'
  99         path = os.path.join(path_bridge, path_port)
 100
 101         with open(os.path.join(path, attribute)) as f:
 102             return f.readline().strip()
 103
 104     def link_exits(self, link):
 105         return os.path.exists(os.path.join('/sys/class/net', link))
 106
 107     def link_remove(self, links):
 108         for link in links:
 109             if os.path.exists(os.path.join('/sys/class/net', link)):
 110                 subprocess.call(['ip', 'link', 'del', 'dev', link])
 111         time.sleep(1)
 112
 113     def read_ipv6_sysctl_attr(self, link, attribute):
 114         with open(os.path.join(os.path.join(network_sysctl_ipv6_path, link), attribute)) as f:
 115             return f.readline().strip()
 116
 117     def read_ipv4_sysctl_attr(self, link, attribute):
 118         with open(os.path.join(os.path.join(network_sysctl_ipv4_path, link), attribute)) as f:
 119             return f.readline().strip()
 120
 121     def copy_unit_to_networkd_unit_path(self, *units):
 122         for unit in units:
 123             shutil.copy(os.path.join(networkd_ci_path, unit), network_unit_file_path)
 124             if (os.path.exists(os.path.join(networkd_ci_path, unit + '.d'))):
 125                 copytree(os.path.join(networkd_ci_path, unit + '.d'), os.path.join(network_unit_file_path, unit + '.d'))
 126
 127     def remove_unit_from_networkd_path(self, units):
 128         for unit in units:
 129             if (os.path.exists(os.path.join(network_unit_file_path, unit))):
 130                 os.remove(os.path.join(network_unit_file_path, unit))
 131                 if (os.path.exists(os.path.join(network_unit_file_path, unit + '.d'))):
 132                     shutil.rmtree(os.path.join(network_unit_file_path, unit + '.d'))
 133
 134     def start_dnsmasq(self, additional_options=''):
 135         dnsmasq_command = '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 --port=0 ' + additional_options
 136         subprocess.check_call(dnsmasq_command, shell=True)
 137
 138         time.sleep(10)
 139
 140     def stop_dnsmasq(self, pid_file):
 141         if os.path.exists(pid_file):
 142             with open(pid_file, 'r') as f:
 143                 pid = f.read().rstrip(' \t\r\n\0')
 144                 os.kill(int(pid), signal.SIGTERM)
 145
 146             os.remove(pid_file)
 147
 148     def search_words_in_dnsmasq_log(self, words, show_all=False):
 149         if os.path.exists(dnsmasq_log_file):
 150             with open (dnsmasq_log_file) as in_file:
 151                 contents = in_file.read()
 152                 if show_all:
 153                     print(contents)
 154                 for line in contents.split('\n'):
 155                     if words in line:
 156                         in_file.close()
 157                         print("%s, %s" % (words, line))
 158                         return True
 159         return False
 160
 161     def remove_lease_file(self):
 162         if os.path.exists(os.path.join(networkd_ci_path, 'lease')):
 163             os.remove(os.path.join(networkd_ci_path, 'lease'))
 164
 165     def remove_log_file(self):
 166         if os.path.exists(dnsmasq_log_file):
 167             os.remove(dnsmasq_log_file)
 168
 169     def start_networkd(self):
 170         if (os.path.exists(os.path.join(networkd_runtime_directory, 'state'))):
 171             subprocess.check_call('systemctl stop systemd-networkd', shell=True)
 172             os.remove(os.path.join(networkd_runtime_directory, 'state'))
 173             subprocess.check_call('systemctl start systemd-networkd', shell=True)
 174         else:
 175             subprocess.check_call('systemctl restart systemd-networkd', shell=True)
 176         time.sleep(5)
 177         print()
 178
 179 class NetworkdNetDevTests(unittest.TestCase, Utilities):
 180
 181     links =[
 182         '6rdtun99',
 183         'bond99',
 184         'bridge99',
 185         'dropin-test',
 186         'dummy98',
 187         'erspan-test',
 188         'geneve99',
 189         'gretap99',
 190         'gretun99',
 191         'ip6gretap99',
 192         'ip6tnl99',
 193         'ipiptun99',
 194         'ipvlan99',
 195         'isataptun99',
 196         'macvlan99',
 197         'macvtap99',
 198         'sittun99',
 199         'tap99',
 200         'test1',
 201         'tun99',
 202         'vcan99',
 203         'veth99',
 204         'vlan99',
 205         'vrf99',
 206         'vti6tun99',
 207         'vtitun99',
 208         'vxlan99',
 209         'wg98',
 210         'wg99']
 211
 212     units = [
 213         '10-dropin-test.netdev',
 214         '11-dummy.netdev',
 215         '12-dummy.netdev',
 216         '21-macvlan.netdev',
 217         '21-macvtap.netdev',
 218         '21-vlan.netdev',
 219         '21-vlan.network',
 220         '25-6rd-tunnel.netdev',
 221         '25-bond.netdev',
 222         '25-bond-balanced-tlb.netdev',
 223         '25-bridge.netdev',
 224         '25-erspan-tunnel.netdev',
 225         '25-geneve.netdev',
 226         '25-gretap-tunnel.netdev',
 227         '25-gre-tunnel.netdev',
 228         '25-ip6gre-tunnel.netdev',
 229         '25-ip6tnl-tunnel.netdev',
 230         '25-ipip-tunnel-independent.netdev',
 231         '25-ipip-tunnel.netdev',
 232         '25-ipvlan.netdev',
 233         '25-isatap-tunnel.netdev',
 234         '25-sit-tunnel.netdev',
 235         '25-tap.netdev',
 236         '25-tun.netdev',
 237         '25-vcan.netdev',
 238         '25-veth.netdev',
 239         '25-vrf.netdev',
 240         '25-vti6-tunnel.netdev',
 241         '25-vti-tunnel.netdev',
 242         '25-vxlan.netdev',
 243         '25-wireguard-23-peers.netdev',
 244         '25-wireguard-23-peers.network',
 245         '25-wireguard.netdev',
 246         '6rd.network',
 247         'gre.network',
 248         'gretap.network',
 249         'gretun.network',
 250         'ip6gretap.network',
 251         'ip6tnl.network',
 252         'ipip.network',
 253         'ipvlan.network',
 254         'isatap.network',
 255         'macvlan.network',
 256         'macvtap.network',
 257         'sit.network',
 258         'vti6.network',
 259         'vti.network',
 260         'vxlan.network']
 261
 262     def setUp(self):
 263         self.link_remove(self.links)
 264
 265     def tearDown(self):
 266         self.link_remove(self.links)
 267         self.remove_unit_from_networkd_path(self.units)
 268
 269     def test_dropin(self):
 270         self.copy_unit_to_networkd_unit_path('10-dropin-test.netdev')
 271         self.start_networkd()
 272
 273         self.assertTrue(self.link_exits('dropin-test'))
 274
 275         output = subprocess.check_output(['ip', 'link', 'show', 'dropin-test']).rstrip().decode('utf-8')
 276         print(output)
 277         self.assertRegex(output, '00:50:56:c0:00:28')
 278
 279     def test_bridge(self):
 280         self.copy_unit_to_networkd_unit_path('25-bridge.netdev')
 281         self.start_networkd()
 282
 283         self.assertTrue(self.link_exits('bridge99'))
 284
 285         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge', 'hello_time'))
 286         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge', 'max_age'))
 287         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge','forward_delay'))
 288         self.assertEqual('900', self.read_link_attr('bridge99', 'bridge','ageing_time'))
 289         self.assertEqual('9',   self.read_link_attr('bridge99', 'bridge','priority'))
 290         self.assertEqual('1',   self.read_link_attr('bridge99', 'bridge','multicast_querier'))
 291         self.assertEqual('1',   self.read_link_attr('bridge99', 'bridge','multicast_snooping'))
 292         self.assertEqual('1',   self.read_link_attr('bridge99', 'bridge','stp_state'))
 293
 294     def test_bond(self):
 295         self.copy_unit_to_networkd_unit_path('25-bond.netdev')
 296         self.start_networkd()
 297
 298         self.assertTrue(self.link_exits('bond99'))
 299
 300         self.assertEqual('802.3ad 4',         self.read_link_attr('bond99', 'bonding', 'mode'))
 301         self.assertEqual('layer3+4 1',        self.read_link_attr('bond99', 'bonding', 'xmit_hash_policy'))
 302         self.assertEqual('1000',              self.read_link_attr('bond99', 'bonding', 'miimon'))
 303         self.assertEqual('fast 1',            self.read_link_attr('bond99', 'bonding', 'lacp_rate'))
 304         self.assertEqual('2000',              self.read_link_attr('bond99', 'bonding', 'updelay'))
 305         self.assertEqual('2000',              self.read_link_attr('bond99', 'bonding', 'downdelay'))
 306         self.assertEqual('4',                 self.read_link_attr('bond99', 'bonding', 'resend_igmp'))
 307         self.assertEqual('1',                 self.read_link_attr('bond99', 'bonding', 'min_links'))
 308         self.assertEqual('1218',              self.read_link_attr('bond99', 'bonding', 'ad_actor_sys_prio'))
 309         self.assertEqual('811',               self.read_link_attr('bond99', 'bonding', 'ad_user_port_key'))
 310         self.assertEqual('00:11:22:33:44:55', self.read_link_attr('bond99', 'bonding', 'ad_actor_system'))
 311
 312     def test_bond_balanced_tlb(self):
 313         self.copy_unit_to_networkd_unit_path('25-bond-balanced-tlb.netdev')
 314         self.start_networkd()
 315
 316         self.assertTrue(self.link_exits('bond99'))
 317
 318         self.assertEqual('balance-tlb 5', self.read_link_attr('bond99', 'bonding', 'mode'))
 319         self.assertEqual('1',             self.read_link_attr('bond99', 'bonding', 'tlb_dynamic_lb'))
 320
 321     def test_vlan(self):
 322         self.copy_unit_to_networkd_unit_path('21-vlan.netdev', '11-dummy.netdev', '21-vlan.network')
 323         self.start_networkd()
 324
 325         self.assertTrue(self.link_exits('vlan99'))
 326
 327         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vlan99']).rstrip().decode('utf-8')
 328         print(output)
 329         self.assertTrue(output, 'REORDER_HDR')
 330         self.assertTrue(output, 'LOOSE_BINDING')
 331         self.assertTrue(output, 'GVRP')
 332         self.assertTrue(output, 'MVRP')
 333         self.assertTrue(output, '99')
 334
 335     def test_macvtap(self):
 336         self.copy_unit_to_networkd_unit_path('21-macvtap.netdev', '11-dummy.netdev', 'macvtap.network')
 337         self.start_networkd()
 338
 339         self.assertTrue(self.link_exits('macvtap99'))
 340
 341     def test_macvlan(self):
 342         self.copy_unit_to_networkd_unit_path('21-macvlan.netdev', '11-dummy.netdev', 'macvlan.network')
 343         self.start_networkd()
 344
 345         self.assertTrue(self.link_exits('macvlan99'))
 346
 347     @expectedFailureIfModuleIsNotAvailable('ipvlan')
 348     def test_ipvlan(self):
 349         self.copy_unit_to_networkd_unit_path('25-ipvlan.netdev', '11-dummy.netdev', 'ipvlan.network')
 350         self.start_networkd()
 351
 352         self.assertTrue(self.link_exits('ipvlan99'))
 353
 354     def test_veth(self):
 355         self.copy_unit_to_networkd_unit_path('25-veth.netdev')
 356         self.start_networkd()
 357
 358         self.assertTrue(self.link_exits('veth99'))
 359
 360     def test_dummy(self):
 361         self.copy_unit_to_networkd_unit_path('11-dummy.netdev')
 362         self.start_networkd()
 363
 364         self.assertTrue(self.link_exits('test1'))
 365
 366     def test_tun(self):
 367         self.copy_unit_to_networkd_unit_path('25-tun.netdev')
 368         self.start_networkd()
 369
 370         self.assertTrue(self.link_exits('tun99'))
 371
 372     def test_tap(self):
 373         self.copy_unit_to_networkd_unit_path('25-tap.netdev')
 374         self.start_networkd()
 375
 376         self.assertTrue(self.link_exits('tap99'))
 377
 378     @expectedFailureIfModuleIsNotAvailable('vrf')
 379     def test_vrf(self):
 380         self.copy_unit_to_networkd_unit_path('25-vrf.netdev')
 381         self.start_networkd()
 382
 383         self.assertTrue(self.link_exits('vrf99'))
 384
 385     @expectedFailureIfModuleIsNotAvailable('vcan')
 386     def test_vcan(self):
 387         self.copy_unit_to_networkd_unit_path('25-vcan.netdev')
 388         self.start_networkd()
 389
 390         self.assertTrue(self.link_exits('vcan99'))
 391
 392     @expectedFailureIfModuleIsNotAvailable('wireguard')
 393     def test_wireguard(self):
 394         self.copy_unit_to_networkd_unit_path('25-wireguard.netdev')
 395         self.start_networkd()
 396
 397         if shutil.which('wg'):
 398             subprocess.call('wg')
 399             output = subprocess.check_output(['wg', 'show', 'wg99', 'listen-port']).rstrip().decode('utf-8')
 400             self.assertTrue(output, '51820')
 401             output = subprocess.check_output(['wg', 'show', 'wg99', 'fwmark']).rstrip().decode('utf-8')
 402             self.assertTrue(output, '0x4d2')
 403             output = subprocess.check_output(['wg', 'show', 'wg99', 'allowed-ips']).rstrip().decode('utf-8')
 404             self.assertTrue(output, 'RDf+LSpeEre7YEIKaxg+wbpsNV7du+ktR99uBEtIiCA=\t192.168.26.0/24 fd31:bf08:57cb::/48')
 405             output = subprocess.check_output(['wg', 'show', 'wg99', 'persistent-keepalive']).rstrip().decode('utf-8')
 406             self.assertTrue(output, 'RDf+LSpeEre7YEIKaxg+wbpsNV7du+ktR99uBEtIiCA=\t20')
 407             output = subprocess.check_output(['wg', 'show', 'wg99', 'endpoints']).rstrip().decode('utf-8')
 408             self.assertTrue(output, 'RDf+LSpeEre7YEIKaxg+wbpsNV7du+ktR99uBEtIiCA=\t192.168.27.3:51820')
 409
 410         self.assertTrue(self.link_exits('wg99'))
 411
 412     @expectedFailureIfModuleIsNotAvailable('wireguard')
 413     def test_wireguard_23_peers(self):
 414         self.copy_unit_to_networkd_unit_path('25-wireguard-23-peers.netdev', '25-wireguard-23-peers.network')
 415         self.start_networkd()
 416
 417         if shutil.which('wg'):
 418             subprocess.call('wg')
 419
 420         self.assertTrue(self.link_exits('wg98'))
 421
 422     def test_geneve(self):
 423         self.copy_unit_to_networkd_unit_path('25-geneve.netdev')
 424         self.start_networkd()
 425
 426         self.assertTrue(self.link_exits('geneve99'))
 427
 428         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'geneve99']).rstrip().decode('utf-8')
 429         print(output)
 430         self.assertTrue(output, '192.168.22.1')
 431         self.assertTrue(output, '6082')
 432         self.assertTrue(output, 'udpcsum')
 433         self.assertTrue(output, 'udp6zerocsumrx')
 434
 435     def test_ipip_tunnel(self):
 436         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ipip-tunnel.netdev', 'ipip.network')
 437         self.start_networkd()
 438
 439         self.assertTrue(self.link_exits('dummy98'))
 440         self.assertTrue(self.link_exits('ipiptun99'))
 441
 442     def test_gre_tunnel(self):
 443         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gre-tunnel.netdev', 'gretun.network')
 444         self.start_networkd()
 445
 446         self.assertTrue(self.link_exits('dummy98'))
 447         self.assertTrue(self.link_exits('gretun99'))
 448
 449     def test_gretap_tunnel(self):
 450         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-gretap-tunnel.netdev', 'gretap.network')
 451         self.start_networkd()
 452
 453         self.assertTrue(self.link_exits('dummy98'))
 454         self.assertTrue(self.link_exits('gretap99'))
 455
 456     def test_ip6gretap_tunnel(self):
 457         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6gre-tunnel.netdev', 'ip6gretap.network')
 458         self.start_networkd()
 459
 460         self.assertTrue(self.link_exits('dummy98'))
 461         self.assertTrue(self.link_exits('ip6gretap99'))
 462
 463     def test_vti_tunnel(self):
 464         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti-tunnel.netdev', 'vti.network')
 465         self.start_networkd()
 466
 467         self.assertTrue(self.link_exits('dummy98'))
 468         self.assertTrue(self.link_exits('vtitun99'))
 469
 470     def test_vti6_tunnel(self):
 471         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-vti6-tunnel.netdev', 'vti6.network')
 472         self.start_networkd()
 473
 474         self.assertTrue(self.link_exits('dummy98'))
 475         self.assertTrue(self.link_exits('vti6tun99'))
 476
 477     def test_ip6tnl_tunnel(self):
 478         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-ip6tnl-tunnel.netdev', 'ip6tnl.network')
 479         self.start_networkd()
 480
 481         self.assertTrue(self.link_exits('dummy98'))
 482         self.assertTrue(self.link_exits('ip6tnl99'))
 483
 484     def test_sit_tunnel(self):
 485         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-sit-tunnel.netdev', 'sit.network')
 486         self.start_networkd()
 487
 488         self.assertTrue(self.link_exits('dummy98'))
 489         self.assertTrue(self.link_exits('sittun99'))
 490
 491     def test_isatap_tunnel(self):
 492         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-isatap-tunnel.netdev', 'isatap.network')
 493         self.start_networkd()
 494
 495         self.assertTrue(self.link_exits('dummy98'))
 496         self.assertTrue(self.link_exits('isataptun99'))
 497
 498         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'isataptun99']).rstrip().decode('utf-8')
 499         print(output)
 500         self.assertRegex(output, "isatap ")
 501
 502     def test_6rd_tunnel(self):
 503         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '25-6rd-tunnel.netdev', '6rd.network')
 504         self.start_networkd()
 505
 506         self.assertTrue(self.link_exits('dummy98'))
 507         self.assertTrue(self.link_exits('sittun99'))
 508
 509     @expectedFailureIfERSPANModuleIsNotAvailable()
 510     def test_erspan_tunnel(self):
 511         self.copy_unit_to_networkd_unit_path('25-erspan-tunnel.netdev')
 512         self.start_networkd()
 513
 514         self.assertTrue(self.link_exits('erspan-test'))
 515
 516         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'erspan-test']).rstrip().decode('utf-8')
 517         print(output)
 518         self.assertTrue(output, '172.16.1.200')
 519         self.assertTrue(output, '172.16.1.100')
 520         self.assertTrue(output, '101')
 521
 522     def test_tunnel_independent(self):
 523         self.copy_unit_to_networkd_unit_path('25-ipip-tunnel-independent.netdev')
 524         self.start_networkd()
 525
 526         self.assertTrue(self.link_exits('ipiptun99'))
 527
 528     def test_vxlan(self):
 529         self.copy_unit_to_networkd_unit_path('25-vxlan.netdev', 'vxlan.network','11-dummy.netdev')
 530         self.start_networkd()
 531
 532         self.assertTrue(self.link_exits('vxlan99'))
 533
 534         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'vxlan99']).rstrip().decode('utf-8')
 535         print(output)
 536         self.assertRegex(output, "999")
 537         self.assertRegex(output, '5555')
 538         self.assertRegex(output, 'l2miss')
 539         self.assertRegex(output, 'l3miss')
 540         self.assertRegex(output, 'udpcsum')
 541         self.assertRegex(output, 'udp6zerocsumtx')
 542         self.assertRegex(output, 'udp6zerocsumrx')
 543         self.assertRegex(output, 'remcsumtx')
 544         self.assertRegex(output, 'remcsumrx')
 545         self.assertRegex(output, 'gbp')
 546
 547 class NetworkdNetWorkTests(unittest.TestCase, Utilities):
 548     links = [
 549         'bond199',
 550         'dummy98',
 551         'test1']
 552
 553     units = [
 554         '11-dummy.netdev',
 555         '12-dummy.netdev',
 556         '23-active-slave.network',
 557         '23-bond199.network',
 558         '23-primary-slave.network',
 559         '23-test1-bond199.network',
 560         '25-address-link-section.network',
 561         '25-address-section-miscellaneous.network',
 562         '25-address-section.network',
 563         '25-bond-active-backup-slave.netdev',
 564         '25-fibrule-invert.network',
 565         '25-fibrule-port-range.network',
 566         '25-ipv6-address-label-section.network',
 567         '25-neighbor-section.network',
 568         '25-link-section-unmanaged.network',
 569         '25-route-gateway.network',
 570         '25-route-gateway-on-link.network',
 571         '25-route-ipv6-src.network',
 572         '25-route-reverse-order.network',
 573         '25-route-section.network',
 574         '25-route-tcp-window-settings.network',
 575         '25-route-type.network',
 576         '25-sysctl.network',
 577         'configure-without-carrier.network',
 578         'routing-policy-rule.network',
 579         'test-static.network']
 580
 581     def setUp(self):
 582         self.link_remove(self.links)
 583
 584     def tearDown(self):
 585         self.link_remove(self.links)
 586         self.remove_unit_from_networkd_path(self.units)
 587
 588     def test_static_address(self):
 589         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', 'test-static.network')
 590         self.start_networkd()
 591
 592         self.assertTrue(self.link_exits('dummy98'))
 593
 594         output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
 595         print(output)
 596         self.assertRegex(output, '192.168.0.15')
 597         self.assertRegex(output, '192.168.0.1')
 598         self.assertRegex(output, 'routable')
 599
 600     def test_configure_without_carrier(self):
 601         self.copy_unit_to_networkd_unit_path('configure-without-carrier.network', '11-dummy.netdev')
 602         self.start_networkd()
 603
 604         self.assertTrue(self.link_exits('test1'))
 605
 606         output = subprocess.check_output(['networkctl', 'status', 'test1']).rstrip().decode('utf-8')
 607         print(output)
 608         self.assertRegex(output, '192.168.0.15')
 609         self.assertRegex(output, '192.168.0.1')
 610         self.assertRegex(output, 'routable')
 611
 612     def test_bond_active_slave(self):
 613         self.copy_unit_to_networkd_unit_path('23-active-slave.network', '23-bond199.network', '25-bond-active-backup-slave.netdev', '12-dummy.netdev')
 614         self.start_networkd()
 615
 616         self.assertTrue(self.link_exits('dummy98'))
 617         self.assertTrue(self.link_exits('bond199'))
 618
 619         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'bond199']).rstrip().decode('utf-8')
 620         print(output)
 621         self.assertRegex(output, 'active_slave dummy98')
 622
 623     def test_bond_primary_slave(self):
 624         self.copy_unit_to_networkd_unit_path('23-primary-slave.network', '23-test1-bond199.network', '25-bond-active-backup-slave.netdev', '11-dummy.netdev')
 625         self.start_networkd()
 626
 627         self.assertTrue(self.link_exits('test1'))
 628         self.assertTrue(self.link_exits('bond199'))
 629
 630         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'bond199']).rstrip().decode('utf-8')
 631         print(output)
 632         self.assertRegex(output, 'primary test1')
 633
 634     def test_routing_policy_rule(self):
 635         self.copy_unit_to_networkd_unit_path('routing-policy-rule.network', '11-dummy.netdev')
 636         self.start_networkd()
 637
 638         self.assertTrue(self.link_exits('test1'))
 639
 640         output = subprocess.check_output(['ip', 'rule']).rstrip().decode('utf-8')
 641         print(output)
 642         self.assertRegex(output, '111')
 643         self.assertRegex(output, 'from 192.168.100.18')
 644         self.assertRegex(output, r'tos (?:0x08|throughput)\s')
 645         self.assertRegex(output, 'iif test1')
 646         self.assertRegex(output, 'oif test1')
 647         self.assertRegex(output, 'lookup 7')
 648
 649         subprocess.call(['ip', 'rule', 'del', 'table', '7'])
 650
 651     @expectedFailureIfRoutingPolicyPortRangeIsNotAvailable()
 652     def test_routing_policy_rule_port_range(self):
 653         self.copy_unit_to_networkd_unit_path('25-fibrule-port-range.network', '11-dummy.netdev')
 654         self.start_networkd()
 655
 656         self.assertTrue(self.link_exits('test1'))
 657
 658         output = subprocess.check_output(['ip', 'rule']).rstrip().decode('utf-8')
 659         print(output)
 660         self.assertRegex(output, '111')
 661         self.assertRegex(output, 'from 192.168.100.18')
 662         self.assertRegex(output, '1123-1150')
 663         self.assertRegex(output, '3224-3290')
 664         self.assertRegex(output, 'tcp')
 665         self.assertRegex(output, 'lookup 7')
 666
 667         subprocess.call(['ip', 'rule', 'del', 'table', '7'])
 668
 669     @expectedFailureIfRoutingPolicyIPProtoIsNotAvailable()
 670     def test_routing_policy_rule_invert(self):
 671         self.copy_unit_to_networkd_unit_path('25-fibrule-invert.network', '11-dummy.netdev')
 672         self.start_networkd()
 673
 674         self.assertTrue(self.link_exits('test1'))
 675
 676         output = subprocess.check_output(['ip', 'rule']).rstrip().decode('utf-8')
 677         print(output)
 678         self.assertRegex(output, '111')
 679         self.assertRegex(output, 'not.*?from.*?192.168.100.18')
 680         self.assertRegex(output, 'tcp')
 681         self.assertRegex(output, 'lookup 7')
 682
 683         subprocess.call(['ip', 'rule', 'del', 'table', '7'])
 684
 685     def test_address_peer(self):
 686         self.copy_unit_to_networkd_unit_path('25-address-section.network', '12-dummy.netdev')
 687         self.start_networkd()
 688
 689         self.assertTrue(self.link_exits('dummy98'))
 690
 691         output = subprocess.check_output(['ip', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
 692         print(output)
 693         self.assertRegex(output, 'inet 10.2.3.4 peer 10.2.3.5/16 scope global 32')
 694         self.assertRegex(output, 'inet 10.6.7.8/16 brd 10.6.255.255 scope global 33')
 695         self.assertRegex(output, 'inet6 2001:db8::20 peer 2001:db8::10/128 scope global')
 696
 697         output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
 698         print(output)
 699         self.assertRegex(output, 'State: routable \(configured\)')
 700
 701     def test_address_preferred_lifetime_zero_ipv6(self):
 702         self.copy_unit_to_networkd_unit_path('25-address-section-miscellaneous.network', '12-dummy.netdev')
 703         self.start_networkd()
 704
 705         self.assertTrue(self.link_exits('dummy98'))
 706
 707         output = subprocess.check_output(['ip', 'address', 'show', 'dummy98']).rstrip().decode('utf-8')
 708         print(output)
 709         self.assertRegex(output, 'inet 10.2.3.4/16 brd 10.2.255.255 scope link deprecated dummy98')
 710         self.assertRegex(output, 'inet6 2001:db8:0:f101::1/64 scope global')
 711
 712     def test_ip_route(self):
 713         self.copy_unit_to_networkd_unit_path('25-route-section.network', '12-dummy.netdev')
 714         self.start_networkd()
 715
 716         self.assertTrue(self.link_exits('dummy98'))
 717
 718         output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'dummy98']).rstrip().decode('utf-8')
 719         print(output)
 720         self.assertRegex(output, '192.168.0.1')
 721         self.assertRegex(output, 'static')
 722         self.assertRegex(output, '192.168.0.0/24')
 723
 724     def test_ip_route_reverse(self):
 725         self.copy_unit_to_networkd_unit_path('25-route-reverse-order.network', '12-dummy.netdev')
 726         self.start_networkd()
 727
 728         self.assertTrue(self.link_exits('dummy98'))
 729
 730         output = subprocess.check_output(['ip', '-6', 'route', 'show', 'dev', 'dummy98']).rstrip().decode('utf-8')
 731         print(output)
 732         self.assertRegex(output, '2001:1234:5:8fff:ff:ff:ff:ff')
 733         self.assertRegex(output, '2001:1234:5:8f63::1')
 734
 735     def test_ip_route_blackhole_unreachable_prohibit(self):
 736         self.copy_unit_to_networkd_unit_path('25-route-type.network', '12-dummy.netdev')
 737         self.start_networkd()
 738
 739         self.assertTrue(self.link_exits('dummy98'))
 740
 741         output = subprocess.check_output(['ip', 'route', 'list']).rstrip().decode('utf-8')
 742         print(output)
 743         self.assertRegex(output, 'blackhole')
 744         self.assertRegex(output, 'unreachable')
 745         self.assertRegex(output, 'prohibit')
 746
 747         subprocess.call(['ip', 'route', 'del', 'blackhole', '202.54.1.2'])
 748         subprocess.call(['ip', 'route', 'del', 'unreachable', '202.54.1.3'])
 749         subprocess.call(['ip', 'route', 'del', 'prohibit', '202.54.1.4'])
 750
 751     def test_ip_route_tcp_window(self):
 752         self.copy_unit_to_networkd_unit_path('25-route-tcp-window-settings.network', '11-dummy.netdev')
 753         self.start_networkd()
 754
 755         self.assertTrue(self.link_exits('test1'))
 756
 757         output = subprocess.check_output(['ip', 'route', 'list']).rstrip().decode('utf-8')
 758         print(output)
 759         self.assertRegex(output, 'initcwnd 20')
 760         self.assertRegex(output, 'initrwnd 30')
 761
 762     def test_ip_route_gateway(self):
 763         self.copy_unit_to_networkd_unit_path('25-route-gateway.network', '12-dummy.netdev')
 764         self.start_networkd()
 765
 766         self.assertTrue(self.link_exits('dummy98'))
 767
 768         output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'dummy98', 'default']).rstrip().decode('utf-8')
 769         print(output)
 770         self.assertRegex(output, 'default')
 771         self.assertRegex(output, 'via')
 772         self.assertRegex(output, '149.10.124.64')
 773         self.assertRegex(output, 'proto')
 774         self.assertRegex(output, 'static')
 775
 776         output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'dummy98', 'src', '149.10.124.58']).rstrip().decode('utf-8')
 777         print(output)
 778         self.assertRegex(output, '149.10.124.48/28')
 779         self.assertRegex(output, 'proto')
 780         self.assertRegex(output, 'kernel')
 781         self.assertRegex(output, 'scope')
 782         self.assertRegex(output, 'link')
 783
 784     def test_ip_route_gateway_on_link(self):
 785         self.copy_unit_to_networkd_unit_path('25-route-gateway-on-link.network', '12-dummy.netdev')
 786         self.start_networkd()
 787
 788         self.assertTrue(self.link_exits('dummy98'))
 789
 790         output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'dummy98', 'default']).rstrip().decode('utf-8')
 791         print(output)
 792         self.assertRegex(output, 'default')
 793         self.assertRegex(output, 'via')
 794         self.assertRegex(output, '149.10.125.65')
 795         self.assertRegex(output, 'proto')
 796         self.assertRegex(output, 'static')
 797         self.assertRegex(output, 'onlink')
 798
 799         output = subprocess.check_output(['ip', 'route', 'list', 'dev', 'dummy98', 'src', '149.10.124.58']).rstrip().decode('utf-8')
 800         print(output)
 801         self.assertRegex(output, '149.10.124.48/28')
 802         self.assertRegex(output, 'proto')
 803         self.assertRegex(output, 'kernel')
 804         self.assertRegex(output, 'scope')
 805         self.assertRegex(output, 'link')
 806
 807     def test_ip_route_ipv6_src_route(self):
 808         # a dummy device does not make the addresses go through tentative state, so we
 809         # reuse a bond from an earlier test, which does make the addresses go through
 810         # tentative state, and do our test on that
 811         self.copy_unit_to_networkd_unit_path('23-active-slave.network', '25-route-ipv6-src.network', '25-bond-active-backup-slave.netdev', '12-dummy.netdev')
 812         self.start_networkd()
 813
 814         self.assertTrue(self.link_exits('dummy98'))
 815         self.assertTrue(self.link_exits('bond199'))
 816
 817         output = subprocess.check_output(['ip', '-6', 'route', 'list', 'dev', 'bond199']).rstrip().decode('utf-8')
 818         print(output)
 819         self.assertRegex(output, 'abcd::/16')
 820         self.assertRegex(output, 'src')
 821         self.assertRegex(output, '2001:1234:56:8f63::2')
 822
 823     def test_ip_link_mac_address(self):
 824         self.copy_unit_to_networkd_unit_path('25-address-link-section.network', '12-dummy.netdev')
 825         self.start_networkd()
 826
 827         self.assertTrue(self.link_exits('dummy98'))
 828
 829         output = subprocess.check_output(['ip', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
 830         print(output)
 831         self.assertRegex(output, '00:01:02:aa:bb:cc')
 832
 833     def test_ip_link_unmanaged(self):
 834         self.copy_unit_to_networkd_unit_path('25-link-section-unmanaged.network', '12-dummy.netdev')
 835         self.start_networkd()
 836
 837         self.assertTrue(self.link_exits('dummy98'))
 838
 839         output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
 840         print(output)
 841         self.assertRegex(output, 'unmanaged')
 842
 843     def test_ipv6_address_label(self):
 844         self.copy_unit_to_networkd_unit_path('25-ipv6-address-label-section.network', '12-dummy.netdev')
 845         self.start_networkd()
 846
 847         self.assertTrue(self.link_exits('dummy98'))
 848
 849         output = subprocess.check_output(['ip', 'addrlabel', 'list']).rstrip().decode('utf-8')
 850         print(output)
 851         self.assertRegex(output, '2004:da8:1::/64')
 852
 853     def test_ipv6_neighbor(self):
 854         self.copy_unit_to_networkd_unit_path('25-neighbor-section.network', '12-dummy.netdev')
 855         self.start_networkd()
 856
 857         self.assertTrue(self.link_exits('dummy98'))
 858
 859         output = subprocess.check_output(['ip', 'neigh', 'list']).rstrip().decode('utf-8')
 860         print(output)
 861         self.assertRegex(output, '192.168.10.1.*00:00:5e:00:02:65.*PERMANENT')
 862         self.assertRegex(output, '2004:da8:1::1.*00:00:5e:00:02:66.*PERMANENT')
 863
 864     def test_sysctl(self):
 865         self.copy_unit_to_networkd_unit_path('25-sysctl.network', '12-dummy.netdev')
 866         self.start_networkd()
 867
 868         self.assertTrue(self.link_exits('dummy98'))
 869
 870         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'forwarding'), '1')
 871         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'use_tempaddr'), '2')
 872         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'dad_transmits'), '3')
 873         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'hop_limit'), '5')
 874         self.assertEqual(self.read_ipv6_sysctl_attr('dummy98', 'proxy_ndp'), '1')
 875         self.assertEqual(self.read_ipv4_sysctl_attr('dummy98', 'forwarding'),'1')
 876         self.assertEqual(self.read_ipv4_sysctl_attr('dummy98', 'proxy_arp'), '1')
 877
 878 class NetworkdNetWorkBridgeTests(unittest.TestCase, Utilities):
 879     links = [
 880         'bridge99',
 881         'dummy98',
 882         'test1']
 883
 884     units = [
 885         '11-dummy.netdev',
 886         '12-dummy.netdev',
 887         '26-bridge.netdev',
 888         '26-bridge-slave-interface-1.network',
 889         '26-bridge-slave-interface-2.network',
 890         'bridge99.network']
 891
 892     def setUp(self):
 893         self.link_remove(self.links)
 894
 895     def tearDown(self):
 896         self.link_remove(self.links)
 897         self.remove_unit_from_networkd_path(self.units)
 898
 899     def test_bridge_property(self):
 900         self.copy_unit_to_networkd_unit_path('11-dummy.netdev', '12-dummy.netdev', '26-bridge.netdev',
 901                                              '26-bridge-slave-interface-1.network', '26-bridge-slave-interface-2.network',
 902                                              'bridge99.network')
 903         self.start_networkd()
 904
 905         self.assertTrue(self.link_exits('dummy98'))
 906         self.assertTrue(self.link_exits('test1'))
 907         self.assertTrue(self.link_exits('bridge99'))
 908
 909         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'test1']).rstrip().decode('utf-8')
 910         print(output)
 911         self.assertRegex(output, 'master')
 912         self.assertRegex(output, 'bridge')
 913
 914         output = subprocess.check_output(['ip', '-d', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
 915         print(output)
 916         self.assertRegex(output, 'master')
 917         self.assertRegex(output, 'bridge')
 918
 919         output = subprocess.check_output(['ip', 'addr', 'show', 'bridge99']).rstrip().decode('utf-8')
 920         print(output)
 921         self.assertRegex(output, '192.168.0.15')
 922         self.assertRegex(output, '192.168.0.1')
 923
 924         output = subprocess.check_output(['bridge', '-d', 'link', 'show', 'dummy98']).rstrip().decode('utf-8')
 925         print(output)
 926         self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'hairpin_mode'), '1')
 927         self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'path_cost'), '400')
 928         self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'unicast_flood'), '1')
 929         self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'multicast_fast_leave'), '1')
 930
 931         # CONFIG_BRIDGE_IGMP_SNOOPING=y
 932         if (os.path.exists('/sys/devices/virtual/net/bridge00/lower_dummy98/brport/multicast_to_unicast')):
 933             self.assertEqual(self.read_bridge_port_attr('bridge99', 'dummy98', 'multicast_to_unicast'), '1')
 934
 935 class NetworkdNetWorkLLDPTests(unittest.TestCase, Utilities):
 936     links = ['veth99']
 937
 938     units = [
 939         '23-emit-lldp.network',
 940         '24-lldp.network',
 941         '25-veth.netdev']
 942
 943     def setUp(self):
 944         self.link_remove(self.links)
 945
 946     def tearDown(self):
 947         self.link_remove(self.links)
 948         self.remove_unit_from_networkd_path(self.units)
 949
 950     def test_lldp(self):
 951         self.copy_unit_to_networkd_unit_path('23-emit-lldp.network', '24-lldp.network', '25-veth.netdev')
 952         self.start_networkd()
 953
 954         self.assertTrue(self.link_exits('veth99'))
 955
 956         output = subprocess.check_output(['networkctl', 'lldp']).rstrip().decode('utf-8')
 957         print(output)
 958         self.assertRegex(output, 'veth-peer')
 959         self.assertRegex(output, 'veth99')
 960
 961 class NetworkdNetworkRATests(unittest.TestCase, Utilities):
 962     links = ['veth99']
 963
 964     units = [
 965         '25-veth.netdev',
 966         'ipv6-prefix.network',
 967         'ipv6-prefix-veth.network']
 968
 969     def setUp(self):
 970         self.link_remove(self.links)
 971
 972     def tearDown(self):
 973         self.link_remove(self.links)
 974         self.remove_unit_from_networkd_path(self.units)
 975
 976     def test_ipv6_prefix_delegation(self):
 977         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'ipv6-prefix.network', 'ipv6-prefix-veth.network')
 978         self.start_networkd()
 979
 980         self.assertTrue(self.link_exits('veth99'))
 981
 982         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
 983         print(output)
 984         self.assertRegex(output, '2002:da8:1:0')
 985
 986 class NetworkdNetworkDHCPServerTests(unittest.TestCase, Utilities):
 987     links = [
 988         'dummy98',
 989         'veth99']
 990
 991     units = [
 992         '12-dummy.netdev',
 993         '24-search-domain.network',
 994         '25-veth.netdev',
 995         'dhcp-client.network',
 996         'dhcp-client-timezone-router.network',
 997         'dhcp-server.network',
 998         'dhcp-server-timezone-router.network']
 999
1000     def setUp(self):
1001         self.link_remove(self.links)
1002
1003     def tearDown(self):
1004         self.link_remove(self.links)
1005         self.remove_unit_from_networkd_path(self.units)
1006
1007     def test_dhcp_server(self):
1008         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client.network', 'dhcp-server.network')
1009         self.start_networkd()
1010
1011         self.assertTrue(self.link_exits('veth99'))
1012
1013         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
1014         print(output)
1015         self.assertRegex(output, '192.168.5.*')
1016         self.assertRegex(output, 'Gateway: 192.168.5.1')
1017         self.assertRegex(output, 'DNS: 192.168.5.1')
1018         self.assertRegex(output, 'NTP: 192.168.5.1')
1019
1020     def test_domain(self):
1021         self.copy_unit_to_networkd_unit_path('12-dummy.netdev', '24-search-domain.network')
1022         self.start_networkd()
1023
1024         self.assertTrue(self.link_exits('dummy98'))
1025
1026         output = subprocess.check_output(['networkctl', 'status', 'dummy98']).rstrip().decode('utf-8')
1027         print(output)
1028         self.assertRegex(output, 'Address: 192.168.42.100')
1029         self.assertRegex(output, 'DNS: 192.168.42.1')
1030         self.assertRegex(output, 'Search Domains: one')
1031
1032     def test_emit_router_timezone(self):
1033         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-client-timezone-router.network', 'dhcp-server-timezone-router.network')
1034         self.start_networkd()
1035
1036         self.assertTrue(self.link_exits('veth99'))
1037
1038         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
1039         print(output)
1040         self.assertRegex(output, 'Gateway: 192.168.5.*')
1041         self.assertRegex(output, '192.168.5.*')
1042         self.assertRegex(output, 'Europe/Berlin')
1043
1044 class NetworkdNetworkDHCPClientTests(unittest.TestCase, Utilities):
1045     links = [
1046         'dummy98',
1047         'veth99']
1048
1049     units = [
1050         '25-veth.netdev',
1051         'dhcp-client-anonymize.network',
1052         'dhcp-client-critical-connection.network',
1053         'dhcp-client-ipv4-dhcp-settings.network',
1054         'dhcp-client-ipv4-only-ipv6-disabled.network',
1055         'dhcp-client-ipv4-only.network',
1056         'dhcp-client-ipv6-only.network',
1057         'dhcp-client-ipv6-rapid-commit.network',
1058         'dhcp-client-listen-port.network',
1059         'dhcp-client-route-metric.network',
1060         'dhcp-client-route-table.network',
1061         'dhcp-client.network',
1062         'dhcp-server-veth-peer.network',
1063         'dhcp-v4-server-veth-peer.network',
1064         'static.network']
1065
1066     def setUp(self):
1067         self.link_remove(self.links)
1068         self.stop_dnsmasq(dnsmasq_pid_file)
1069
1070     def tearDown(self):
1071         self.link_remove(self.links)
1072         self.remove_unit_from_networkd_path(self.units)
1073         self.stop_dnsmasq(dnsmasq_pid_file)
1074         self.remove_lease_file()
1075         self.remove_log_file()
1076
1077     def test_dhcp_client_ipv6_only(self):
1078         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network')
1079         self.start_networkd()
1080
1081         self.assertTrue(self.link_exits('veth99'))
1082
1083         self.start_dnsmasq()
1084
1085         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
1086         print(output)
1087         self.assertRegex(output, '2600::')
1088         self.assertNotRegex(output, '192.168.5')
1089
1090     def test_dhcp_client_ipv4_only(self):
1091         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-only-ipv6-disabled.network')
1092         self.start_networkd()
1093
1094         self.assertTrue(self.link_exits('veth99'))
1095
1096         self.start_dnsmasq()
1097
1098         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
1099         print(output)
1100         self.assertNotRegex(output, '2600::')
1101         self.assertRegex(output, '192.168.5')
1102
1103     def test_dhcp_client_ipv4_ipv6(self):
1104         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network',
1105                                              'dhcp-client-ipv4-only.network')
1106         self.start_networkd()
1107
1108         self.assertTrue(self.link_exits('veth99'))
1109
1110         self.start_dnsmasq()
1111
1112         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
1113         print(output)
1114         self.assertRegex(output, '2600::')
1115         self.assertRegex(output, '192.168.5')
1116
1117     def test_dhcp_client_settings(self):
1118         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv4-dhcp-settings.network')
1119         self.start_networkd()
1120
1121         self.assertTrue(self.link_exits('veth99'))
1122
1123         self.start_dnsmasq()
1124
1125         output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
1126         print(output)
1127         self.assertRegex(output, '12:34:56:78:9a:bc')
1128         self.assertRegex(output, '192.168.5')
1129         self.assertRegex(output, '1492')
1130
1131         output = subprocess.check_output(['ip', 'route']).rstrip().decode('utf-8')
1132         print(output)
1133         self.assertRegex(output, 'default.*dev veth99 proto dhcp')
1134
1135         self.assertTrue(self.search_words_in_dnsmasq_log('vendor class: SusantVendorTest', True))
1136         self.assertTrue(self.search_words_in_dnsmasq_log('DHCPDISCOVER(veth-peer) 12:34:56:78:9a:bc'))
1137         self.assertTrue(self.search_words_in_dnsmasq_log('client provides name: test-hostname'))
1138         self.assertTrue(self.search_words_in_dnsmasq_log('26:mtu'))
1139
1140     def test_dhcp6_client_settings_rapidcommit_true(self):
1141         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-only.network')
1142         self.start_networkd()
1143
1144         self.assertTrue(self.link_exits('veth99'))
1145
1146         self.start_dnsmasq()
1147
1148         output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
1149         print(output)
1150         self.assertRegex(output, '12:34:56:78:9a:bc')
1151         self.assertTrue(self.search_words_in_dnsmasq_log('14:rapid-commit', True))
1152
1153     def test_dhcp6_client_settings_rapidcommit_false(self):
1154         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-ipv6-rapid-commit.network')
1155         self.start_networkd()
1156
1157         self.assertTrue(self.link_exits('veth99'))
1158
1159         self.start_dnsmasq()
1160
1161         output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
1162         print(output)
1163         self.assertRegex(output, '12:34:56:78:9a:bc')
1164         self.assertFalse(self.search_words_in_dnsmasq_log('14:rapid-commit', True))
1165
1166     def test_dhcp_client_settings_anonymize(self):
1167         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-anonymize.network')
1168         self.start_networkd()
1169
1170         self.assertTrue(self.link_exits('veth99'))
1171
1172         self.start_dnsmasq()
1173
1174         self.assertFalse(self.search_words_in_dnsmasq_log('VendorClassIdentifier=SusantVendorTest', True))
1175         self.assertFalse(self.search_words_in_dnsmasq_log('test-hostname'))
1176         self.assertFalse(self.search_words_in_dnsmasq_log('26:mtu'))
1177
1178     def test_dhcp_client_listen_port(self):
1179         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client-listen-port.network')
1180         self.start_networkd()
1181
1182         self.assertTrue(self.link_exits('veth99'))
1183
1184         self.start_dnsmasq('--dhcp-alternate-port=67,5555')
1185
1186         output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
1187         print(output)
1188         self.assertRegex(output, '192.168.5.* dynamic')
1189
1190     def test_dhcp_route_table_id(self):
1191         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-table.network')
1192         self.start_networkd()
1193
1194         self.assertTrue(self.link_exits('veth99'))
1195
1196         self.start_dnsmasq()
1197
1198         output = subprocess.check_output(['ip', 'route', 'show', 'table', '12']).rstrip().decode('utf-8')
1199         print(output)
1200         self.assertRegex(output, 'veth99 proto dhcp')
1201         self.assertRegex(output, '192.168.5.1')
1202
1203     def test_dhcp_route_metric(self):
1204         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-route-metric.network')
1205         self.start_networkd()
1206
1207         self.assertTrue(self.link_exits('veth99'))
1208
1209         self.start_dnsmasq()
1210
1211         output = subprocess.check_output(['ip', 'route', 'show', 'dev', 'veth99']).rstrip().decode('utf-8')
1212         print(output)
1213         self.assertRegex(output, 'metric 24')
1214
1215     def test_dhcp_route_criticalconnection_true(self):
1216         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-v4-server-veth-peer.network', 'dhcp-client-critical-connection.network')
1217         self.start_networkd()
1218
1219         self.assertTrue(self.link_exits('veth99'))
1220
1221         self.start_dnsmasq()
1222
1223         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
1224         print(output)
1225         self.assertRegex(output, '192.168.5.*')
1226
1227         # Stoping dnsmasq as networkd won't be allowed to renew the DHCP lease.
1228         self.stop_dnsmasq(dnsmasq_pid_file)
1229
1230         # Sleep for 120 sec as the dnsmasq minimum lease time can only be set to 120
1231         time.sleep(125)
1232
1233         output = subprocess.check_output(['networkctl', 'status', 'veth99']).rstrip().decode('utf-8')
1234         print(output)
1235         self.assertRegex(output, '192.168.5.*')
1236
1237     def test_dhcp_client_reuse_address_as_static(self):
1238         self.copy_unit_to_networkd_unit_path('25-veth.netdev', 'dhcp-server-veth-peer.network', 'dhcp-client.network')
1239         self.start_networkd()
1240
1241         self.assertTrue(self.link_exits('veth99'))
1242
1243         self.start_dnsmasq()
1244
1245         output = subprocess.check_output(['ip', 'address', 'show', 'dev', 'veth99', 'scope', 'global']).rstrip().decode('utf-8')
1246         print(output)
1247         self.assertRegex(output, '192.168.5')
1248         self.assertRegex(output, '2600::')
1249
1250         ipv4_address = re.search('192\.168\.5\.[0-9]*/24', output)
1251         ipv6_address = re.search('2600::[0-9a-f:]*/128', output)
1252         static_network = '\n'.join(['[Match]', 'Name=veth99', '[Network]', 'IPv6AcceptRA=no', 'Address=' + ipv4_address.group(), 'Address=' + ipv6_address.group()])
1253         print(static_network)
1254
1255         self.remove_unit_from_networkd_path(['dhcp-client.network'])
1256
1257         with open(os.path.join(network_unit_file_path, 'static.network'), mode='w') as f:
1258             f.write(static_network)
1259
1260         self.start_networkd()
1261
1262         self.assertTrue(self.link_exits('veth99'))
1263
1264         output = subprocess.check_output(['ip', '-4', 'address', 'show', 'dev', 'veth99', 'scope', 'global']).rstrip().decode('utf-8')
1265         print(output)
1266         self.assertRegex(output, '192.168.5')
1267         self.assertRegex(output, 'valid_lft forever preferred_lft forever')
1268
1269         output = subprocess.check_output(['ip', '-6', 'address', 'show', 'dev', 'veth99', 'scope', 'global']).rstrip().decode('utf-8')
1270         print(output)
1271         self.assertRegex(output, '2600::')
1272         self.assertRegex(output, 'valid_lft forever preferred_lft forever')
1273
1274 if __name__ == '__main__':
1275     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
1276                                                      verbosity=3))