1 # Copyright (c) 2015 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import logging 6 import time 7 from autotest_lib.client.common_lib import error 8 from autotest_lib.client.common_lib.cros.network import iw_runner 9 from autotest_lib.client.common_lib.cros.network import xmlrpc_datatypes 10 from autotest_lib.server.cros.network import hostap_config 11 from autotest_lib.server.cros.network import wifi_cell_test_base 12 13 class network_WiFi_RoamDbus(wifi_cell_test_base.WiFiCellTestBase): 14 """Tests an intentional client-driven roam between APs 15 16 This test seeks to associate the DUT with an AP with a set of 17 association parameters, create a second AP with a second set of 18 parameters but the same SSID, and send roam command to shill. After 19 that shill will send a dbus roam command to wpa_supplicant. We seek 20 to observe that the DUT successfully connects to the second AP in 21 a reasonable amount of time. 22 """ 23 24 version = 1 25 TIMEOUT_SECONDS = 15 26 27 def dut_sees_bss(self, bssid): 28 """ 29 Check if a DUT can see a BSS in scan results. 30 31 @param bssid: string bssid of AP we expect to see in scan results. 32 @return True iff scan results from DUT include the specified BSS. 33 34 """ 35 runner = iw_runner.IwRunner(remote_host=self.context.client.host) 36 is_requested_bss = lambda iw_bss: iw_bss.bss == bssid 37 scan_results = runner.scan(self.context.client.wifi_if) 38 return scan_results and filter(is_requested_bss, scan_results) 39 40 41 def retry(self, func, reason, timeout_seconds=TIMEOUT_SECONDS): 42 """ 43 Retry a function until it returns true or we time out. 44 45 @param func: function that takes no parameters. 46 @param reason: string concise description of what the function does. 47 @param timeout_seconds: int number of seconds to wait for a True 48 response from |func|. 49 50 """ 51 logging.info('Waiting for %s.', reason) 52 start_time = time.time() 53 while time.time() - start_time < timeout_seconds: 54 if func(): 55 return 56 time.sleep(1) 57 else: 58 raise error.TestFail('Timed out waiting for %s.' % reason) 59 60 61 def run_once(self,host): 62 """Test body.""" 63 self._router0_conf = hostap_config.HostapConfig(channel=48, 64 mode=hostap_config.HostapConfig.MODE_11A) 65 self._router1_conf = hostap_config.HostapConfig(channel=1) 66 self._client_conf = xmlrpc_datatypes.AssociationParameters() 67 68 # Configure the inital AP. 69 self.context.configure(self._router0_conf) 70 router_ssid = self.context.router.get_ssid() 71 72 # Connect to the inital AP. 73 self._client_conf.ssid = router_ssid 74 self.context.assert_connect_wifi(self._client_conf) 75 76 # Setup a second AP with the same SSID. 77 self._router1_conf.ssid = router_ssid 78 self.context.configure(self._router1_conf, multi_interface=True) 79 80 # Get BSSIDs of the two APs 81 bssid0 = self.context.router.get_hostapd_mac(0) 82 bssid1 = self.context.router.get_hostapd_mac(1) 83 84 # Wait for DUT to see the second AP 85 self.retry(lambda: self.dut_sees_bss(bssid1), 'DUT to see second AP') 86 87 # Check which AP we are currently connected. 88 # This is to include the case that wpa_supplicant 89 # automatically roam to AP2 during the scan. 90 interface = self.context.client.wifi_if 91 current_bssid = self.context.client.iw_runner.get_current_bssid(interface) 92 if current_bssid == bssid0: 93 roam_to_bssid = bssid1 94 else: 95 roam_to_bssid = bssid0 96 # Send roam command to shill, 97 # and shill will send dbus roam command to wpa_supplicant 98 self.context.client.request_roam_dbus(roam_to_bssid, interface) 99 100 # Expect that the DUT will re-connect to the new AP. 101 if not self.context.client.wait_for_roam( 102 roam_to_bssid, timeout_seconds=self.TIMEOUT_SECONDS): 103 raise error.TestFail('Failed to roam.') 104 self.context.router.deconfig() 105
