diff options
Diffstat (limited to 'debian/tests/nm')
| -rwxr-xr-x | debian/tests/nm | 848 |
1 files changed, 848 insertions, 0 deletions
diff --git a/debian/tests/nm b/debian/tests/nm new file mode 100755 index 00000000..e44728f2 --- /dev/null +++ b/debian/tests/nm @@ -0,0 +1,848 @@ +#!/usr/bin/python3 +# Test NetworkManager on simulated network devices +# For an interactive shell test, run "nm ColdplugWifi.shell", see below + +__author__ = 'Martin Pitt <martin.pitt@ubuntu.com>' +__copyright__ = '(C) 2013 Canonical Ltd.' +__license__ = 'GPL v2 or later' + +import sys +import os +import os.path +import time +import subprocess +import socket +import unittest +import ctypes + +try: + from dbusmock import DBusTestCase +except ImportError: + DBusTestCase = object # dummy so that the class declaration works + +from gi.repository import NetworkManager, NMClient, GLib + +sys.path.append(os.path.dirname(__file__)) +import network_test_base + +SSID = 'fake net' + +# If True, NetworkManager logs directly to stdout, to watch logs in real time +NM_LOG_STDOUT = os.getenv('NM_LOG_STDOUT', False) + +# avoid accidentally destroying any real config +os.environ['GSETTINGS_BACKEND'] = 'memory' + +# we currently get a lot of WARNINGs/CRITICALs from GI (leaked objects from +# previous test runs/main loops?) Redirect them to stdout, to avoid failing +# autopkgtests +os.dup2(sys.stdout.fileno(), sys.stderr.fileno()) + + +class NetworkManagerTest(network_test_base.NetworkTestBase): + '''Provide common functionality for NM tests''' + + def start_nm(self, wait_iface=None, auto_connect=True): + '''Start NetworkManager and initialize client object + + If wait_iface is given, wait until NM recognizes that interface. + Otherwise, just wait until NM has initialized (for coldplug mode). + + If auto_connect is False, set the "no-auto-default=*" option to avoid + auto-connecting to wired devices. + ''' + # mount tmpfses over system directories, to avoid destroying the + # production configuration, and isolating tests from each other + if not os.path.exists('/run/NetworkManager'): + os.mkdir('/run/NetworkManager') + for d in ['/etc/NetworkManager', '/var/lib/NetworkManager', + '/run/NetworkManager']: + subprocess.check_call(['mount', '-n', '-t', 'tmpfs', 'none', d]) + self.addCleanup(subprocess.call, ['umount', d]) + os.mkdir('/etc/NetworkManager/system-connections') + + # create local configuration; this allows us to have full control, and + # we also need to blacklist the AP device so that NM does not tear it + # down; we also blacklist any existing real interface to avoid + # interfering with it, and for getting predictable results + blacklist = '' + for iface in os.listdir('/sys/class/net'): + if iface == "bonding_masters": + continue + if iface != self.dev_w_client and iface != self.dev_e_client: + with open('/sys/class/net/%s/address' % iface) as f: + if blacklist: + blacklist += ';' + blacklist += 'mac:%s' % f.read().strip() + + conf = os.path.join(self.workdir, 'NetworkManager.conf') + extra_main = '' + if not auto_connect: + extra_main += 'no-auto-default=*\n' + + with open(conf, 'w') as f: + f.write('[main]\nplugins=keyfile\n%s\n[keyfile]\nunmanaged-devices=%s\n' % + (extra_main, blacklist)) + + if NM_LOG_STDOUT: + f_log = None + else: + log = os.path.join(self.workdir, 'NetworkManager.log') + f_log = os.open(log, os.O_CREAT | os.O_WRONLY | os.O_SYNC) + + # build NM command line + argv = ['NetworkManager', '--log-level=debug', '--debug', '--config=' + conf] + # allow specifying extra arguments + argv += os.environ.get('NM_TEST_DAEMON_ARGS', '').strip().split() + + p = subprocess.Popen(argv, stdout=f_log, stderr=subprocess.STDOUT) + # automatically terminate process at end of test case + self.addCleanup(p.wait) + self.addCleanup(p.terminate) + self.addCleanup(self.shutdown_connections) + + if NM_LOG_STDOUT: + # let it initialize, then print a marker + time.sleep(1) + print('******* NM initialized *********\n\n') + else: + self.addCleanup(os.close, f_log) + + # this should be fast, give it 2 s to initialize + if wait_iface: + self.poll_text(log, 'manager: (%s): new' % wait_iface, timeout=100) + + self.nmclient = NMClient.Client.new() + self.assertTrue(self.nmclient.networking_get_enabled()) + + # FIXME: This certainly ought to be true, but isn't + #self.assertTrue(self.nmclient.get_manager_running()) + + # determine device objects + for d in self.nmclient.get_devices(): + if d.props.interface == self.dev_w_ap: + self.assertEqual(d.get_device_type(), NetworkManager.DeviceType.WIFI) + self.assertEqual(d.get_driver(), 'mac80211_hwsim') + self.assertEqual(d.get_hw_address(), self.mac_w_ap) + self.nmdev_w_ap = d + elif d.props.interface == self.dev_w_client: + self.assertEqual(d.get_device_type(), NetworkManager.DeviceType.WIFI) + self.assertEqual(d.get_driver(), 'mac80211_hwsim') + # NM ≥ 1.4 randomizes MAC addresses by default, so we can't + # test for equality, just make sure it's not our AP + self.assertNotEqual(d.get_hw_address(), self.mac_w_ap) + self.nmdev_w = d + elif d.props.interface == self.dev_e_client: + self.assertEqual(d.get_device_type(), NetworkManager.DeviceType.ETHERNET) + self.assertEqual(d.get_driver(), 'veth') + self.assertEqual(d.get_hw_address(), self.mac_e_client) + self.nmdev_e = d + + self.assertTrue(hasattr(self, 'nmdev_w_ap'), 'Could not determine wifi AP NM device') + self.assertTrue(hasattr(self, 'nmdev_w'), 'Could not determine wifi client NM device') + self.assertTrue(hasattr(self, 'nmdev_e'), 'Could not determine eth client NM device') + + self.process_glib_events() + + def shutdown_connections(self): + '''Shut down all active NM connections.''' + + if NM_LOG_STDOUT: + print('\n\n******* Shutting down NM connections *********') + + # remove all created connections + for active_conn in self.nmclient.get_active_connections(): + self.nmclient.deactivate_connection(active_conn) + self.assertEventually(lambda: self.nmclient.get_active_connections() == [], + timeout=20) + + # verify that NM properly deconfigures the devices + self.assert_iface_down(self.dev_w_client) + self.assert_iface_down(self.dev_e_client) + + @classmethod + def process_glib_events(klass): + '''Process pending GLib main loop events''' + + context = GLib.MainContext.default() + while context.iteration(False): + pass + + def assertEventually(self, condition, message=None, timeout=50): + '''Assert that condition function eventually returns True. + + timeout is in deciseconds, defaulting to 50 (5 seconds). message is + printed on failure. + ''' + while timeout >= 0: + self.process_glib_events() + if condition(): + break + timeout -= 1 + time.sleep(0.1) + else: + self.fail(message or 'timed out waiting for ' + str(condition)) + + def assert_iface_down(self, iface): + '''Assert that client interface is down''' + + out = subprocess.check_output(['ip', 'a', 'show', 'dev', iface], + universal_newlines=True) + self.assertIn('state DOWN', out) + self.assertNotIn('inet 192', out) + self.assertNotIn('inet6 2600', out) + + if iface == self.dev_w_client: + out = subprocess.check_output(['iw', 'dev', iface, 'link'], + universal_newlines=True) + self.assertIn('Not connected', out) + + # but AP device should never be touched by NM + out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.dev_w_ap], + universal_newlines=True) + self.assertIn('state UP', out) + + def assert_iface_up(self, iface, expected_ip_a=None, unexpected_ip_a=None): + '''Assert that client interface is up''' + + out = subprocess.check_output(['ip', 'a', 'show', 'dev', iface], + universal_newlines=True) + self.assertIn('state UP', out) + if expected_ip_a: + for r in expected_ip_a: + self.assertRegex(out, r) + if unexpected_ip_a: + for r in unexpected_ip_a: + self.assertNotRegex(out, r) + + if iface == self.dev_w_client: + out = subprocess.check_output(['iw', 'dev', iface, 'link'], + universal_newlines=True) + self.assertIn('Connected to ' + self.mac_w_ap, out) + self.assertIn('SSID: ' + SSID, out) + + def wait_ap(self, timeout): + '''Wait for AccessPoint NM object to appear, and return it''' + + self.assertEventually(lambda: len(self.nmdev_w.get_access_points()) > 0, + 'timed out waiting for AP to be detected', + timeout=timeout) + + return self.nmdev_w.get_access_points()[0] + + def connect_to_ap(self, ap, secret, ipv6_mode, ip6_privacy): + '''Connect to an NMAccessPoint. + + secret should be None for open networks, and a string with the password + for WEP/WPA. + + ip6_privacy is a NetworkManager.SettingIP6ConfigPrivacy flag. + + Return (NMConnection, NMActiveConnection) objects. + ''' + + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_DISABLED + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_IGNORE + if ipv6_mode is None: + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_AUTO + else: + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_AUTO + + # If we have a secret, supply it to the new connection right away; + # adding it afterwards with update_secrets() does not work, and we + # can't implement a SecretAgent as get_secrets() would need to build a + # map of a map of gpointers to gpointers which is too much for PyGI + partial_conn = NetworkManager.Connection.new() + partial_conn.add_setting(NetworkManager.SettingIP4Config(method=ip4_method)) + if secret: + partial_conn.add_setting(NetworkManager.SettingWirelessSecurity.new()) + # FIXME: needs update for other auth types + partial_conn.update_secrets(NetworkManager.SETTING_WIRELESS_SECURITY_SETTING_NAME, + {'psk': secret}) + if ip6_privacy is not None: + partial_conn.add_setting(NetworkManager.SettingIP6Config(ip6_privacy=ip6_privacy, + method=ip6_method)) + + ml = GLib.MainLoop() + self.cb_conn = None + + def add_activate_cb(client, conn, conn_path, error, data): + self.cb_conn = conn + self.cb_error = error + ml.quit() + self.nmclient.add_and_activate_connection(partial_conn, self.nmdev_w, ap.get_path(), add_activate_cb, None) + ml.run() + self.assertEqual(self.cb_error, None) + active_conn = self.cb_conn + self.cb_conn = None + + conn = self.conn_from_active_conn(active_conn) + self.assertTrue(conn.verify()) + + # verify need_secrets() + needed_secrets = conn.need_secrets() + if secret is None: + self.assertEqual(needed_secrets, (None, [])) + else: + self.assertEqual(needed_secrets[0], NetworkManager.SETTING_WIRELESS_SECURITY_SETTING_NAME) + self.assertEqual(type(needed_secrets[1]), list) + self.assertGreaterEqual(len(needed_secrets[1]), 1) + # FIXME: needs update for other auth types + self.assertIn(needed_secrets[1][0], [NetworkManager.SETTING_WIRELESS_SECURITY_PSK]) + + # we are usually ACTIVATING at this point; wait for completion + # TODO: 5s is not enough, argh slow DHCP client + self.assertEventually(lambda: active_conn.get_state() == NetworkManager.ActiveConnectionState.ACTIVATED, + 'timed out waiting for %s to get activated' % active_conn.get_connection(), + timeout=600) + self.assertEqual(self.nmdev_w.get_state(), NetworkManager.DeviceState.ACTIVATED) + return (conn, active_conn) + + def conn_from_active_conn(self, active_conn): + '''Get NMConnection object for an NMActiveConnection object''' + + # this sometimes takes a second try, when the corresponding + # NMConnection object is not yet available + tries = 3 + while tries > 0: + self.process_glib_events() + path = active_conn.get_connection() + for dev in active_conn.get_devices(): + for c in dev.get_available_connections(): + if c.get_path() == path: + return c + time.sleep(0.1) + tries -= 1 + + self.fail('Could not find NMConnection object for %s' % path) + + def check_connected_device_config(self, ipv6_mode, nmdev): + '''Check NMDevice configuration state after being connected''' + + time.sleep(10) + if ipv6_mode is not None: + # FIXME: why do we need to wait here, if state is already ACTIVATED? + self.assertEventually(lambda: nmdev.get_ip6_config() is not None, timeout=50) + #self.assertEqual(nmdev.get_ip4_config(), None) + conf = nmdev.get_ip6_config() + self.assertNotEqual(conf, None) + # we expect at least a link-local and a RA prefix or DHCP assigned + # address + addrs = conf.get_addresses() + self.assertGreaterEqual(len(addrs), 2, [a.get_address() for a in addrs]) + # note, we cannot call IP6Address.get_address(), as that returns a + # raw gpointer; check address with low-level tools only + else: + # FIXME: why do we need to wait here, if state is already ACTIVATED? + self.assertEventually(lambda: nmdev.get_ip4_config() is not None, timeout=50) + conf = nmdev.get_ip4_config() + self.assertNotEqual(conf, None) + self.assertEqual(len(conf.get_addresses()), 1) + self.assertEqual(socket.ntohl(conf.get_addresses()[0].get_address()) & 0xFFFFFF00, + 0xC0A80500) # 192.168.5.x + + def check_low_level_config(self, iface, ipv6_mode, ip6_privacy): + '''Check actual hardware state with ip/iw after being connected''' + + # list of expected regexps in "ip a" output + expected_ip_a = [] + unexpected_ip_a = [] + + if ipv6_mode is not None: + if ipv6_mode in ('', 'slaac'): + # has global address from our DHCP server + expected_ip_a.append('inet6 2600::[0-9a-f]+/') + else: + # has address with our prefix and MAC + expected_ip_a.append('inet6 2600::[0-9a-f:]+/64 scope global (?:tentative )?(?:mngtmpaddr )?(?:noprefixroute )?dynamic') + # has address with our prefix and random IP (Privacy + # Extension), if requested + priv_re = 'inet6 2600:[0-9a-f:]+/64 scope global temporary (?:tentative )?(?:mngtmpaddr )?dynamic' + if ip6_privacy in (NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR, + NetworkManager.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR): + expected_ip_a.append(priv_re) + else: + # FIXME: add a negative test here + pass + #unexpected_ip_a.append(priv_re) + + # has a link-local address + expected_ip_a.append('inet6 fe80::[0-9a-f:]+/64 scope link') + else: + expected_ip_a.append('inet 192.168.5.\d+/24') + + self.assert_iface_up(iface, expected_ip_a, unexpected_ip_a) + + +class ColdplugWifi(NetworkManagerTest): + '''Wifi: In these tests NM starts after setting up the AP''' + + # not run by default; run "nm-wifi ColdplugWifi.shell" to get this + @network_test_base.run_in_subprocess + def shell(self): + '''Start AP and NM, then run a shell (for debugging)''' + + self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + self.start_nm(self.dev_w_client) + print(''' + +client interface: %s, access point interface: %s, AP SSID: "%s" + +You can now run commands like "nmcli dev" or "nmcli dev wifi connect '%s'". +Logs are in '%s'. When done, exit the shell. + +''' % (self.dev_w_client, self.dev_w_ap, SSID, SSID, self.workdir)) + subprocess.call(['bash', '-i']) + + @network_test_base.run_in_subprocess + def test_no_ap(self): + '''no available access point''' + + self.start_nm(self.dev_w_client) + self.assertEventually(self.nmclient.networking_get_enabled, timeout=20) + + # state independent properties + self.assertEqual(self.nmdev_w.props.device_type, NetworkManager.DeviceType.WIFI) + self.assertTrue(self.nmdev_w.props.managed) + self.assertFalse(self.nmdev_w.props.firmware_missing) + self.assertTrue(self.nmdev_w.props.udi.startswith('/sys/devices/'), self.nmdev_w.props.udi) + + # get_version() plausibility check + out = subprocess.check_output(['nmcli', '--version'], universal_newlines=True) + cli_version = out.split()[-1] + self.assertTrue(cli_version[0].isdigit()) + self.assertEqual(self.nmclient.get_version(), cli_version) + + # state dependent properties (disconnected) + self.assertIn(self.nmdev_w.get_state(), + [NetworkManager.DeviceState.DISCONNECTED, NetworkManager.DeviceState.UNAVAILABLE]) + self.assertEqual(self.nmdev_w.get_access_points(), []) + self.assertEqual(self.nmdev_w.get_available_connections(), []) + + def test_open_b_ip4(self): + '''Open network, 802.11b, IPv4''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, None, 11000) + + def test_open_b_ip6_raonly_tmpaddr(self): + '''Open network, 802.11b, IPv6 with only RA, preferring temp address''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + + def test_open_b_ip6_raonly_pubaddr(self): + '''Open network, 802.11b, IPv6 with only RA, preferring public address''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR) + + def test_open_b_ip6_raonly_no_pe(self): + '''Open network, 802.11b, IPv6 with only RA, PE disabled''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.DISABLED) + + def test_open_b_ip6_dhcp(self): + '''Open network, 802.11b, IPv6 with DHCP, preferring temp address''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, '', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN) + + def test_open_g_ip4(self): + '''Open network, 802.11g, IPv4''' + + self.do_test('hw_mode=g\nchannel=1\nssid=' + SSID, None, 54000) + + def test_wpa1_ip4(self): + '''WPA1, 802.11g, IPv4''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=1 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=TKIP +wpa_passphrase=12345678 +''' % SSID, None, 54000, '12345678') + + def test_wpa2_ip4(self): + '''WPA2, 802.11g, IPv4''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=2 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=CCMP +wpa_passphrase=12345678 +''' % SSID, None, 54000, '12345678') + + def test_wpa2_ip6(self): + '''WPA2, 802.11g, IPv6 with only RA''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=2 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=CCMP +wpa_passphrase=12345678 +''' % SSID, 'ra-only', 54000, '12345678', + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + + @network_test_base.run_in_subprocess + def test_rfkill(self): + '''shut down connection on killswitch, restore it on unblock''' + + self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + self.start_nm(self.dev_w_client) + ap = self.wait_ap(timeout=1800) + (conn, active_conn) = self.connect_to_ap(ap, None, None, None) + + self.assertFalse(self.get_rfkill(self.dev_w_client)) + self.assertFalse(self.get_rfkill(self.dev_w_ap)) + + # now block the client interface + self.set_rfkill(self.dev_w_client, True) + # disabling should be fast, give it ten seconds + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.UNAVAILABLE, + timeout=100) + + # dev_w_client should be down now + self.assert_iface_down(self.dev_w_client) + + # turn it back on + self.set_rfkill(self.dev_w_client, False) + # this involves DHCP, use same timeout as for regular connection + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.ACTIVATED, + timeout=200) + + # dev_w_client should be back up + self.assert_iface_up(self.dev_w_client, ['inet 192.168.5.\d+/24']) + + # + # Common test code + # + + # libnm-glib has a lot of internal persistent state (private D-BUS + # connections and such); as it is very brittle and hard to track down + # all remaining references to any NM* object after a test, we rather + # run each test in a separate subprocess + @network_test_base.run_in_subprocess + def do_test(self, hostapd_conf, ipv6_mode, expected_max_bitrate, + secret=None, ip6_privacy=None): + '''Actual test code, parameterized for the particular test case''' + + self.setup_ap(hostapd_conf, ipv6_mode) + self.start_nm(self.dev_w_client) + + # on coldplug we expect the AP to be picked out fast + ap = self.wait_ap(timeout=100) + self.assertTrue(ap.get_path().startswith('/org/freedesktop/NetworkManager')) + self.assertEqual(ap.get_mode(), getattr(NetworkManager, '80211Mode').INFRA) + self.assertEqual(ap.get_max_bitrate(), expected_max_bitrate) + #self.assertEqual(ap.get_flags(), ) + + # should not auto-connect + self.assertEqual(self.nmclient.get_active_connections(), []) + + # connect to that AP + (conn, active_conn) = self.connect_to_ap(ap, secret, ipv6_mode, ip6_privacy) + + # check NMActiveConnection object + self.assertIn(active_conn.get_uuid(), [c.get_uuid() for c in self.nmclient.get_active_connections()]) + self.assertEqual([d.get_udi() for d in active_conn.get_devices()], [self.nmdev_w.get_udi()]) + + self.check_connected_device_config(ipv6_mode, self.nmdev_w) + + # check corresponding NMConnection object + wireless_setting = conn.get_setting_wireless() + self.assertEqual(wireless_setting.get_ssid(), SSID.encode()) + self.assertEqual(wireless_setting.get_hidden(), False) + if secret: + self.assertEqual(wireless_setting.get_security(), NetworkManager.SETTING_WIRELESS_SECURITY_SETTING_NAME) + else: + self.assertEqual(wireless_setting.get_security(), None) + # for debugging + #conn.dump() + + # for IPv6, check privacy setting + if ipv6_mode is not None and ip6_privacy != NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN: + assert ip6_privacy is not None, 'for IPv6 tests you need to specify ip6_privacy flag' + ip6_setting = conn.get_setting_ip6_config() + self.assertEqual(ip6_setting.props.ip6_privacy, ip6_privacy) + + self.check_low_level_config(self.dev_w_client, ipv6_mode, ip6_privacy) + + +class ColdplugEthernet(NetworkManagerTest): + '''Ethernet: In these tests NM starts after setting up the router''' + + # not run by default; run "nm-wifi ColdplugEthernet.shell" to get this + @network_test_base.run_in_subprocess + def shell(self): + '''Start router and NM, then run a shell (for debugging)''' + + self.setup_eth(None) + self.start_nm(self.dev_e_client) + print(''' + +client interface: %s, router interface: %s + +You can now run commands like "nmcli dev". +Logs are in '%s'. When done, exit the shell. + +''' % (self.dev_e_client, self.dev_e_ap, self.workdir)) + subprocess.call(['bash', '-i']) + + def test_auto_ip4(self): + '''ethernet: auto-connection, IPv4''' + + self.do_test(None, auto_connect=True) + + def test_auto_ip6_raonly_no_pe(self): + '''ethernet: auto-connection, IPv6 with only RA, PE disabled''' + + self.do_test('ra-only', auto_connect=True, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.DISABLED) + + def test_auto_ip6_dhcp(self): + '''ethernet: auto-connection, IPv6 with DHCP''' + + self.do_test('', auto_connect=True, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN) + + def test_manual_ip4(self): + '''ethernet: manual connection, IPv4''' + + self.do_test(None, auto_connect=False) + + def test_manual_ip6_raonly_tmpaddr(self): + '''ethernet: manual connection, IPv6 with only RA, preferring temp address''' + + self.do_test('ra-only', auto_connect=False, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + + def test_manual_ip6_raonly_pubaddr(self): + '''ethernet: manual connection, IPv6 with only RA, preferring public address''' + + self.do_test('ra-only', auto_connect=False, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR) + + # + # Common test code + # + + @network_test_base.run_in_subprocess + def do_test(self, ipv6_mode, ip6_privacy=None, auto_connect=True): + '''Actual test code, parameterized for the particular test case''' + + self.setup_eth(ipv6_mode) + self.start_nm(self.dev_e_client, auto_connect=auto_connect) + + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_DISABLED + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_IGNORE + if ipv6_mode is None: + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_AUTO + else: + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_AUTO + + if auto_connect: + # ethernet should auto-connect quickly without an existing defined connection + self.assertEventually(lambda: len(self.nmclient.get_active_connections()) > 0, + timeout=100) + active_conn = self.nmclient.get_active_connections()[0] + else: + # auto-connection was disabled, set up manual connection + partial_conn = NetworkManager.Connection.new() + partial_conn.add_setting(NetworkManager.SettingIP4Config(method=ip4_method)) + if ip6_privacy is not None: + partial_conn.add_setting(NetworkManager.SettingIP6Config(ip6_privacy=ip6_privacy, + method=ip6_method)) + + ml = GLib.MainLoop() + self.cb_conn = None + + def add_activate_cb(client, conn, conn_path, error, data): + self.cb_conn = conn + self.cb_error = error + ml.quit() + self.nmclient.add_and_activate_connection(partial_conn, self.nmdev_e, None, add_activate_cb, None) + ml.run() + self.assertEqual(self.cb_error, None) + active_conn = self.cb_conn + self.cb_conn = None + + # we are usually ACTIVATING at this point; wait for completion + # TODO: 5s is not enough, argh slow DHCP client + self.assertEventually(lambda: active_conn.get_state() == NetworkManager.ActiveConnectionState.ACTIVATED, + 'timed out waiting for %s to get activated' % active_conn.get_connection(), + timeout=150) + self.assertEqual(self.nmdev_e.get_state(), NetworkManager.DeviceState.ACTIVATED) + + conn = self.conn_from_active_conn(active_conn) + self.assertTrue(conn.verify()) + + # check NMActiveConnection object + self.assertIn(active_conn.get_uuid(), [c.get_uuid() for c in self.nmclient.get_active_connections()]) + self.assertEqual([d.get_udi() for d in active_conn.get_devices()], [self.nmdev_e.get_udi()]) + + self.check_connected_device_config(ipv6_mode, self.nmdev_e) + + # for IPv6, check privacy setting + if ipv6_mode is not None: + assert ip6_privacy is not None, 'for IPv6 tests you need to specify ip6_privacy flag' + if ip6_privacy not in (NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN, + NetworkManager.SettingIP6ConfigPrivacy.DISABLED): + ip6_setting = conn.get_setting_ip6_config() + self.assertEqual(ip6_setting.props.ip6_privacy, ip6_privacy) + + self.check_low_level_config(self.dev_e_client, ipv6_mode, ip6_privacy) + + +class Hotplug(NetworkManagerTest): + '''In these tests APs are set up while NM is already running''' + + @network_test_base.run_in_subprocess + @unittest.expectedFailure + def test_auto_detect_ap(self): + '''new AP is being detected automatically within 30s''' + + self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + self.start_nm() + ap = self.wait_ap(timeout=300) + # get_ssid returns a byte array + self.assertEqual(ap.get_ssid(), SSID.encode()) + self.assertEqual(self.nmdev_w.get_active_access_point(), None) + + @network_test_base.run_in_subprocess + @unittest.expectedFailure + def test_auto_detect_eth(self): + '''new eth router is being detected automatically within 30s''' + + self.start_nm() + self.setup_eth(None) + self.assertEventually(lambda: len(self.nmclient.get_active_connections()) > 0, + timeout=300) + active_conn = self.nmclient.get_active_connections()[0] + + self.assertEventually(lambda: active_conn.get_state() == NetworkManager.ActiveConnectionState.ACTIVATED, + 'timed out waiting for %s to get activated' % active_conn.get_connection(), + timeout=80) + self.assertEqual(self.nmdev_e.get_state(), NetworkManager.DeviceState.ACTIVATED) + + conn = self.conn_from_active_conn(active_conn) + self.assertTrue(conn.verify()) + + +@unittest.skipIf(DBusTestCase is object, + 'WARNING: python-dbusmock not installed, skipping suspend tests; get it from https://pypi.python.org/pypi/python-dbusmock') +class Suspend(NetworkManagerTest, DBusTestCase): + '''These tests run under a mock logind on a private system D-BUS''' + + @classmethod + def setUpClass(klass): + klass.start_system_bus() + NetworkManagerTest.setUpClass() + + @classmethod + def tearDownClass(klass): + NetworkManagerTest.tearDownClass() + DBusTestCase.tearDownClass() + + def setUp(self): + NetworkManagerTest.setUp(self) + + # start mock polkit and logind processes, so that we can + # intercept/control suspend + (p_polkit, self.obj_polkit) = self.spawn_server_template('polkitd', {}, stdout=subprocess.PIPE) + # by default we are not concerned about restricting access in the tests + self.obj_polkit.AllowUnknown(True) + self.addCleanup(p_polkit.wait) + self.addCleanup(p_polkit.terminate) + + (p_logind, self.obj_logind) = self.spawn_server_template('logind', {}, stdout=subprocess.PIPE) + self.addCleanup(p_logind.wait) + self.addCleanup(p_logind.terminate) + + # we have to manually start wpa_supplicant, as D-BUS activation does + # not happen for the fake D-BUS + log = os.path.join(self.workdir, 'wpasupplicant.log') + p_wpasupp = subprocess.Popen(['wpa_supplicant', '-u', '-d', '-e', '-K', + self.entropy_file, '-f', log]) + self.addCleanup(p_wpasupp.wait) + self.addCleanup(p_wpasupp.terminate) + + def fixme_test_active_ip4(self): + '''suspend during active IPv4 connection''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, None, + ['inet 192.168.5.\d+/24']) + + def fixme_test_active_ip6(self): + '''suspend during active IPv6 connection''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', + ['inet6 2600::']) + + # + # Common test code + # + + @network_test_base.run_in_subprocess + def do_test(self, hostapd_conf, ipv6_mode, expected_ip_a): + '''Actual test code, parameterized for the particular test case''' + + self.setup_ap(hostapd_conf, ipv6_mode) + self.start_nm(self.dev_w_client) + ap = self.wait_ap(timeout=1800) + (conn, active_conn) = self.connect_to_ap(ap, None, ipv6_mode, None) + + # send logind signal that we are about to suspend + self.obj_logind.EmitSignal('', 'PrepareForSleep', 'b', [True]) + + # disabling should be fast, give it one second + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.UNMANAGED, + timeout=10) + self.assert_iface_down(self.dev_w_client) + + # send logind signal that we resumed + self.obj_logind.EmitSignal('', 'PrepareForSleep', 'b', [False]) + + # this involves DHCP, use same timeout as for regular connection + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.ACTIVATED, + timeout=100) + + # dev_w_client should be back up + self.assert_iface_up(self.dev_w_client, expected_ip_a) + + +# avoid unintelligible error messages, and breaking "make check" when not being +# root +if os.getuid() != 0: + sys.stderr.write('This integration test suite needs to be run as root\n') + sys.exit(1) + +# AppArmor currently does not allow us to access the system D-BUS from an +# unshared file system. Hack the policy to allow that until that gets fixed +# properly. See https://launchpad.net/bugs/1244157 +subprocess.check_call("sed '/nm-dhcp-client.action {/ s/{/flags=(attach_disconnected) {/'" + " /etc/apparmor.d/sbin.dhclient > $ADTTMP/sbin.dhclient", + shell=True) +subprocess.check_call('apparmor_parser -Kr $ADTTMP/sbin.dhclient', shell=True) + +# unshare the mount namespace, so that our tmpfs mounts are guaranteed to get +# cleaned up, and don't influence the production system +libc6 = ctypes.cdll.LoadLibrary('libc.so.6') +assert libc6.unshare(ctypes.c_int(0x00020000)) == 0, 'failed to unshare mount namespace' + +# stop system-wide NetworkManager to avoid interfering with tests +nm_running = subprocess.call('service network-manager stop 2>&1', shell=True) == 0 + +# write to stdout, not stderr +runner = unittest.TextTestRunner(stream=sys.stdout, verbosity=2) +try: + unittest.main(testRunner=runner) +finally: + subprocess.call('dhclient eth0', shell=True) + subprocess.call('sleep 10', shell=True) |