test/networkd-test.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: LGPL-2.1+
   3 #
   4 # networkd integration test
   5 # This uses temporary configuration in /run and temporary veth devices, and
   6 # does not write anything on disk or change any system configuration;
   7 # but it assumes (and checks at the beginning) that networkd is not currently
   8 # running.
   9 #
  10 # This can be run on a normal installation, in QEMU, nspawn (with
  11 # --private-network), LXD (with "--config raw.lxc=lxc.aa_profile=unconfined"),
  12 # or LXC system containers. You need at least the "ip" tool from the iproute
  13 # package; it is recommended to install dnsmasq too to get full test coverage.
  14 #
  15 # ATTENTION: This uses the *installed* networkd, not the one from the built
  16 # source tree.
  17 #
  18 # © 2015 Canonical Ltd.
  19 # Author: Martin Pitt <martin.pitt@ubuntu.com>
  20
  21 import errno
  22 import os
  23 import shutil
  24 import socket
  25 import subprocess
  26 import sys
  27 import tempfile
  28 import time
  29 import unittest
  30
  31 HAVE_DNSMASQ = shutil.which('dnsmasq') is not None
  32 IS_CONTAINER = subprocess.call(['systemd-detect-virt', '--quiet', '--container']) == 0
  33
  34 NETWORK_UNITDIR = '/run/systemd/network'
  35
  36 NETWORKD_WAIT_ONLINE = shutil.which('systemd-networkd-wait-online',
  37                                     path='/usr/lib/systemd:/lib/systemd')
  38
  39 RESOLV_CONF = '/run/systemd/resolve/resolv.conf'
  40
  41 tmpmounts = []
  42 running_units = []
  43 stopped_units = []
  44
  45
  46 def setUpModule():
  47     global tmpmounts
  48
  49     """Initialize the environment, and perform sanity checks on it."""
  50     if NETWORKD_WAIT_ONLINE is None:
  51         raise OSError(errno.ENOENT, 'systemd-networkd-wait-online not found')
  52
  53     # Do not run any tests if the system is using networkd already and it's not virtualized
  54     if (subprocess.call(['systemctl', 'is-active', '--quiet', 'systemd-networkd.service']) == 0 and
  55             subprocess.call(['systemd-detect-virt', '--quiet']) != 0):
  56         raise unittest.SkipTest('not virtualized and networkd is already active')
  57
  58     # Ensure we don't mess with an existing networkd config
  59     for u in ['systemd-networkd.socket', 'systemd-networkd', 'systemd-resolved']:
  60         if subprocess.call(['systemctl', 'is-active', '--quiet', u]) == 0:
  61             subprocess.call(['systemctl', 'stop', u])
  62             running_units.append(u)
  63         else:
  64             stopped_units.append(u)
  65
  66     # create static systemd-network user for networkd-test-router.service (it
  67     # needs to do some stuff as root and can't start as user; but networkd
  68     # still insists on the user)
  69     if subprocess.call(['getent', 'passwd', 'systemd-network']) != 0:
  70         subprocess.call(['useradd', '--system', '--no-create-home', 'systemd-network'])
  71
  72     for d in ['/etc/systemd/network', '/run/systemd/network',
  73               '/run/systemd/netif', '/run/systemd/resolve']:
  74         if os.path.isdir(d):
  75             subprocess.check_call(["mount", "-t", "tmpfs", "none", d])
  76             tmpmounts.append(d)
  77     if os.path.isdir('/run/systemd/resolve'):
  78         os.chmod('/run/systemd/resolve', 0o755)
  79         shutil.chown('/run/systemd/resolve', 'systemd-resolve', 'systemd-resolve')
  80     if os.path.isdir('/run/systemd/netif'):
  81         os.chmod('/run/systemd/netif', 0o755)
  82         shutil.chown('/run/systemd/netif', 'systemd-network', 'systemd-network')
  83
  84     # Avoid "Failed to open /dev/tty" errors in containers.
  85     os.environ['SYSTEMD_LOG_TARGET'] = 'journal'
  86
  87     # Ensure the unit directory exists so tests can dump files into it.
  88     os.makedirs(NETWORK_UNITDIR, exist_ok=True)
  89
  90
  91 def tearDownModule():
  92     global tmpmounts
  93     for d in tmpmounts:
  94         subprocess.check_call(["umount", d])
  95     for u in stopped_units:
  96         subprocess.call(["systemctl", "stop", u])
  97     for u in running_units:
  98         subprocess.call(["systemctl", "restart", u])
  99
 100
 101 class NetworkdTestingUtilities:
 102     """Provide a set of utility functions to facilitate networkd tests.
 103
 104     This class must be inherited along with unittest.TestCase to define
 105     some required methods.
 106     """
 107
 108     def add_veth_pair(self, veth, peer, veth_options=(), peer_options=()):
 109         """Add a veth interface pair, and queue them to be removed."""
 110         subprocess.check_call(['ip', 'link', 'add', 'name', veth] +
 111                               list(veth_options) +
 112                               ['type', 'veth', 'peer', 'name', peer] +
 113                               list(peer_options))
 114         self.addCleanup(subprocess.call, ['ip', 'link', 'del', 'dev', peer])
 115
 116     def write_config(self, path, contents):
 117         """"Write a configuration file, and queue it to be removed."""
 118
 119         with open(path, 'w') as f:
 120             f.write(contents)
 121
 122         self.addCleanup(os.remove, path)
 123
 124     def write_network(self, unit_name, contents):
 125         """Write a network unit file, and queue it to be removed."""
 126         self.write_config(os.path.join(NETWORK_UNITDIR, unit_name), contents)
 127
 128     def write_network_dropin(self, unit_name, dropin_name, contents):
 129         """Write a network unit drop-in, and queue it to be removed."""
 130         dropin_dir = os.path.join(NETWORK_UNITDIR, "{}.d".format(unit_name))
 131         dropin_path = os.path.join(dropin_dir, "{}.conf".format(dropin_name))
 132
 133         os.makedirs(dropin_dir, exist_ok=True)
 134         self.addCleanup(os.rmdir, dropin_dir)
 135         with open(dropin_path, 'w') as dropin:
 136             dropin.write(contents)
 137         self.addCleanup(os.remove, dropin_path)
 138
 139     def read_attr(self, link, attribute):
 140         """Read a link attributed from the sysfs."""
 141         # Note we we don't want to check if interface `link' is managed, we
 142         # want to evaluate link variable and pass the value of the link to
 143         # assert_link_states e.g. eth0=managed.
 144         self.assert_link_states(**{link:'managed'})
 145         with open(os.path.join('/sys/class/net', link, attribute)) as f:
 146             return f.readline().strip()
 147
 148     def assert_link_states(self, **kwargs):
 149         """Match networkctl link states to the given ones.
 150
 151         Each keyword argument should be the name of a network interface
 152         with its expected value of the "SETUP" column in output from
 153         networkctl.  The interfaces have five seconds to come online
 154         before the check is performed.  Every specified interface must
 155         be present in the output, and any other interfaces found in the
 156         output are ignored.
 157
 158         A special interface state "managed" is supported, which matches
 159         any value in the "SETUP" column other than "unmanaged".
 160         """
 161         if not kwargs:
 162             return
 163         interfaces = set(kwargs)
 164
 165         # Wait for the requested interfaces, but don't fail for them.
 166         subprocess.call([NETWORKD_WAIT_ONLINE, '--timeout=5'] +
 167                         ['--interface={}'.format(iface) for iface in kwargs])
 168
 169         # Validate each link state found in the networkctl output.
 170         out = subprocess.check_output(['networkctl', '--no-legend']).rstrip()
 171         for line in out.decode('utf-8').split('\n'):
 172             fields = line.split()
 173             if len(fields) >= 5 and fields[1] in kwargs:
 174                 iface = fields[1]
 175                 expected = kwargs[iface]
 176                 actual = fields[-1]
 177                 if (actual != expected and
 178                         not (expected == 'managed' and actual != 'unmanaged')):
 179                     self.fail("Link {} expects state {}, found {}".format(iface, expected, actual))
 180                 interfaces.remove(iface)
 181
 182         # Ensure that all requested interfaces have been covered.
 183         if interfaces:
 184             self.fail("Missing links in status output: {}".format(interfaces))
 185
 186
 187 class BridgeTest(NetworkdTestingUtilities, unittest.TestCase):
 188     """Provide common methods for testing networkd against servers."""
 189
 190     def setUp(self):
 191         self.write_network('port1.netdev', '''\
 192 [NetDev]
 193 Name=port1
 194 Kind=dummy
 195 MACAddress=12:34:56:78:9a:bc''')
 196         self.write_network('port2.netdev', '''\
 197 [NetDev]
 198 Name=port2
 199 Kind=dummy
 200 MACAddress=12:34:56:78:9a:bd''')
 201         self.write_network('mybridge.netdev', '''\
 202 [NetDev]
 203 Name=mybridge
 204 Kind=bridge''')
 205         self.write_network('port1.network', '''\
 206 [Match]
 207 Name=port1
 208 [Network]
 209 Bridge=mybridge''')
 210         self.write_network('port2.network', '''\
 211 [Match]
 212 Name=port2
 213 [Network]
 214 Bridge=mybridge''')
 215         self.write_network('mybridge.network', '''\
 216 [Match]
 217 Name=mybridge
 218 [Network]
 219 DNS=192.168.250.1
 220 Address=192.168.250.33/24
 221 Gateway=192.168.250.1''')
 222         subprocess.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
 223         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 224
 225     def tearDown(self):
 226         subprocess.check_call(['systemctl', 'stop', 'systemd-networkd'])
 227         subprocess.check_call(['ip', 'link', 'del', 'mybridge'])
 228         subprocess.check_call(['ip', 'link', 'del', 'port1'])
 229         subprocess.check_call(['ip', 'link', 'del', 'port2'])
 230
 231     def test_bridge_init(self):
 232         self.assert_link_states(
 233             port1='managed',
 234             port2='managed',
 235             mybridge='managed')
 236
 237     def test_bridge_port_priority(self):
 238         self.assertEqual(self.read_attr('port1', 'brport/priority'), '32')
 239         self.write_network_dropin('port1.network', 'priority', '''\
 240 [Bridge]
 241 Priority=28
 242 ''')
 243         subprocess.check_call(['systemctl', 'restart', 'systemd-networkd'])
 244         self.assertEqual(self.read_attr('port1', 'brport/priority'), '28')
 245
 246     def test_bridge_port_priority_set_zero(self):
 247         """It should be possible to set the bridge port priority to 0"""
 248         self.assertEqual(self.read_attr('port2', 'brport/priority'), '32')
 249         self.write_network_dropin('port2.network', 'priority', '''\
 250 [Bridge]
 251 Priority=0
 252 ''')
 253         subprocess.check_call(['systemctl', 'restart', 'systemd-networkd'])
 254         self.assertEqual(self.read_attr('port2', 'brport/priority'), '0')
 255
 256     def test_bridge_port_property(self):
 257         """Test the "[Bridge]" section keys"""
 258         self.assertEqual(self.read_attr('port2', 'brport/priority'), '32')
 259         self.write_network_dropin('port2.network', 'property', '''\
 260 [Bridge]
 261 UnicastFlood=true
 262 HairPin=true
 263 UseBPDU=true
 264 FastLeave=true
 265 AllowPortToBeRoot=true
 266 Cost=555
 267 Priority=23
 268 ''')
 269         subprocess.check_call(['systemctl', 'restart', 'systemd-networkd'])
 270
 271         self.assertEqual(self.read_attr('port2', 'brport/priority'), '23')
 272         self.assertEqual(self.read_attr('port2', 'brport/hairpin_mode'), '1')
 273         self.assertEqual(self.read_attr('port2', 'brport/path_cost'), '555')
 274         self.assertEqual(self.read_attr('port2', 'brport/multicast_fast_leave'), '1')
 275         self.assertEqual(self.read_attr('port2', 'brport/unicast_flood'), '1')
 276         self.assertEqual(self.read_attr('port2', 'brport/bpdu_guard'), '1')
 277         self.assertEqual(self.read_attr('port2', 'brport/root_block'), '1')
 278
 279 class ClientTestBase(NetworkdTestingUtilities):
 280     """Provide common methods for testing networkd against servers."""
 281
 282     @classmethod
 283     def setUpClass(klass):
 284         klass.orig_log_level = subprocess.check_output(
 285             ['systemctl', 'show', '--value', '--property', 'LogLevel'],
 286             universal_newlines=True).strip()
 287         subprocess.check_call(['systemd-analyze', 'log-level', 'debug'])
 288
 289     @classmethod
 290     def tearDownClass(klass):
 291         subprocess.check_call(['systemd-analyze', 'log-level', klass.orig_log_level])
 292
 293     def setUp(self):
 294         self.iface = 'test_eth42'
 295         self.if_router = 'router_eth42'
 296         self.workdir_obj = tempfile.TemporaryDirectory()
 297         self.workdir = self.workdir_obj.name
 298         self.config = 'test_eth42.network'
 299
 300         # get current journal cursor
 301         subprocess.check_output(['journalctl', '--sync'])
 302         out = subprocess.check_output(['journalctl', '-b', '--quiet',
 303                                        '--no-pager', '-n0', '--show-cursor'],
 304                                       universal_newlines=True)
 305         self.assertTrue(out.startswith('-- cursor:'))
 306         self.journal_cursor = out.split()[-1]
 307
 308         subprocess.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
 309
 310     def tearDown(self):
 311         self.shutdown_iface()
 312         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 313         subprocess.call(['ip', 'link', 'del', 'dummy0'],
 314                         stderr=subprocess.DEVNULL)
 315
 316     def show_journal(self, unit):
 317         '''Show journal of given unit since start of the test'''
 318
 319         print('---- {} ----'.format(unit))
 320         subprocess.check_output(['journalctl', '--sync'])
 321         sys.stdout.flush()
 322         subprocess.call(['journalctl', '-b', '--no-pager', '--quiet',
 323                          '--cursor', self.journal_cursor, '-u', unit])
 324
 325     def create_iface(self, ipv6=False):
 326         '''Create test interface with DHCP server behind it'''
 327
 328         raise NotImplementedError('must be implemented by a subclass')
 329
 330     def shutdown_iface(self):
 331         '''Remove test interface and stop DHCP server'''
 332
 333         raise NotImplementedError('must be implemented by a subclass')
 334
 335     def print_server_log(self):
 336         '''Print DHCP server log for debugging failures'''
 337
 338         raise NotImplementedError('must be implemented by a subclass')
 339
 340     def start_unit(self, unit):
 341         try:
 342             subprocess.check_call(['systemctl', 'start', unit])
 343         except subprocess.CalledProcessError:
 344             self.show_journal(unit)
 345             raise
 346
 347     def do_test(self, coldplug=True, ipv6=False, extra_opts='',
 348                 online_timeout=10, dhcp_mode='yes'):
 349         self.start_unit('systemd-resolved')
 350         self.write_network(self.config, '''\
 351 [Match]
 352 Name={}
 353 [Network]
 354 DHCP={}
 355 {}'''.format(self.iface, dhcp_mode, extra_opts))
 356
 357         if coldplug:
 358             # create interface first, then start networkd
 359             self.create_iface(ipv6=ipv6)
 360             self.start_unit('systemd-networkd')
 361         elif coldplug is not None:
 362             # start networkd first, then create interface
 363             self.start_unit('systemd-networkd')
 364             self.create_iface(ipv6=ipv6)
 365         else:
 366             # "None" means test sets up interface by itself
 367             self.start_unit('systemd-networkd')
 368
 369         try:
 370             subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface',
 371                                    self.iface, '--timeout=%i' % online_timeout])
 372
 373             if ipv6:
 374                 # check iface state and IP 6 address; FIXME: we need to wait a bit
 375                 # longer, as the iface is "configured" already with IPv4 *or*
 376                 # IPv6, but we want to wait for both
 377                 for _ in range(10):
 378                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
 379                     if b'state UP' in out and b'inet6 2600' in out and b'inet 192.168' in out and b'tentative' not in out:
 380                         break
 381                     time.sleep(1)
 382                 else:
 383                     self.fail('timed out waiting for IPv6 configuration')
 384
 385                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
 386                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 387             else:
 388                 # should have link-local address on IPv6 only
 389                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
 390                 self.assertRegex(out, br'inet6 fe80::.* scope link')
 391                 self.assertNotIn(b'scope global', out)
 392
 393             # should have IPv4 address
 394             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 395             self.assertIn(b'state UP', out)
 396             self.assertRegex(out, br'inet 192.168.5.\d+/.* scope global dynamic')
 397
 398             # check networkctl state
 399             out = subprocess.check_output(['networkctl'])
 400             self.assertRegex(out, (r'{}\s+ether\s+[a-z-]+\s+unmanaged'.format(self.if_router)).encode())
 401             self.assertRegex(out, (r'{}\s+ether\s+routable\s+configured'.format(self.iface)).encode())
 402
 403             out = subprocess.check_output(['networkctl', '-n', '0', 'status', self.iface])
 404             self.assertRegex(out, br'Type:\s+ether')
 405             self.assertRegex(out, br'State:\s+routable.*configured')
 406             self.assertRegex(out, br'Address:\s+192.168.5.\d+')
 407             if ipv6:
 408                 self.assertRegex(out, br'2600::')
 409             else:
 410                 self.assertNotIn(br'2600::', out)
 411             self.assertRegex(out, br'fe80::')
 412             self.assertRegex(out, br'Gateway:\s+192.168.5.1')
 413             self.assertRegex(out, br'DNS:\s+192.168.5.1')
 414         except (AssertionError, subprocess.CalledProcessError):
 415             # show networkd status, journal, and DHCP server log on failure
 416             with open(os.path.join(NETWORK_UNITDIR, self.config)) as f:
 417                 print('\n---- {} ----\n{}'.format(self.config, f.read()))
 418             print('---- interface status ----')
 419             sys.stdout.flush()
 420             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
 421             print('---- networkctl status {} ----'.format(self.iface))
 422             sys.stdout.flush()
 423             rc = subprocess.call(['networkctl', '-n', '0', 'status', self.iface])
 424             if rc != 0:
 425                 print("'networkctl status' exited with an unexpected code {}".format(rc))
 426             self.show_journal('systemd-networkd.service')
 427             self.print_server_log()
 428             raise
 429
 430         for timeout in range(50):
 431             with open(RESOLV_CONF) as f:
 432                 contents = f.read()
 433             if 'nameserver 192.168.5.1\n' in contents:
 434                 break
 435             time.sleep(0.1)
 436         else:
 437             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
 438
 439         if coldplug is False:
 440             # check post-down.d hook
 441             self.shutdown_iface()
 442
 443     def test_coldplug_dhcp_yes_ip4(self):
 444         # we have a 12s timeout on RA, so we need to wait longer
 445         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
 446
 447     def test_coldplug_dhcp_yes_ip4_no_ra(self):
 448         # with disabling RA explicitly things should be fast
 449         self.do_test(coldplug=True, ipv6=False,
 450                      extra_opts='IPv6AcceptRA=False')
 451
 452     def test_coldplug_dhcp_ip4_only(self):
 453         # we have a 12s timeout on RA, so we need to wait longer
 454         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 455                      online_timeout=15)
 456
 457     def test_coldplug_dhcp_ip4_only_no_ra(self):
 458         # with disabling RA explicitly things should be fast
 459         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 460                      extra_opts='IPv6AcceptRA=False')
 461
 462     def test_coldplug_dhcp_ip6(self):
 463         self.do_test(coldplug=True, ipv6=True)
 464
 465     def test_hotplug_dhcp_ip4(self):
 466         # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
 467         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
 468
 469     def test_hotplug_dhcp_ip6(self):
 470         self.do_test(coldplug=False, ipv6=True)
 471
 472     def test_route_only_dns(self):
 473         self.write_network('myvpn.netdev', '''\
 474 [NetDev]
 475 Name=dummy0
 476 Kind=dummy
 477 MACAddress=12:34:56:78:9a:bc''')
 478         self.write_network('myvpn.network', '''\
 479 [Match]
 480 Name=dummy0
 481 [Network]
 482 Address=192.168.42.100/24
 483 DNS=192.168.42.1
 484 Domains= ~company''')
 485
 486         try:
 487             self.do_test(coldplug=True, ipv6=False,
 488                          extra_opts='IPv6AcceptRouterAdvertisements=False')
 489         except subprocess.CalledProcessError as e:
 490             # networkd often fails to start in LXC: https://github.com/systemd/systemd/issues/11848
 491             if IS_CONTAINER and e.cmd == ['systemctl', 'start', 'systemd-networkd']:
 492                 raise unittest.SkipTest('https://github.com/systemd/systemd/issues/11848')
 493             else:
 494                 raise
 495
 496         with open(RESOLV_CONF) as f:
 497             contents = f.read()
 498             # ~company is not a search domain, only a routing domain
 499             self.assertNotRegex(contents, 'search.*company')
 500             # our global server should appear
 501             self.assertIn('nameserver 192.168.5.1\n', contents)
 502             # should not have domain-restricted server as global server
 503             self.assertNotIn('nameserver 192.168.42.1\n', contents)
 504
 505     def test_route_only_dns_all_domains(self):
 506         self.write_network('myvpn.netdev', '''[NetDev]
 507 Name=dummy0
 508 Kind=dummy
 509 MACAddress=12:34:56:78:9a:bc''')
 510         self.write_network('myvpn.network', '''[Match]
 511 Name=dummy0
 512 [Network]
 513 Address=192.168.42.100/24
 514 DNS=192.168.42.1
 515 Domains= ~company ~.''')
 516
 517         try:
 518             self.do_test(coldplug=True, ipv6=False,
 519                          extra_opts='IPv6AcceptRouterAdvertisements=False')
 520         except subprocess.CalledProcessError as e:
 521             # networkd often fails to start in LXC: https://github.com/systemd/systemd/issues/11848
 522             if IS_CONTAINER and e.cmd == ['systemctl', 'start', 'systemd-networkd']:
 523                 raise unittest.SkipTest('https://github.com/systemd/systemd/issues/11848')
 524             else:
 525                 raise
 526
 527         with open(RESOLV_CONF) as f:
 528             contents = f.read()
 529
 530         # ~company is not a search domain, only a routing domain
 531         self.assertNotRegex(contents, 'search.*company')
 532
 533         # our global server should appear
 534         self.assertIn('nameserver 192.168.5.1\n', contents)
 535         # should have company server as global server due to ~.
 536         self.assertIn('nameserver 192.168.42.1\n', contents)
 537
 538
 539 @unittest.skipUnless(HAVE_DNSMASQ, 'dnsmasq not installed')
 540 class DnsmasqClientTest(ClientTestBase, unittest.TestCase):
 541     '''Test networkd client against dnsmasq'''
 542
 543     def setUp(self):
 544         super().setUp()
 545         self.dnsmasq = None
 546         self.iface_mac = 'de:ad:be:ef:47:11'
 547
 548     def create_iface(self, ipv6=False, dnsmasq_opts=None):
 549         '''Create test interface with DHCP server behind it'''
 550
 551         # add veth pair
 552         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface,
 553                                'address', self.iface_mac,
 554                                'type', 'veth', 'peer', 'name', self.if_router])
 555
 556         # give our router an IP
 557         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
 558         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
 559         if ipv6:
 560             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
 561         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
 562
 563         # add DHCP server
 564         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
 565         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
 566         if ipv6:
 567             extra_opts = ['--enable-ra', '--dhcp-range=2600::10,2600::20']
 568         else:
 569             extra_opts = []
 570         if dnsmasq_opts:
 571             extra_opts += dnsmasq_opts
 572         self.dnsmasq = subprocess.Popen(
 573             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 574              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
 575              '--dhcp-leasefile=' + lease_file, '--bind-interfaces',
 576              '--interface=' + self.if_router, '--except-interface=lo',
 577              '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts)
 578
 579     def shutdown_iface(self):
 580         '''Remove test interface and stop DHCP server'''
 581
 582         if self.if_router:
 583             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
 584             self.if_router = None
 585         if self.dnsmasq:
 586             self.dnsmasq.kill()
 587             self.dnsmasq.wait()
 588             self.dnsmasq = None
 589
 590     def print_server_log(self):
 591         '''Print DHCP server log for debugging failures'''
 592
 593         with open(self.dnsmasq_log) as f:
 594             sys.stdout.write('\n\n---- dnsmasq log ----\n{}\n------\n\n'.format(f.read()))
 595
 596     def test_resolved_domain_restricted_dns(self):
 597         '''resolved: domain-restricted DNS servers'''
 598
 599         # FIXME: resolvectl query fails with enabled DNSSEC against our dnsmasq
 600         conf = '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
 601         os.makedirs(os.path.dirname(conf), exist_ok=True)
 602         with open(conf, 'w') as f:
 603             f.write('[Resolve]\nDNSSEC=no\n')
 604         self.addCleanup(os.remove, conf)
 605
 606         # create interface for generic connections; this will map all DNS names
 607         # to 192.168.42.1
 608         self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1'])
 609         self.write_network('general.network', '''\
 610 [Match]
 611 Name={}
 612 [Network]
 613 DHCP=ipv4
 614 IPv6AcceptRA=False'''.format(self.iface))
 615
 616         # create second device/dnsmasq for a .company/.lab VPN interface
 617         # static IPs for simplicity
 618         self.add_veth_pair('testvpnclient', 'testvpnrouter')
 619         subprocess.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
 620         subprocess.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
 621         subprocess.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
 622
 623         vpn_dnsmasq_log = os.path.join(self.workdir, 'dnsmasq-vpn.log')
 624         vpn_dnsmasq = subprocess.Popen(
 625             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 626              '--log-facility=' + vpn_dnsmasq_log, '--conf-file=/dev/null',
 627              '--dhcp-leasefile=/dev/null', '--bind-interfaces',
 628              '--interface=testvpnrouter', '--except-interface=lo',
 629              '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
 630         self.addCleanup(vpn_dnsmasq.wait)
 631         self.addCleanup(vpn_dnsmasq.kill)
 632
 633         self.write_network('vpn.network', '''\
 634 [Match]
 635 Name=testvpnclient
 636 [Network]
 637 IPv6AcceptRA=False
 638 Address=10.241.3.2/24
 639 DNS=10.241.3.1
 640 Domains= ~company ~lab''')
 641
 642         self.start_unit('systemd-networkd')
 643         subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface', self.iface,
 644                                '--interface=testvpnclient', '--timeout=20'])
 645
 646         # ensure we start fresh with every test
 647         subprocess.check_call(['systemctl', 'restart', 'systemd-resolved'])
 648
 649         # test vpnclient specific domains; these should *not* be answered by
 650         # the general DNS
 651         out = subprocess.check_output(['resolvectl', 'query', 'math.lab'])
 652         self.assertIn(b'math.lab: 10.241.3.3', out)
 653         out = subprocess.check_output(['resolvectl', 'query', 'kettle.cantina.company'])
 654         self.assertIn(b'kettle.cantina.company: 10.241.4.4', out)
 655
 656         # test general domains
 657         out = subprocess.check_output(['resolvectl', 'query', 'megasearch.net'])
 658         self.assertIn(b'megasearch.net: 192.168.42.1', out)
 659
 660         with open(self.dnsmasq_log) as f:
 661             general_log = f.read()
 662         with open(vpn_dnsmasq_log) as f:
 663             vpn_log = f.read()
 664
 665         # VPN domains should only be sent to VPN DNS
 666         self.assertRegex(vpn_log, 'query.*math.lab')
 667         self.assertRegex(vpn_log, 'query.*cantina.company')
 668         self.assertNotIn('.lab', general_log)
 669         self.assertNotIn('.company', general_log)
 670
 671         # general domains should not be sent to the VPN DNS
 672         self.assertRegex(general_log, 'query.*megasearch.net')
 673         self.assertNotIn('megasearch.net', vpn_log)
 674
 675     def test_resolved_etc_hosts(self):
 676         '''resolved queries to /etc/hosts'''
 677
 678         # FIXME: -t MX query fails with enabled DNSSEC (even when using
 679         # the known negative trust anchor .internal instead of .example.com)
 680         conf = '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
 681         os.makedirs(os.path.dirname(conf), exist_ok=True)
 682         with open(conf, 'w') as f:
 683             f.write('[Resolve]\nDNSSEC=no\nLLMNR=no\nMulticastDNS=no\n')
 684         self.addCleanup(os.remove, conf)
 685
 686         # create /etc/hosts bind mount which resolves my.example.com for IPv4
 687         hosts = os.path.join(self.workdir, 'hosts')
 688         with open(hosts, 'w') as f:
 689             f.write('172.16.99.99  my.example.com\n')
 690         subprocess.check_call(['mount', '--bind', hosts, '/etc/hosts'])
 691         self.addCleanup(subprocess.call, ['umount', '/etc/hosts'])
 692         subprocess.check_call(['systemctl', 'stop', 'systemd-resolved.service'])
 693
 694         # note: different IPv4 address here, so that it's easy to tell apart
 695         # what resolved the query
 696         self.create_iface(dnsmasq_opts=['--host-record=my.example.com,172.16.99.1,2600::99:99',
 697                                         '--host-record=other.example.com,172.16.0.42,2600::42',
 698                                         '--mx-host=example.com,mail.example.com'],
 699                           ipv6=True)
 700         self.do_test(coldplug=None, ipv6=True)
 701
 702         try:
 703             # family specific queries
 704             out = subprocess.check_output(['resolvectl', 'query', '-4', 'my.example.com'])
 705             self.assertIn(b'my.example.com: 172.16.99.99', out)
 706             # we don't expect an IPv6 answer; if /etc/hosts has any IP address,
 707             # it's considered a sufficient source
 708             self.assertNotEqual(subprocess.call(['resolvectl', 'query', '-6', 'my.example.com']), 0)
 709             # "any family" query; IPv4 should come from /etc/hosts
 710             out = subprocess.check_output(['resolvectl', 'query', 'my.example.com'])
 711             self.assertIn(b'my.example.com: 172.16.99.99', out)
 712             # IP → name lookup; again, takes the /etc/hosts one
 713             out = subprocess.check_output(['resolvectl', 'query', '172.16.99.99'])
 714             self.assertIn(b'172.16.99.99: my.example.com', out)
 715
 716             # non-address RRs should fall back to DNS
 717             out = subprocess.check_output(['resolvectl', 'query', '--type=MX', 'example.com'])
 718             self.assertIn(b'example.com IN MX 1 mail.example.com', out)
 719
 720             # other domains query DNS
 721             out = subprocess.check_output(['resolvectl', 'query', 'other.example.com'])
 722             self.assertIn(b'172.16.0.42', out)
 723             out = subprocess.check_output(['resolvectl', 'query', '172.16.0.42'])
 724             self.assertIn(b'172.16.0.42: other.example.com', out)
 725         except (AssertionError, subprocess.CalledProcessError):
 726             self.show_journal('systemd-resolved.service')
 727             self.print_server_log()
 728             raise
 729
 730     def test_transient_hostname(self):
 731         '''networkd sets transient hostname from DHCP'''
 732
 733         orig_hostname = socket.gethostname()
 734         self.addCleanup(socket.sethostname, orig_hostname)
 735         # temporarily move /etc/hostname away; restart hostnamed to pick it up
 736         if os.path.exists('/etc/hostname'):
 737             subprocess.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
 738             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
 739         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 740         self.addCleanup(subprocess.call, ['systemctl', 'stop', 'systemd-hostnamed.service'])
 741
 742         self.create_iface(dnsmasq_opts=['--dhcp-host={},192.168.5.210,testgreen'.format(self.iface_mac)])
 743         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 744
 745         try:
 746             # should have received the fixed IP above
 747             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 748             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 749             # should have set transient hostname in hostnamed; this is
 750             # sometimes a bit lagging (issue #4753), so retry a few times
 751             for retry in range(1, 6):
 752                 out = subprocess.check_output(['hostnamectl'])
 753                 if b'testgreen' in out:
 754                     break
 755                 time.sleep(5)
 756                 sys.stdout.write('[retry %i] ' % retry)
 757                 sys.stdout.flush()
 758             else:
 759                 self.fail('Transient hostname not found in hostnamectl:\n{}'.format(out.decode()))
 760             # and also applied to the system
 761             self.assertEqual(socket.gethostname(), 'testgreen')
 762         except AssertionError:
 763             self.show_journal('systemd-networkd.service')
 764             self.show_journal('systemd-hostnamed.service')
 765             self.print_server_log()
 766             raise
 767
 768     def test_transient_hostname_with_static(self):
 769         '''transient hostname is not applied if static hostname exists'''
 770
 771         orig_hostname = socket.gethostname()
 772         self.addCleanup(socket.sethostname, orig_hostname)
 773
 774         if not os.path.exists('/etc/hostname'):
 775             self.write_config('/etc/hostname', "foobarqux")
 776         else:
 777             self.write_config('/run/hostname.tmp', "foobarqux")
 778             subprocess.check_call(['mount', '--bind', '/run/hostname.tmp', '/etc/hostname'])
 779             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
 780
 781         socket.sethostname("foobarqux");
 782
 783         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 784         self.addCleanup(subprocess.call, ['systemctl', 'stop', 'systemd-hostnamed.service'])
 785
 786         self.create_iface(dnsmasq_opts=['--dhcp-host={},192.168.5.210,testgreen'.format(self.iface_mac)])
 787         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 788
 789         try:
 790             # should have received the fixed IP above
 791             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 792             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 793             # static hostname wins over transient one, thus *not* applied
 794             self.assertEqual(socket.gethostname(), "foobarqux")
 795         except AssertionError:
 796             self.show_journal('systemd-networkd.service')
 797             self.show_journal('systemd-hostnamed.service')
 798             self.print_server_log()
 799             raise
 800
 801
 802 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 803     '''Test networkd client against networkd server'''
 804
 805     def setUp(self):
 806         super().setUp()
 807         self.dnsmasq = None
 808
 809     def create_iface(self, ipv6=False, dhcpserver_opts=None):
 810         '''Create test interface with DHCP server behind it'''
 811
 812         # run "router-side" networkd in own mount namespace to shield it from
 813         # "client-side" configuration and networkd
 814         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 815         self.addCleanup(os.remove, script)
 816         with os.fdopen(fd, 'w+') as f:
 817             f.write('''\
 818 #!/bin/sh
 819 set -eu
 820 mkdir -p /run/systemd/network
 821 mkdir -p /run/systemd/netif
 822 mount -t tmpfs none /run/systemd/network
 823 mount -t tmpfs none /run/systemd/netif
 824 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 825 # create router/client veth pair
 826 cat << EOF > /run/systemd/network/test.netdev
 827 [NetDev]
 828 Name=%(ifr)s
 829 Kind=veth
 830
 831 [Peer]
 832 Name=%(ifc)s
 833 EOF
 834
 835 cat << EOF > /run/systemd/network/test.network
 836 [Match]
 837 Name=%(ifr)s
 838
 839 [Network]
 840 Address=192.168.5.1/24
 841 %(addr6)s
 842 DHCPServer=yes
 843
 844 [DHCPServer]
 845 PoolOffset=10
 846 PoolSize=50
 847 DNS=192.168.5.1
 848 %(dhopts)s
 849 EOF
 850
 851 # run networkd as in systemd-networkd.service
 852 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; s/^[@+-]//; s/^!*//; p}')
 853 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or '',
 854        'dhopts': dhcpserver_opts or ''})
 855
 856             os.fchmod(fd, 0o755)
 857
 858         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 859                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 860                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 861                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 862                                '--service-type=notify', script])
 863
 864         # wait until devices got created
 865         for _ in range(50):
 866             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 867             if b'state UP' in out and b'scope global' in out:
 868                 break
 869             time.sleep(0.1)
 870
 871     def shutdown_iface(self):
 872         '''Remove test interface and stop DHCP server'''
 873
 874         if self.if_router:
 875             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 876             # ensure failed transient unit does not stay around
 877             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 878             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 879             self.if_router = None
 880
 881     def print_server_log(self):
 882         '''Print DHCP server log for debugging failures'''
 883
 884         self.show_journal('networkd-test-router.service')
 885
 886     @unittest.skip('networkd does not have DHCPv6 server support')
 887     def test_hotplug_dhcp_ip6(self):
 888         pass
 889
 890     @unittest.skip('networkd does not have DHCPv6 server support')
 891     def test_coldplug_dhcp_ip6(self):
 892         pass
 893
 894     def test_search_domains(self):
 895
 896         # we don't use this interface for this test
 897         self.if_router = None
 898
 899         self.write_network('test.netdev', '''\
 900 [NetDev]
 901 Name=dummy0
 902 Kind=dummy
 903 MACAddress=12:34:56:78:9a:bc''')
 904         self.write_network('test.network', '''\
 905 [Match]
 906 Name=dummy0
 907 [Network]
 908 Address=192.168.42.100/24
 909 DNS=192.168.42.1
 910 Domains= one two three four five six seven eight nine ten''')
 911
 912         self.start_unit('systemd-networkd')
 913
 914         for timeout in range(50):
 915             with open(RESOLV_CONF) as f:
 916                 contents = f.read()
 917             if ' one' in contents:
 918                 break
 919             time.sleep(0.1)
 920         self.assertRegex(contents, 'search .*one two three four')
 921         self.assertNotIn('seven\n', contents)
 922         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 923
 924     def test_search_domains_too_long(self):
 925
 926         # we don't use this interface for this test
 927         self.if_router = None
 928
 929         name_prefix = 'a' * 60
 930
 931         self.write_network('test.netdev', '''\
 932 [NetDev]
 933 Name=dummy0
 934 Kind=dummy
 935 MACAddress=12:34:56:78:9a:bc''')
 936         self.write_network('test.network', '''\
 937 [Match]
 938 Name=dummy0
 939 [Network]
 940 Address=192.168.42.100/24
 941 DNS=192.168.42.1
 942 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 943
 944         self.start_unit('systemd-networkd')
 945
 946         for timeout in range(50):
 947             with open(RESOLV_CONF) as f:
 948                 contents = f.read()
 949             if ' one' in contents:
 950                 break
 951             time.sleep(0.1)
 952         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 953         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 954
 955     def test_dropin(self):
 956         # we don't use this interface for this test
 957         self.if_router = None
 958
 959         self.write_network('test.netdev', '''\
 960 [NetDev]
 961 Name=dummy0
 962 Kind=dummy
 963 MACAddress=12:34:56:78:9a:bc''')
 964         self.write_network('test.network', '''\
 965 [Match]
 966 Name=dummy0
 967 [Network]
 968 Address=192.168.42.100/24
 969 DNS=192.168.42.1''')
 970         self.write_network_dropin('test.network', 'dns', '''\
 971 [Network]
 972 DNS=127.0.0.1''')
 973
 974         self.start_unit('systemd-resolved')
 975         self.start_unit('systemd-networkd')
 976
 977         for timeout in range(50):
 978             with open(RESOLV_CONF) as f:
 979                 contents = f.read()
 980             if ' 127.0.0.1' in contents and '192.168.42.1' in contents:
 981                 break
 982             time.sleep(0.1)
 983         self.assertIn('nameserver 192.168.42.1\n', contents)
 984         self.assertIn('nameserver 127.0.0.1\n', contents)
 985
 986     def test_dhcp_timezone(self):
 987         '''networkd sets time zone from DHCP'''
 988
 989         def get_tz():
 990             out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
 991                                            '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
 992             assert out.startswith(b's "')
 993             out = out.strip()
 994             assert out.endswith(b'"')
 995             return out[3:-1].decode()
 996
 997         orig_timezone = get_tz()
 998         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
 999
1000         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
1001         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='ipv4')
1002
1003         # should have applied the received timezone
1004         try:
1005             self.assertEqual(get_tz(), 'Pacific/Honolulu')
1006         except AssertionError:
1007             self.show_journal('systemd-networkd.service')
1008             self.show_journal('systemd-hostnamed.service')
1009             raise
1010
1011
1012 class MatchClientTest(unittest.TestCase, NetworkdTestingUtilities):
1013     """Test [Match] sections in .network files.
1014
1015     Be aware that matching the test host's interfaces will wipe their
1016     configuration, so as a precaution, all network files should have a
1017     restrictive [Match] section to only ever interfere with the
1018     temporary veth interfaces created here.
1019     """
1020
1021     def tearDown(self):
1022         """Stop networkd."""
1023         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
1024
1025     def test_basic_matching(self):
1026         """Verify the Name= line works throughout this class."""
1027         self.add_veth_pair('test_if1', 'fake_if2')
1028         self.write_network('test.network', "[Match]\nName=test_*\n[Network]")
1029         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1030         self.assert_link_states(test_if1='managed', fake_if2='unmanaged')
1031
1032     def test_inverted_matching(self):
1033         """Verify that a '!'-prefixed value inverts the match."""
1034         # Use a MAC address as the interfaces' common matching attribute
1035         # to avoid depending on udev, to support testing in containers.
1036         mac = '00:01:02:03:98:99'
1037         self.add_veth_pair('test_veth', 'test_peer',
1038                            ['addr', mac], ['addr', mac])
1039         self.write_network('no-veth.network', """\
1040 [Match]
1041 MACAddress={}
1042 Name=!nonexistent *peer*
1043 [Network]""".format(mac))
1044         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1045         self.assert_link_states(test_veth='managed', test_peer='unmanaged')
1046
1047
1048 class UnmanagedClientTest(unittest.TestCase, NetworkdTestingUtilities):
1049     """Test if networkd manages the correct interfaces."""
1050
1051     def setUp(self):
1052         """Write .network files to match the named veth devices."""
1053         # Define the veth+peer pairs to be created.
1054         # Their pairing doesn't actually matter, only their names do.
1055         self.veths = {
1056             'm1def': 'm0unm',
1057             'm1man': 'm1unm',
1058         }
1059
1060         # Define the contents of .network files to be read in order.
1061         self.configs = (
1062             "[Match]\nName=m1def\n",
1063             "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
1064             "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
1065         )
1066
1067         # Write out the .network files to be cleaned up automatically.
1068         for i, config in enumerate(self.configs):
1069             self.write_network("%02d-test.network" % i, config)
1070
1071     def tearDown(self):
1072         """Stop networkd."""
1073         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
1074
1075     def create_iface(self):
1076         """Create temporary veth pairs for interface matching."""
1077         for veth, peer in self.veths.items():
1078             self.add_veth_pair(veth, peer)
1079
1080     def test_unmanaged_setting(self):
1081         """Verify link states with Unmanaged= settings, hot-plug."""
1082         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1083         self.create_iface()
1084         self.assert_link_states(m1def='managed',
1085                                 m1man='managed',
1086                                 m1unm='unmanaged',
1087                                 m0unm='unmanaged')
1088
1089     def test_unmanaged_setting_coldplug(self):
1090         """Verify link states with Unmanaged= settings, cold-plug."""
1091         self.create_iface()
1092         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1093         self.assert_link_states(m1def='managed',
1094                                 m1man='managed',
1095                                 m1unm='unmanaged',
1096                                 m0unm='unmanaged')
1097
1098     def test_catchall_config(self):
1099         """Verify link states with a catch-all config, hot-plug."""
1100         # Don't actually catch ALL interfaces.  It messes up the host.
1101         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1102         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1103         self.create_iface()
1104         self.assert_link_states(m1def='managed',
1105                                 m1man='managed',
1106                                 m1unm='unmanaged',
1107                                 m0unm='managed')
1108
1109     def test_catchall_config_coldplug(self):
1110         """Verify link states with a catch-all config, cold-plug."""
1111         # Don't actually catch ALL interfaces.  It messes up the host.
1112         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1113         self.create_iface()
1114         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1115         self.assert_link_states(m1def='managed',
1116                                 m1man='managed',
1117                                 m1unm='unmanaged',
1118                                 m0unm='managed')
1119
1120
1121 if __name__ == '__main__':
1122     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
1123                                                      verbosity=2))