summaryrefslogtreecommitdiff
path: root/test/networkd-test.py
diff options
context:
space:
mode:
authorDavid Michael <david.michael@coreos.com>2016-09-27 15:18:14 -0700
committerDavid Michael <david.michael@coreos.com>2016-12-01 14:41:51 -0800
commita09dc5467a3d289a53ef3ea87d3b155d8d0551c9 (patch)
tree85a69dde6004c10cc899c9010dc4e649509c1cfc /test/networkd-test.py
parentec89276c2ab345b84c2dab4c35826de41aa6fd0f (diff)
networkd: support marking links unmanaged
Diffstat (limited to 'test/networkd-test.py')
-rwxr-xr-xtest/networkd-test.py116
1 files changed, 116 insertions, 0 deletions
diff --git a/test/networkd-test.py b/test/networkd-test.py
index 4837c71652..aed5139275 100755
--- a/test/networkd-test.py
+++ b/test/networkd-test.py
@@ -92,6 +92,45 @@ class NetworkdTestingUtilities:
dropin.write(contents)
self.addCleanup(os.remove, dropin_path)
+ def assert_link_states(self, **kwargs):
+ """Match networkctl link states to the given ones.
+
+ Each keyword argument should be the name of a network interface
+ with its expected value of the "SETUP" column in output from
+ networkctl. The interfaces have five seconds to come online
+ before the check is performed. Every specified interface must
+ be present in the output, and any other interfaces found in the
+ output are ignored.
+
+ A special interface state "managed" is supported, which matches
+ any value in the "SETUP" column other than "unmanaged".
+ """
+ if not kwargs:
+ return
+ interfaces = set(kwargs)
+
+ # Wait for the requested interfaces, but don't fail for them.
+ subprocess.call([NETWORKD_WAIT_ONLINE, '--timeout=5'] +
+ ['--interface=%s' % iface for iface in kwargs])
+
+ # Validate each link state found in the networkctl output.
+ out = subprocess.check_output(['networkctl', '--no-legend']).rstrip()
+ for line in out.decode('utf-8').split('\n'):
+ fields = line.split()
+ if len(fields) >= 5 and fields[1] in kwargs:
+ iface = fields[1]
+ expected = kwargs[iface]
+ actual = fields[-1]
+ if (actual != expected and
+ not (expected == 'managed' and actual != 'unmanaged')):
+ self.fail("Link %s expects state %s, found %s" %
+ (iface, expected, actual))
+ interfaces.remove(iface)
+
+ # Ensure that all requested interfaces have been covered.
+ if interfaces:
+ self.fail("Missing links in status output: %s" % interfaces)
+
class ClientTestBase(NetworkdTestingUtilities):
"""Provide common methods for testing networkd against servers."""
@@ -720,6 +759,83 @@ DNS=127.0.0.1''')
raise
+class UnmanagedClientTest(unittest.TestCase, NetworkdTestingUtilities):
+ """Test if networkd manages the correct interfaces."""
+
+ def setUp(self):
+ """Write .network files to match the named veth devices."""
+ # Define the veth+peer pairs to be created.
+ # Their pairing doesn't actually matter, only their names do.
+ self.veths = {
+ 'm1def': 'm0unm',
+ 'm1man': 'm1unm',
+ }
+
+ # Define the contents of .network files to be read in order.
+ self.configs = (
+ "[Match]\nName=m1def\n",
+ "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
+ "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
+ )
+
+ # Write out the .network files to be cleaned up automatically.
+ for i, config in enumerate(self.configs):
+ self.write_network("%02d-test.network" % i, config)
+
+ def tearDown(self):
+ """Stop networkd."""
+ subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
+
+ def create_iface(self):
+ """Create temporary veth pairs for interface matching."""
+ for veth, peer in self.veths.items():
+ subprocess.check_call(['ip', 'link', 'add',
+ 'name', veth, 'type', 'veth',
+ 'peer', 'name', peer])
+ self.addCleanup(subprocess.call,
+ ['ip', 'link', 'del', 'dev', peer])
+
+ def test_unmanaged_setting(self):
+ """Verify link states with Unmanaged= settings, hot-plug."""
+ subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
+ self.create_iface()
+ self.assert_link_states(m1def='managed',
+ m1man='managed',
+ m1unm='unmanaged',
+ m0unm='unmanaged')
+
+ def test_unmanaged_setting_coldplug(self):
+ """Verify link states with Unmanaged= settings, cold-plug."""
+ self.create_iface()
+ subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
+ self.assert_link_states(m1def='managed',
+ m1man='managed',
+ m1unm='unmanaged',
+ m0unm='unmanaged')
+
+ def test_catchall_config(self):
+ """Verify link states with a catch-all config, hot-plug."""
+ # Don't actually catch ALL interfaces. It messes up the host.
+ self.write_network('all.network', "[Match]\nName=m[01]???\n")
+ subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
+ self.create_iface()
+ self.assert_link_states(m1def='managed',
+ m1man='managed',
+ m1unm='unmanaged',
+ m0unm='managed')
+
+ def test_catchall_config_coldplug(self):
+ """Verify link states with a catch-all config, cold-plug."""
+ # Don't actually catch ALL interfaces. It messes up the host.
+ self.write_network('all.network', "[Match]\nName=m[01]???\n")
+ self.create_iface()
+ subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
+ self.assert_link_states(m1def='managed',
+ m1man='managed',
+ m1unm='unmanaged',
+ m0unm='managed')
+
+
if __name__ == '__main__':
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
verbosity=2))