# Lint as: python2, python3 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. import contextlib import logging import math import re import time from contextlib import contextmanager from collections import namedtuple from autotest_lib.client.common_lib import error from autotest_lib.client.common_lib import utils from autotest_lib.client.common_lib.cros.network import interface from autotest_lib.client.common_lib.cros.network import iw_runner from autotest_lib.client.common_lib.cros.network import ping_runner from autotest_lib.client.cros import constants from autotest_lib.server import autotest from autotest_lib.server import site_linux_system from autotest_lib.server.cros.network import wpa_cli_proxy from autotest_lib.server.cros.network import wpa_mon from autotest_lib.server.hosts import cast_os_host # Wake-on-WiFi feature strings WAKE_ON_WIFI_NONE = 'none' WAKE_ON_WIFI_PACKET = 'packet' WAKE_ON_WIFI_DARKCONNECT = 'darkconnect' WAKE_ON_WIFI_PACKET_DARKCONNECT = 'packet_and_darkconnect' WAKE_ON_WIFI_NOT_SUPPORTED = 'not_supported' # Wake-on-WiFi test timing constants SUSPEND_WAIT_TIME_SECONDS = 10 RECEIVE_PACKET_WAIT_TIME_SECONDS = 10 DARK_RESUME_WAIT_TIME_SECONDS = 25 WAKE_TO_SCAN_PERIOD_SECONDS = 30 NET_DETECT_SCAN_WAIT_TIME_SECONDS = 15 WAIT_UP_TIMEOUT_SECONDS = 10 DISCONNECT_WAIT_TIME_SECONDS = 10 INTERFACE_DOWN_WAIT_TIME_SECONDS = 10 ConnectTime = namedtuple('ConnectTime', 'state, time') XMLRPC_BRINGUP_TIMEOUT_SECONDS = 60 SHILL_XMLRPC_LOG_PATH = '/var/log/shill_xmlrpc_server.log' def _is_eureka_host(host): return host.get_os_type() == cast_os_host.OS_TYPE_CAST_OS def get_xmlrpc_proxy(host): """Get a shill XMLRPC proxy for |host|. The returned object has no particular type. Instead, when you call a method on the object, it marshalls the objects passed as arguments and uses them to make RPCs on the remote server. Thus, you should read shill_xmlrpc_server.py to find out what methods are supported. @param host: host object representing a remote device. @return proxy object for remote XMLRPC server. """ # Make sure the client library is on the device so that the proxy # code is there when we try to call it. if host.is_client_install_supported: client_at = autotest.Autotest(host) client_at.install() # This is the default port for shill xmlrpc server. server_port = constants.SHILL_XMLRPC_SERVER_PORT xmlrpc_server_command = constants.SHILL_XMLRPC_SERVER_COMMAND log_path = SHILL_XMLRPC_LOG_PATH command_name = constants.SHILL_XMLRPC_SERVER_CLEANUP_PATTERN rpc_server_host = host # Start up the XMLRPC proxy on the client proxy = rpc_server_host.rpc_server_tracker.xmlrpc_connect( xmlrpc_server_command, server_port, command_name=command_name, ready_test_name=constants.SHILL_XMLRPC_SERVER_READY_METHOD, timeout_seconds=XMLRPC_BRINGUP_TIMEOUT_SECONDS, logfile=log_path ) return proxy def _is_conductive(host): """Determine if the host is conductive based on AFE labels. @param host: A Host object. """ info = host.host_info_store.get() conductive = info.get_label_value('conductive') return conductive.lower() == 'true' class WiFiClient(site_linux_system.LinuxSystem): """WiFiClient is a thin layer of logic over a remote DUT in wifitests.""" DEFAULT_PING_COUNT = 10 COMMAND_PING = 'ping' MAX_SERVICE_GONE_TIMEOUT_SECONDS = 60 # List of interface names we won't consider for use as "the" WiFi interface # on Android or CastOS hosts. WIFI_IF_BLOCKLIST = ['p2p0', 'wfd0'] UNKNOWN_BOARD_TYPE = 'unknown' # DBus device properties. Wireless interfaces should support these. WAKE_ON_WIFI_FEATURES = 'WakeOnWiFiFeaturesEnabled' NET_DETECT_SCAN_PERIOD = 'NetDetectScanPeriodSeconds' WAKE_TO_SCAN_PERIOD = 'WakeToScanPeriodSeconds' FORCE_WAKE_TO_SCAN_TIMER = 'ForceWakeToScanTimer' MAC_ADDRESS_RANDOMIZATION_SUPPORTED = 'MACAddressRandomizationSupported' MAC_ADDRESS_RANDOMIZATION_ENABLED = 'MACAddressRandomizationEnabled' CONNECTED_STATES = ['portal', 'no-connectivity', 'redirect-found', 'portal-suspected', 'online', 'ready'] @property def machine_id(self): """@return string unique to a particular board/cpu configuration.""" if self._machine_id: return self._machine_id uname_result = self.host.run('uname -m', ignore_status=True) kernel_arch = '' if not uname_result.exit_status and uname_result.stdout.find(' ') < 0: kernel_arch = uname_result.stdout.strip() cpu_info = self.host.run('cat /proc/cpuinfo').stdout.splitlines() cpu_count = len([x for x in cpu_info if x.lower().startswith('bogomips')]) cpu_count_str = '' if cpu_count: cpu_count_str = 'x%d' % cpu_count ghz_value = '' ghz_pattern = re.compile('([0-9.]+GHz)') for line in cpu_info: match = ghz_pattern.search(line) if match is not None: ghz_value = '_' + match.group(1) break return '%s_%s%s%s' % (self.board, kernel_arch, ghz_value, cpu_count_str) @property def powersave_on(self): """@return bool True iff WiFi powersave mode is enabled.""" result = self.host.run("iw dev %s get power_save" % self.wifi_if) output = result.stdout.rstrip() # NB: chop \n # Output should be either "Power save: on" or "Power save: off". find_re = re.compile(r'([^:]+):\s+(\w+)') find_results = find_re.match(output) if not find_results: raise error.TestFail('Failed to find power_save parameter ' 'in iw results.') return find_results.group(2) == 'on' @property def shill(self): """@return shill RPCProxy object.""" return self._shill_proxy @property def client(self): """Deprecated accessor for the client host. The term client is used very loosely in old autotests and this accessor should not be used in new code. Use host() instead. @return host object representing a remote DUT. """ return self.host @property def command_ip(self): """@return string path to ip command.""" return self._command_ip @property def command_iptables(self): """@return string path to iptables command.""" return self._command_iptables @property def command_ping6(self): """@return string path to ping6 command.""" return self._command_ping6 @property def command_wpa_cli(self): """@return string path to wpa_cli command.""" return self._command_wpa_cli @property def conductive(self): """@return True if the rig is conductive; False otherwise.""" if self._conductive is None: self._conductive = _is_conductive(self.host) return self._conductive @conductive.setter def conductive(self, value): """Set the conductive member to True or False. @param value: boolean value to set the conductive member to. """ self._conductive = value @property def module_name(self): """@return Name of kernel module in use by this interface.""" return self._interface.module_name @property def parent_device_name(self): """ @return Path of the parent device for the net device""" return self._interface.parent_device_name @property def wifi_if(self): """@return string wifi device on machine (e.g. mlan0).""" return self._wifi_if @property def wifi_mac(self): """@return string MAC address of self.wifi_if.""" return self._interface.mac_address @property def wifi_ip(self): """@return string IPv4 address of self.wifi_if.""" return self._interface.ipv4_address @property def wifi_ip_subnet(self): """@return string IPv4 subnet prefix of self.wifi_if.""" return self._interface.ipv4_subnet @property def wifi_phy_name(self): """@return wiphy name (e.g., 'phy0') or None""" return self._interface.wiphy_name @property def wifi_signal_level(self): """Returns the signal level of this DUT's WiFi interface. @return int signal level of connected WiFi interface or None (e.g. -67). """ return self._interface.signal_level @property def wifi_signal_level_all_chains(self): """Returns the signal level of all chains of this DUT's WiFi interface. @return int array signal level of each chain of connected WiFi interface or None (e.g. [-67, -60]). """ return self._interface.signal_level_all_chains @staticmethod def assert_bsses_include_ssids(found_bsses, expected_ssids): """Verifies that |found_bsses| includes |expected_ssids|. @param found_bsses list of IwBss objects. @param expected_ssids list of string SSIDs. @raise error.TestFail if any element of |expected_ssids| is not found. """ for ssid in expected_ssids: if not ssid: continue for bss in found_bsses: if bss.ssid == ssid: break else: raise error.TestFail('SSID %s is not in scan results: %r' % (ssid, found_bsses)) def wifi_noise_level(self, frequency_mhz): """Returns the noise level of this DUT's WiFi interface. @param frequency_mhz: frequency at which the noise level should be measured and reported. @return int signal level of connected WiFi interface in dBm (e.g. -67) or None if the value is unavailable. """ return self._interface.noise_level(frequency_mhz) def __init__(self, client_host, result_dir, use_wpa_cli): """ Construct a WiFiClient. @param client_host host object representing a remote host. @param result_dir string directory to store test logs/packet caps. @param use_wpa_cli bool True if we want to use |wpa_cli| commands for Android testing. """ super(WiFiClient, self).__init__(client_host, 'client', inherit_interfaces=True) self._command_ip = 'ip' self._command_iptables = 'iptables -w 5' self._command_ping6 = 'ping6' self._command_wpa_cli = 'wpa_cli' self._machine_id = None self._result_dir = result_dir self._conductive = None if _is_eureka_host(self.host) and use_wpa_cli: # Look up the WiFi device (and its MAC) on the client. devs = self.iw_runner.list_interfaces(desired_if_type='managed') devs = [dev for dev in devs if dev.if_name not in self.WIFI_IF_BLOCKLIST] if not devs: raise error.TestFail('No wlan devices found on %s.' % self.host.hostname) if len(devs) > 1: logging.warning('Warning, found multiple WiFi devices on ' '%s: %r', self.host.hostname, devs) self._wifi_if = devs[0].if_name self._shill_proxy = wpa_cli_proxy.WpaCliProxy( self.host, self._wifi_if) self._wpa_cli_proxy = self._shill_proxy else: self._shill_proxy = get_xmlrpc_proxy(self.host) interfaces = self._shill_proxy.list_controlled_wifi_interfaces() if not interfaces: logging.debug('No interfaces managed by shill. Rebooting host') self.host.reboot() raise error.TestError('No interfaces managed by shill on %s' % self.host.hostname) self._wifi_if = interfaces[0] self._wpa_cli_proxy = wpa_cli_proxy.WpaCliProxy( self.host, self._wifi_if) self._raise_logging_level() self._interface = interface.Interface(self._wifi_if, host=self.host) self._wpa_mon = wpa_mon.WpaMon(self.host, self.wifi_if) logging.debug('WiFi interface is: %r', self._interface.device_description) # All tests that use this object assume the interface starts enabled. self.set_device_enabled(self._wifi_if, True) # Turn off powersave mode by default. self.powersave_switch(False) # Invoke the |capabilities| property defined in the parent |Linuxsystem| # to workaround the lazy loading of the capabilities cache and supported # frequency list. This is needed for tests that may need access to these # when the DUT is unreachable (for ex: suspended). #pylint: disable=pointless-statement self.capabilities def _assert_method_supported(self, method_name): """Raise a TestNAError if the XMLRPC proxy has no method |method_name|. @param method_name: string name of method that should exist on the XMLRPC proxy. """ if not self._supports_method(method_name): raise error.TestNAError('%s() is not supported' % method_name) def _raise_logging_level(self): """Raises logging levels for WiFi on DUT.""" self.host.run('ff_debug --level -2', ignore_status=True) self.host.run('ff_debug +wifi', ignore_status=True) def is_vht_supported(self): """Returns True if VHT supported; False otherwise""" return self.CAPABILITY_VHT in self.capabilities def is_5ghz_supported(self): """Returns True if 5Ghz bands are supported; False otherwise.""" return self.CAPABILITY_5GHZ in self.capabilities def is_ibss_supported(self): """Returns True if IBSS mode is supported; False otherwise.""" return self.CAPABILITY_IBSS in self.capabilities def is_frequency_supported(self, frequency): """Returns True if the given frequency is supported; False otherwise. @param frequency: int Wifi frequency to check if it is supported by DUT. """ return frequency in self.phys_for_frequency def _supports_method(self, method_name): """Checks if |method_name| is supported on the remote XMLRPC proxy. autotest will, for their own reasons, install python files in the autotest client package that correspond the version of the build rather than the version running on the autotest drone. This creates situations where we call methods on the client XMLRPC proxy that don't exist in that version of the code. This detects those situations so that we can degrade more or less gracefully. @param method_name: string name of method that should exist on the XMLRPC proxy. @return True if method is available, False otherwise. """ supported = (_is_eureka_host(self.host) or method_name in self._shill_proxy.system.listMethods()) if not supported: logging.warning('%s() is not supported on older images', method_name) return supported def close(self): """Tear down state associated with the client.""" self.stop_capture() self.powersave_switch(False) self.shill.clean_profiles() super(WiFiClient, self).close() def sync_host_times(self): """Set time on our DUT to match local time.""" epoch_seconds = time.time() self.shill.sync_time_to(epoch_seconds) def collect_debug_info(self, local_save_dir_prefix): """Collect any debug information needed from the DUT This invokes the |collect_debug_info| RPC method to trigger bugreport/logcat collection and then transfers the logs to the server. @param local_save_dir_prefix Used as a prefix for local save directory. """ pass def check_iw_link_value(self, iw_link_key, desired_value): """Assert that the current wireless link property is |desired_value|. @param iw_link_key string one of IW_LINK_KEY_* defined in iw_runner. @param desired_value string desired value of iw link property. """ actual_value = self.get_iw_link_value(iw_link_key) desired_value = str(desired_value) if actual_value != desired_value: raise error.TestFail('Wanted iw link property %s value %s, but ' 'got %s instead.' % (iw_link_key, desired_value, actual_value)) def get_iw_link_value(self, iw_link_key): """Get the current value of a link property for this WiFi interface. @param iw_link_key string one of IW_LINK_KEY_* defined in iw_runner. """ return self.iw_runner.get_link_value(self.wifi_if, iw_link_key) def powersave_switch(self, turn_on): """Toggle powersave mode for the DUT. @param turn_on bool True iff powersave mode should be turned on. """ mode = 'off' if turn_on: mode = 'on' # Turn ON interface and set power_save option. self.host.run('ifconfig %s up' % self.wifi_if) self.host.run('iw dev %s set power_save %s' % (self.wifi_if, mode)) def timed_scan(self, frequencies, ssids, scan_timeout_seconds=10, retry_timeout_seconds=10): """Request timed scan to discover given SSIDs. This method will retry for a default of |retry_timeout_seconds| until it is able to successfully kick off a scan. Sometimes, the device on the DUT claims to be busy and rejects our requests. It will raise error if the scan did not complete within |scan_timeout_seconds| or it was not able to discover the given SSIDs. @param frequencies list of int WiFi frequencies to scan for. @param ssids list of string ssids to probe request for. @param scan_timeout_seconds: float number of seconds the scan operation not to exceed. @param retry_timeout_seconds: float number of seconds to retry scanning if the interface is busy. This does not retry if certain SSIDs are missing from the results. @return time in seconds took to complete scan request. """ # the poll method returns the result of the func scan_result = utils.poll_for_condition( condition=lambda: self.iw_runner.timed_scan( self.wifi_if, frequencies=frequencies, ssids=ssids), exception=error.TestFail('Unable to trigger scan on client'), timeout=retry_timeout_seconds, sleep_interval=0.5) # Verify scan operation completed within given timeout if scan_result.time > scan_timeout_seconds: raise error.TestFail('Scan time %.2fs exceeds the scan timeout' % (scan_result.time)) # Verify all ssids are discovered self.assert_bsses_include_ssids(scan_result.bss_list, ssids) logging.info('Wifi scan completed in %.2f seconds', scan_result.time) return scan_result.time def scan(self, frequencies, ssids, timeout_seconds=10, require_match=True): """Request a scan and (optionally) check that requested SSIDs appear in the results. This method will retry for a default of |timeout_seconds| until it is able to successfully kick off a scan. Sometimes, the device on the DUT claims to be busy and rejects our requests. If |ssids| is non-empty, we will speficially probe for those SSIDs. If |require_match| is True, we will verify that every element of |ssids| was found in scan results. @param frequencies list of int WiFi frequencies to scan for. @param ssids list of string ssids to probe request for. @param timeout_seconds: float number of seconds to retry scanning if the interface is busy. This does not retry if certain SSIDs are missing from the results. @param require_match: bool True if we must find |ssids|. """ bss_list = utils.poll_for_condition( condition=lambda: self.iw_runner.scan( self.wifi_if, frequencies=frequencies, ssids=ssids), exception=error.TestFail('Unable to trigger scan on client'), timeout=timeout_seconds, sleep_interval=0.5) if require_match: self.assert_bsses_include_ssids(bss_list, ssids) def wait_for_bss(self, bssid, timeout_seconds=15): """Wait for a specific BSS to appear in the scan results. @param bssid: string bssid of AP we expect to see in scan results @param timeout_seconds int seconds to wait for BSSes to be discovered """ def dut_sees_bss(): """Check if a DUT can see a BSS in scan results. @return True iff scan results from DUT include the specified BSS. """ is_requested_bss = lambda iw_bss: iw_bss.bss == bssid scan_results = self.iw_runner.scan(self.wifi_if) return scan_results and list(filter(is_requested_bss, scan_results)) try: utils.poll_for_condition( condition=dut_sees_bss, timeout=timeout_seconds, sleep_interval=0.5) except: raise error.TestFail('Failed to discover BSS %s' % bssid) def wait_for_bsses(self, ssid, num_bss_expected, timeout_seconds=15): """Wait for all BSSes associated with given SSID to be discovered in the scan. @param ssid string name of network being queried @param num_bss_expected int number of BSSes expected @param timeout_seconds int seconds to wait for BSSes to be discovered """ # If the scan returns None, return 0, else return the matching count # Wrap num_bss_actual as a mutable object, list, so that an inner function # can update the value without making an assignment to it. Without any # assignment, the inner function will look for the variable in outer scope # instead of creating a new local one. num_bss_actual = [0] def are_all_bsses_discovered(): """Determine if all BSSes associated with the SSID from parent function are discovered in the scan @return boolean representing whether the expected bss count matches how many in the scan match the given ssid """ self.claim_wifi_if() # Stop shill/supplicant scans try: scan_results = self.iw_runner.scan(self.wifi_if, frequencies=[], ssids=[ssid]) if scan_results is None: return False num_bss_actual[0] = sum(ssid == bss.ssid for bss in scan_results) return num_bss_expected == num_bss_actual[0] finally: self.release_wifi_if() try: utils.poll_for_condition(condition=are_all_bsses_discovered, timeout=timeout_seconds, sleep_interval=0.5) except utils.TimeoutError: raise error.TestFail('Failed to discover all BSSes. Found %d,' ' wanted %d with SSID %s' % (num_bss_actual[0], num_bss_expected, ssid)) def wait_for_service_states(self, ssid, states, timeout_seconds): """Waits for a WiFi service to achieve one of |states|. @param ssid string name of network being queried @param states tuple list of states for which the caller is waiting @param timeout_seconds int seconds to wait for a state in |states| """ logging.info('Waiting for %s to reach one of %r...', ssid, states) success, state, duration = self._shill_proxy.wait_for_service_states( ssid, states, timeout_seconds) logging.info('...ended up in state \'%s\' (%s) after %f seconds.', state, 'success' if success else 'failure', duration) return success, state, duration def do_suspend(self, seconds): """Puts the DUT in suspend power state for |seconds| seconds. @param seconds: The number of seconds to suspend the device. """ logging.info('Suspending DUT for %d seconds...', seconds) self._shill_proxy.do_suspend(seconds) logging.info('...done suspending') def do_suspend_bg(self, seconds): """Suspend DUT using the power manager - non-blocking. @param seconds: The number of seconds to suspend the device. """ logging.info('Suspending DUT (in background) for %d seconds...', seconds) self._shill_proxy.do_suspend_bg(seconds) def flush_bss(self, age=0): """Flush supplicant's cached BSS on the DUT. @param age: BSS older than |age| seconds will be removed from the cache. """ result = self._wpa_cli_proxy.run_wpa_cli_cmd('bss_flush %d' % age, check_result=False); logging.info('wpa_cli bss_flush %d: out:%r err:%r', age, result.stdout, result.stderr) return result.stdout, result.stderr def clear_supplicant_blocklist(self): """Clear's the AP blocklist on the DUT. @return stdout and stderror returns passed from wpa_cli command. """ result = self._wpa_cli_proxy.run_wpa_cli_cmd('bssid_ignore clear', check_result=False) logging.info('wpa_cli bssid_ignore clear: out:%r err:%r', result.stdout, result.stderr) return result.stdout, result.stderr def get_active_wifi_SSIDs(self): """Get a list of visible SSID's around the DUT @return list of string SSIDs """ self._assert_method_supported('get_active_wifi_SSIDs') return self._shill_proxy.get_active_wifi_SSIDs() def set_device_enabled(self, wifi_interface, value, fail_on_unsupported=False): """Enable or disable the WiFi device. @param wifi_interface: string name of interface being modified. @param enabled: boolean; true if this device should be enabled, false if this device should be disabled. @return True if it worked; False, otherwise. """ if fail_on_unsupported: self._assert_method_supported('set_device_enabled') elif not self._supports_method('set_device_enabled'): return False return self._shill_proxy.set_device_enabled(wifi_interface, value) def add_arp_entry(self, ip_address, mac_address): """Add an ARP entry to the table associated with the WiFi interface. @param ip_address: string IP address associated with the new ARP entry. @param mac_address: string MAC address associated with the new ARP entry. """ self.host.run('ip neigh add %s lladdr %s dev %s nud perm' % (ip_address, mac_address, self.wifi_if)) def add_wake_packet_source(self, source_ip): """Add |source_ip| as a source that can wake us up with packets. @param source_ip: IP address from which to wake upon receipt of packets @return True if successful, False otherwise. """ return self._shill_proxy.add_wake_packet_source( self.wifi_if, source_ip) def remove_wake_packet_source(self, source_ip): """Remove |source_ip| as a source that can wake us up with packets. @param source_ip: IP address to stop waking on packets from @return True if successful, False otherwise. """ return self._shill_proxy.remove_wake_packet_source( self.wifi_if, source_ip) def remove_all_wake_packet_sources(self): """Remove all IPs as sources that can wake us up with packets. @return True if successful, False otherwise. """ return self._shill_proxy.remove_all_wake_packet_sources(self.wifi_if) def wake_on_wifi_features(self, features): """Shill supports programming the NIC to wake on special kinds of incoming packets, or on changes to the available APs (disconnect, coming in range of a known SSID). This method allows you to configure what wake-on-WiFi mechanisms are active. It returns a context manager, because this is a system-wide setting and we don't want it to persist across different tests. If you enable wake-on-packet, then the IPs registered by add_wake_packet_source will be able to wake the system from suspend. The correct way to use this method is: with client.wake_on_wifi_features(WAKE_ON_WIFI_DARKCONNECT): ... @param features: string from the WAKE_ON_WIFI constants above. @return a context manager for the features. """ return TemporaryDeviceDBusProperty(self._shill_proxy, self.wifi_if, self.WAKE_ON_WIFI_FEATURES, features) def net_detect_scan_period_seconds(self, period): """Sets the period between net detect scans performed by the NIC to look for allowlisted SSIDs to |period|. This setting only takes effect if the NIC is programmed to wake on SSID. The correct way to use this method is: with client.net_detect_scan_period_seconds(60): ... @param period: integer number of seconds between NIC net detect scans @return a context manager for the net detect scan period """ return TemporaryDeviceDBusProperty(self._shill_proxy, self.wifi_if, self.NET_DETECT_SCAN_PERIOD, period) def wake_to_scan_period_seconds(self, period): """Sets the period between RTC timer wakeups where the system is woken from suspend to perform scans. This setting only takes effect if the NIC is programmed to wake on SSID. Use as with net_detect_scan_period_seconds. @param period: integer number of seconds between wake to scan RTC timer wakes. @return a context manager for the net detect scan period """ return TemporaryDeviceDBusProperty(self._shill_proxy, self.wifi_if, self.WAKE_TO_SCAN_PERIOD, period) def force_wake_to_scan_timer(self, is_forced): """Sets the boolean value determining whether or not to force the use of the wake to scan RTC timer, which wakes the system from suspend periodically to scan for networks. @param is_forced: boolean whether or not to force the use of the wake to scan timer @return a context manager for the net detect scan period """ return TemporaryDeviceDBusProperty(self._shill_proxy, self.wifi_if, self.FORCE_WAKE_TO_SCAN_TIMER, is_forced) def mac_address_randomization(self, enabled): """Sets the boolean value determining whether or not to enable MAC address randomization. This instructs the NIC to randomize the last three octets of the MAC address used in probe requests while disconnected to make the DUT harder to track. If MAC address randomization is not supported on this DUT and the caller tries to turn it on, this raises a TestNAError. @param enabled: boolean whether or not to enable MAC address randomization @return a context manager for the MAC address randomization property """ if not self._shill_proxy.get_dbus_property_on_device( self.wifi_if, self.MAC_ADDRESS_RANDOMIZATION_SUPPORTED): if enabled: raise error.TestNAError( 'MAC address randomization not supported') else: # Return a no-op context manager. return contextlib.nested() return TemporaryDeviceDBusProperty( self._shill_proxy, self.wifi_if, self.MAC_ADDRESS_RANDOMIZATION_ENABLED, enabled) def request_roam(self, bssid): """Request that we roam to the specified BSSID. Note that this operation assumes that: 1) We're connected to an SSID for which |bssid| is a member. 2) There is a BSS with an appropriate ID in our scan results. This method does not check for success of either the command or the roaming operation. @param bssid: string MAC address of bss to roam to. """ self._wpa_cli_proxy.run_wpa_cli_cmd('roam %s' % bssid, check_result=False); return True def request_roam_dbus(self, bssid, iface): """Request that we roam to the specified BSSID through dbus. Note that this operation assumes that: 1) We're connected to an SSID for which |bssid| is a member. 2) There is a BSS with an appropriate ID in our scan results. @param bssid: string MAC address of bss to roam to. @param iface: interface to use @return True if the roam was initiated successfully. Note that this does not guarantee the roam completed successfully. """ self._assert_method_supported('request_roam_dbus') return self._shill_proxy.request_roam_dbus(bssid, iface) def wait_for_roam(self, bssid, timeout_seconds=10.0): """Wait for a roam to the given |bssid|. @param bssid: string bssid to expect a roam to (e.g. '00:11:22:33:44:55'). @param timeout_seconds: float number of seconds to wait for a roam. @return True if we detect an association to the given |bssid| within |timeout_seconds|. """ def attempt_roam(): """Perform a roam to the given |bssid| @return True if there is an assocation between the given |bssid| and that association is detected within the timeout frame """ current_bssid = self.iw_runner.get_current_bssid(self.wifi_if) logging.debug('Current BSSID is %s.', current_bssid) return current_bssid == bssid start_time = time.time() duration = 0.0 try: success = utils.poll_for_condition( condition=attempt_roam, timeout=timeout_seconds, sleep_interval=0.5, desc='Wait for a roam to the given bssid') # wait_for_roam should return False on timeout except utils.TimeoutError: success = False duration = time.time() - start_time logging.debug('%s to %s in %f seconds.', 'Roamed ' if success else 'Failed to roam ', bssid, duration) return success def wait_for_ssid_vanish(self, ssid): """Wait for shill to notice that there are no BSS's for an SSID present. Raise a test failing exception if this does not come to pass. @param ssid: string SSID of the network to require be missing. """ def is_missing_yet(): """Determine if the ssid is removed from the service list yet @return True if the ssid is not found in the service list """ visible_ssids = self.get_active_wifi_SSIDs() logging.info('Got service list: %r', visible_ssids) if ssid not in visible_ssids: return True self.scan(frequencies=[], ssids=[], timeout_seconds=30) utils.poll_for_condition( condition=is_missing_yet, # func exception=error.TestFail('shill should mark BSS not present'), timeout=self.MAX_SERVICE_GONE_TIMEOUT_SECONDS, sleep_interval=0) def reassociate(self, timeout_seconds=10): """Reassociate to the connected network. @param timeout_seconds: float number of seconds to wait for operation to complete. """ logging.info('Attempt to reassociate') with self.iw_runner.get_event_logger() as logger: logger.start() # Issue reattach command to wpa_supplicant self._wpa_cli_proxy.run_wpa_cli_cmd('reattach'); # Wait for the timeout seconds for association to complete time.sleep(timeout_seconds) # Stop iw event logger logger.stop() # Get association time based on the iw event log reassociate_time = logger.get_reassociation_time() if reassociate_time is None or reassociate_time > timeout_seconds: raise error.TestFail( 'Failed to reassociate within given timeout') logging.info('Reassociate time: %.2f seconds', reassociate_time) def wait_for_connection(self, ssid, timeout_seconds=30, freq=None, ping_ip=None, desired_subnet=None, source_iface=None): """Verifies a connection to network ssid, optionally verifying frequency, ping connectivity and subnet. @param ssid string ssid of the network to check. @param timeout_seconds int number of seconds to wait for connection on the given frequency. @param freq int frequency of network to check. @param ping_ip string ip address to ping for verification. @param desired_subnet string expected subnet in which client ip address should reside. @returns a named tuple of (state, time) """ start_time = time.time() duration = lambda: time.time() - start_time # Wrap state as a mutable object, list, so that an inner function can # update the value without making an assignment to it. Without any # assignment, the inner function will look for the variable in outer # scope instead of creating a new local one. state = [None] def verify_connection(): """Verify the connection and perform optional operations as defined in the parent function @return False if there is a failure in the connection or the prescribed verification steps The named tuple ConnectTime otherwise, with the state and connection time from waiting for service states """ success, state[0], conn_time = self.wait_for_service_states( ssid, self.CONNECTED_STATES, int(math.ceil(timeout_seconds - duration()))) if not success: return False if freq: actual_freq = self.get_iw_link_value( iw_runner.IW_LINK_KEY_FREQUENCY) if str(freq) != actual_freq: logging.debug( 'Waiting for desired frequency %s (got %s).', freq, actual_freq) return False if desired_subnet: actual_subnet = self.wifi_ip_subnet if actual_subnet != desired_subnet: logging.debug( 'Waiting for desired subnet %s (got %s).', desired_subnet, actual_subnet) return False if ping_ip: ping_config = ping_runner.PingConfig(ping_ip, source_iface=source_iface) self.ping(ping_config) return ConnectTime(state[0], conn_time) freq_error_str = (' on frequency %d Mhz' % freq) if freq else '' try: ret = utils.poll_for_condition( condition=verify_connection, timeout=timeout_seconds, sleep_interval=0) except utils.TimeoutError: raise error.TestFail( 'Failed to connect to "%s"%s in %f seconds (state=%s)' % (ssid, freq_error_str, duration(), state[0])) return ret @contextmanager def assert_disconnect_count(self, count): """Context asserting |count| disconnects for the context lifetime. Creates an iw logger during the lifetime of the context and asserts that the client disconnects exactly |count| times. @param count int the expected number of disconnections. """ with self.iw_runner.get_event_logger() as logger: logger.start() yield logger.stop() if logger.get_disconnect_count() != count: raise error.TestFail( 'Client disconnected %d times; expected %d' % (logger.get_disconnect_count(), count)) def assert_no_disconnects(self): """Context asserting no disconnects for the context lifetime.""" return self.assert_disconnect_count(0) @contextmanager def assert_disconnect_event(self): """Context asserting at least one disconnect for the context lifetime. Creates an iw logger during the lifetime of the context and asserts that the client disconnects at least one time. """ with self.iw_runner.get_event_logger() as logger: logger.start() yield logger.stop() if logger.get_disconnect_count() == 0: raise error.TestFail('Client did not disconnect') def get_num_card_resets(self): """Get card reset count.""" reset_msg = '[m]wifiex_sdio_card_reset' result = self.host.run('grep -c %s /var/log/messages' % reset_msg, ignore_status=True) if result.exit_status == 1: return 0 count = int(result.stdout.strip()) return count def get_disconnect_reasons(self): """Get disconnect reason codes.""" disconnect_reason_msg = "updated DisconnectReason " disconnect_reason_cleared = "clearing DisconnectReason for " result = self.host.run('grep -a -E "(%s|%s)" /var/log/net.log' % (disconnect_reason_msg, disconnect_reason_cleared), ignore_status=True) if result.exit_status == 1: return None lines = result.stdout.strip().split('\n') disconnect_reasons = [] disconnect_reason_regex = re.compile(r' to (\D?\d+)') found = False for line in reversed(lines): match = disconnect_reason_regex.search(line) if match is not None: disconnect_reasons.append(match.group(1)) found = True else: if (found): break return list(reversed(disconnect_reasons)) def release_wifi_if(self): """Release the control over the wifi interface back to normal operation. This will give the ownership of the wifi interface back to shill and wpa_supplicant. """ self.set_device_enabled(self._wifi_if, True) def claim_wifi_if(self): """Claim the control over the wifi interface from this wifi client. This claim the ownership of the wifi interface from shill and wpa_supplicant. The wifi interface should be UP when this call returns. """ # Disabling a wifi device in shill will remove that device from # wpa_supplicant as well. self.set_device_enabled(self._wifi_if, False) # Wait for shill to bring down the wifi interface. is_interface_down = lambda: not self._interface.is_up utils.poll_for_condition( is_interface_down, timeout=INTERFACE_DOWN_WAIT_TIME_SECONDS, sleep_interval=0.5, desc='Timeout waiting for interface to go down.') # Bring up the wifi interface to allow the test to use the interface. self.host.run('%s link set %s up' % (self.cmd_ip, self.wifi_if)) def set_sched_scan(self, enable, fail_on_unsupported=False): """enable/disable scheduled scan. @param enable bool flag indicating to enable/disable scheduled scan. """ if fail_on_unsupported: self._assert_method_supported('set_sched_scan') elif not self._supports_method('set_sched_scan'): return False return self._shill_proxy.set_sched_scan(enable) def check_connected_on_last_resume(self): """Checks whether the DUT was connected on its last resume. Checks that the DUT was connected after waking from suspend by parsing the last instance shill log message that reports shill's connection status on resume. Fails the test if this log message reports that the DUT woke up disconnected. """ # As of build R43 6913.0.0, the shill log message from the function # OnAfterResume is called as soon as shill resumes from suspend, and # will report whether or not shill is connected. The log message will # take one of the following two forms: # # [...] [INFO:wifi.cc(1941)] OnAfterResume: connected # [...] [INFO:wifi.cc(1941)] OnAfterResume: not connected # # where 1941 is an arbitrary PID number. By checking if the last # instance of this message contains the substring "not connected", we # can determine whether or not shill was connected on its last resume. connection_status_log_regex_str = str(r'INFO:wifi\.cc.*OnAfterResume') not_connected_substr = str(r'not connected') connected_substr = str(r'connected') cmd = ('grep -E %s /var/log/net.log | tail -1' % connection_status_log_regex_str) connection_status_log = self.host.run(cmd).stdout if not connection_status_log: raise error.TestFail('Could not find resume connection status log ' 'message.') logging.debug('Connection status message:\n%s', connection_status_log) if not_connected_substr in connection_status_log: raise error.TestFail('Client was not connected upon waking from ' 'suspend.') if not connected_substr in connection_status_log: raise error.TestFail('Last resume log message did not contain ' 'connection status.') logging.info('Client was connected upon waking from suspend.') def check_wake_on_wifi_throttled(self): """ Checks whether wake on WiFi was throttled on the DUT on the last dark resume. Check for this by parsing shill logs for a throttling message. """ # We are looking for an dark resume error log message indicating that # wake on WiFi was throttled. This is an example of the error message: # [...] [ERROR:wake_on_wifi.cc(1304)] OnDarkResume: Too many dark \ # resumes; disabling wake on WiFi temporarily dark_resume_log_regex_str = str(r'ERROR:wake_on_wifi\.cc.*OnDarkResume:.*') throttled_msg_substr = ('Too many dark resumes; disabling wake on ' 'WiFi temporarily') cmd = ('grep -E %s /var/log/net.log | tail -1' % dark_resume_log_regex_str) last_dark_resume_error_log = self.host.run(cmd).stdout if not last_dark_resume_error_log: raise error.TestFail('Could not find a dark resume log message.') logging.debug('Last dark resume log message:\n%s', last_dark_resume_error_log) if not throttled_msg_substr in last_dark_resume_error_log: raise error.TestFail('Wake on WiFi was not throttled on the last ' 'dark resume.') logging.info('Wake on WiFi was throttled on the last dark resume.') def shill_debug_log(self, message): """Logs a message to the shill log (i.e. /var/log/net.log). This message will be logged at the DEBUG level. @param message: the message to be logged. """ logging.info(message) logger_command = ('/usr/bin/logger' ' --tag shill' ' --priority daemon.debug' ' "%s"' % utils.sh_escape(message)) self.host.run(logger_command) def is_wake_on_wifi_supported(self): """Returns true iff wake-on-WiFi is supported by the DUT.""" if (self.shill.get_dbus_property_on_device( self.wifi_if, self.WAKE_ON_WIFI_FEATURES) == WAKE_ON_WIFI_NOT_SUPPORTED): return False return True def set_manager_property(self, prop_name, prop_value): """Sets the given manager property to the value provided. @param prop_name: the property to be set @param prop_value: value to assign to the prop_name @return a context manager for the setting """ return TemporaryManagerDBusProperty(self._shill_proxy, prop_name, prop_value) class TemporaryDeviceDBusProperty: """Utility class to temporarily change a dbus property for the WiFi device. Since dbus properties are global and persistent settings, we want to make sure that we change them back to what they were before the test started. """ def __init__(self, shill_proxy, iface, prop_name, value): """Construct a TemporaryDeviceDBusProperty context manager. @param shill_proxy: the shill proxy to use to communicate via dbus @param iface: device whose property to change (e.g. 'wlan0') @param prop_name: the name of the property we want to set @param value: the desired value of the property """ self._shill = shill_proxy self._interface = iface self._prop_name = prop_name self._value = value self._saved_value = None def __enter__(self): logging.info('- Setting property %s on device %s', self._prop_name, self._interface) self._saved_value = self._shill.get_dbus_property_on_device( self._interface, self._prop_name) if self._saved_value is None: raise error.TestFail('Device or property not found.') if not self._shill.set_dbus_property_on_device(self._interface, self._prop_name, self._value): raise error.TestFail('Could not set property') logging.info('- Changed value from %s to %s', self._saved_value, self._value) def __exit__(self, exception, value, traceback): logging.info('- Resetting property %s', self._prop_name) if not self._shill.set_dbus_property_on_device(self._interface, self._prop_name, self._saved_value): raise error.TestFail('Could not reset property') class TemporaryManagerDBusProperty: """Utility class to temporarily change a Manager dbus property. Since dbus properties are global and persistent settings, we want to make sure that we change them back to what they were before the test started. """ def __init__(self, shill_proxy, prop_name, value): """Construct a TemporaryManagerDBusProperty context manager. @param shill_proxy: the shill proxy to use to communicate via dbus @param prop_name: the name of the property we want to set @param value: the desired value of the property """ self._shill = shill_proxy self._prop_name = prop_name self._value = value self._saved_value = None def __enter__(self): logging.info('- Setting Manager property: %s', self._prop_name) self._saved_value = self._shill.get_manager_property( self._prop_name) if self._saved_value is None: self._saved_value = "" if not self._shill.set_optional_manager_property(self._prop_name, self._value): raise error.TestFail('Could not set optional manager property.') else: setprop_result = self._shill.set_manager_property(self._prop_name, self._value) if not setprop_result: raise error.TestFail('Could not set manager property') logging.info('- Changed value from [%s] to [%s]', self._saved_value, self._value) def __exit__(self, exception, value, traceback): logging.info('- Resetting property %s to [%s]', self._prop_name, self._saved_value) if not self._shill.set_manager_property(self._prop_name, self._saved_value): raise error.TestFail('Could not reset manager property')