1#!/usr/bin/env python3 2# 3# Copyright 2019 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import inspect 18import logging 19 20import acts_contrib.test_utils.wifi.wifi_test_utils as awutils 21from acts.utils import get_interface_ip_addresses 22from acts.utils import adb_shell_ping 23 24from acts import asserts 25from acts.controllers import iperf_client 26from acts.controllers.fuchsia_device import FuchsiaDevice 27from acts.controllers.android_device import AndroidDevice 28 29 30def create_wlan_device(hardware_device): 31 """Creates a generic WLAN device based on type of device that is sent to 32 the functions. 33 34 Args: 35 hardware_device: A WLAN hardware device that is supported by ACTS. 36 """ 37 if isinstance(hardware_device, FuchsiaDevice): 38 return FuchsiaWlanDevice(hardware_device) 39 elif isinstance(hardware_device, AndroidDevice): 40 return AndroidWlanDevice(hardware_device) 41 else: 42 raise ValueError('Unable to create WlanDevice for type %s' % 43 type(hardware_device)) 44 45 46FUCHSIA_VALID_SECURITY_TYPES = {"none", "wep", "wpa", "wpa2", "wpa3"} 47 48 49class WlanDevice(object): 50 """Class representing a generic WLAN device. 51 52 Each object of this class represents a generic WLAN device. 53 Android device and Fuchsia devices are the currently supported devices/ 54 55 Attributes: 56 device: A generic WLAN device. 57 """ 58 59 def __init__(self, device): 60 self.device = device 61 self.log = logging 62 self.identifier = None 63 64 def wifi_toggle_state(self, state): 65 """Base generic WLAN interface. Only called if not overridden by 66 another supported device. 67 """ 68 raise NotImplementedError("{} must be defined.".format( 69 inspect.currentframe().f_code.co_name)) 70 71 def reset_wifi(self): 72 """Base generic WLAN interface. Only called if not overridden by 73 another supported device. 74 """ 75 raise NotImplementedError("{} must be defined.".format( 76 inspect.currentframe().f_code.co_name)) 77 78 def take_bug_report(self, test_name=None, begin_time=None): 79 """Base generic WLAN interface. Only called if not overridden by 80 another supported device. 81 """ 82 raise NotImplementedError("{} must be defined.".format( 83 inspect.currentframe().f_code.co_name)) 84 85 def get_log(self, test_name, begin_time): 86 """Base generic WLAN interface. Only called if not overridden by 87 another supported device. 88 """ 89 raise NotImplementedError("{} must be defined.".format( 90 inspect.currentframe().f_code.co_name)) 91 92 def turn_location_off_and_scan_toggle_off(self): 93 """Base generic WLAN interface. Only called if not overridden by 94 another supported device. 95 """ 96 raise NotImplementedError("{} must be defined.".format( 97 inspect.currentframe().f_code.co_name)) 98 99 def associate(self, 100 target_ssid, 101 target_pwd=None, 102 check_connectivity=True, 103 hidden=False, 104 target_security=None): 105 """Base generic WLAN interface. Only called if not overriden by 106 another supported device. 107 """ 108 raise NotImplementedError("{} must be defined.".format( 109 inspect.currentframe().f_code.co_name)) 110 111 def disconnect(self): 112 """Base generic WLAN interface. Only called if not overridden by 113 another supported device. 114 """ 115 raise NotImplementedError("{} must be defined.".format( 116 inspect.currentframe().f_code.co_name)) 117 118 def get_wlan_interface_id_list(self): 119 """Base generic WLAN interface. Only called if not overridden by 120 another supported device. 121 """ 122 raise NotImplementedError("{} must be defined.".format( 123 inspect.currentframe().f_code.co_name)) 124 125 def get_default_wlan_test_interface(self): 126 raise NotImplementedError("{} must be defined.".format( 127 inspect.currentframe().f_code.co_name)) 128 129 def destroy_wlan_interface(self, iface_id): 130 """Base generic WLAN interface. Only called if not overridden by 131 another supported device. 132 """ 133 raise NotImplementedError("{} must be defined.".format( 134 inspect.currentframe().f_code.co_name)) 135 136 def send_command(self, command): 137 raise NotImplementedError("{} must be defined.".format( 138 inspect.currentframe().f_code.co_name)) 139 140 def get_interface_ip_addresses(self, interface): 141 raise NotImplementedError("{} must be defined.".format( 142 inspect.currentframe().f_code.co_name)) 143 144 def is_connected(self, ssid=None): 145 raise NotImplementedError("{} must be defined.".format( 146 inspect.currentframe().f_code.co_name)) 147 148 def can_ping(self, 149 dest_ip, 150 count=3, 151 interval=1000, 152 timeout=1000, 153 size=25, 154 additional_ping_params=None): 155 raise NotImplementedError("{} must be defined.".format( 156 inspect.currentframe().f_code.co_name)) 157 158 def ping(self, 159 dest_ip, 160 count=3, 161 interval=1000, 162 timeout=1000, 163 size=25, 164 additional_ping_params=None): 165 raise NotImplementedError("{} must be defined.".format( 166 inspect.currentframe().f_code.co_name)) 167 168 def hard_power_cycle(self, pdus=None): 169 raise NotImplementedError("{} must be defined.".format( 170 inspect.currentframe().f_code.co_name)) 171 172 def save_network(self, ssid): 173 raise NotImplementedError("{} must be defined.".format( 174 inspect.currentframe().f_code.co_name)) 175 176 def clear_saved_networks(self): 177 raise NotImplementedError("{} must be defined.".format( 178 inspect.currentframe().f_code.co_name)) 179 180 def create_iperf_client(self, test_interface=None): 181 raise NotImplementedError("{} must be defined.".format( 182 inspect.currentframe().f_code.co_name)) 183 184 185class AndroidWlanDevice(WlanDevice): 186 """Class wrapper for an Android WLAN device. 187 188 Each object of this class represents a generic WLAN device. 189 Android device and Fuchsia devices are the currently supported devices/ 190 191 Attributes: 192 android_device: An Android WLAN device. 193 """ 194 195 def __init__(self, android_device): 196 super().__init__(android_device) 197 self.identifier = android_device.serial 198 199 def wifi_toggle_state(self, state): 200 awutils.wifi_toggle_state(self.device, state) 201 202 def reset_wifi(self): 203 awutils.reset_wifi(self.device) 204 205 def take_bug_report(self, test_name=None, begin_time=None): 206 self.device.take_bug_report(test_name, begin_time) 207 208 def get_log(self, test_name, begin_time): 209 self.device.cat_adb_log(test_name, begin_time) 210 211 def turn_location_off_and_scan_toggle_off(self): 212 awutils.turn_location_off_and_scan_toggle_off(self.device) 213 214 def associate(self, 215 target_ssid, 216 target_pwd=None, 217 key_mgmt=None, 218 check_connectivity=True, 219 hidden=False, 220 target_security=None): 221 """Function to associate an Android WLAN device. 222 223 Args: 224 target_ssid: SSID to associate to. 225 target_pwd: Password for the SSID, if necessary. 226 key_mgmt: The hostapd wpa_key_mgmt value, distinguishes wpa3 from 227 wpa2 for android tests. 228 check_connectivity: Whether to check for internet connectivity. 229 hidden: Whether the network is hidden. 230 Returns: 231 True if successfully connected to WLAN, False if not. 232 """ 233 network = {'SSID': target_ssid, 'hiddenSSID': hidden} 234 if target_pwd: 235 network['password'] = target_pwd 236 if key_mgmt: 237 network['security'] = key_mgmt 238 try: 239 awutils.connect_to_wifi_network( 240 self.device, 241 network, 242 check_connectivity=check_connectivity, 243 hidden=hidden) 244 return True 245 except Exception as e: 246 self.device.log.info('Failed to associated (%s)' % e) 247 return False 248 249 def disconnect(self): 250 awutils.turn_location_off_and_scan_toggle_off(self.device) 251 252 def get_wlan_interface_id_list(self): 253 pass 254 255 def get_default_wlan_test_interface(self): 256 return 'wlan0' 257 258 def destroy_wlan_interface(self, iface_id): 259 pass 260 261 def send_command(self, command): 262 return self.device.adb.shell(str(command)) 263 264 def get_interface_ip_addresses(self, interface): 265 return get_interface_ip_addresses(self.device, interface) 266 267 def is_connected(self, ssid=None): 268 wifi_info = self.device.droid.wifiGetConnectionInfo() 269 if ssid: 270 return 'BSSID' in wifi_info and wifi_info['SSID'] == ssid 271 return 'BSSID' in wifi_info 272 273 def can_ping(self, 274 dest_ip, 275 count=3, 276 interval=1000, 277 timeout=1000, 278 size=25, 279 additional_ping_params=None): 280 return adb_shell_ping(self.device, 281 dest_ip=dest_ip, 282 count=count, 283 timeout=timeout) 284 285 def ping(self, dest_ip, count=3, interval=1000, timeout=1000, size=25): 286 pass 287 288 def hard_power_cycle(self, pdus): 289 pass 290 291 def save_network(self, ssid): 292 pass 293 294 def clear_saved_networks(self): 295 pass 296 297 def create_iperf_client(self, test_interface=None): 298 """ Returns an iperf client on the Android, without requiring a 299 specific config. 300 301 Args: 302 test_interface: optional, string, name of test interface. 303 304 Returns: 305 IPerfClient object 306 """ 307 if not test_interface: 308 test_interface = self.get_default_wlan_test_interface() 309 310 return iperf_client.IPerfClientOverAdb( 311 android_device_or_serial=self.device, 312 test_interface=test_interface) 313 314 315class FuchsiaWlanDevice(WlanDevice): 316 """Class wrapper for an Fuchsia WLAN device. 317 318 Each object of this class represents a generic WLAN device. 319 Android device and Fuchsia devices are the currently supported devices/ 320 321 Attributes: 322 fuchsia_device: A Fuchsia WLAN device. 323 """ 324 325 def __init__(self, fuchsia_device): 326 super().__init__(fuchsia_device) 327 self.identifier = fuchsia_device.ip 328 self.device.configure_wlan() 329 330 def wifi_toggle_state(self, state): 331 """Stub for Fuchsia implementation.""" 332 pass 333 334 def reset_wifi(self): 335 """Stub for Fuchsia implementation.""" 336 pass 337 338 def take_bug_report(self, test_name=None, begin_time=None): 339 """Stub for Fuchsia implementation.""" 340 self.device.take_bug_report(test_name, begin_time) 341 342 def get_log(self, test_name, begin_time): 343 """Stub for Fuchsia implementation.""" 344 pass 345 346 def turn_location_off_and_scan_toggle_off(self): 347 """Stub for Fuchsia implementation.""" 348 pass 349 350 def associate(self, 351 target_ssid, 352 target_pwd=None, 353 key_mgmt=None, 354 check_connectivity=True, 355 hidden=False, 356 target_security=None): 357 """Function to associate a Fuchsia WLAN device. 358 359 Args: 360 target_ssid: SSID to associate to. 361 target_pwd: Password for the SSID, if necessary. 362 key_mgmt: the hostapd wpa_key_mgmt, if specified. 363 check_connectivity: Whether to check for internet connectivity. 364 hidden: Whether the network is hidden. 365 target_security: string, target security for network, used to 366 save the network in policy connects (see wlan_policy_lib) 367 Returns: 368 True if successfully connected to WLAN, False if not. 369 """ 370 if self.device.association_mechanism == 'drivers': 371 bss_scan_response = self.device.wlan_lib.wlanScanForBSSInfo() 372 if bss_scan_response.get('error'): 373 self.log.error('Scan for BSS info failed. Err: %s' % 374 bss_scan_response['error']) 375 return False 376 377 bss_descs_for_ssid = bss_scan_response['result'].get( 378 target_ssid, None) 379 if not bss_descs_for_ssid or len(bss_descs_for_ssid) < 1: 380 self.log.error( 381 'Scan failed to find a BSS description for target_ssid %s' 382 % target_ssid) 383 return False 384 385 connection_response = self.device.wlan_lib.wlanConnectToNetwork( 386 target_ssid, bss_descs_for_ssid[0], target_pwd=target_pwd) 387 return self.device.check_connect_response(connection_response) 388 else: 389 return self.device.wlan_policy_controller.save_and_connect( 390 target_ssid, target_security, password=target_pwd) 391 392 def disconnect(self): 393 """Function to disconnect from a Fuchsia WLAN device. 394 Asserts if disconnect was not successful. 395 """ 396 if self.device.association_mechanism == 'drivers': 397 disconnect_response = self.device.wlan_lib.wlanDisconnect() 398 return self.device.check_disconnect_response(disconnect_response) 399 else: 400 return self.device.wlan_policy_controller.remove_all_networks_and_wait_for_no_connections( 401 ) 402 403 def status(self): 404 return self.device.wlan_lib.wlanStatus() 405 406 def can_ping(self, 407 dest_ip, 408 count=3, 409 interval=1000, 410 timeout=1000, 411 size=25, 412 additional_ping_params=None): 413 return self.device.can_ping( 414 dest_ip, 415 count=count, 416 interval=interval, 417 timeout=timeout, 418 size=size, 419 additional_ping_params=additional_ping_params) 420 421 def ping(self, 422 dest_ip, 423 count=3, 424 interval=1000, 425 timeout=1000, 426 size=25, 427 additional_ping_params=None): 428 return self.device.ping(dest_ip, 429 count=count, 430 interval=interval, 431 timeout=timeout, 432 size=size, 433 additional_ping_params=additional_ping_params) 434 435 def get_wlan_interface_id_list(self): 436 """Function to list available WLAN interfaces. 437 438 Returns: 439 A list of wlan interface IDs. 440 """ 441 return self.device.wlan_lib.wlanGetIfaceIdList().get('result') 442 443 def get_default_wlan_test_interface(self): 444 """Returns name of the WLAN client interface""" 445 return self.device.wlan_client_test_interface_name 446 447 def destroy_wlan_interface(self, iface_id): 448 """Function to associate a Fuchsia WLAN device. 449 450 Args: 451 target_ssid: SSID to associate to. 452 target_pwd: Password for the SSID, if necessary. 453 check_connectivity: Whether to check for internet connectivity. 454 hidden: Whether the network is hidden. 455 Returns: 456 True if successfully destroyed wlan interface, False if not. 457 """ 458 result = self.device.wlan_lib.wlanDestroyIface(iface_id) 459 if result.get('error') is None: 460 return True 461 else: 462 self.log.error("Failed to destroy interface with: {}".format( 463 result.get('error'))) 464 return False 465 466 def send_command(self, command): 467 return self.device.send_command_ssh(str(command)).stdout 468 469 def get_interface_ip_addresses(self, interface): 470 return get_interface_ip_addresses(self.device, interface) 471 472 def is_connected(self, ssid=None): 473 """ Determines if wlan_device is connected to wlan network. 474 475 Args: 476 ssid (optional): string, to check if device is connect to a specific 477 network. 478 479 Returns: 480 True, if connected to a network or to the correct network when SSID 481 is provided. 482 False, if not connected or connect to incorrect network when SSID is 483 provided. 484 """ 485 response = self.status() 486 if response.get('error'): 487 raise ConnectionError( 488 'Failed to get client network connection status') 489 result = response.get('result') 490 if isinstance(result, dict): 491 connected_to = result.get('Connected') 492 # TODO(https://fxbug.dev/85938): Remove backwards compatibility once 493 # ACTS is versioned with Fuchsia. 494 if not connected_to: 495 connected_to = result.get('connected_to') 496 if not connected_to: 497 return False 498 499 if ssid: 500 # Replace encoding errors instead of raising an exception. 501 # Since `ssid` is a string, this will not affect the test 502 # for equality. 503 connected_ssid = bytearray(connected_to['ssid']).decode( 504 encoding='utf-8', errors='replace') 505 return ssid == connected_ssid 506 return True 507 return False 508 509 def hard_power_cycle(self, pdus): 510 self.device.reboot(reboot_type='hard', testbed_pdus=pdus) 511 512 def save_network(self, target_ssid, security_type=None, target_pwd=None): 513 if self.device.association_mechanism == 'drivers': 514 raise EnvironmentError( 515 'Cannot save network using the drivers. Saved networks are a ' 516 'policy layer concept.') 517 if security_type and security_type not in FUCHSIA_VALID_SECURITY_TYPES: 518 raise TypeError('Invalid security type: %s' % security_type) 519 if not self.device.wlan_policy_controller.save_network( 520 target_ssid, security_type, password=target_pwd): 521 raise EnvironmentError('Failed to save network: %s' % target_ssid) 522 523 def clear_saved_networks(self): 524 if self.device.association_mechanism == 'drivers': 525 raise EnvironmentError( 526 'Cannot clear saved network using the drivers. Saved networks ' 527 'are a policy layer concept.') 528 if not self.device.wlan_policy_controller.remove_all_networks(): 529 raise EnvironmentError('Failed to clear saved networks') 530 531 def create_iperf_client(self, test_interface=None): 532 """ Returns an iperf client on the FuchsiaDevice, without requiring a 533 specific config. 534 535 Args: 536 test_interface: optional, string, name of test interface. Defaults 537 to first found wlan client interface. 538 539 Returns: 540 IPerfClient object 541 """ 542 if not test_interface: 543 test_interface = self.get_default_wlan_test_interface() 544 return iperf_client.IPerfClientOverSsh( 545 { 546 'user': 'fuchsia', 547 'host': self.device.ip, 548 'ssh_config': self.device.ssh_config 549 }, 550 use_paramiko=True, 551 test_interface=test_interface) 552