test/networkd-test.py

   1 #!/usr/bin/env python3
   2 #
   3 # networkd integration test
   4 # This uses temporary configuration in /run and temporary veth devices, and
   5 # does not write anything on disk or change any system configuration;
   6 # but it assumes (and checks at the beginning) that networkd is not currently
   7 # running.
   8 #
   9 # This can be run on a normal installation, in QEMU, nspawn (with
  10 # --private-network), LXD (with "--config raw.lxc=lxc.aa_profile=unconfined"),
  11 # or LXC system containers. You need at least the "ip" tool from the iproute
  12 # package; it is recommended to install dnsmasq too to get full test coverage.
  13 #
  14 # ATTENTION: This uses the *installed* networkd, not the one from the built
  15 # source tree.
  16 #
  17 # (C) 2015 Canonical Ltd.
  18 # Author: Martin Pitt <martin.pitt@ubuntu.com>
  19 #
  20 # systemd is free software; you can redistribute it and/or modify it
  21 # under the terms of the GNU Lesser General Public License as published by
  22 # the Free Software Foundation; either version 2.1 of the License, or
  23 # (at your option) any later version.
  24
  25 # systemd is distributed in the hope that it will be useful, but
  26 # WITHOUT ANY WARRANTY; without even the implied warranty of
  27 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  28 # Lesser General Public License for more details.
  29 #
  30 # You should have received a copy of the GNU Lesser General Public License
  31 # along with systemd; If not, see <http://www.gnu.org/licenses/>.
  32
  33 import os
  34 import sys
  35 import time
  36 import unittest
  37 import tempfile
  38 import subprocess
  39 import shutil
  40 import socket
  41
  42 networkd_active = subprocess.call(['systemctl', 'is-active', '--quiet',
  43                                    'systemd-networkd']) == 0
  44 have_dnsmasq = shutil.which('dnsmasq')
  45
  46 RESOLV_CONF = '/run/systemd/resolve/resolv.conf'
  47
  48
  49 @unittest.skipIf(networkd_active,
  50                  'networkd is already active')
  51 class ClientTestBase:
  52     def setUp(self):
  53         self.iface = 'test_eth42'
  54         self.if_router = 'router_eth42'
  55         self.workdir_obj = tempfile.TemporaryDirectory()
  56         self.workdir = self.workdir_obj.name
  57         self.config = '/run/systemd/network/test_eth42.network'
  58
  59         # avoid "Failed to open /dev/tty" errors in containers
  60         os.environ['SYSTEMD_LOG_TARGET'] = 'journal'
  61
  62         # determine path to systemd-networkd-wait-online
  63         for p in ['/usr/lib/systemd/systemd-networkd-wait-online',
  64                   '/lib/systemd/systemd-networkd-wait-online']:
  65             if os.path.exists(p):
  66                 self.networkd_wait_online = p
  67                 break
  68         else:
  69             self.fail('systemd-networkd-wait-online not found')
  70
  71         # get current journal cursor
  72         out = subprocess.check_output(['journalctl', '-b', '--quiet',
  73                                        '--no-pager', '-n0', '--show-cursor'],
  74                                       universal_newlines=True)
  75         self.assertTrue(out.startswith('-- cursor:'))
  76         self.journal_cursor = out.split()[-1]
  77
  78     def tearDown(self):
  79         self.shutdown_iface()
  80         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
  81         subprocess.call(['ip', 'link', 'del', 'dummy0'],
  82                         stderr=subprocess.DEVNULL)
  83
  84     def writeConfig(self, fname, contents):
  85         os.makedirs(os.path.dirname(fname), exist_ok=True)
  86         with open(fname, 'w') as f:
  87             f.write(contents)
  88         self.addCleanup(os.remove, fname)
  89
  90     def show_journal(self, unit):
  91         '''Show journal of given unit since start of the test'''
  92
  93         print('---- %s ----' % unit)
  94         sys.stdout.flush()
  95         subprocess.call(['journalctl', '-b', '--no-pager', '--quiet',
  96                          '--cursor', self.journal_cursor, '-u', unit])
  97
  98     def create_iface(self, ipv6=False):
  99         '''Create test interface with DHCP server behind it'''
 100
 101         raise NotImplementedError('must be implemented by a subclass')
 102
 103     def shutdown_iface(self):
 104         '''Remove test interface and stop DHCP server'''
 105
 106         raise NotImplementedError('must be implemented by a subclass')
 107
 108     def print_server_log(self):
 109         '''Print DHCP server log for debugging failures'''
 110
 111         raise NotImplementedError('must be implemented by a subclass')
 112
 113     def do_test(self, coldplug=True, ipv6=False, extra_opts='',
 114                 online_timeout=10, dhcp_mode='yes'):
 115         subprocess.check_call(['systemctl', 'start', 'systemd-resolved'])
 116         self.writeConfig(self.config, '''\
 117 [Match]
 118 Name=%s
 119 [Network]
 120 DHCP=%s
 121 %s''' % (self.iface, dhcp_mode, extra_opts))
 122
 123         if coldplug:
 124             # create interface first, then start networkd
 125             self.create_iface(ipv6=ipv6)
 126             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 127         elif coldplug is not None:
 128             # start networkd first, then create interface
 129             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 130             self.create_iface(ipv6=ipv6)
 131         else:
 132             # "None" means test sets up interface by itself
 133             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 134
 135         try:
 136             subprocess.check_call([self.networkd_wait_online, '--interface',
 137                                    self.iface, '--timeout=%i' % online_timeout])
 138
 139             if ipv6:
 140                 # check iface state and IP 6 address; FIXME: we need to wait a bit
 141                 # longer, as the iface is "configured" already with IPv4 *or*
 142                 # IPv6, but we want to wait for both
 143                 for timeout in range(10):
 144                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
 145                     if b'state UP' in out and b'inet6 2600' in out and b'inet 192.168' in out:
 146                         break
 147                     time.sleep(1)
 148                 else:
 149                     self.fail('timed out waiting for IPv6 configuration')
 150
 151                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
 152                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 153             else:
 154                 # should have link-local address on IPv6 only
 155                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
 156                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 157                 self.assertNotIn(b'scope global', out)
 158
 159             # should have IPv4 address
 160             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 161             self.assertIn(b'state UP', out)
 162             self.assertRegex(out, b'inet 192.168.5.\d+/.* scope global dynamic')
 163
 164             # check networkctl state
 165             out = subprocess.check_output(['networkctl'])
 166             self.assertRegex(out, ('%s\s+ether\s+routable\s+unmanaged' % self.if_router).encode())
 167             self.assertRegex(out, ('%s\s+ether\s+routable\s+configured' % self.iface).encode())
 168
 169             out = subprocess.check_output(['networkctl', 'status', self.iface])
 170             self.assertRegex(out, b'Type:\s+ether')
 171             self.assertRegex(out, b'State:\s+routable.*configured')
 172             self.assertRegex(out, b'Address:\s+192.168.5.\d+')
 173             if ipv6:
 174                 self.assertRegex(out, b'2600::')
 175             else:
 176                 self.assertNotIn(b'2600::', out)
 177             self.assertRegex(out, b'fe80::')
 178             self.assertRegex(out, b'Gateway:\s+192.168.5.1')
 179             self.assertRegex(out, b'DNS:\s+192.168.5.1')
 180         except (AssertionError, subprocess.CalledProcessError):
 181             # show networkd status, journal, and DHCP server log on failure
 182             with open(self.config) as f:
 183                 print('\n---- %s ----\n%s' % (self.config, f.read()))
 184             print('---- interface status ----')
 185             sys.stdout.flush()
 186             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
 187             print('---- networkctl status %s ----' % self.iface)
 188             sys.stdout.flush()
 189             subprocess.call(['networkctl', 'status', self.iface])
 190             self.show_journal('systemd-networkd.service')
 191             self.print_server_log()
 192             raise
 193
 194         for timeout in range(50):
 195             with open(RESOLV_CONF) as f:
 196                 contents = f.read()
 197             if 'nameserver 192.168.5.1\n' in contents:
 198                 break
 199             time.sleep(0.1)
 200         else:
 201             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
 202
 203         if coldplug is False:
 204             # check post-down.d hook
 205             self.shutdown_iface()
 206
 207     def test_coldplug_dhcp_yes_ip4(self):
 208         # we have a 12s timeout on RA, so we need to wait longer
 209         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
 210
 211     def test_coldplug_dhcp_yes_ip4_no_ra(self):
 212         # with disabling RA explicitly things should be fast
 213         self.do_test(coldplug=True, ipv6=False,
 214                      extra_opts='IPv6AcceptRA=False')
 215
 216     def test_coldplug_dhcp_ip4_only(self):
 217         # we have a 12s timeout on RA, so we need to wait longer
 218         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 219                      online_timeout=15)
 220
 221     def test_coldplug_dhcp_ip4_only_no_ra(self):
 222         # with disabling RA explicitly things should be fast
 223         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 224                      extra_opts='IPv6AcceptRA=False')
 225
 226     def test_coldplug_dhcp_ip6(self):
 227         self.do_test(coldplug=True, ipv6=True)
 228
 229     def test_hotplug_dhcp_ip4(self):
 230         # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
 231         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
 232
 233     def test_hotplug_dhcp_ip6(self):
 234         self.do_test(coldplug=False, ipv6=True)
 235
 236     def test_route_only_dns(self):
 237         self.writeConfig('/run/systemd/network/myvpn.netdev', '''\
 238 [NetDev]
 239 Name=dummy0
 240 Kind=dummy
 241 MACAddress=12:34:56:78:9a:bc''')
 242         self.writeConfig('/run/systemd/network/myvpn.network', '''\
 243 [Match]
 244 Name=dummy0
 245 [Network]
 246 Address=192.168.42.100
 247 DNS=192.168.42.1
 248 Domains= ~company''')
 249
 250         self.do_test(coldplug=True, ipv6=False,
 251                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 252
 253         with open(RESOLV_CONF) as f:
 254             contents = f.read()
 255             # ~company is not a search domain, only a routing domain
 256             self.assertNotRegex(contents, 'search.*company')
 257             # our global server should appear
 258             self.assertIn('nameserver 192.168.5.1\n', contents)
 259             # should not have domain-restricted server as global server
 260             self.assertNotIn('nameserver 192.168.42.1\n', contents)
 261
 262     def test_route_only_dns_all_domains(self):
 263         with open('/run/systemd/network/myvpn.netdev', 'w') as f:
 264             f.write('''[NetDev]
 265 Name=dummy0
 266 Kind=dummy
 267 MACAddress=12:34:56:78:9a:bc''')
 268         with open('/run/systemd/network/myvpn.network', 'w') as f:
 269             f.write('''[Match]
 270 Name=dummy0
 271 [Network]
 272 Address=192.168.42.100
 273 DNS=192.168.42.1
 274 Domains= ~company ~.''')
 275         self.addCleanup(os.remove, '/run/systemd/network/myvpn.netdev')
 276         self.addCleanup(os.remove, '/run/systemd/network/myvpn.network')
 277
 278         self.do_test(coldplug=True, ipv6=False,
 279                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 280
 281         with open(RESOLV_CONF) as f:
 282             contents = f.read()
 283
 284         # ~company is not a search domain, only a routing domain
 285         self.assertNotRegex(contents, 'search.*company')
 286
 287         # our global server should appear
 288         self.assertIn('nameserver 192.168.5.1\n', contents)
 289         # should have company server as global server due to ~.
 290         self.assertIn('nameserver 192.168.42.1\n', contents)
 291
 292
 293 @unittest.skipUnless(have_dnsmasq, 'dnsmasq not installed')
 294 class DnsmasqClientTest(ClientTestBase, unittest.TestCase):
 295     '''Test networkd client against dnsmasq'''
 296
 297     def setUp(self):
 298         super().setUp()
 299         self.dnsmasq = None
 300         self.iface_mac = 'de:ad:be:ef:47:11'
 301
 302     def create_iface(self, ipv6=False, dnsmasq_opts=None):
 303         '''Create test interface with DHCP server behind it'''
 304
 305         # add veth pair
 306         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface,
 307                                'address', self.iface_mac,
 308                                'type', 'veth', 'peer', 'name', self.if_router])
 309
 310         # give our router an IP
 311         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
 312         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
 313         if ipv6:
 314             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
 315         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
 316
 317         # add DHCP server
 318         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
 319         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
 320         if ipv6:
 321             extra_opts = ['--enable-ra', '--dhcp-range=2600::10,2600::20']
 322         else:
 323             extra_opts = []
 324         if dnsmasq_opts:
 325             extra_opts += dnsmasq_opts
 326         self.dnsmasq = subprocess.Popen(
 327             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 328              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
 329              '--dhcp-leasefile=' + lease_file, '--bind-interfaces',
 330              '--interface=' + self.if_router, '--except-interface=lo',
 331              '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts)
 332
 333     def shutdown_iface(self):
 334         '''Remove test interface and stop DHCP server'''
 335
 336         if self.if_router:
 337             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
 338             self.if_router = None
 339         if self.dnsmasq:
 340             self.dnsmasq.kill()
 341             self.dnsmasq.wait()
 342             self.dnsmasq = None
 343
 344     def print_server_log(self):
 345         '''Print DHCP server log for debugging failures'''
 346
 347         with open(self.dnsmasq_log) as f:
 348             sys.stdout.write('\n\n---- dnsmasq log ----\n%s\n------\n\n' % f.read())
 349
 350     def test_resolved_domain_restricted_dns(self):
 351         '''resolved: domain-restricted DNS servers'''
 352
 353         # create interface for generic connections; this will map all DNS names
 354         # to 192.168.42.1
 355         self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1'])
 356         self.writeConfig('/run/systemd/network/general.network', '''\
 357 [Match]
 358 Name=%s
 359 [Network]
 360 DHCP=ipv4
 361 IPv6AcceptRA=False''' % self.iface)
 362
 363         # create second device/dnsmasq for a .company/.lab VPN interface
 364         # static IPs for simplicity
 365         subprocess.check_call(['ip', 'link', 'add', 'name', 'testvpnclient', 'type',
 366                                'veth', 'peer', 'name', 'testvpnrouter'])
 367         self.addCleanup(subprocess.call, ['ip', 'link', 'del', 'dev', 'testvpnrouter'])
 368         subprocess.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
 369         subprocess.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
 370         subprocess.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
 371
 372         vpn_dnsmasq_log = os.path.join(self.workdir, 'dnsmasq-vpn.log')
 373         vpn_dnsmasq = subprocess.Popen(
 374             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 375              '--log-facility=' + vpn_dnsmasq_log, '--conf-file=/dev/null',
 376              '--dhcp-leasefile=/dev/null', '--bind-interfaces',
 377              '--interface=testvpnrouter', '--except-interface=lo',
 378              '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
 379         self.addCleanup(vpn_dnsmasq.wait)
 380         self.addCleanup(vpn_dnsmasq.kill)
 381
 382         self.writeConfig('/run/systemd/network/vpn.network', '''\
 383 [Match]
 384 Name=testvpnclient
 385 [Network]
 386 IPv6AcceptRA=False
 387 Address=10.241.3.2/24
 388 DNS=10.241.3.1
 389 Domains= ~company ~lab''')
 390
 391         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 392         subprocess.check_call([self.networkd_wait_online, '--interface', self.iface,
 393                                '--interface=testvpnclient', '--timeout=20'])
 394
 395         # ensure we start fresh with every test
 396         subprocess.check_call(['systemctl', 'restart', 'systemd-resolved'])
 397
 398         # test vpnclient specific domains; these should *not* be answered by
 399         # the general DNS
 400         out = subprocess.check_output(['systemd-resolve', 'math.lab'])
 401         self.assertIn(b'math.lab: 10.241.3.3', out)
 402         out = subprocess.check_output(['systemd-resolve', 'kettle.cantina.company'])
 403         self.assertIn(b'kettle.cantina.company: 10.241.4.4', out)
 404
 405         # test general domains
 406         out = subprocess.check_output(['systemd-resolve', 'megasearch.net'])
 407         self.assertIn(b'megasearch.net: 192.168.42.1', out)
 408
 409         with open(self.dnsmasq_log) as f:
 410             general_log = f.read()
 411         with open(vpn_dnsmasq_log) as f:
 412             vpn_log = f.read()
 413
 414         # VPN domains should only be sent to VPN DNS
 415         self.assertRegex(vpn_log, 'query.*math.lab')
 416         self.assertRegex(vpn_log, 'query.*cantina.company')
 417         self.assertNotIn('lab', general_log)
 418         self.assertNotIn('company', general_log)
 419
 420         # general domains should not be sent to the VPN DNS
 421         self.assertRegex(general_log, 'query.*megasearch.net')
 422         self.assertNotIn('megasearch.net', vpn_log)
 423
 424     def test_transient_hostname(self):
 425         '''networkd sets transient hostname from DHCP'''
 426
 427         orig_hostname = socket.gethostname()
 428         self.addCleanup(socket.sethostname, orig_hostname)
 429         # temporarily move /etc/hostname away; restart hostnamed to pick it up
 430         if os.path.exists('/etc/hostname'):
 431             subprocess.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
 432             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
 433         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 434
 435         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 436         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 437
 438         # should have received the fixed IP above
 439         out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 440         self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 441         # should have set transient hostname in hostnamed
 442         self.assertIn(b'testgreen', subprocess.check_output(['hostnamectl']))
 443         # and also applied to the system
 444         self.assertEqual(socket.gethostname(), 'testgreen')
 445
 446     def test_transient_hostname_with_static(self):
 447         '''transient hostname is not applied if static hostname exists'''
 448
 449         orig_hostname = socket.gethostname()
 450         self.addCleanup(socket.sethostname, orig_hostname)
 451         if not os.path.exists('/etc/hostname'):
 452             self.writeConfig('/etc/hostname', orig_hostname)
 453         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 454
 455         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 456         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 457
 458         # should have received the fixed IP above
 459         out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 460         self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 461         # static hostname wins over transient one, thus *not* applied
 462         self.assertEqual(socket.gethostname(), orig_hostname)
 463
 464
 465 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 466     '''Test networkd client against networkd server'''
 467
 468     def setUp(self):
 469         super().setUp()
 470         self.dnsmasq = None
 471
 472     def create_iface(self, ipv6=False, dhcpserver_opts=None):
 473         '''Create test interface with DHCP server behind it'''
 474
 475         # run "router-side" networkd in own mount namespace to shield it from
 476         # "client-side" configuration and networkd
 477         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 478         self.addCleanup(os.remove, script)
 479         with os.fdopen(fd, 'w+') as f:
 480             f.write('''\
 481 #!/bin/sh -eu
 482 mkdir -p /run/systemd/network
 483 mkdir -p /run/systemd/netif
 484 mount -t tmpfs none /run/systemd/network
 485 mount -t tmpfs none /run/systemd/netif
 486 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 487 # create router/client veth pair
 488 cat << EOF > /run/systemd/network/test.netdev
 489 [NetDev]
 490 Name=%(ifr)s
 491 Kind=veth
 492
 493 [Peer]
 494 Name=%(ifc)s
 495 EOF
 496
 497 cat << EOF > /run/systemd/network/test.network
 498 [Match]
 499 Name=%(ifr)s
 500
 501 [Network]
 502 Address=192.168.5.1/24
 503 %(addr6)s
 504 DHCPServer=yes
 505
 506 [DHCPServer]
 507 PoolOffset=10
 508 PoolSize=50
 509 DNS=192.168.5.1
 510 %(dhopts)s
 511 EOF
 512
 513 # run networkd as in systemd-networkd.service
 514 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; p}')
 515 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or '',
 516        'dhopts': dhcpserver_opts or ''})
 517
 518             os.fchmod(fd, 0o755)
 519
 520         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 521                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 522                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 523                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 524                                '--service-type=notify', script])
 525
 526         # wait until devices got created
 527         for timeout in range(50):
 528             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 529             if b'state UP' in out and b'scope global' in out:
 530                 break
 531             time.sleep(0.1)
 532
 533     def shutdown_iface(self):
 534         '''Remove test interface and stop DHCP server'''
 535
 536         if self.if_router:
 537             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 538             # ensure failed transient unit does not stay around
 539             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 540             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 541             self.if_router = None
 542
 543     def print_server_log(self):
 544         '''Print DHCP server log for debugging failures'''
 545
 546         self.show_journal('networkd-test-router.service')
 547
 548     @unittest.skip('networkd does not have DHCPv6 server support')
 549     def test_hotplug_dhcp_ip6(self):
 550         pass
 551
 552     @unittest.skip('networkd does not have DHCPv6 server support')
 553     def test_coldplug_dhcp_ip6(self):
 554         pass
 555
 556     def test_search_domains(self):
 557
 558         # we don't use this interface for this test
 559         self.if_router = None
 560
 561         self.writeConfig('/run/systemd/network/test.netdev', '''\
 562 [NetDev]
 563 Name=dummy0
 564 Kind=dummy
 565 MACAddress=12:34:56:78:9a:bc''')
 566         self.writeConfig('/run/systemd/network/test.network', '''\
 567 [Match]
 568 Name=dummy0
 569 [Network]
 570 Address=192.168.42.100
 571 DNS=192.168.42.1
 572 Domains= one two three four five six seven eight nine ten''')
 573
 574         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 575
 576         for timeout in range(50):
 577             with open(RESOLV_CONF) as f:
 578                 contents = f.read()
 579             if ' one' in contents:
 580                 break
 581             time.sleep(0.1)
 582         self.assertRegex(contents, 'search .*one two three four')
 583         self.assertNotIn('seven\n', contents)
 584         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 585
 586     def test_search_domains_too_long(self):
 587
 588         # we don't use this interface for this test
 589         self.if_router = None
 590
 591         name_prefix = 'a' * 60
 592
 593         self.writeConfig('/run/systemd/network/test.netdev', '''\
 594 [NetDev]
 595 Name=dummy0
 596 Kind=dummy
 597 MACAddress=12:34:56:78:9a:bc''')
 598         self.writeConfig('/run/systemd/network/test.network', '''\
 599 [Match]
 600 Name=dummy0
 601 [Network]
 602 Address=192.168.42.100
 603 DNS=192.168.42.1
 604 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 605
 606         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 607
 608         for timeout in range(50):
 609             with open(RESOLV_CONF) as f:
 610                 contents = f.read()
 611             if ' one' in contents:
 612                 break
 613             time.sleep(0.1)
 614         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 615         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 616
 617     def test_dropin(self):
 618         # we don't use this interface for this test
 619         self.if_router = None
 620
 621         self.writeConfig('/run/systemd/network/test.netdev', '''\
 622 [NetDev]
 623 Name=dummy0
 624 Kind=dummy
 625 MACAddress=12:34:56:78:9a:bc''')
 626         self.writeConfig('/run/systemd/network/test.network', '''\
 627 [Match]
 628 Name=dummy0
 629 [Network]
 630 Address=192.168.42.100
 631 DNS=192.168.42.1''')
 632         self.writeConfig('/run/systemd/network/test.network.d/dns.conf', '''\
 633 [Network]
 634 DNS=127.0.0.1''')
 635
 636         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 637
 638         for timeout in range(50):
 639             with open(RESOLV_CONF) as f:
 640                 contents = f.read()
 641             if ' 127.0.0.1' in contents:
 642                 break
 643             time.sleep(0.1)
 644         self.assertIn('nameserver 192.168.42.1\n', contents)
 645         self.assertIn('nameserver 127.0.0.1\n', contents)
 646
 647     def test_dhcp_timezone(self):
 648         '''networkd sets time zone from DHCP'''
 649
 650         def get_tz():
 651             out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
 652                                            '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
 653             assert out.startswith(b's "')
 654             out = out.strip()
 655             assert out.endswith(b'"')
 656             return out[3:-1].decode()
 657
 658         orig_timezone = get_tz()
 659         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
 660
 661         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
 662         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='ipv4')
 663
 664         # should have applied the received timezone
 665         try:
 666             self.assertEqual(get_tz(), 'Pacific/Honolulu')
 667         except AssertionError:
 668             self.show_journal('systemd-networkd.service')
 669             self.show_journal('systemd-hostnamed.service')
 670             raise
 671
 672
 673 if __name__ == '__main__':
 674     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
 675                                                      verbosity=2))