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
  41 networkd_active = subprocess.call(['systemctl', 'is-active', '--quiet',
  42                                    'systemd-networkd']) == 0
  43 have_dnsmasq = shutil.which('dnsmasq')
  44
  45 RESOLV_CONF = '/run/systemd/resolve/resolv.conf'
  46
  47
  48 @unittest.skipIf(networkd_active,
  49                  'networkd is already active')
  50 class ClientTestBase:
  51     def setUp(self):
  52         self.iface = 'test_eth42'
  53         self.if_router = 'router_eth42'
  54         self.workdir_obj = tempfile.TemporaryDirectory()
  55         self.workdir = self.workdir_obj.name
  56         self.config = '/run/systemd/network/test_eth42.network'
  57
  58         # avoid "Failed to open /dev/tty" errors in containers
  59         os.environ['SYSTEMD_LOG_TARGET'] = 'journal'
  60
  61         # determine path to systemd-networkd-wait-online
  62         for p in ['/usr/lib/systemd/systemd-networkd-wait-online',
  63                   '/lib/systemd/systemd-networkd-wait-online']:
  64             if os.path.exists(p):
  65                 self.networkd_wait_online = p
  66                 break
  67         else:
  68             self.fail('systemd-networkd-wait-online not found')
  69
  70         # get current journal cursor
  71         out = subprocess.check_output(['journalctl', '-b', '--quiet',
  72                                        '--no-pager', '-n0', '--show-cursor'],
  73                                       universal_newlines=True)
  74         self.assertTrue(out.startswith('-- cursor:'))
  75         self.journal_cursor = out.split()[-1]
  76
  77     def tearDown(self):
  78         self.shutdown_iface()
  79         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
  80
  81     def writeConfig(self, fname, contents):
  82         os.makedirs(os.path.dirname(fname), exist_ok=True)
  83         with open(fname, 'w') as f:
  84             f.write(contents)
  85         self.addCleanup(os.remove, fname)
  86
  87     def show_journal(self, unit):
  88         '''Show journal of given unit since start of the test'''
  89
  90         print('---- %s ----' % unit)
  91         sys.stdout.flush()
  92         subprocess.call(['journalctl', '-b', '--no-pager', '--quiet',
  93                          '--cursor', self.journal_cursor, '-u', unit])
  94
  95     def create_iface(self, ipv6=False):
  96         '''Create test interface with DHCP server behind it'''
  97
  98         raise NotImplementedError('must be implemented by a subclass')
  99
 100     def shutdown_iface(self):
 101         '''Remove test interface and stop DHCP server'''
 102
 103         raise NotImplementedError('must be implemented by a subclass')
 104
 105     def print_server_log(self):
 106         '''Print DHCP server log for debugging failures'''
 107
 108         raise NotImplementedError('must be implemented by a subclass')
 109
 110     def do_test(self, coldplug=True, ipv6=False, extra_opts='',
 111                 online_timeout=10, dhcp_mode='yes'):
 112         subprocess.check_call(['systemctl', 'start', 'systemd-resolved'])
 113         self.writeConfig(self.config, '''\
 114 [Match]
 115 Name=%s
 116 [Network]
 117 DHCP=%s
 118 %s''' % (self.iface, dhcp_mode, extra_opts))
 119
 120         if coldplug:
 121             # create interface first, then start networkd
 122             self.create_iface(ipv6=ipv6)
 123             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 124         else:
 125             # start networkd first, then create interface
 126             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 127             self.create_iface(ipv6=ipv6)
 128
 129         try:
 130             subprocess.check_call([self.networkd_wait_online, '--interface',
 131                                    self.iface, '--timeout=%i' % online_timeout])
 132
 133             if ipv6:
 134                 # check iface state and IP 6 address; FIXME: we need to wait a bit
 135                 # longer, as the iface is "configured" already with IPv4 *or*
 136                 # IPv6, but we want to wait for both
 137                 for timeout in range(10):
 138                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
 139                     if b'state UP' in out and b'inet6 2600' in out and b'inet 192.168' in out:
 140                         break
 141                     time.sleep(1)
 142                 else:
 143                     self.fail('timed out waiting for IPv6 configuration')
 144
 145                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
 146                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 147             else:
 148                 # should have link-local address on IPv6 only
 149                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
 150                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 151                 self.assertNotIn(b'scope global', out)
 152
 153             # should have IPv4 address
 154             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 155             self.assertIn(b'state UP', out)
 156             self.assertRegex(out, b'inet 192.168.5.\d+/.* scope global dynamic')
 157
 158             # check networkctl state
 159             out = subprocess.check_output(['networkctl'])
 160             self.assertRegex(out, ('%s\s+ether\s+routable\s+unmanaged' % self.if_router).encode())
 161             self.assertRegex(out, ('%s\s+ether\s+routable\s+configured' % self.iface).encode())
 162
 163             out = subprocess.check_output(['networkctl', 'status', self.iface])
 164             self.assertRegex(out, b'Type:\s+ether')
 165             self.assertRegex(out, b'State:\s+routable.*configured')
 166             self.assertRegex(out, b'Address:\s+192.168.5.\d+')
 167             if ipv6:
 168                 self.assertRegex(out, b'2600::')
 169             else:
 170                 self.assertNotIn(b'2600::', out)
 171             self.assertRegex(out, b'fe80::')
 172             self.assertRegex(out, b'Gateway:\s+192.168.5.1')
 173             self.assertRegex(out, b'DNS:\s+192.168.5.1')
 174         except (AssertionError, subprocess.CalledProcessError):
 175             # show networkd status, journal, and DHCP server log on failure
 176             with open(self.config) as f:
 177                 print('\n---- %s ----\n%s' % (self.config, f.read()))
 178             print('---- interface status ----')
 179             sys.stdout.flush()
 180             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
 181             print('---- networkctl status %s ----' % self.iface)
 182             sys.stdout.flush()
 183             subprocess.call(['networkctl', 'status', self.iface])
 184             self.show_journal('systemd-networkd.service')
 185             self.print_server_log()
 186             raise
 187
 188         for timeout in range(50):
 189             with open(RESOLV_CONF) as f:
 190                 contents = f.read()
 191             if 'nameserver 192.168.5.1\n' in contents:
 192                 break
 193             time.sleep(0.1)
 194         else:
 195             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
 196
 197         if not coldplug:
 198             # check post-down.d hook
 199             self.shutdown_iface()
 200
 201     def test_coldplug_dhcp_yes_ip4(self):
 202         # we have a 12s timeout on RA, so we need to wait longer
 203         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
 204
 205     def test_coldplug_dhcp_yes_ip4_no_ra(self):
 206         # with disabling RA explicitly things should be fast
 207         self.do_test(coldplug=True, ipv6=False,
 208                      extra_opts='IPv6AcceptRA=False')
 209
 210     def test_coldplug_dhcp_ip4_only(self):
 211         # we have a 12s timeout on RA, so we need to wait longer
 212         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 213                      online_timeout=15)
 214
 215     def test_coldplug_dhcp_ip4_only_no_ra(self):
 216         # with disabling RA explicitly things should be fast
 217         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 218                      extra_opts='IPv6AcceptRA=False')
 219
 220     def test_coldplug_dhcp_ip6(self):
 221         self.do_test(coldplug=True, ipv6=True)
 222
 223     def test_hotplug_dhcp_ip4(self):
 224         # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
 225         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
 226
 227     def test_hotplug_dhcp_ip6(self):
 228         self.do_test(coldplug=False, ipv6=True)
 229
 230     def test_route_only_dns(self):
 231         self.writeConfig('/run/systemd/network/myvpn.netdev', '''\
 232 [NetDev]
 233 Name=dummy0
 234 Kind=dummy
 235 MACAddress=12:34:56:78:9a:bc''')
 236         self.writeConfig('/run/systemd/network/myvpn.network', '''\
 237 [Match]
 238 Name=dummy0
 239 [Network]
 240 Address=192.168.42.100
 241 DNS=192.168.42.1
 242 Domains= ~company''')
 243
 244         self.do_test(coldplug=True, ipv6=False,
 245                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 246
 247         with open(RESOLV_CONF) as f:
 248             contents = f.read()
 249             # ~company is not a search domain, only a routing domain
 250             self.assertNotRegex(contents, 'search.*company')
 251             # our global server should appear
 252             self.assertIn('nameserver 192.168.5.1\n', contents)
 253             # should not have domain-restricted server as global server
 254             self.assertNotIn('nameserver 192.168.42.1\n', contents)
 255
 256     def test_route_only_dns_all_domains(self):
 257         with open('/run/systemd/network/myvpn.netdev', 'w') as f:
 258             f.write('''[NetDev]
 259 Name=dummy0
 260 Kind=dummy
 261 MACAddress=12:34:56:78:9a:bc''')
 262         with open('/run/systemd/network/myvpn.network', 'w') as f:
 263             f.write('''[Match]
 264 Name=dummy0
 265 [Network]
 266 Address=192.168.42.100
 267 DNS=192.168.42.1
 268 Domains= ~company ~.''')
 269         self.addCleanup(os.remove, '/run/systemd/network/myvpn.netdev')
 270         self.addCleanup(os.remove, '/run/systemd/network/myvpn.network')
 271
 272         self.do_test(coldplug=True, ipv6=False,
 273                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 274
 275         with open(RESOLV_CONF) as f:
 276             contents = f.read()
 277
 278         # ~company is not a search domain, only a routing domain
 279         self.assertNotRegex(contents, 'search.*company')
 280
 281         # our global server should appear
 282         self.assertIn('nameserver 192.168.5.1\n', contents)
 283         # should have company server as global server due to ~.
 284         self.assertIn('nameserver 192.168.42.1\n', contents)
 285
 286
 287 @unittest.skipUnless(have_dnsmasq, 'dnsmasq not installed')
 288 class DnsmasqClientTest(ClientTestBase, unittest.TestCase):
 289     '''Test networkd client against dnsmasq'''
 290
 291     def setUp(self):
 292         super().setUp()
 293         self.dnsmasq = None
 294
 295     def create_iface(self, ipv6=False, dnsmasq_opts=None):
 296         '''Create test interface with DHCP server behind it'''
 297
 298         # add veth pair
 299         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface, 'type',
 300                                'veth', 'peer', 'name', self.if_router])
 301
 302         # give our router an IP
 303         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
 304         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
 305         if ipv6:
 306             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
 307         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
 308
 309         # add DHCP server
 310         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
 311         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
 312         if ipv6:
 313             extra_opts = ['--enable-ra', '--dhcp-range=2600::10,2600::20']
 314         else:
 315             extra_opts = []
 316         if dnsmasq_opts:
 317             extra_opts += dnsmasq_opts
 318         self.dnsmasq = subprocess.Popen(
 319             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 320              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
 321              '--dhcp-leasefile=' + lease_file, '--bind-interfaces',
 322              '--interface=' + self.if_router, '--except-interface=lo',
 323              '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts)
 324
 325     def shutdown_iface(self):
 326         '''Remove test interface and stop DHCP server'''
 327
 328         if self.if_router:
 329             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
 330             self.if_router = None
 331         if self.dnsmasq:
 332             self.dnsmasq.kill()
 333             self.dnsmasq.wait()
 334             self.dnsmasq = None
 335
 336     def print_server_log(self):
 337         '''Print DHCP server log for debugging failures'''
 338
 339         with open(self.dnsmasq_log) as f:
 340             sys.stdout.write('\n\n---- dnsmasq log ----\n%s\n------\n\n' % f.read())
 341
 342     def test_resolved_domain_restricted_dns(self):
 343         '''resolved: domain-restricted DNS servers'''
 344
 345         # create interface for generic connections; this will map all DNS names
 346         # to 192.168.42.1
 347         self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1'])
 348         self.writeConfig('/run/systemd/network/general.network', '''\
 349 [Match]
 350 Name=%s
 351 [Network]
 352 DHCP=ipv4
 353 IPv6AcceptRA=False''' % self.iface)
 354
 355         # create second device/dnsmasq for a .company/.lab VPN interface
 356         # static IPs for simplicity
 357         subprocess.check_call(['ip', 'link', 'add', 'name', 'testvpnclient', 'type',
 358                                'veth', 'peer', 'name', 'testvpnrouter'])
 359         self.addCleanup(subprocess.call, ['ip', 'link', 'del', 'dev', 'testvpnrouter'])
 360         subprocess.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
 361         subprocess.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
 362         subprocess.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
 363
 364         vpn_dnsmasq_log = os.path.join(self.workdir, 'dnsmasq-vpn.log')
 365         vpn_dnsmasq = subprocess.Popen(
 366             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 367              '--log-facility=' + vpn_dnsmasq_log, '--conf-file=/dev/null',
 368              '--dhcp-leasefile=/dev/null', '--bind-interfaces',
 369              '--interface=testvpnrouter', '--except-interface=lo',
 370              '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
 371         self.addCleanup(vpn_dnsmasq.wait)
 372         self.addCleanup(vpn_dnsmasq.kill)
 373
 374         self.writeConfig('/run/systemd/network/vpn.network', '''\
 375 [Match]
 376 Name=testvpnclient
 377 [Network]
 378 IPv6AcceptRA=False
 379 Address=10.241.3.2/24
 380 DNS=10.241.3.1
 381 Domains= ~company ~lab''')
 382
 383         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 384         subprocess.check_call([self.networkd_wait_online, '--interface', self.iface,
 385                                '--interface=testvpnclient', '--timeout=20'])
 386
 387         # ensure we start fresh with every test
 388         subprocess.check_call(['systemctl', 'restart', 'systemd-resolved'])
 389
 390         # test vpnclient specific domains; these should *not* be answered by
 391         # the general DNS
 392         out = subprocess.check_output(['systemd-resolve', 'math.lab'])
 393         self.assertIn(b'math.lab: 10.241.3.3', out)
 394         out = subprocess.check_output(['systemd-resolve', 'kettle.cantina.company'])
 395         self.assertIn(b'kettle.cantina.company: 10.241.4.4', out)
 396
 397         # test general domains
 398         out = subprocess.check_output(['systemd-resolve', 'megasearch.net'])
 399         self.assertIn(b'megasearch.net: 192.168.42.1', out)
 400
 401         with open(self.dnsmasq_log) as f:
 402             general_log = f.read()
 403         with open(vpn_dnsmasq_log) as f:
 404             vpn_log = f.read()
 405
 406         # VPN domains should only be sent to VPN DNS
 407         self.assertRegex(vpn_log, 'query.*math.lab')
 408         self.assertRegex(vpn_log, 'query.*cantina.company')
 409         self.assertNotIn('lab', general_log)
 410         self.assertNotIn('company', general_log)
 411
 412         # general domains should not be sent to the VPN DNS
 413         self.assertRegex(general_log, 'query.*megasearch.net')
 414         self.assertNotIn('megasearch.net', vpn_log)
 415
 416
 417 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 418     '''Test networkd client against networkd server'''
 419
 420     def setUp(self):
 421         super().setUp()
 422         self.dnsmasq = None
 423
 424     def create_iface(self, ipv6=False):
 425         '''Create test interface with DHCP server behind it'''
 426
 427         # run "router-side" networkd in own mount namespace to shield it from
 428         # "client-side" configuration and networkd
 429         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 430         self.addCleanup(os.remove, script)
 431         with os.fdopen(fd, 'w+') as f:
 432             f.write('''\
 433 #!/bin/sh -eu
 434 mkdir -p /run/systemd/network
 435 mkdir -p /run/systemd/netif
 436 mount -t tmpfs none /run/systemd/network
 437 mount -t tmpfs none /run/systemd/netif
 438 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 439 # create router/client veth pair
 440 cat << EOF > /run/systemd/network/test.netdev
 441 [NetDev]
 442 Name=%(ifr)s
 443 Kind=veth
 444
 445 [Peer]
 446 Name=%(ifc)s
 447 EOF
 448
 449 cat << EOF > /run/systemd/network/test.network
 450 [Match]
 451 Name=%(ifr)s
 452
 453 [Network]
 454 Address=192.168.5.1/24
 455 %(addr6)s
 456 DHCPServer=yes
 457
 458 [DHCPServer]
 459 PoolOffset=10
 460 PoolSize=50
 461 DNS=192.168.5.1
 462 EOF
 463
 464 # run networkd as in systemd-networkd.service
 465 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; p}')
 466 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or ''})
 467
 468             os.fchmod(fd, 0o755)
 469
 470         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 471                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 472                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 473                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 474                                '--service-type=notify', script])
 475
 476         # wait until devices got created
 477         for timeout in range(50):
 478             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 479             if b'state UP' in out and b'scope global' in out:
 480                 break
 481             time.sleep(0.1)
 482
 483     def shutdown_iface(self):
 484         '''Remove test interface and stop DHCP server'''
 485
 486         if self.if_router:
 487             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 488             # ensure failed transient unit does not stay around
 489             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 490             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 491             self.if_router = None
 492
 493     def print_server_log(self):
 494         '''Print DHCP server log for debugging failures'''
 495
 496         self.show_journal('networkd-test-router.service')
 497
 498     @unittest.skip('networkd does not have DHCPv6 server support')
 499     def test_hotplug_dhcp_ip6(self):
 500         pass
 501
 502     @unittest.skip('networkd does not have DHCPv6 server support')
 503     def test_coldplug_dhcp_ip6(self):
 504         pass
 505
 506     def test_search_domains(self):
 507
 508         # we don't use this interface for this test
 509         self.if_router = None
 510
 511         self.writeConfig('/run/systemd/network/test.netdev', '''\
 512 [NetDev]
 513 Name=dummy0
 514 Kind=dummy
 515 MACAddress=12:34:56:78:9a:bc''')
 516         self.writeConfig('/run/systemd/network/test.network', '''\
 517 [Match]
 518 Name=dummy0
 519 [Network]
 520 Address=192.168.42.100
 521 DNS=192.168.42.1
 522 Domains= one two three four five six seven eight nine ten''')
 523
 524         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 525
 526         for timeout in range(50):
 527             with open(RESOLV_CONF) as f:
 528                 contents = f.read()
 529             if ' one' in contents:
 530                 break
 531             time.sleep(0.1)
 532         self.assertRegex(contents, 'search .*one two three four')
 533         self.assertNotIn('seven\n', contents)
 534         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 535
 536     def test_search_domains_too_long(self):
 537
 538         # we don't use this interface for this test
 539         self.if_router = None
 540
 541         name_prefix = 'a' * 60
 542
 543         self.writeConfig('/run/systemd/network/test.netdev', '''\
 544 [NetDev]
 545 Name=dummy0
 546 Kind=dummy
 547 MACAddress=12:34:56:78:9a:bc''')
 548         self.writeConfig('/run/systemd/network/test.network', '''\
 549 [Match]
 550 Name=dummy0
 551 [Network]
 552 Address=192.168.42.100
 553 DNS=192.168.42.1
 554 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 555
 556         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 557
 558         for timeout in range(50):
 559             with open(RESOLV_CONF) as f:
 560                 contents = f.read()
 561             if ' one' in contents:
 562                 break
 563             time.sleep(0.1)
 564         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 565         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 566
 567     def test_dropin(self):
 568         # we don't use this interface for this test
 569         self.if_router = None
 570
 571         self.writeConfig('/run/systemd/network/test.netdev', '''\
 572 [NetDev]
 573 Name=dummy0
 574 Kind=dummy
 575 MACAddress=12:34:56:78:9a:bc''')
 576         self.writeConfig('/run/systemd/network/test.network', '''\
 577 [Match]
 578 Name=dummy0
 579 [Network]
 580 Address=192.168.42.100
 581 DNS=192.168.42.1''')
 582         self.writeConfig('/run/systemd/network/test.network.d/dns.conf', '''\
 583 [Network]
 584 DNS=127.0.0.1''')
 585
 586         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 587
 588         for timeout in range(50):
 589             with open(RESOLV_CONF) as f:
 590                 contents = f.read()
 591             if ' 127.0.0.1' in contents:
 592                 break
 593             time.sleep(0.1)
 594         self.assertIn('nameserver 192.168.42.1\n', contents)
 595         self.assertIn('nameserver 127.0.0.1\n', contents)
 596
 597 if __name__ == '__main__':
 598     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
 599                                                      verbosity=2))