diff options
Diffstat (limited to 'debian/tests')
| -rw-r--r-- | debian/tests/network_test_base.py | 302 | ||||
| -rwxr-xr-x | debian/tests/nm.py | 630 |
2 files changed, 568 insertions, 364 deletions
diff --git a/debian/tests/network_test_base.py b/debian/tests/network_test_base.py index ffbd60df..2adbbd4b 100644 --- a/debian/tests/network_test_base.py +++ b/debian/tests/network_test_base.py @@ -1,14 +1,14 @@ -''' +""" Base class for network related tests. This provides fake wifi devices with mac80211_hwsim and hostapd, test ethernet devices with veth, utility functions to start wpasupplicant, dnsmasq, get/set rfkill status, and some utility functions. -''' +""" -__author__ = 'Martin Pitt <martin.pitt@ubuntu.com>' -__copyright__ = '(C) 2013 Canonical Ltd.' -__license__ = 'GPL v2 or later' +__author__ = "Martin Pitt <martin.pitt@ubuntu.com>" +__copyright__ = "(C) 2013 Canonical Ltd." +__license__ = "GPL v2 or later" import sys import os @@ -24,14 +24,17 @@ from glob import glob # check availability of programs, and cleanly skip test if they are not # available -for program in ['wpa_supplicant', 'hostapd', 'dnsmasq', 'dhclient']: - if subprocess.call(['which', program], stdout=subprocess.PIPE) != 0: - sys.stderr.write('%s is required for this test suite, but not available. Skipping\n' % program) +for program in ["wpa_supplicant", "hostapd", "dnsmasq", "dhclient"]: + if subprocess.call(["which", program], stdout=subprocess.PIPE) != 0: + sys.stderr.write( + "%s is required for this test suite, but not available. Skipping\n" + % program + ) sys.exit(0) class NetworkTestBase(unittest.TestCase): - '''Common functionality for network test cases + """Common functionality for network test cases setUp() creates two test wlan devices, one for a simulated access point (self.dev_w_ap), the other for a simulated client device @@ -40,59 +43,79 @@ class NetworkTestBase(unittest.TestCase): Each test should call self.setup_ap() or self.setup_eth() with the desired configuration. - ''' + """ + @classmethod def setUpClass(klass): # ensure we have this so that iw works - subprocess.check_call(['modprobe', 'cfg80211']) + subprocess.check_call(["modprobe", "cfg80211"]) # set regulatory domain "EU", so that we can use 80211.a 5 GHz channels - out = subprocess.check_output(['iw', 'reg', 'get'], universal_newlines=True) - m = re.match('^(?:global\n)?country (\S+):', out) + out = subprocess.check_output(["iw", "reg", "get"], universal_newlines=True) + m = re.match("^(?:global\n)?country (\S+):", out) assert m klass.orig_country = m.group(1) - subprocess.check_call(['iw', 'reg', 'set', 'EU']) + subprocess.check_call(["iw", "reg", "set", "EU"]) @classmethod def tearDownClass(klass): - subprocess.check_call(['iw', 'reg', 'set', klass.orig_country]) - os.remove('/run/udev/rules.d/99-nm-veth-test.rules') + subprocess.check_call(["iw", "reg", "set", klass.orig_country]) + os.remove("/run/udev/rules.d/99-nm-veth-test.rules") @classmethod def create_devices(klass): - '''Create Access Point and Client devices with mac80211_hwsim and veth''' + """Create Access Point and Client devices with mac80211_hwsim and veth""" - klass.dev_e_ap = 'veth42' - klass.dev_e_client = 'eth42' + klass.dev_e_ap = "veth42" + klass.dev_e_client = "eth42" - if os.path.exists('/sys/module/mac80211_hwsim'): - raise SystemError('mac80211_hwsim module already loaded') - if os.path.exists('/sys/class/net/' + klass.dev_e_client): - raise SystemError('%s interface already exists' % klass.dev_e_client) + if os.path.exists("/sys/module/mac80211_hwsim"): + raise SystemError("mac80211_hwsim module already loaded") + if os.path.exists("/sys/class/net/" + klass.dev_e_client): + raise SystemError("%s interface already exists" % klass.dev_e_client) # ensure NM can manage our fake eths - os.makedirs('/run/udev/rules.d', exist_ok=True) - with open('/run/udev/rules.d/99-nm-veth-test.rules', 'w') as f: - f.write('ENV{ID_NET_DRIVER}=="veth", ENV{INTERFACE}=="%s", ENV{NM_UNMANAGED}="0"\n' % klass.dev_e_client) - subprocess.check_call(['udevadm', 'control', '--reload']) + os.makedirs("/run/udev/rules.d", exist_ok=True) + with open("/run/udev/rules.d/99-nm-veth-test.rules", "w") as f: + f.write( + 'ENV{ID_NET_DRIVER}=="veth", ENV{INTERFACE}=="%s", ENV{NM_UNMANAGED}="0"\n' + % klass.dev_e_client + ) + subprocess.check_call(["udevadm", "control", "--reload"]) # create virtual ethernet devs - subprocess.check_call(['ip', 'link', 'add', 'name', klass.dev_e_client, 'type', - 'veth', 'peer', 'name', klass.dev_e_ap]) + subprocess.check_call( + [ + "ip", + "link", + "add", + "name", + klass.dev_e_client, + "type", + "veth", + "peer", + "name", + klass.dev_e_ap, + ] + ) # create virtual wlan devs - before_wlan = set([c for c in os.listdir('/sys/class/net') if c.startswith('wlan')]) - subprocess.check_call(['modprobe', 'mac80211_hwsim']) + before_wlan = set( + [c for c in os.listdir("/sys/class/net") if c.startswith("wlan")] + ) + subprocess.check_call(["modprobe", "mac80211_hwsim"]) # wait 5 seconds for fake devices to appear timeout = 50 while timeout > 0: - after_wlan = set([c for c in os.listdir('/sys/class/net') if c.startswith('wlan')]) + after_wlan = set( + [c for c in os.listdir("/sys/class/net") if c.startswith("wlan")] + ) if len(after_wlan) - len(before_wlan) >= 2: break timeout -= 1 time.sleep(0.1) else: - raise SystemError('timed out waiting for fake devices to appear') + raise SystemError("timed out waiting for fake devices to appear") devs = list(after_wlan - before_wlan) klass.dev_w_ap = devs[0] @@ -104,22 +127,22 @@ class NetworkTestBase(unittest.TestCase): # was created and networkd took control. Give it some time, so we read # the correct MAC address time.sleep(0.1) - with open('/sys/class/net/%s/address' % klass.dev_w_ap) as f: + with open("/sys/class/net/%s/address" % klass.dev_w_ap) as f: klass.mac_w_ap = f.read().strip().upper() - with open('/sys/class/net/%s/address' % klass.dev_w_client) as f: + with open("/sys/class/net/%s/address" % klass.dev_w_client) as f: klass.mac_w_client = f.read().strip().upper() - with open('/sys/class/net/%s/address' % klass.dev_e_ap) as f: + with open("/sys/class/net/%s/address" % klass.dev_e_ap) as f: klass.mac_e_ap = f.read().strip().upper() - with open('/sys/class/net/%s/address' % klass.dev_e_client) as f: + with open("/sys/class/net/%s/address" % klass.dev_e_client) as f: klass.mac_e_client = f.read().strip().upper() - #print('Created fake devices: AP: %s, client: %s' % (klass.dev_w_ap, klass.dev_w_client)) + # print('Created fake devices: AP: %s, client: %s' % (klass.dev_w_ap, klass.dev_w_client)) @classmethod def shutdown_devices(klass): - '''Remove test wlan devices''' + """Remove test wlan devices""" - subprocess.check_call(['rmmod', 'mac80211_hwsim']) - subprocess.check_call(['ip', 'link', 'del', 'dev', klass.dev_e_ap]) + subprocess.check_call(["rmmod", "mac80211_hwsim"]) + subprocess.check_call(["ip", "link", "del", "dev", klass.dev_e_ap]) klass.dev_w_ap = None klass.dev_w_client = None klass.dev_e_ap = None @@ -127,45 +150,47 @@ class NetworkTestBase(unittest.TestCase): @classmethod def get_rfkill(klass, interface): - '''Get rfkill status of an interface. + """Get rfkill status of an interface. Returns whether the interface is blocked, i. e. "True" for blocked, "False" for enabled. - ''' + """ with open(klass._rfkill_attribute(interface)) as f: val = f.read() - return val == '1' + return val == "1" @classmethod def set_rfkill(klass, interface, block): - '''Set rfkill status of an interface + """Set rfkill status of an interface Use block==True for disabling ("killswitching") an interface, block==False to re-enable. - ''' - with open(klass._rfkill_attribute(interface), 'w') as f: - f.write(block and '1' or '0') + """ + with open(klass._rfkill_attribute(interface), "w") as f: + f.write(block and "1" or "0") def run(self, result=None): - '''Show log files on failed tests''' + """Show log files on failed tests""" if result: orig_err_fail = len(result.errors) + len(result.failures) super().run(result) - if hasattr(self, 'workdir'): - logs = glob(os.path.join(self.workdir, '*.log')) + if hasattr(self, "workdir"): + logs = glob(os.path.join(self.workdir, "*.log")) if result and len(result.errors) + len(result.failures) > orig_err_fail: for log_file in logs: with open(log_file) as f: - print('\n----- %s -----\n%s\n------\n' - % (os.path.basename(log_file), f.read())) + print( + "\n----- %s -----\n%s\n------\n" + % (os.path.basename(log_file), f.read()) + ) # clean up log files, so that we don't see ones from previous tests for log_file in logs: os.unlink(log_file) def setUp(self): - '''Create test devices and workdir''' + """Create test devices and workdir""" self.create_devices() self.addCleanup(self.shutdown_devices) @@ -173,74 +198,94 @@ class NetworkTestBase(unittest.TestCase): self.workdir = self.workdir_obj.name # create static entropy file to avoid draining/blocking on /dev/random - self.entropy_file = os.path.join(self.workdir, 'entropy') - with open(self.entropy_file, 'wb') as f: - f.write(b'012345678901234567890') + self.entropy_file = os.path.join(self.workdir, "entropy") + with open(self.entropy_file, "wb") as f: + f.write(b"012345678901234567890") def setup_ap(self, hostapd_conf, ipv6_mode): - '''Set up simulated access point + """Set up simulated access point On self.dev_w_ap, run hostapd with given configuration. Setup dnsmasq according to ipv6_mode, see start_dnsmasq(). This is torn down automatically at the end of the test. - ''' + """ # give our AP an IP - subprocess.check_call(['ip', 'a', 'flush', 'dev', self.dev_w_ap]) + subprocess.check_call(["ip", "a", "flush", "dev", self.dev_w_ap]) if ipv6_mode is not None: - subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.dev_w_ap]) + subprocess.check_call( + ["ip", "a", "add", "2600::1/64", "dev", self.dev_w_ap] + ) else: - subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.dev_w_ap]) + subprocess.check_call( + ["ip", "a", "add", "192.168.5.1/24", "dev", self.dev_w_ap] + ) self.start_hostapd(hostapd_conf) self.start_dnsmasq(ipv6_mode, self.dev_w_ap) def setup_eth(self, ipv6_mode, start_dnsmasq=True): - '''Set up simulated ethernet router + """Set up simulated ethernet router On self.dev_e_ap, run dnsmasq according to ipv6_mode, see start_dnsmasq(). This is torn down automatically at the end of the test. - ''' + """ # give our router an IP - subprocess.check_call(['ip', 'a', 'flush', 'dev', self.dev_e_ap]) + subprocess.check_call(["ip", "a", "flush", "dev", self.dev_e_ap]) if ipv6_mode is not None: - subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.dev_e_ap]) + subprocess.check_call( + ["ip", "a", "add", "2600::1/64", "dev", self.dev_e_ap] + ) else: - subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.dev_e_ap]) - subprocess.check_call(['ip', 'link', 'set', self.dev_e_ap, 'up']) + subprocess.check_call( + ["ip", "a", "add", "192.168.5.1/24", "dev", self.dev_e_ap] + ) + subprocess.check_call(["ip", "link", "set", self.dev_e_ap, "up"]) # we don't really want to up the client iface already, but veth doesn't # work otherwise (no link detected) - subprocess.check_call(['ip', 'link', 'set', self.dev_e_client, 'up']) + subprocess.check_call(["ip", "link", "set", self.dev_e_client, "up"]) if start_dnsmasq: self.start_dnsmasq(ipv6_mode, self.dev_e_ap) def start_wpasupp(self, conf): - '''Start wpa_supplicant on client interface''' - - w_conf = os.path.join(self.workdir, 'wpasupplicant.conf') - with open(w_conf, 'w') as f: - f.write('ctrl_interface=%s\nnetwork={\n%s\n}\n' % (self.workdir, conf)) - log = os.path.join(self.workdir, 'wpasupp.log') - p = subprocess.Popen(['wpa_supplicant', '-Dwext', '-i', self.dev_w_client, - '-e', self.entropy_file, '-c', w_conf, '-f', log], - stderr=subprocess.PIPE) + """Start wpa_supplicant on client interface""" + + w_conf = os.path.join(self.workdir, "wpasupplicant.conf") + with open(w_conf, "w") as f: + f.write("ctrl_interface=%s\nnetwork={\n%s\n}\n" % (self.workdir, conf)) + log = os.path.join(self.workdir, "wpasupp.log") + p = subprocess.Popen( + [ + "wpa_supplicant", + "-Dwext", + "-i", + self.dev_w_client, + "-e", + self.entropy_file, + "-c", + w_conf, + "-f", + log, + ], + stderr=subprocess.PIPE, + ) self.addCleanup(p.wait) self.addCleanup(p.terminate) # TODO: why does this sometimes take so long? - self.poll_text(log, 'CTRL-EVENT-CONNECTED', timeout=200) + self.poll_text(log, "CTRL-EVENT-CONNECTED", timeout=200) def wrap_process(self, fn, *args, **kwargs): - '''Run a test method in a separate process. + """Run a test method in a separate process. Run test method fn(*args, **kwargs) in a child process. If that raises any exception, it gets propagated to the main process and wrap_process() fails with that exception. - ''' + """ # exception from subprocess is propagated through this file - exc_path = os.path.join(self.workdir, 'exc') + exc_path = os.path.join(self.workdir, "exc") try: os.unlink(exc_path) except OSError: @@ -256,7 +301,7 @@ class NetworkTestBase(unittest.TestCase): try: fn(*args, **kwargs) except: - with open(exc_path, 'w') as f: + with open(exc_path, "w") as f: f.write(traceback.format_exc()) raise else: @@ -273,17 +318,17 @@ class NetworkTestBase(unittest.TestCase): @classmethod def poll_text(klass, logpath, string, timeout=50): - '''Poll log file for a given string with a timeout. + """Poll log file for a given string with a timeout. Timeout is given in deciseconds. - ''' - log = '' + """ + log = "" while timeout > 0: if os.path.exists(logpath): break timeout -= 1 time.sleep(0.1) - assert timeout > 0, 'Timed out waiting for file %s to appear' % logpath + assert timeout > 0, "Timed out waiting for file %s to appear" % logpath with open(logpath) as f: while timeout > 0: @@ -296,70 +341,85 @@ class NetworkTestBase(unittest.TestCase): timeout -= 1 time.sleep(0.1) - assert timeout > 0, 'Timed out waiting for "%s":\n------------\n%s\n-------\n' % (string, log) + assert ( + timeout > 0 + ), 'Timed out waiting for "%s":\n------------\n%s\n-------\n' % (string, log) def start_hostapd(self, conf): - hostapd_conf = os.path.join(self.workdir, 'hostapd.conf') - with open(hostapd_conf, 'w') as f: - f.write('interface=%s\ndriver=nl80211\n' % self.dev_w_ap) + hostapd_conf = os.path.join(self.workdir, "hostapd.conf") + with open(hostapd_conf, "w") as f: + f.write("interface=%s\ndriver=nl80211\n" % self.dev_w_ap) f.write(conf) - log = os.path.join(self.workdir, 'hostapd.log') - p = subprocess.Popen(['hostapd', '-e', self.entropy_file, '-f', log, hostapd_conf], - stdout=subprocess.PIPE) + log = os.path.join(self.workdir, "hostapd.log") + p = subprocess.Popen( + ["hostapd", "-e", self.entropy_file, "-f", log, hostapd_conf], + stdout=subprocess.PIPE, + ) self.addCleanup(p.wait) self.addCleanup(p.terminate) - self.poll_text(log, '' + self.dev_w_ap + ': AP-ENABLED') + self.poll_text(log, "" + self.dev_w_ap + ": AP-ENABLED") def start_dnsmasq(self, ipv6_mode, iface): - '''Start dnsmasq. + """Start dnsmasq. If ipv6_mode is None, IPv4 is set up with DHCP. If it is not None, it must be a valid dnsmasq mode, i. e. a combination of "ra-only", "slaac", "ra-stateless", and "ra-names". See dnsmasq(8). - ''' + """ if ipv6_mode is None: - dhcp_range = '192.168.5.10,192.168.5.200' + dhcp_range = "192.168.5.10,192.168.5.200" else: - dhcp_range = '2600::10,2600::20' + dhcp_range = "2600::10,2600::20" if ipv6_mode: - dhcp_range += ',' + ipv6_mode - - self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log') - lease_file = os.path.join(self.workdir, 'dnsmasq.leases') - - p = subprocess.Popen(['dnsmasq', '--keep-in-foreground', '--log-queries', - '--log-facility=' + self.dnsmasq_log, - '--conf-file=/dev/null', - '--dhcp-leasefile=' + lease_file, - '--bind-interfaces', - '--interface=' + iface, - '--except-interface=lo', - '--enable-ra', - '--dhcp-range=' + dhcp_range]) + dhcp_range += "," + ipv6_mode + + self.dnsmasq_log = os.path.join(self.workdir, "dnsmasq.log") + lease_file = os.path.join(self.workdir, "dnsmasq.leases") + + p = subprocess.Popen( + [ + "dnsmasq", + "--keep-in-foreground", + "--log-queries", + "--log-facility=" + self.dnsmasq_log, + "--conf-file=/dev/null", + "--dhcp-leasefile=" + lease_file, + "--bind-interfaces", + "--interface=" + iface, + "--except-interface=lo", + "--enable-ra", + "--dhcp-range=" + dhcp_range, + ] + ) self.addCleanup(p.wait) self.addCleanup(p.terminate) if ipv6_mode is not None: - self.poll_text(self.dnsmasq_log, 'IPv6 router advertisement enabled') + self.poll_text(self.dnsmasq_log, "IPv6 router advertisement enabled") else: - self.poll_text(self.dnsmasq_log, 'DHCP, IP range') + self.poll_text(self.dnsmasq_log, "DHCP, IP range") @classmethod def _rfkill_attribute(klass, interface): - '''Return the path to interface's rfkill soft toggle in sysfs.''' - - g = glob('/sys/class/net/%s/phy80211/rfkill*/soft' % interface) - assert len(g) == 1, 'Did not find exactly one "soft" rfkill attribute for %s: %s' % ( - interface, str(g)) + """Return the path to interface's rfkill soft toggle in sysfs.""" + + g = glob("/sys/class/net/%s/phy80211/rfkill*/soft" % interface) + assert ( + len(g) == 1 + ), 'Did not find exactly one "soft" rfkill attribute for %s: %s' % ( + interface, + str(g), + ) return g[0] def run_in_subprocess(fn): - '''Decorator for running fn in a child process''' + """Decorator for running fn in a child process""" @functools.wraps(fn) def wrapped(*args, **kwargs): # args[0] is self args[0].wrap_process(fn, *args, **kwargs) + return wrapped diff --git a/debian/tests/nm.py b/debian/tests/nm.py index 529f4cb4..32b6bad7 100755 --- a/debian/tests/nm.py +++ b/debian/tests/nm.py @@ -2,9 +2,9 @@ # 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' +__author__ = "Martin Pitt <martin.pitt@ubuntu.com>" +__copyright__ = "(C) 2013 Canonical Ltd." +__license__ = "GPL v2 or later" import sys import os @@ -24,19 +24,19 @@ except ImportError: import gi -gi.require_version('NM', '1.0') +gi.require_version("NM", "1.0") from gi.repository import NM, GLib, Gio sys.path.append(os.path.dirname(__file__)) import network_test_base -SSID = 'fake net' +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) +NM_LOG_STDOUT = os.getenv("NM_LOG_STDOUT", False) # avoid accidentally destroying any real config -os.environ['GSETTINGS_BACKEND'] = 'memory' +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 @@ -45,60 +45,65 @@ os.dup2(sys.stdout.fileno(), sys.stderr.fileno()) class NetworkManagerTest(network_test_base.NetworkTestBase): - '''Provide common functionality for NM tests''' + """Provide common functionality for NM tests""" def start_nm(self, wait_iface=None, auto_connect=True): - '''Start NetworkManager and initialize client object + """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') + 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'): + 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: + with open("/sys/class/net/%s/address" % iface) as f: if blacklist: - blacklist += ';' - blacklist += 'mac:%s' % f.read().strip() + blacklist += ";" + blacklist += "mac:%s" % f.read().strip() - conf = os.path.join(self.workdir, 'NetworkManager.conf') - extra_main = '' + conf = os.path.join(self.workdir, "NetworkManager.conf") + extra_main = "" if not auto_connect: - extra_main += 'no-auto-default=*\n' + 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)) + 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') + 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] + argv = ["NetworkManager", "--log-level=debug", "--debug", "--config=" + conf] # allow specifying extra arguments - argv += os.environ.get('NM_TEST_DAEMON_ARGS', '').strip().split() + 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 @@ -109,57 +114,64 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): if NM_LOG_STDOUT: # let it initialize, then print a marker time.sleep(1) - print('******* NM initialized *********\n\n') + 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.poll_text(log, "manager: (%s): new" % wait_iface, timeout=100) self.nmclient = NM.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()) + # 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(), NM.DeviceType.WIFI) - self.assertEqual(d.get_driver(), 'mac80211_hwsim') + 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(), NM.DeviceType.WIFI) - self.assertEqual(d.get_driver(), 'mac80211_hwsim') + 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(), NM.DeviceType.VETH) - self.assertEqual(d.get_driver(), 'veth') + 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.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.''' + """Shut down all active NM connections.""" if NM_LOG_STDOUT: - print('\n\n******* Shutting down NM connections *********') + 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) + 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) @@ -167,18 +179,18 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): @classmethod def process_glib_events(klass): - '''Process pending GLib main loop events''' + """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. + """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(): @@ -186,32 +198,36 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): timeout -= 1 time.sleep(0.1) else: - self.fail(message or 'timed out waiting for ' + str(condition)) + self.fail(message or "timed out waiting for " + str(condition)) def assert_iface_down(self, iface): - '''Assert that client interface is down''' + """Assert that client interface is down""" - out = subprocess.check_output(['ip', 'a', 'show', 'dev', iface], - universal_newlines=True) - self.assertNotIn('inet 192', out) - self.assertNotIn('inet6 2600', out) + out = subprocess.check_output( + ["ip", "a", "show", "dev", iface], universal_newlines=True + ) + 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) + 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) + 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''' + """Assert that client interface is up""" - out = subprocess.check_output(['ip', 'a', 'show', 'dev', iface], - universal_newlines=True) - self.assertIn('state UP', out) + 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) @@ -220,22 +236,25 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): 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) + 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''' + """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) + 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. + """Connect to an NMAccessPoint. secret should be None for open networks, and a string with the password for WEP/WPA. @@ -243,7 +262,7 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): ip6_privacy is a NM.SettingIP6ConfigPrivacy flag. Return (NMConnection, NMActiveConnection) objects. - ''' + """ ip4_method = NM.SETTING_IP4_CONFIG_METHOD_DISABLED ip6_method = NM.SETTING_IP6_CONFIG_METHOD_IGNORE @@ -261,13 +280,14 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): if secret: partial_conn.add_setting(NM.SettingWirelessSecurity.new()) # FIXME: needs update for other auth types - partial_conn.update_secrets(NM.SETTING_WIRELESS_SECURITY_SETTING_NAME, - GLib.Variant('a{sv}', { - 'psk': GLib.Variant('s', secret) - })) + partial_conn.update_secrets( + NM.SETTING_WIRELESS_SECURITY_SETTING_NAME, + GLib.Variant("a{sv}", {"psk": GLib.Variant("s", secret)}), + ) if ip6_privacy is not None: - partial_conn.add_setting(NM.SettingIP6Config(ip6_privacy=ip6_privacy, - method=ip6_method)) + partial_conn.add_setting( + NM.SettingIP6Config(ip6_privacy=ip6_privacy, method=ip6_method) + ) ml = GLib.MainLoop() self.cb_conn = None @@ -275,17 +295,18 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): self.timeout_tag = 0 def add_activate_cb(client, res, data): - if (self.timeout_tag > 0): + if self.timeout_tag > 0: GLib.source_remove(self.timeout_tag) self.timeout_tag = 0 try: - self.cb_conn = \ - self.nmclient.add_and_activate_connection_finish(res) + self.cb_conn = self.nmclient.add_and_activate_connection_finish(res) except gi.repository.GLib.Error as e: # Check if the error is "Operation was cancelled" - if (e.domain != "g-io-error-quark" or e.code != 19): - self.fail("add_and_activate_connection failed: %s (%s, %d)" % - (e.message, e.domain, e.code)) + if e.domain != "g-io-error-quark" or e.code != 19: + self.fail( + "add_and_activate_connection failed: %s (%s, %d)" + % (e.message, e.domain, e.code) + ) ml.quit() def timeout_cb(): @@ -294,12 +315,19 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): ml.quit() return GLib.SOURCE_REMOVE - self.nmclient.add_and_activate_connection_async(partial_conn, self.nmdev_w, ap.get_path(), self.cancel, add_activate_cb, None) + self.nmclient.add_and_activate_connection_async( + partial_conn, + self.nmdev_w, + ap.get_path(), + self.cancel, + add_activate_cb, + None, + ) self.timeout_tag = GLib.timeout_add_seconds(300, timeout_cb) ml.run() - if (self.timeout_tag < 0): + if self.timeout_tag < 0: self.timeout_tag = 0 - self.fail('Main loop for adding connection timed out!') + self.fail("Main loop for adding connection timed out!") self.assertNotEqual(self.cb_conn, None) active_conn = self.cb_conn self.cb_conn = None @@ -312,7 +340,9 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): if secret is None: self.assertEqual(needed_secrets, (None, [])) else: - self.assertEqual(needed_secrets[0], NM.SETTING_WIRELESS_SECURITY_SETTING_NAME) + self.assertEqual( + needed_secrets[0], NM.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 @@ -320,14 +350,16 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): # 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() == NM.ActiveConnectionState.ACTIVATED, - 'timed out waiting for %s to get activated' % active_conn.get_connection(), - timeout=600) + self.assertEventually( + lambda: active_conn.get_state() == NM.ActiveConnectionState.ACTIVATED, + "timed out waiting for %s to get activated" % active_conn.get_connection(), + timeout=600, + ) self.assertEqual(self.nmdev_w.get_state(), NM.DeviceState.ACTIVATED) return (conn, active_conn) def conn_from_active_conn(self, active_conn): - '''Get NMConnection object for an NMActiveConnection object''' + """Get NMConnection object for an NMActiveConnection object""" # this sometimes takes a second try, when the corresponding # NMConnection object is not yet available @@ -342,64 +374,71 @@ class NetworkManagerTest(network_test_base.NetworkTestBase): time.sleep(0.1) tries -= 1 - self.fail('Could not find NMConnection object for %s' % path) + self.fail("Could not find NMConnection object for %s" % path) def check_low_level_config(self, iface, ipv6_mode, ip6_privacy): - '''Check actual hardware state with ip/iw after being connected''' + """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'): + if ipv6_mode in ("", "slaac"): # has global address from our DHCP server - expected_ip_a.append('inet6 2600::[0-9a-f]+/') + 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|\n\s*valid_lft forever preferred_lft forever)') + expected_ip_a.append( + "inet6 2600::[0-9a-f:]+/64 scope global (?:tentative )?(?:mngtmpaddr )?(?:noprefixroute )?(dynamic|\n\s*valid_lft forever preferred_lft forever)" + ) # 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 (NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR, - NM.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR): + priv_re = "inet6 2600:[0-9a-f:]+/64 scope global temporary (?:tentative )?(?:mngtmpaddr )?dynamic" + if ip6_privacy in ( + NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR, + NM.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR, + ): expected_ip_a.append(priv_re) else: # FIXME: add a negative test here pass - #unexpected_ip_a.append(priv_re) + # unexpected_ip_a.append(priv_re) # has a link-local address - expected_ip_a.append('inet6 fe80::[0-9a-f:]+/64 scope link') + expected_ip_a.append("inet6 fe80::[0-9a-f:]+/64 scope link") else: - expected_ip_a.append('inet 192.168.5.\d+/24') + 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''' + """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)''' + """Start AP and NM, then run a shell (for debugging)""" - self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + self.setup_ap("hw_mode=b\nchannel=1\nssid=" + SSID, None) self.start_nm(self.dev_w_client) - print(''' + 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']) +""" + % (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''' + """no available access point""" self.start_nm(self.dev_w_client) # Give the interfaces a bit more time to intialized; it may be needed @@ -411,96 +450,134 @@ Logs are in '%s'. When done, exit the shell. self.assertEqual(self.nmdev_w.props.device_type, NM.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) + 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) + 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(), - [NM.DeviceState.DISCONNECTED, NM.DeviceState.UNAVAILABLE]) + self.assertIn( + self.nmdev_w.get_state(), + [NM.DeviceState.DISCONNECTED, NM.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''' + """Open network, 802.11b, IPv4""" - self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, None, 11000) + 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''' + """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=NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + self.do_test( + "hw_mode=b\nchannel=1\nssid=" + SSID, + "ra-only", + 11000, + ip6_privacy=NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR, + ) def test_open_b_ip6_raonly_pubaddr(self): - '''Open network, 802.11b, IPv6 with only RA, preferring public address''' + """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=NM.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR) + self.do_test( + "hw_mode=b\nchannel=1\nssid=" + SSID, + "ra-only", + 11000, + ip6_privacy=NM.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR, + ) def test_open_b_ip6_raonly_no_pe(self): - '''Open network, 802.11b, IPv6 with only RA, PE disabled''' + """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=NM.SettingIP6ConfigPrivacy.DISABLED) + self.do_test( + "hw_mode=b\nchannel=1\nssid=" + SSID, + "ra-only", + 11000, + ip6_privacy=NM.SettingIP6ConfigPrivacy.DISABLED, + ) def test_open_b_ip6_dhcp(self): - '''Open network, 802.11b, IPv6 with DHCP, preferring temp address''' + """Open network, 802.11b, IPv6 with DHCP, preferring temp address""" - self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, '', 11000, - ip6_privacy=NM.SettingIP6ConfigPrivacy.UNKNOWN) + self.do_test( + "hw_mode=b\nchannel=1\nssid=" + SSID, + "", + 11000, + ip6_privacy=NM.SettingIP6ConfigPrivacy.UNKNOWN, + ) def test_open_g_ip4(self): - '''Open network, 802.11g, IPv4''' + """Open network, 802.11g, IPv4""" - self.do_test('hw_mode=g\nchannel=1\nssid=' + SSID, None, 54000) + self.do_test("hw_mode=g\nchannel=1\nssid=" + SSID, None, 54000) def test_wpa1_ip4(self): - '''WPA1, 802.11g, IPv4''' + """WPA1, 802.11g, IPv4""" - self.do_test('''hw_mode=g + 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') +""" + % SSID, + None, + 54000, + "12345678", + ) def test_wpa2_ip4(self): - '''WPA2, 802.11g, IPv4''' + """WPA2, 802.11g, IPv4""" - self.do_test('''hw_mode=g + 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') +""" + % SSID, + None, + 54000, + "12345678", + ) def test_wpa2_ip6(self): - '''WPA2, 802.11g, IPv6 with only RA''' + """WPA2, 802.11g, IPv6 with only RA""" - self.do_test('''hw_mode=g + 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=NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) +""" + % SSID, + "ra-only", + 54000, + "12345678", + ip6_privacy=NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR, + ) @network_test_base.run_in_subprocess def test_rfkill(self): - '''shut down connection on killswitch, restore it on unblock''' + """shut down connection on killswitch, restore it on unblock""" - self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + 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) @@ -511,8 +588,9 @@ wpa_passphrase=12345678 # 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() == NM.DeviceState.UNAVAILABLE, - timeout=100) + self.assertEventually( + lambda: self.nmdev_w.get_state() == NM.DeviceState.UNAVAILABLE, timeout=100 + ) # dev_w_client should be down now self.assert_iface_down(self.dev_w_client) @@ -520,11 +598,12 @@ wpa_passphrase=12345678 # 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() == NM.DeviceState.ACTIVATED, - timeout=200) + self.assertEventually( + lambda: self.nmdev_w.get_state() == NM.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']) + self.assert_iface_up(self.dev_w_client, ["inet 192.168.5.\d+/24"]) # # Common test code @@ -535,19 +614,25 @@ wpa_passphrase=12345678 # 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''' + 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(NM, '80211Mode').INFRA) + self.assertTrue(ap.get_path().startswith("/org/freedesktop/NetworkManager")) + self.assertEqual(ap.get_mode(), getattr(NM, "80211Mode").INFRA) self.assertEqual(ap.get_max_bitrate(), expected_max_bitrate) - #self.assertEqual(ap.get_flags(), ) + # self.assertEqual(ap.get_flags(), ) # should not auto-connect self.assertEqual(self.nmclient.get_active_connections(), []) @@ -556,23 +641,33 @@ wpa_passphrase=12345678 (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.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()] + ) # check corresponding NMConnection object wireless_setting = conn.get_setting_wireless() self.assertEqual(wireless_setting.get_ssid().get_data(), SSID.encode()) self.assertEqual(wireless_setting.get_hidden(), False) if secret: - self.assertEqual(conn.get_setting_wireless_security().get_name(), NM.SETTING_WIRELESS_SECURITY_SETTING_NAME) + self.assertEqual( + conn.get_setting_wireless_security().get_name(), + NM.SETTING_WIRELESS_SECURITY_SETTING_NAME, + ) else: self.assertEqual(conn.get_setting_wireless_security(), None) # for debugging - #conn.dump() + # conn.dump() # for IPv6, check privacy setting if ipv6_mode is not None and ip6_privacy != NM.SettingIP6ConfigPrivacy.UNKNOWN: - assert ip6_privacy is not None, 'for IPv6 tests you need to specify ip6_privacy flag' + 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) @@ -580,58 +675,71 @@ wpa_passphrase=12345678 class ColdplugEthernet(NetworkManagerTest): - '''Ethernet: In these tests NM starts after setting up the router''' + """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)''' + """Start router and NM, then run a shell (for debugging)""" self.setup_eth(None) self.start_nm(self.dev_e_client) - print(''' + 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']) +""" + % (self.dev_e_client, self.dev_e_ap, self.workdir) + ) + subprocess.call(["bash", "-i"]) def test_auto_ip4(self): - '''ethernet: auto-connection, IPv4''' + """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''' + """ethernet: auto-connection, IPv6 with only RA, PE disabled""" - self.do_test('ra-only', auto_connect=True, - ip6_privacy=NM.SettingIP6ConfigPrivacy.DISABLED) + self.do_test( + "ra-only", + auto_connect=True, + ip6_privacy=NM.SettingIP6ConfigPrivacy.DISABLED, + ) def test_auto_ip6_dhcp(self): - '''ethernet: auto-connection, IPv6 with DHCP''' + """ethernet: auto-connection, IPv6 with DHCP""" - self.do_test('', auto_connect=True, - ip6_privacy=NM.SettingIP6ConfigPrivacy.UNKNOWN) + self.do_test( + "", auto_connect=True, ip6_privacy=NM.SettingIP6ConfigPrivacy.UNKNOWN + ) def test_manual_ip4(self): - '''ethernet: manual connection, IPv4''' + """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''' + """ethernet: manual connection, IPv6 with only RA, preferring temp address""" - self.do_test('ra-only', auto_connect=False, - ip6_privacy=NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + self.do_test( + "ra-only", + auto_connect=False, + ip6_privacy=NM.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR, + ) def test_manual_ip6_raonly_pubaddr(self): - '''ethernet: manual connection, IPv6 with only RA, preferring public address''' + """ethernet: manual connection, IPv6 with only RA, preferring public address""" - self.do_test('ra-only', auto_connect=False, - ip6_privacy=NM.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR) + self.do_test( + "ra-only", + auto_connect=False, + ip6_privacy=NM.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR, + ) # # Common test code @@ -639,7 +747,7 @@ Logs are in '%s'. When done, exit the shell. @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''' + """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) @@ -653,17 +761,20 @@ Logs are in '%s'. When done, exit the shell. if auto_connect: # ethernet should auto-connect quickly without an existing defined connection - self.assertEventually(lambda: len(self.nmclient.get_active_connections()) > 0, - 'timed out waiting for active connections', - timeout=100) + self.assertEventually( + lambda: len(self.nmclient.get_active_connections()) > 0, + "timed out waiting for active connections", + timeout=100, + ) active_conn = self.nmclient.get_active_connections()[0] else: # auto-connection was disabled, set up manual connection partial_conn = NM.SimpleConnection.new() partial_conn.add_setting(NM.SettingIP4Config(method=ip4_method)) if ip6_privacy is not None: - partial_conn.add_setting(NM.SettingIP6Config(ip6_privacy=ip6_privacy, - method=ip6_method)) + partial_conn.add_setting( + NM.SettingIP6Config(ip6_privacy=ip6_privacy, method=ip6_method) + ) ml = GLib.MainLoop() self.cb_conn = None @@ -671,17 +782,18 @@ Logs are in '%s'. When done, exit the shell. self.timeout_tag = 0 def add_activate_cb(client, res, data): - if (self.timeout_tag > 0): + if self.timeout_tag > 0: GLib.source_remove(self.timeout_tag) self.timeout_tag = 0 try: - self.cb_conn = \ - self.nmclient.add_and_activate_connection_finish(res) + self.cb_conn = self.nmclient.add_and_activate_connection_finish(res) except gi.repository.GLib.Error as e: # Check if the error is "Operation was cancelled" - if (e.domain != "g-io-error-quark" or e.code != 19): - self.fail("add_and_activate_connection failed: %s (%s, %d)" % - (e.message, e.domain, e.code)) + if e.domain != "g-io-error-quark" or e.code != 19: + self.fail( + "add_and_activate_connection failed: %s (%s, %d)" + % (e.message, e.domain, e.code) + ) ml.quit() def timeout_cb(): @@ -690,35 +802,48 @@ Logs are in '%s'. When done, exit the shell. ml.quit() return GLib.SOURCE_REMOVE - self.nmclient.add_and_activate_connection_async(partial_conn, self.nmdev_e, None, self.cancel, add_activate_cb, None) + self.nmclient.add_and_activate_connection_async( + partial_conn, self.nmdev_e, None, self.cancel, add_activate_cb, None + ) self.timeout_tag = GLib.timeout_add_seconds(300, timeout_cb) ml.run() - if (self.timeout_tag < 0): + if self.timeout_tag < 0: self.timeout_tag = 0 - self.fail('Main loop for adding connection timed out!') + self.fail("Main loop for adding connection timed out!") self.assertNotEqual(self.cb_conn, 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() == NM.ActiveConnectionState.ACTIVATED, - 'timed out waiting for %s to get activated' % active_conn.get_connection(), - timeout=150) + self.assertEventually( + lambda: active_conn.get_state() == NM.ActiveConnectionState.ACTIVATED, + "timed out waiting for %s to get activated" % active_conn.get_connection(), + timeout=150, + ) self.assertEqual(self.nmdev_e.get_state(), NM.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.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()] + ) # 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 (NM.SettingIP6ConfigPrivacy.UNKNOWN, - NM.SettingIP6ConfigPrivacy.DISABLED): + assert ( + ip6_privacy is not None + ), "for IPv6 tests you need to specify ip6_privacy flag" + if ip6_privacy not in ( + NM.SettingIP6ConfigPrivacy.UNKNOWN, + NM.SettingIP6ConfigPrivacy.DISABLED, + ): ip6_setting = conn.get_setting_ip6_config() self.assertEqual(ip6_setting.props.ip6_privacy, ip6_privacy) @@ -726,14 +851,14 @@ Logs are in '%s'. When done, exit the shell. class Hotplug(NetworkManagerTest): - '''In these tests APs are set up while NM is already running''' + """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''' + """new AP is being detected automatically within 30s""" - self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + 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 @@ -743,27 +868,32 @@ class Hotplug(NetworkManagerTest): @network_test_base.run_in_subprocess @unittest.expectedFailure def test_auto_detect_eth(self): - '''new eth router is being detected automatically within 30s''' + """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) + 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() == NM.ActiveConnectionState.ACTIVATED, - 'timed out waiting for %s to get activated' % active_conn.get_connection(), - timeout=80) + self.assertEventually( + lambda: active_conn.get_state() == NM.ActiveConnectionState.ACTIVATED, + "timed out waiting for %s to get activated" % active_conn.get_connection(), + timeout=80, + ) self.assertEqual(self.nmdev_e.get_state(), NM.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') +@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''' + """These tests run under a mock logind on a private system D-BUS""" @classmethod def setUpClass(klass): @@ -780,35 +910,40 @@ class Suspend(NetworkManagerTest, DBusTestCase): # 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) + (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) + (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]) + 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''' + """suspend during active IPv4 connection""" - self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, None, - ['inet 192.168.5.\d+/24']) + 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''' + """suspend during active IPv6 connection""" - self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', - ['inet6 2600::']) + self.do_test("hw_mode=b\nchannel=1\nssid=" + SSID, "ra-only", ["inet6 2600::"]) # # Common test code @@ -816,7 +951,7 @@ class Suspend(NetworkManagerTest, DBusTestCase): @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''' + """Actual test code, parameterized for the particular test case""" self.setup_ap(hostapd_conf, ipv6_mode) self.start_nm(self.dev_w_client) @@ -824,19 +959,21 @@ class Suspend(NetworkManagerTest, DBusTestCase): (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]) + self.obj_logind.EmitSignal("", "PrepareForSleep", "b", [True]) # disabling should be fast, give it one second - self.assertEventually(lambda: self.nmdev_w.get_state() == NM.DeviceState.UNMANAGED, - timeout=10) + self.assertEventually( + lambda: self.nmdev_w.get_state() == NM.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]) + 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() == NM.DeviceState.ACTIVATED, - timeout=100) + self.assertEventually( + lambda: self.nmdev_w.get_state() == NM.DeviceState.ACTIVATED, timeout=100 + ) # dev_w_client should be back up self.assert_iface_up(self.dev_w_client, expected_ip_a) @@ -846,33 +983,40 @@ def setUpModule(): # 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) + 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' + 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 NetworkManager stop 2>&1', shell=True) == 0 + nm_running = subprocess.call("service NetworkManager stop 2>&1", shell=True) == 0 def tearDownModule(): - subprocess.call('dhclient eth0', shell=True) - subprocess.call('sleep 10', shell=True) + subprocess.call("dhclient eth0", shell=True) + subprocess.call("sleep 10", shell=True) -if __name__ == '__main__': +if __name__ == "__main__": # 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.stderr.write("This integration test suite needs to be run as root\n") sys.exit(1) - if re.search(b's390', subprocess.run(['dpkg', '--print-architecture'], capture_output=True).stdout): + if re.search( + b"s390", + subprocess.run(["dpkg", "--print-architecture"], capture_output=True).stdout, + ): print("s390 arch has no wireless support, skipping") sys.exit(77) |