diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/networkd-test.py | 119 |
1 files changed, 72 insertions, 47 deletions
diff --git a/test/networkd-test.py b/test/networkd-test.py index f8914a7895..4837c71652 100755 --- a/test/networkd-test.py +++ b/test/networkd-test.py @@ -30,6 +30,7 @@ # You should have received a copy of the GNU Lesser General Public License # along with systemd; If not, see <http://www.gnu.org/licenses/>. +import errno import os import sys import time @@ -39,16 +40,62 @@ import subprocess import shutil import socket -networkd_active = subprocess.call(['systemctl', 'is-active', '--quiet', - 'systemd-networkd']) == 0 -have_dnsmasq = shutil.which('dnsmasq') +HAVE_DNSMASQ = shutil.which('dnsmasq') is not None + +NETWORK_UNITDIR = '/run/systemd/network' + +NETWORKD_WAIT_ONLINE = shutil.which('systemd-networkd-wait-online', + path='/usr/lib/systemd:/lib/systemd') RESOLV_CONF = '/run/systemd/resolve/resolv.conf' -@unittest.skipIf(networkd_active, - 'networkd is already active') -class ClientTestBase: +def setUpModule(): + """Initialize the environment, and perform sanity checks on it.""" + if NETWORKD_WAIT_ONLINE is None: + raise OSError(errno.ENOENT, 'systemd-networkd-wait-online not found') + + # Do not run any tests if the system is using networkd already. + if subprocess.call(['systemctl', 'is-active', '--quiet', + 'systemd-networkd.service']) == 0: + raise unittest.SkipTest('networkd is already active') + + # Avoid "Failed to open /dev/tty" errors in containers. + os.environ['SYSTEMD_LOG_TARGET'] = 'journal' + + # Ensure the unit directory exists so tests can dump files into it. + os.makedirs(NETWORK_UNITDIR, exist_ok=True) + + +class NetworkdTestingUtilities: + """Provide a set of utility functions to facilitate networkd tests. + + This class must be inherited along with unittest.TestCase to define + some required methods. + """ + + def write_network(self, unit_name, contents): + """Write a network unit file, and queue it to be removed.""" + unit_path = os.path.join(NETWORK_UNITDIR, unit_name) + + with open(unit_path, 'w') as unit: + unit.write(contents) + self.addCleanup(os.remove, unit_path) + + def write_network_dropin(self, unit_name, dropin_name, contents): + """Write a network unit drop-in, and queue it to be removed.""" + dropin_dir = os.path.join(NETWORK_UNITDIR, "%s.d" % unit_name) + dropin_path = os.path.join(dropin_dir, "%s.conf" % dropin_name) + + os.makedirs(dropin_dir, exist_ok=True) + with open(dropin_path, 'w') as dropin: + dropin.write(contents) + self.addCleanup(os.remove, dropin_path) + + +class ClientTestBase(NetworkdTestingUtilities): + """Provide common methods for testing networkd against servers.""" + @classmethod def setUpClass(klass): klass.orig_log_level = subprocess.check_output( @@ -65,19 +112,7 @@ class ClientTestBase: self.if_router = 'router_eth42' self.workdir_obj = tempfile.TemporaryDirectory() self.workdir = self.workdir_obj.name - self.config = '/run/systemd/network/test_eth42.network' - - # avoid "Failed to open /dev/tty" errors in containers - os.environ['SYSTEMD_LOG_TARGET'] = 'journal' - - # determine path to systemd-networkd-wait-online - for p in ['/usr/lib/systemd/systemd-networkd-wait-online', - '/lib/systemd/systemd-networkd-wait-online']: - if os.path.exists(p): - self.networkd_wait_online = p - break - else: - self.fail('systemd-networkd-wait-online not found') + self.config = 'test_eth42.network' # get current journal cursor subprocess.check_output(['journalctl', '--sync']) @@ -93,12 +128,6 @@ class ClientTestBase: subprocess.call(['ip', 'link', 'del', 'dummy0'], stderr=subprocess.DEVNULL) - def writeConfig(self, fname, contents): - os.makedirs(os.path.dirname(fname), exist_ok=True) - with open(fname, 'w') as f: - f.write(contents) - self.addCleanup(os.remove, fname) - def show_journal(self, unit): '''Show journal of given unit since start of the test''' @@ -126,7 +155,7 @@ class ClientTestBase: def do_test(self, coldplug=True, ipv6=False, extra_opts='', online_timeout=10, dhcp_mode='yes'): subprocess.check_call(['systemctl', 'start', 'systemd-resolved']) - self.writeConfig(self.config, '''\ + self.write_network(self.config, '''\ [Match] Name=%s [Network] @@ -146,7 +175,7 @@ DHCP=%s subprocess.check_call(['systemctl', 'start', 'systemd-networkd']) try: - subprocess.check_call([self.networkd_wait_online, '--interface', + subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface', self.iface, '--timeout=%i' % online_timeout]) if ipv6: @@ -192,7 +221,7 @@ DHCP=%s self.assertRegex(out, b'DNS:\s+192.168.5.1') except (AssertionError, subprocess.CalledProcessError): # show networkd status, journal, and DHCP server log on failure - with open(self.config) as f: + with open(os.path.join(NETWORK_UNITDIR, self.config)) as f: print('\n---- %s ----\n%s' % (self.config, f.read())) print('---- interface status ----') sys.stdout.flush() @@ -247,12 +276,12 @@ DHCP=%s self.do_test(coldplug=False, ipv6=True) def test_route_only_dns(self): - self.writeConfig('/run/systemd/network/myvpn.netdev', '''\ + self.write_network('myvpn.netdev', '''\ [NetDev] Name=dummy0 Kind=dummy MACAddress=12:34:56:78:9a:bc''') - self.writeConfig('/run/systemd/network/myvpn.network', '''\ + self.write_network('myvpn.network', '''\ [Match] Name=dummy0 [Network] @@ -273,20 +302,16 @@ Domains= ~company''') self.assertNotIn('nameserver 192.168.42.1\n', contents) def test_route_only_dns_all_domains(self): - with open('/run/systemd/network/myvpn.netdev', 'w') as f: - f.write('''[NetDev] + self.write_network('myvpn.netdev', '''[NetDev] Name=dummy0 Kind=dummy MACAddress=12:34:56:78:9a:bc''') - with open('/run/systemd/network/myvpn.network', 'w') as f: - f.write('''[Match] + self.write_network('myvpn.network', '''[Match] Name=dummy0 [Network] Address=192.168.42.100 DNS=192.168.42.1 Domains= ~company ~.''') - self.addCleanup(os.remove, '/run/systemd/network/myvpn.netdev') - self.addCleanup(os.remove, '/run/systemd/network/myvpn.network') self.do_test(coldplug=True, ipv6=False, extra_opts='IPv6AcceptRouterAdvertisements=False') @@ -303,7 +328,7 @@ Domains= ~company ~.''') self.assertIn('nameserver 192.168.42.1\n', contents) -@unittest.skipUnless(have_dnsmasq, 'dnsmasq not installed') +@unittest.skipUnless(HAVE_DNSMASQ, 'dnsmasq not installed') class DnsmasqClientTest(ClientTestBase, unittest.TestCase): '''Test networkd client against dnsmasq''' @@ -366,7 +391,7 @@ class DnsmasqClientTest(ClientTestBase, unittest.TestCase): # create interface for generic connections; this will map all DNS names # to 192.168.42.1 self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1']) - self.writeConfig('/run/systemd/network/general.network', '''\ + self.write_network('general.network', '''\ [Match] Name=%s [Network] @@ -392,7 +417,7 @@ IPv6AcceptRA=False''' % self.iface) self.addCleanup(vpn_dnsmasq.wait) self.addCleanup(vpn_dnsmasq.kill) - self.writeConfig('/run/systemd/network/vpn.network', '''\ + self.write_network('vpn.network', '''\ [Match] Name=testvpnclient [Network] @@ -402,7 +427,7 @@ DNS=10.241.3.1 Domains= ~company ~lab''') subprocess.check_call(['systemctl', 'start', 'systemd-networkd']) - subprocess.check_call([self.networkd_wait_online, '--interface', self.iface, + subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface', self.iface, '--interface=testvpnclient', '--timeout=20']) # ensure we start fresh with every test @@ -583,12 +608,12 @@ exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=// # we don't use this interface for this test self.if_router = None - self.writeConfig('/run/systemd/network/test.netdev', '''\ + self.write_network('test.netdev', '''\ [NetDev] Name=dummy0 Kind=dummy MACAddress=12:34:56:78:9a:bc''') - self.writeConfig('/run/systemd/network/test.network', '''\ + self.write_network('test.network', '''\ [Match] Name=dummy0 [Network] @@ -615,12 +640,12 @@ Domains= one two three four five six seven eight nine ten''') name_prefix = 'a' * 60 - self.writeConfig('/run/systemd/network/test.netdev', '''\ + self.write_network('test.netdev', '''\ [NetDev] Name=dummy0 Kind=dummy MACAddress=12:34:56:78:9a:bc''') - self.writeConfig('/run/systemd/network/test.network', '''\ + self.write_network('test.network', '''\ [Match] Name=dummy0 [Network] @@ -643,18 +668,18 @@ Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix)) # we don't use this interface for this test self.if_router = None - self.writeConfig('/run/systemd/network/test.netdev', '''\ + self.write_network('test.netdev', '''\ [NetDev] Name=dummy0 Kind=dummy MACAddress=12:34:56:78:9a:bc''') - self.writeConfig('/run/systemd/network/test.network', '''\ + self.write_network('test.network', '''\ [Match] Name=dummy0 [Network] Address=192.168.42.100 DNS=192.168.42.1''') - self.writeConfig('/run/systemd/network/test.network.d/dns.conf', '''\ + self.write_network_dropin('test.network', 'dns', '''\ [Network] DNS=127.0.0.1''') |