108     def add_veth_pair(self, veth, peer, veth_options=(), peer_options=()):  argument
114         self.addCleanup(subprocess.call, ['ip', 'link', 'del', 'dev', peer])
116     def write_config(self, path, contents):  argument
122         self.addCleanup(os.remove, path)
124     def write_network(self, unit_name, contents):  argument
126         self.write_config(os.path.join(NETWORK_UNITDIR, unit_name), contents)
128     def write_network_dropin(self, unit_name, dropin_name, contents):  argument
134         self.addCleanup(os.rmdir, dropin_dir)
137         self.addCleanup(os.remove, dropin_path)
139     def read_attr(self, link, attribute):  argument
144         self.assert_link_states(**{link:'managed'})
148     def assert_link_states(self, **kwargs):  argument
179                     self.fail("Link {} expects state {}, found {}".format(iface, expected, actual))
184             self.fail("Missing links in status output: {}".format(interfaces))
190     def setUp(self):  argument
191         self.write_network('port1.netdev', '''\
197         self.write_network('port2.netdev', '''\
203         self.write_network('mybridge.netdev', '''\
208         self.write_network('port1.network', '''\
214         self.write_network('port2.network', '''\
220         self.write_network('mybridge.network', '''\
231     def tearDown(self):  argument
238     def test_bridge_init(self):  argument
239         self.assert_link_states(
244     def test_bridge_port_priority(self):  argument
245         self.assertEqual(self.read_attr('port1', 'brport/priority'), '32')
246         self.write_network_dropin('port1.network', 'priority', '''\
254         self.assertEqual(self.read_attr('port1', 'brport/priority'), '28')
256     def test_bridge_port_priority_set_zero(self):  argument
258         self.assertEqual(self.read_attr('port2', 'brport/priority'), '32')
259         self.write_network_dropin('port2.network', 'priority', '''\
267         self.assertEqual(self.read_attr('port2', 'brport/priority'), '0')
269     def test_bridge_port_property(self):  argument
271         self.assertEqual(self.read_attr('port2', 'brport/priority'), '32')
272         self.write_network_dropin('port2.network', 'property', '''\
288         self.assertEqual(self.read_attr('port2', 'brport/priority'), '23')
289         self.assertEqual(self.read_attr('port2', 'brport/hairpin_mode'), '1')
290         self.assertEqual(self.read_attr('port2', 'brport/isolated'), '1')
291         self.assertEqual(self.read_attr('port2', 'brport/path_cost'), '555')
292         self.assertEqual(self.read_attr('port2', 'brport/multicast_fast_leave'), '1')
293         self.assertEqual(self.read_attr('port2', 'brport/unicast_flood'), '1')
294         self.assertEqual(self.read_attr('port2', 'brport/bpdu_guard'), '1')
295         self.assertEqual(self.read_attr('port2', 'brport/root_block'), '1')
311     def setUp(self):  argument
312         self.iface = 'test_eth42'
313         self.if_router = 'router_eth42'
314         self.workdir_obj = tempfile.TemporaryDirectory()
315         self.workdir = self.workdir_obj.name
316         self.config = 'test_eth42.network'
323         self.assertTrue(out.startswith('-- cursor:'))
324         self.journal_cursor = out.split()[-1]
328     def tearDown(self):  argument
329         self.shutdown_iface()
335     def show_journal(self, unit):  argument
342                          '--cursor', self.journal_cursor, '-u', unit])
344     def create_iface(self, ipv6=False):  argument
349     def shutdown_iface(self):  argument
354     def print_server_log(self):  argument
359     def start_unit(self, unit):  argument
363             self.show_journal(unit)
366     def do_test(self, coldplug=True, ipv6=False, extra_opts='',  argument
368         self.start_unit('systemd-resolved')
369         self.write_network(self.config, '''\
375 '''.format(iface=self.iface, dhcp_mode=dhcp_mode, extra_opts=extra_opts))
379             self.create_iface(ipv6=ipv6)
380             self.start_unit('systemd-networkd')
383             self.start_unit('systemd-networkd')
384             self.create_iface(ipv6=ipv6)
387             self.start_unit('systemd-networkd')
391                                    self.iface, '--timeout=%i' % online_timeout])
398                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
403                     self.fail('timed out waiting for IPv6 configuration')
405                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
406                 self.assertRegex(out, b'inet6 fe80::.* scope link')
409                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
410                 self.assertRegex(out, br'inet6 fe80::.* scope link')
411                 self.assertNotIn(b'scope global', out)
414             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
415             self.assertIn(b'state UP', out)
416             self.assertRegex(out, br'inet 192.168.5.\d+/.* scope global dynamic')
420 …       self.assertRegex(out, (r'{}\s+ether\s+[a-z-]+\s+unmanaged'.format(self.if_router)).encode())
421 …         self.assertRegex(out, (r'{}\s+ether\s+routable\s+configured'.format(self.iface)).encode())
423             out = subprocess.check_output(['networkctl', '-n', '0', 'status', self.iface])
424             self.assertRegex(out, br'Type:\s+ether')
425             self.assertRegex(out, br'State:\s+routable.*configured')
426             self.assertRegex(out, br'Online state:\s+online')
427             self.assertRegex(out, br'Address:\s+192.168.5.\d+')
429                 self.assertRegex(out, br'2600::')
431                 self.assertNotIn(br'2600::', out)
432             self.assertRegex(out, br'fe80::')
433             self.assertRegex(out, br'Gateway:\s+192.168.5.1')
434             self.assertRegex(out, br'DNS:\s+192.168.5.1')
437             with open(os.path.join(NETWORK_UNITDIR, self.config)) as f:
438                 print('\n---- {} ----\n{}'.format(self.config, f.read()))
441             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
442             print('---- networkctl status {} ----'.format(self.iface))
444             rc = subprocess.call(['networkctl', '-n', '0', 'status', self.iface])
447             self.show_journal('systemd-networkd.service')
448             self.print_server_log()
458             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
462             self.shutdown_iface()
464     def test_coldplug_dhcp_yes_ip4(self):  argument
466         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
468     def test_coldplug_dhcp_yes_ip4_no_ra(self):  argument
470         self.do_test(coldplug=True, ipv6=False,
473     def test_coldplug_dhcp_ip4_only(self):  argument
475         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
478     def test_coldplug_dhcp_ip4_only_no_ra(self):  argument
480         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
483     def test_coldplug_dhcp_ip6(self):  argument
484         self.do_test(coldplug=True, ipv6=True)
486     def test_hotplug_dhcp_ip4(self):  argument
488         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
490     def test_hotplug_dhcp_ip6(self):  argument
491         self.do_test(coldplug=False, ipv6=True)
493     def test_route_only_dns(self):  argument
494         self.write_network('myvpn.netdev', '''\
500         self.write_network('myvpn.network', '''\
510             self.do_test(coldplug=True, ipv6=False,
522             self.assertNotRegex(contents, 'search.*company')
524             self.assertIn('nameserver 192.168.5.1\n', contents)
526             self.assertNotIn('nameserver 192.168.42.1\n', contents)
528     def test_route_only_dns_all_domains(self):  argument
529         self.write_network('myvpn.netdev', '''[NetDev]
534         self.write_network('myvpn.network', '''[Match]
543             self.do_test(coldplug=True, ipv6=False,
556         self.assertNotRegex(contents, 'search.*company')
559         self.assertIn('nameserver 192.168.5.1\n', contents)
561         self.assertIn('nameserver 192.168.42.1\n', contents)
568     def setUp(self):  argument
570         self.dnsmasq = None
571         self.iface_mac = 'de:ad:be:ef:47:11'
573     def create_iface(self, ipv6=False, dnsmasq_opts=None):  argument
577         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface,
578                                'address', self.iface_mac,
579                                'type', 'veth', 'peer', 'name', self.if_router])
582         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
583         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
585             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
586         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
589         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
590         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
597         self.dnsmasq = subprocess.Popen(
599              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
601              '--interface=' + self.if_router, '--except-interface=lo',
604     def shutdown_iface(self):  argument
607         if self.if_router:
608             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
609             self.if_router = None
610         if self.dnsmasq:
611             self.dnsmasq.kill()
612             self.dnsmasq.wait()
613             self.dnsmasq = None
615     def print_server_log(self):  argument
618         with open(self.dnsmasq_log) as f:
621     def test_resolved_domain_restricted_dns(self):  argument
629         self.addCleanup(os.remove, conf)
633         self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1'])
634         self.write_network('general.network', '''\
641 '''.format(self.iface))
645         self.add_veth_pair('testvpnclient', 'testvpnrouter')
650         vpn_dnsmasq_log = os.path.join(self.workdir, 'dnsmasq-vpn.log')
657         self.addCleanup(vpn_dnsmasq.wait)
658         self.addCleanup(vpn_dnsmasq.kill)
660         self.write_network('vpn.network', '''\
671         self.start_unit('systemd-networkd')
672         subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface', self.iface,
681         self.assertIn(b'math.lab: 10.241.3.3', out)
683         self.assertIn(b'kettle.cantina.company: 10.241.4.4', out)
687         self.assertIn(b'search.example.com: 192.168.42.1', out)
689         with open(self.dnsmasq_log) as f:
695         self.assertRegex(vpn_log, 'query.*math.lab')
696         self.assertRegex(vpn_log, 'query.*cantina.company')
697         self.assertNotIn('.lab', general_log)
698         self.assertNotIn('.company', general_log)
701         self.assertRegex(general_log, 'query.*search.example.com')
702         self.assertNotIn('search.example.com', vpn_log)
704     def test_resolved_etc_hosts(self):  argument
712         self.addCleanup(os.remove, conf)
719         self.addCleanup(os.remove, negative)
722         hosts = os.path.join(self.workdir, 'hosts')
726         self.addCleanup(subprocess.call, ['umount', '/etc/hosts'])
731         self.create_iface(dnsmasq_opts=['--host-record=my.example.com,172.16.99.1,2600::99:99',
735         self.do_test(coldplug=None, ipv6=True)
740             self.assertIn(b'my.example.com: 172.16.99.99', out)
743             self.assertNotEqual(subprocess.call(['resolvectl', 'query', '-6', 'my.example.com']), 0)
746             self.assertIn(b'my.example.com: 172.16.99.99', out)
749             self.assertIn(b'172.16.99.99: my.example.com', out)
753             self.assertIn(b'example.com IN MX 1 mail.example.com', out)
757             self.assertIn(b'172.16.0.42', out)
759             self.assertIn(b'172.16.0.42: other.example.com', out)
761             self.show_journal('systemd-resolved.service')
762             self.print_server_log()
765     def test_transient_hostname(self):  argument
769         self.addCleanup(socket.sethostname, orig_hostname)
773             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
775         self.addCleanup(subprocess.call, ['systemctl', 'stop', 'systemd-hostnamed.service'])
777 …  self.create_iface(dnsmasq_opts=['--dhcp-host={},192.168.5.210,testgreen'.format(self.iface_mac)])
778         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
782             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
783             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
794                 self.fail('Transient hostname not found in hostnamectl:\n{}'.format(out.decode()))
796             self.assertEqual(socket.gethostname(), 'testgreen')
798             self.show_journal('systemd-networkd.service')
799             self.show_journal('systemd-hostnamed.service')
800             self.print_server_log()
803     def test_transient_hostname_with_static(self):  argument
807         self.addCleanup(socket.sethostname, orig_hostname)
810             self.write_config('/etc/hostname', "foobarqux")
812             self.write_config('/run/hostname.tmp', "foobarqux")
814             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
819         self.addCleanup(subprocess.call, ['systemctl', 'stop', 'systemd-hostnamed.service'])
821 …  self.create_iface(dnsmasq_opts=['--dhcp-host={},192.168.5.210,testgreen'.format(self.iface_mac)])
822         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
826             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
827             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
829             self.assertEqual(socket.gethostname(), "foobarqux")
831             self.show_journal('systemd-networkd.service')
832             self.show_journal('systemd-hostnamed.service')
833             self.print_server_log()
840     def setUp(self):  argument
842         self.dnsmasq = None
844     def create_iface(self, ipv6=False, dhcpserver_opts=None):  argument
850         self.addCleanup(os.remove, script)
888 '''.format(ifr=self.if_router,
889            ifc=self.iface,
903             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
908     def shutdown_iface(self):  argument
911         if self.if_router:
915             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
916             self.if_router = None
918     def print_server_log(self):  argument
921         self.show_journal('networkd-test-router.service')
924     def test_hotplug_dhcp_ip6(self):  argument
928     def test_coldplug_dhcp_ip6(self):  argument
931     def test_search_domains(self):  argument
934         self.if_router = None
936         self.write_network('test.netdev', '''\
942         self.write_network('test.network', '''\
951         self.start_unit('systemd-networkd')
959         self.assertRegex(contents, 'search .*one two three four five six seven eight nine ten')
961     def test_dropin(self):  argument
963         self.if_router = None
965         self.write_network('test.netdev', '''\
971         self.write_network('test.network', '''\
978         self.write_network_dropin('test.network', 'dns', '''\
983         self.start_unit('systemd-resolved')
984         self.start_unit('systemd-networkd')
992         self.assertIn('nameserver 192.168.42.1\n', contents)
993         self.assertIn('nameserver 127.0.0.1\n', contents)
995     def test_dhcp_timezone(self):  argument
1007         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
1009         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
1010 …self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='…
1014             self.assertEqual(get_tz(), 'Pacific/Honolulu')
1016             self.show_journal('systemd-networkd.service')
1017             self.show_journal('systemd-hostnamed.service')
1030     def tearDown(self):  argument
1035     def test_basic_matching(self):  argument
1037         self.add_veth_pair('test_if1', 'fake_if2')
1038         self.write_network('test.network', "[Match]\nName=test_*\n[Network]")
1040         self.assert_link_states(test_if1='managed', fake_if2='unmanaged')
1042     def test_inverted_matching(self):  argument
1047         self.add_veth_pair('test_veth', 'test_peer',
1049         self.write_network('no-veth.network', """\
1055         self.assert_link_states(test_veth='managed', test_peer='unmanaged')
1061     def setUp(self):  argument
1065         self.veths = {
1071         self.configs = (
1078         for i, config in enumerate(self.configs):
1079             self.write_network("%02d-test.network" % i, config)
1081     def tearDown(self):  argument
1086     def create_iface(self):  argument
1088         for veth, peer in self.veths.items():
1089             self.add_veth_pair(veth, peer)
1091     def test_unmanaged_setting(self):  argument
1094         self.create_iface()
1095         self.assert_link_states(m1def='managed',
1100     def test_unmanaged_setting_coldplug(self):  argument
1102         self.create_iface()
1104         self.assert_link_states(m1def='managed',
1109     def test_catchall_config(self):  argument
1112         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1114         self.create_iface()
1115         self.assert_link_states(m1def='managed',
1120     def test_catchall_config_coldplug(self):  argument
1123         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1124         self.create_iface()
1126         self.assert_link_states(m1def='managed',